
'''simple docstring'''
import csv
import json
import logging
import os
import re
import subprocess
import tempfile
import unittest
import zipfile
from pathlib import Path
from typing import Optional
from unittest import mock

import torch

# We use TF to parse the logs
from accelerate import Accelerator
from accelerate.test_utils.testing import (
    MockingTestCase,
    TempDirTestCase,
    require_comet_ml,
    require_tensorboard,
    require_wandb,
    skip,
)
from accelerate.tracking import CometMLTracker, GeneralTracker
from accelerate.utils import ProjectConfiguration, is_comet_ml_available, is_tensorboard_available


if is_comet_ml_available():
     from comet_ml import OfflineExperiment

if is_tensorboard_available():
     import struct

     import tensorboard.compat.proto.event_pba as event_pba

A__ :		Dict					=logging.getLogger(__name__)







@require_tensorboard
class 			__A					(			unittest.TestCase      ):
 def 		lowercase_( self      :     Optional[int]							):



   """simple docstring"""





   __A		:  Tuple       				=		"""test_project_with_config"""
   with tempfile.TemporaryDirectory() as dirpath:
     __A		:  Any       				=		Accelerator(log_with="""tensorboard"""					, project_dir=lowerCamelCase							)
     __A		:  Union[str, Any]       				=		{"""num_iterations""": 12, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
     accelerator.init_trackers(lowerCamelCase					, lowerCamelCase							)
     accelerator.end_training()
     for child in Path(f"{dirpath}/{project_name}"							).glob("""*/**"""							):
       __A		:  int       				=		list(filter(lambda lowerCamelCase							: x.is_file()					, child.iterdir()							)							)[0]
     self.assertNotEqual(str(lowerCamelCase							)					, """"""							)
 def 		lowercase_( self      :     Dict							):



   """simple docstring"""





   __A		:  List[str]       				=		"""test_project_with_log"""
   with tempfile.TemporaryDirectory() as dirpath:
     __A		:  Union[str, Any]       				=		Accelerator(log_with="""tensorboard"""					, project_dir=lowerCamelCase							)
     accelerator.init_trackers(lowerCamelCase							)
     __A		:  int       				=		{"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
     accelerator.log(lowerCamelCase					, step=0							)
     accelerator.end_training()
     # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
     # Names are randomly generated each time
     __A		:  List[Any]       				=		list(filter(lambda lowerCamelCase							: x.is_file()					, Path(f"{dirpath}/{project_name}"							).iterdir()							)							)[0]
     self.assertNotEqual(str(lowerCamelCase							)					, """"""							)
 def 		lowercase_( self      :     List[Any]							):



   """simple docstring"""





   __A		:  str       				=		"""test_project_with_log"""
   with tempfile.TemporaryDirectory() as dirpath:
     __A		:  Tuple       				=		Accelerator(log_with="""tensorboard"""					, project_dir=lowerCamelCase							)
     accelerator.init_trackers(lowerCamelCase							)
     __A		:  Optional[int]       				=		{"""tensor""": torch.tensor(1							)}
     accelerator.log(lowerCamelCase					, step=0							)
     accelerator.end_training()
     # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
     # Names are randomly generated each time
     __A		:  List[str]       				=		list(filter(lambda lowerCamelCase							: x.is_file()					, Path(f"{dirpath}/{project_name}"							).iterdir()							)							)[0]
     # Reading implementation based on https://github.com/pytorch/pytorch/issues/45327#issuecomment-703757685
     with open(lowerCamelCase					, """rb"""							) as f:
       __A		:  Optional[Any]       				=		f.read()
     __A		:  str       				=		False
     while data:
       __A		:  Union[str, Any]       				=		struct.unpack("""Q"""					, data[:8]							)

       __A		:  Optional[int]       				=		data[12 : 12 + int(header[0]							)]  # 8+4
       __A		:  str       				=		data[12 + int(header[0]							) + 4 :]
       __A		:  Any       				=		event_pba.Event()

       event.ParseFromString(lowerCamelCase							)
       if event.HasField("""summary"""							):
         for value in event.summary.value:
           if value.simple_value == 1.0 and value.tag == "tensor":
             __A		:  List[str]       				=		True
     self.assertTrue(lowerCamelCase					, """Converted tensor was not found in the log file!"""							)
 def 		lowercase_( self      :     Optional[int]							):



   """simple docstring"""





   with self.assertRaisesRegex(lowerCamelCase					, """Logging with `tensorboard` requires a `logging_dir`"""							):
     __A		:  Tuple       				=		Accelerator(log_with="""tensorboard"""							)
   with tempfile.TemporaryDirectory() as dirpath:
     __A		:  Optional[int]       				=		Accelerator(log_with="""tensorboard"""					, project_dir=lowerCamelCase							)






 def 		lowercase_( self      :     Tuple							):



   """simple docstring"""





   __A		:  Dict       				=		ProjectConfiguration(total_limit=30							)
   with tempfile.TemporaryDirectory() as dirpath:
     __A		:  str       				=		Accelerator(log_with="""tensorboard"""					, project_dir=lowerCamelCase					, project_config=lowerCamelCase							)







@require_wandb
@mock.patch.dict(os.environ     ,    {'''WANDB_MODE''': '''offline'''}      )
class 			__A					(			_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE      ):
 def 		lowercase_( self      :     List[str]							):



   """simple docstring"""





   super().setUp()
   # wandb let's us override where logs are stored to via the WANDB_DIR env var
   self.add_mocks(mock.patch.dict(os.environ					, {"""WANDB_DIR""": self.tmpdir}							)							)
 @staticmethod
 def 		lowercase_( lowerCamelCase      :     str					, lowerCamelCase      :     str					, lowerCamelCase      :     bool = True							):



   """simple docstring"""





   __A		:  str       				=		rf"{section} ([\S\s]*?)\n\n"
   if record:
     __A		:  Any       				=		rf"Record: {pattern}"
   __A		:  int       				=		re.findall(lowerCamelCase					, lowerCamelCase							)[0]
   # A config
   if section == "config" or section == "history":
     __A		:  Tuple       				=		re.findall(r"""\"([a-zA-Z0-9_.,]+)"""					, lowerCamelCase							)
     return {key: val for key, val in zip(cleaned_record[0::2]					, cleaned_record[1::2]							)}
   # Everything else
   else:
     return dict(re.findall(r"""(\w+): \"([^\s]+)\""""					, lowerCamelCase							)							)






 @skip
 def 		lowercase_( self      :     Dict							):



   """simple docstring"""





   __A		:  Tuple       				=		"""test_project_with_config"""
   __A		:  Optional[Any]       				=		Accelerator(log_with="""wandb"""							)
   __A		:  Any       				=		{"""num_iterations""": 12, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
   __A		:  str       				=		{"""wandb""": {"""tags""": ["""my_tag"""]}}
   accelerator.init_trackers(lowerCamelCase					, lowerCamelCase					, lowerCamelCase							)
   __A		:  Optional[Any]       				=		{"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
   accelerator.log(lowerCamelCase					, step=0							)
   accelerator.end_training()
   # The latest offline log is stored at wandb/latest-run/*.wandb
   for child in Path(f"{self.tmpdir}/wandb/latest-run"							).glob("""*"""							):
     if child.is_file() and child.suffix == ".wandb":
       __A		:  Optional[Any]       				=		subprocess.check_output(
           ["""wandb""", """sync""", """--view""", """--verbose""", str(lowerCamelCase							)]					, env=os.environ.copy()							).decode("""utf8"""					, """ignore"""							)
       break

        # Check HPS through careful parsing and cleaning
   __A		:  Dict       				=		self.parse_log(lowerCamelCase					, """config"""							)
   self.assertEqual(logged_items["""num_iterations"""]					, """12"""							)
   self.assertEqual(logged_items["""learning_rate"""]					, """0.01"""							)
   self.assertEqual(logged_items["""some_boolean"""]					, """false"""							)
   self.assertEqual(logged_items["""some_string"""]					, """some_value"""							)
   self.assertEqual(logged_items["""some_string"""]					, """some_value"""							)

   # Run tags
   __A		:  Tuple       				=		self.parse_log(lowerCamelCase					, """run"""					, lowerCamelCase							)
   self.assertEqual(logged_items["""tags"""]					, """my_tag"""							)

   # Actual logging
   __A		:  Optional[Any]       				=		self.parse_log(lowerCamelCase					, """history"""							)
   self.assertEqual(logged_items["""total_loss"""]					, """0.1"""							)
   self.assertEqual(logged_items["""iteration"""]					, """1"""							)
   self.assertEqual(logged_items["""my_text"""]					, """some_value"""							)
   self.assertEqual(logged_items["""_step"""]					, """0"""							)







def 							A_  (    self    :	str      ,  __SCREAMING_SNAKE_CASE    :	str      ,  __SCREAMING_SNAKE_CASE    :	str			)					->							Optional[int]:





  """simple docstring"""

  __A		:  Optional[Any]       				=		run_name
  __A		:  Optional[int]       				=		OfflineExperiment(project_name=__SCREAMING_SNAKE_CASE      ,  offline_directory=__SCREAMING_SNAKE_CASE			)
  logger.info(F"Initialized offline CometML project {self.run_name}"			)
  logger.info("""Make sure to log any initial configurations with `self.store_init_configuration` before training!"""			)







@require_comet_ml
@mock.patch.object(_SCREAMING_SNAKE_CASE     ,    '''__init__'''     ,    _SCREAMING_SNAKE_CASE      )
class 			__A					(			unittest.TestCase      ):
 @staticmethod
 def 		lowercase_( lowerCamelCase      :     int					, lowerCamelCase      :     str					, lowerCamelCase      :     bool = False							):



   """simple docstring"""





   for log in log_list:
     __A		:  Dict       				=		json.loads(lowerCamelCase							)["""payload"""]
     if is_param and "param" in j.keys():
       if j["param"]["paramName"] == key:
         return j["param"]["paramValue"]
     if "log_other" in j.keys():
       if j["log_other"]["key"] == key:
         return j["log_other"]["val"]
     if "metric" in j.keys():
       if j["metric"]["metricName"] == key:
         return j["metric"]["metricValue"]
 def 		lowercase_( self      :     Tuple							):



   """simple docstring"""





   with tempfile.TemporaryDirectory() as d:
     __A		:  Optional[int]       				=		CometMLTracker("""test_project_with_config"""					, lowerCamelCase							)
     __A		:  List[str]       				=		Accelerator(log_with=lowerCamelCase							)
     __A		:  List[Any]       				=		{"""num_iterations""": 12, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
     accelerator.init_trackers(lowerCamelCase					, lowerCamelCase							)
     accelerator.end_training()
     __A		:  Tuple       				=		os.listdir(lowerCamelCase							)[0]  # Comet is nice, it's just a zip file here
     # We parse the raw logs
     __A		:  str       				=		os.path.join(lowerCamelCase					, lowerCamelCase							)
     __A		:  Optional[int]       				=		zipfile.ZipFile(lowerCamelCase					, """r"""							)
     __A		:  Tuple       				=		archive.open("""messages.json"""							).read().decode("""utf-8"""							)
   __A		:  Union[str, Any]       				=		log.split("""\n"""							)[:-1]
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """num_iterations"""					, lowerCamelCase							)					, 12							)
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """learning_rate"""					, lowerCamelCase							)					, 0.01							)
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """some_boolean"""					, lowerCamelCase							)					, lowerCamelCase							)
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """some_string"""					, lowerCamelCase							)					, """some_value"""							)






 def 		lowercase_( self      :     Tuple							):



   """simple docstring"""





   with tempfile.TemporaryDirectory() as d:
     __A		:  Optional[int]       				=		CometMLTracker("""test_project_with_config"""					, lowerCamelCase							)
     __A		:  List[Any]       				=		Accelerator(log_with=lowerCamelCase							)
     accelerator.init_trackers(lowerCamelCase							)
     __A		:  str       				=		{"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
     accelerator.log(lowerCamelCase					, step=0							)
     accelerator.end_training()
     __A		:  Dict       				=		os.listdir(lowerCamelCase							)[0]  # Comet is nice, it's just a zip file here
     # We parse the raw logs
     __A		:  Union[str, Any]       				=		os.path.join(lowerCamelCase					, lowerCamelCase							)
     __A		:  List[str]       				=		zipfile.ZipFile(lowerCamelCase					, """r"""							)
     __A		:  Optional[int]       				=		archive.open("""messages.json"""							).read().decode("""utf-8"""							)
   __A		:  Dict       				=		log.split("""\n"""							)[:-1]
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """curr_step"""					, lowerCamelCase							)					, 0							)
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """total_loss"""							)					, 0.1							)
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """iteration"""							)					, 1							)
   self.assertEqual(self.get_value_from_key(lowerCamelCase					, """my_text"""							)					, """some_value"""							)







class 			__A					(			_SCREAMING_SNAKE_CASE      ):
 lowerCamelCase												=[
     '''total_loss''',
     '''iteration''',
     '''my_text''',
     '''learning_rate''',
     '''num_iterations''',
     '''some_boolean''',
     '''some_string''',
 ]

 lowerCamelCase												='''my_custom_tracker'''
 lowerCamelCase												=False
 def __init__( self      :     Any					, lowerCamelCase      :     str							):



   """simple docstring"""





   __A		:  Optional[int]       				=		open(f"{dir}/log.csv"					, """w+"""							)
   __A		:  Tuple       				=		csv.DictWriter(self.f					, fieldnames=self._col_names							)
   self.writer.writeheader()
 @property
 def 		lowercase_( self      :     str							):



   """simple docstring"""





   return self.writer
 def 		lowercase_( self      :     Union[str, Any]					, lowerCamelCase      :     dict							):



   """simple docstring"""





   logger.info("""Call init"""							)
   self.writer.writerow(lowerCamelCase							)
 def 		lowercase_( self      :     Optional[int]					, lowerCamelCase      :     dict					, lowerCamelCase      :     Optional[int]							):



   """simple docstring"""





   logger.info("""Call log"""							)
   self.writer.writerow(lowerCamelCase							)






 def 		lowercase_( self      :     Optional[int]							):



   """simple docstring"""





   self.f.close()







class 			__A					(			unittest.TestCase      ):
 def 		lowercase_( self      :     List[Any]							):



   """simple docstring"""





   with tempfile.TemporaryDirectory() as d:
     __A		:  Optional[Any]       				=		MyCustomTracker(lowerCamelCase							)
     __A		:  str       				=		Accelerator(log_with=lowerCamelCase							)
     __A		:  Optional[Any]       				=		{"""num_iterations""": 12, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
     accelerator.init_trackers("""Some name"""					, lowerCamelCase							)
     accelerator.end_training()
     with open(f"{d}/log.csv"					, """r"""							) as f:
       __A		:  Dict       				=		csv.DictReader(lowerCamelCase							)
       __A		:  Tuple       				=		next(lowerCamelCase							)
       __A		:  List[str]       				=		{
           """total_loss""": """""",
           """iteration""": """""",
           """my_text""": """""",
           """learning_rate""": """0.01""",
           """num_iterations""": """12""",
           """some_boolean""": """False""",
           """some_string""": """some_value""",
       }
       self.assertDictEqual(lowerCamelCase					, lowerCamelCase							)






 def 		lowercase_( self      :     str							):



   """simple docstring"""





   with tempfile.TemporaryDirectory() as d:
     __A		:  str       				=		MyCustomTracker(lowerCamelCase							)
     __A		:  Optional[int]       				=		Accelerator(log_with=lowerCamelCase							)
     accelerator.init_trackers("""Some name"""							)
     __A		:  Tuple       				=		{"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
     accelerator.log(lowerCamelCase					, step=0							)
     accelerator.end_training()
     with open(f"{d}/log.csv"					, """r"""							) as f:
       __A		:  int       				=		csv.DictReader(lowerCamelCase							)
       __A		:  List[Any]       				=		next(lowerCamelCase							)
       __A		:  Optional[Any]       				=		{
           """total_loss""": """0.1""",
           """iteration""": """1""",
           """my_text""": """some_value""",
           """learning_rate""": """""",
           """num_iterations""": """""",
           """some_boolean""": """""",
           """some_string""": """""",
       }
       self.assertDictEqual(lowerCamelCase					, lowerCamelCase							)





'''simple docstring'''
import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


A__ :		str					=logging.get_logger(__name__)

A__ :		Optional[int]					={
    'facebook/sam-vit-huge': 'https://huggingface.co/facebook/sam-vit-huge/resolve/main/config.json',
    'facebook/sam-vit-large': 'https://huggingface.co/facebook/sam-vit-large/resolve/main/config.json',
    'facebook/sam-vit-base': 'https://huggingface.co/facebook/sam-vit-base/resolve/main/config.json',
}







class 			__A					(			_SCREAMING_SNAKE_CASE      ):
 def __init__( self      :     Tuple					, lowerCamelCase      :     Tuple=2_56					, lowerCamelCase      :     Union[str, Any]=10_24					, lowerCamelCase      :     List[str]=16					, lowerCamelCase      :     str=16					, lowerCamelCase      :     Union[str, Any]=4					, lowerCamelCase      :     Optional[Any]="gelu"					, lowerCamelCase      :     Union[str, Any]=1e-6					, **lowerCamelCase      :     Union[str, Any]					, ):



   """simple docstring"""





   super().__init__(**lowerCamelCase							)
   __A		:  List[str]       				=		hidden_size
   __A		:  Any       				=		image_size
   __A		:  Dict       				=		patch_size
   __A		:  Union[str, Any]       				=		image_size // patch_size
   __A		:  List[Any]       				=		mask_input_channels
   __A		:  str       				=		num_point_embeddings
   __A		:  Tuple       				=		hidden_act
   __A		:  List[str]       				=		layer_norm_eps







class 			__A					(			_SCREAMING_SNAKE_CASE      ):
 def __init__( self      :     Optional[int]					, lowerCamelCase      :     int=2_56					, lowerCamelCase      :     Tuple="relu"					, lowerCamelCase      :     Optional[Any]=20_48					, lowerCamelCase      :     int=2					, lowerCamelCase      :     Tuple=8					, lowerCamelCase      :     int=2					, lowerCamelCase      :     List[str]=3					, lowerCamelCase      :     Dict=3					, lowerCamelCase      :     Any=2_56					, lowerCamelCase      :     int=1e-6					, **lowerCamelCase      :     Tuple					, ):



   """simple docstring"""





   super().__init__(**lowerCamelCase							)
   __A		:  List[Any]       				=		hidden_size
   __A		:  Union[str, Any]       				=		hidden_act
   __A		:  Tuple       				=		mlp_dim
   __A		:  Any       				=		num_hidden_layers
   __A		:  Dict       				=		num_attention_heads
   __A		:  Optional[Any]       				=		attention_downsample_rate
   __A		:  List[Any]       				=		num_multimask_outputs
   __A		:  Optional[Any]       				=		iou_head_depth
   __A		:  Optional[int]       				=		iou_head_hidden_dim
   __A		:  Any       				=		layer_norm_eps







class 			__A					(			_SCREAMING_SNAKE_CASE      ):
 def __init__( self      :     Optional[int]					, lowerCamelCase      :     List[str]=7_68					, lowerCamelCase      :     Tuple=2_56					, lowerCamelCase      :     Optional[Any]=12					, lowerCamelCase      :     Tuple=12					, lowerCamelCase      :     int=3					, lowerCamelCase      :     Dict=10_24					, lowerCamelCase      :     int=16					, lowerCamelCase      :     Any="gelu"					, lowerCamelCase      :     Tuple=1e-0_6					, lowerCamelCase      :     Optional[int]=0.0					, lowerCamelCase      :     List[str]=1e-1_0					, lowerCamelCase      :     Optional[int]=True					, lowerCamelCase      :     Union[str, Any]=4.0					, lowerCamelCase      :     List[Any]=True					, lowerCamelCase      :     Optional[Any]=True					, lowerCamelCase      :     str=14					, lowerCamelCase      :     str=[2, 5, 8, 11]					, lowerCamelCase      :     Dict=1_28					, lowerCamelCase      :     Any=None					, **lowerCamelCase      :     Optional[int]					, ):



   """simple docstring"""





   super().__init__(**lowerCamelCase							)

   __A		:  Any       				=		hidden_size
   __A		:  Union[str, Any]       				=		output_channels
   __A		:  int       				=		num_hidden_layers
   __A		:  Any       				=		num_attention_heads
   __A		:  Tuple       				=		num_channels
   __A		:  Tuple       				=		image_size
   __A		:  Optional[Any]       				=		patch_size
   __A		:  List[str]       				=		hidden_act
   __A		:  Dict       				=		layer_norm_eps
   __A		:  Optional[int]       				=		attention_dropout
   __A		:  Dict       				=		initializer_range
   __A		:  Union[str, Any]       				=		qkv_bias
   __A		:  Optional[Any]       				=		mlp_ratio
   __A		:  Optional[int]       				=		use_abs_pos
   __A		:  int       				=		use_rel_pos
   __A		:  str       				=		window_size
   __A		:  Union[str, Any]       				=		global_attn_indexes
   __A		:  Dict       				=		num_pos_feats
   __A		:  int       				=		int(hidden_size * mlp_ratio							) if mlp_dim is None else mlp_dim







class 			__A					(			_SCREAMING_SNAKE_CASE      ):
 lowerCamelCase												='''sam'''
 lowerCamelCase												=True
 def __init__( self      :     Optional[int]					, lowerCamelCase      :     Dict=None					, lowerCamelCase      :     Dict=None					, lowerCamelCase      :     Optional[Any]=None					, lowerCamelCase      :     List[Any]=0.02					, **lowerCamelCase      :     Any					, ):



   """simple docstring"""





   super().__init__(**lowerCamelCase							)
   __A		:  str       				=		vision_config if vision_config is not None else {}
   __A		:  int       				=		prompt_encoder_config if prompt_encoder_config is not None else {}
   __A		:  Dict       				=		mask_decoder_config if mask_decoder_config is not None else {}

   if isinstance(lowerCamelCase					, lowerCamelCase							):
     __A		:  List[str]       				=		vision_config.to_dict()
   if isinstance(lowerCamelCase					, lowerCamelCase							):
     __A		:  Any       				=		prompt_encoder_config.to_dict()
   if isinstance(lowerCamelCase					, lowerCamelCase							):
     __A		:  Any       				=		mask_decoder_config.to_dict()

   __A		:  Any       				=		SamVisionConfig(**lowerCamelCase							)
   __A		:  List[Any]       				=		SamPromptEncoderConfig(**lowerCamelCase							)
   __A		:  Union[str, Any]       				=		SamMaskDecoderConfig(**lowerCamelCase							)
   __A		:  Any       				=		initializer_range






 def 		lowercase_( self      :     List[Any]							):



   """simple docstring"""





   __A		:  str       				=		copy.deepcopy(self.__dict__							)
   __A		:  Dict       				=		self.vision_config.to_dict()
   __A		:  Any       				=		self.prompt_encoder_config.to_dict()
   __A		:  str       				=		self.mask_decoder_config.to_dict()
   __A		:  Optional[int]       				=		self.__class__.model_type
   return output



