



'''simple docstring'''






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        snake_case_   (	lowerCAmelCase_			,  lowerCAmelCase_						)->						Any:





       '''simple docstring'''


       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 				lowercase			(    torch.optim.Optimizer							):


 """simple docstring"""


 def __init__(				self    ,a_    ,a_=True    ,a_=None				)						->  Dict:
        _UpperCAmelCase    :		Any     	=					optimizer
        _UpperCAmelCase    :		Optional[int]     	=					scaler
        _UpperCAmelCase    :		List[Any]     	=					AcceleratorState()
        _UpperCAmelCase    :		Any     	=					GradientState()
        _UpperCAmelCase    :		List[str]     	=					device_placement
        _UpperCAmelCase    :		str     	=					False
        _UpperCAmelCase    :		Union[str, Any]     	=					None

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







 @property
 def _snake_case					(				self				)						->  Dict:
        return self.optimizer.state







 @state.setter
 def _snake_case					(				self    ,a_				)						->  int:
        _UpperCAmelCase    :		Optional[int]     	=					state







 @property
 def _snake_case					(				self				)						->  int:
        return self.optimizer.param_groups







 @param_groups.setter
 def _snake_case					(				self    ,a_				)						->  Optional[int]:
        _UpperCAmelCase    :		int     	=					param_groups







 @property
 def _snake_case					(				self				)						->  List[Any]:
        return self.optimizer.defaults







 @defaults.setter
 def _snake_case					(				self    ,a_				)						->  Dict:
        _UpperCAmelCase    :		Optional[Any]     	=					defaults







 def _snake_case					(				self    ,a_				)						->  Optional[Any]:
        self.optimizer.add_param_group(a_				)







 def _snake_case					(				self    ,a_				)						->  Dict:
        if self.accelerator_state.distributed_type == DistributedType.TPU and self.device_placement:
               xm.send_cpu_data_to_device(a_    ,self.accelerator_state.device				)
        self.optimizer.load_state_dict(a_				)







 def _snake_case					(				self				)						->  Dict:
        return self.optimizer.state_dict()







 def _snake_case					(				self    ,a_=None				)						->  Optional[int]:
        if self.gradient_state.sync_gradients:
               _UpperCAmelCase    :		List[Any]     	=					"""set_to_none""" in inspect.signature(self.optimizer.zero_grad				).parameters
               if accept_arg:
                      if set_to_none is None:
                             _UpperCAmelCase    :		Tuple     	=					False
                      self.optimizer.zero_grad(set_to_none=a_				)
               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 _snake_case					(				self    ,a_=None				)						->  str:
        if self.gradient_state.sync_gradients:
               if self.accelerator_state.distributed_type == DistributedType.TPU:
                      _UpperCAmelCase    :		str     	=					{"""closure""": closure} if closure is not None else {}
                      xm.optimizer_step(self.optimizer    ,optimizer_args=a_				)
               elif self.scaler is not None:
                      _UpperCAmelCase    :		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
                             _UpperCAmelCase    :		Optional[Any]     	=					self.scaler.get_scale()
                             _UpperCAmelCase    :		Any     	=					True
                      self.scaler.step(self.optimizer    ,a_				)
                      self.scaler.update()
                      _UpperCAmelCase    :		int     	=					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.
                             _UpperCAmelCase    :		List[Any]     	=					scale_after < self._last_scale
                      _UpperCAmelCase    :		List[str]     	=					scale_after

               else:
                      self.optimizer.step(a_				)







 def _snake_case					(				self    ,a_				)						->  List[Any]:
        for param_group in self.optimizer.param_groups:
               _UpperCAmelCase    :		Union[str, Any]     	=					[parameters_map.get(a_    ,a_				) for p in param_group["""params"""]]







 @property
 def _snake_case					(				self				)						->  Union[str, Any]:
        warnings.warn(
            """The `is_overflow` property is deprecated and will be removed in version 1.0 of Accelerate use """
            """`optimizer.step_was_skipped` instead."""    ,a_    ,)
        return self._is_overflow







 @property
 def _snake_case					(				self				)						->  str:
        return self._is_overflow







 def __getstate__(				self				)						->  str:
        return self.__dict__.copy()







 def __setstate__(				self    ,a_				)						->  str:
        self.__dict__.update(a_				)








'''simple docstring'''






from unittest.mock import patch

import datasets
from datasets import Dataset




def        snake_case_   (	)->						str:





       '''simple docstring'''


       _UpperCAmelCase    :		Tuple     	=					Dataset.from_dict({"""col_1""": [3, 2, 0, 1]}						)

       with patch("""tqdm.auto.tqdm"""						) as mock_tqdm:
              datasets.disable_progress_bar()
              dset.map(lambda lowerCAmelCase_						: {"col_2": x["col_1"] + 1}						)
              mock_tqdm.assert_not_called()

              mock_tqdm.reset_mock()

              datasets.enable_progress_bar()
              dset.map(lambda lowerCAmelCase_						: {"col_2": x["col_1"] + 1}						)
              mock_tqdm.assert_called()



