




import inspect
import warnings

import torch

from .state import AcceleratorState, GradientState
from .utils import DistributedType, honor_type, is_tpu_available


if is_tpu_available(check_device=False):
     import torch_xla.core.xla_model as xm



def    UpperCAmelCase__	(  lowerCamelCase,      lowerCamelCase ):
  if isinstance(lowerCamelCase,      (list, tuple) ):
    return honor_type(lowerCamelCase,      (move_to_device(lowerCamelCase,      lowerCamelCase ) for t in state) )
  elif isinstance(lowerCamelCase,      lowerCamelCase ):
    return type(lowerCamelCase )({k: move_to_device(lowerCamelCase,      lowerCamelCase ) for k, v in state.items()} )
  elif isinstance(lowerCamelCase,      torch.Tensor ):
    return state.to(lowerCamelCase )
  return state







class  __lowerCAmelCase							(				torch.optim.Optimizer):






       def __init__( self:       Optional[int]							,       _lowerCAmelCase:       str							,       _lowerCAmelCase:       Optional[int]=True							,       _lowerCAmelCase:       Optional[int]=None					):
         lowercase     :List[Any]   						=    optimizer
         lowercase     :Union[str, Any]   						=    scaler
         lowercase     :Optional[Any]   						=    AcceleratorState()
         lowercase     :int   						=    GradientState()
         lowercase     :List[str]   						=    device_placement
         lowercase     :Optional[int]   						=    False
         lowercase     :Any   						=    None

         # Handle device placement
         if device_placement:
           lowercase     :List[str]   						=    self.optimizer.state_dict()
           if self.accelerator_state.distributed_type == DistributedType.TPU:
             xm.send_cpu_data_to_device(_lowerCAmelCase							,       self.accelerator_state.device					)
           else:
             lowercase     :Optional[Any]   						=    move_to_device(_lowerCAmelCase							,       self.accelerator_state.device					)
           self.optimizer.load_state_dict(_lowerCAmelCase					)






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]					):
         return self.optimizer.state






       @state.setter
       def        SCREAMING_SNAKE_CASE     ( self:       Any							,       _lowerCAmelCase:       Optional[int]					):
         lowercase     :int   						=    state






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       List[str]					):
         return self.optimizer.param_groups






       @param_groups.setter
       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]							,       _lowerCAmelCase:       Optional[int]					):
         lowercase     :Tuple   						=    param_groups






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       List[str]					):
         return self.optimizer.defaults






       @defaults.setter
       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]							,       _lowerCAmelCase:       List[str]					):
         lowercase     :int   						=    defaults






       def        SCREAMING_SNAKE_CASE     ( self:       str							,       _lowerCAmelCase:       Any					):
         self.optimizer.add_param_group(_lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]							,       _lowerCAmelCase:       Optional[Any]					):
         if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
           xm.send_cpu_data_to_device(_lowerCAmelCase							,       self.accelerator_state.device					)
         self.optimizer.load_state_dict(_lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       str					):
         return self.optimizer.state_dict()






       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]							,       _lowerCAmelCase:       Optional[Any]=None					):
         if self.gradient_state.sync_gradients:
           lowercase     :List[str]   						=    "set_to_none" in inspect.signature(self.optimizer.zero_grad					).parameters
           if accept_arg:
             if set_to_none is None:
               lowercase     :Optional[Any]   						=    False
             self.optimizer.zero_grad(set_to_none=_lowerCAmelCase					)
           else:
             if set_to_none is not None:
               raise ValueError("`set_to_none` for Optimizer.zero_grad` is not supported by this optimizer."					)
             self.optimizer.zero_grad()






       def        SCREAMING_SNAKE_CASE     ( self:       str							,       _lowerCAmelCase:       List[str]=None					):
         if self.gradient_state.sync_gradients:
           if self.accelerator_state.distributed_type == DistributedType.TPU:
             lowercase     :Optional[Any]   						=    {"closure": closure} if closure is not None else {}
             xm.optimizer_step(self.optimizer							,       optimizer_args=_lowerCAmelCase					)
           elif self.scaler is not None:
             lowercase     :Tuple   						=    False
             if self._last_scale is None:
               # `get_scale` is an async operation requiring full synchronization
               # on CPU and GPUs before finishing. As a result, we store away
               # the prior one to reduce the call overhead
               lowercase     :Optional[Any]   						=    self.scaler.get_scale()
               lowercase     :Union[str, Any]   						=    True
             self.scaler.step(self.optimizer							,       _lowerCAmelCase					)
             self.scaler.update()
             lowercase     :List[Any]   						=    self.scaler.get_scale()
             if not new_scale:
               # If we reduced the loss scale, it means the optimizer step was skipped because of gradient overflow.
               lowercase     :int   						=    scale_after < self._last_scale
             lowercase     :List[Any]   						=    scale_after

           else:
             self.optimizer.step(_lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]							,       _lowerCAmelCase:       List[str]					):
         for param_group in self.optimizer.param_groups:
           lowercase     :str   						=    [parameters_map.get(_lowerCAmelCase							,       _lowerCAmelCase					) for p in param_group["params"]]






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]					):
         warnings.warn(
             "The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use "
             "`optimizer.step_was_skipped` instead."							,       _lowerCAmelCase							,       )
         return self._is_overflow






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       Optional[Any]					):
         return self._is_overflow






       def __getstate__( self:       List[Any]					):
         return self.__dict__.copy()






       def __setstate__( self:       Any							,       _lowerCAmelCase:       str					):
         self.__dict__.update(_lowerCAmelCase					)









import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


_UpperCAmelCase						:    List[str]      =  2 << 20  # 2mb
_UpperCAmelCase						:    Optional[int]      =  get_logger(__name__)




class  __lowerCAmelCase							(				lowerCAmelCase):






       def        SCREAMING_SNAKE_CASE     ( self:       str					):
         lowercase     :str   						=    self._options.get_all_options().get("runner"					)
         return runner in [None, "DirectRunner", "PortableRunner"]







def    UpperCAmelCase__	(  lowerCamelCase,      lowerCamelCase,      lowerCamelCase=False ):
  lowercase     :str   						=    FileSystems
  if fs.exists(lowerCamelCase ):
    if force_upload:
      logger.info(F"Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True." )
    else:
      logger.info(F"Remote path already exist: {remote_file_path}. Skipping it as force_upload=False." )
      return
  with fs.create(lowerCamelCase ) as remote_file:
    with open(lowerCamelCase,      "rb" ) as local_file:
      lowercase     :int   						=    local_file.read(lowerCamelCase )
      while chunk:
        remote_file.write(lowerCamelCase )
        lowercase     :List[Any]   						=    local_file.read(lowerCamelCase )







def    UpperCAmelCase__	(  lowerCamelCase,      lowerCamelCase,      lowerCamelCase=False ):
  lowercase     :Union[str, Any]   						=    FileSystems
  if os.path.exists(lowerCamelCase ):
    if force_download:
      logger.info(F"Local path already exist: {remote_file_path}. Overwriting it as force_upload=True." )
    else:
      logger.info(F"Local path already exist: {remote_file_path}. Skipping it as force_upload=False." )
      return
  with fs.open(lowerCamelCase ) as remote_file:
    with open(lowerCamelCase,      "wb" ) as local_file:
      lowercase     :Optional[Any]   						=    remote_file.read(lowerCamelCase )
      while chunk:
        local_file.write(lowerCamelCase )
        lowercase     :Dict   						=    remote_file.read(lowerCamelCase )



