




'''simple docstring'''





import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


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

from .logging import get_logger
from .state import PartialState


UpperCamelCase_		:    List[str]									=		get_logger(__name__)





def 			__a			(     _UpperCamelCase:					str      ,		_UpperCamelCase:					List[dict]      ,		_UpperCamelCase:					list      ,		_UpperCamelCase:					list      ,		_UpperCamelCase:					int      ,		_UpperCamelCase:					GradScaler = None      ,		)		->				List[str]:

     """simple docstring"""




     for i, state in enumerate(_UpperCamelCase					):
          _snake_case							=   F"""{MODEL_NAME}.bin""" if i == 0 else F"""{MODEL_NAME}_{i}.bin"""
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          save(_UpperCamelCase      ,		_UpperCamelCase					)
          logger.info(F"""Model weights saved in {output_model_file}"""					)
     # Optimizer states
     for i, opt in enumerate(_UpperCamelCase					):
          _snake_case							=   opt.state_dict()
          _snake_case							=   F"""{OPTIMIZER_NAME}.bin""" if i == 0 else F"""{OPTIMIZER_NAME}_{i}.bin"""
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          save(_UpperCamelCase      ,		_UpperCamelCase					)
          logger.info(F"""Optimizer state saved in {output_optimizer_file}"""					)
     # Scheduler states
     for i, scheduler in enumerate(_UpperCamelCase					):
          _snake_case							=   scheduler.state_dict()
          _snake_case							=   F"""{SCHEDULER_NAME}.bin""" if i == 0 else F"""{SCHEDULER_NAME}_{i}.bin"""
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          save(_UpperCamelCase      ,		_UpperCamelCase					)
          logger.info(F"""Scheduler state saved in {output_scheduler_file}"""					)
     # GradScaler state
     if scaler is not None:
          _snake_case							=   scaler.state_dict()
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          torch.save(_UpperCamelCase      ,		_UpperCamelCase					)
          logger.info(F"""Gradient scaler state saved in {output_scaler_file}"""					)
     # Random number generator states
     _snake_case							=   {}
     _snake_case							=   F"""{RNG_STATE_NAME}_{process_index}.pkl"""
     _snake_case							=   random.getstate()
     _snake_case							=   np.random.get_state()
     _snake_case							=   torch.get_rng_state()
     if is_xpu_available():
          _snake_case							=   torch.xpu.get_rng_state_all()
     else:
          _snake_case							=   torch.cuda.get_rng_state_all()
     if is_tpu_available():
          _snake_case							=   xm.get_rng_state()
     _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
     torch.save(_UpperCamelCase      ,		_UpperCamelCase					)
     logger.info(F"""Random states saved in {output_states_file}"""					)
     return output_dir





def 			__a			(     _UpperCamelCase:					Optional[Any]      ,		_UpperCamelCase:					List[Any]      ,		_UpperCamelCase:					Optional[Any]      ,		_UpperCamelCase:					List[Any]      ,		_UpperCamelCase:					Any      ,		_UpperCamelCase:					List[Any]=None      ,		_UpperCamelCase:					int=None      ,		**_UpperCamelCase:					Union[str, Any]      ,		)		->				Optional[Any]:

     """simple docstring"""




     if map_location not in [None, "cpu", "on_device"]:
          raise TypeError(
              "Unsupported optimizer map location passed, please choose one of `None`, `'cpu'`, or `'on_device'`"					)
     if map_location is None:
          _snake_case							=   "cpu"
     elif map_location == "on_device":
          _snake_case							=   PartialState().device
     # Model states
     for i, model in enumerate(_UpperCamelCase					):
          _snake_case							=   F"""{MODEL_NAME}.bin""" if i == 0 else F"""{MODEL_NAME}_{i}.bin"""
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          models[i].load_state_dict(torch.load(_UpperCamelCase      ,		map_location=_UpperCamelCase					)      ,		**_UpperCamelCase					)
     logger.info("All model weights loaded successfully"					)

     # Optimizer states
     for i, opt in enumerate(_UpperCamelCase					):
          _snake_case							=   F"""{OPTIMIZER_NAME}.bin""" if i == 0 else F"""{OPTIMIZER_NAME}_{i}.bin"""
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          _snake_case							=   torch.load(_UpperCamelCase      ,		map_location=_UpperCamelCase					)
          optimizers[i].load_state_dict(_UpperCamelCase					)
     logger.info("All optimizer states loaded successfully"					)

     # Scheduler states
     for i, scheduler in enumerate(_UpperCamelCase					):
          _snake_case							=   F"""{SCHEDULER_NAME}.bin""" if i == 0 else F"""{SCHEDULER_NAME}_{i}.bin"""
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          scheduler.load_state_dict(torch.load(_UpperCamelCase					)					)
     logger.info("All scheduler states loaded successfully"					)

     # GradScaler state
     if scaler is not None:
          _snake_case							=   os.path.join(_UpperCamelCase      ,		_UpperCamelCase					)
          scaler.load_state_dict(torch.load(_UpperCamelCase					)					)
          logger.info("GradScaler state loaded successfully"					)

     # Random states
     try:
          _snake_case							=   torch.load(os.path.join(_UpperCamelCase      ,		F"""{RNG_STATE_NAME}_{process_index}.pkl"""					)					)
          random.setstate(states["random_state"]					)
          np.random.set_state(states["numpy_random_seed"]					)
          torch.set_rng_state(states["torch_manual_seed"]					)
          if is_xpu_available():
               torch.xpu.set_rng_state_all(states["torch_xpu_manual_seed"]					)
          else:
               torch.cuda.set_rng_state_all(states["torch_cuda_manual_seed"]					)
          if is_tpu_available():
               xm.set_rng_state(states["xm_seed"]					)
          logger.info("All random states loaded successfully"					)
     except Exception:
          logger.info("Could not load random states"					)





def 			__a			(     _UpperCamelCase:					List[str]      ,		_UpperCamelCase:					Optional[int]      ,		_UpperCamelCase:					int = 0					)		->				str:

     """simple docstring"""




     _snake_case							=   Path(_UpperCamelCase					) / F"""custom_checkpoint_{index}.pkl"""
     logger.info(F"""Saving the state of {get_pretty_name(_UpperCamelCase					)} to {save_location}"""					)
     torch.save(obj.state_dict()      ,		_UpperCamelCase					)





def 			__a			(     _UpperCamelCase:					Optional[int]      ,		_UpperCamelCase:					Union[str, Any]      ,		_UpperCamelCase:					int = 0					)		->				str:

     """simple docstring"""




     _snake_case							=   F"""{path}/custom_checkpoint_{index}.pkl"""
     logger.info(F"""Loading the state of {get_pretty_name(_UpperCamelCase					)} from {load_location}"""					)
     obj.load_state_dict(torch.load(_UpperCamelCase      ,		map_location="cpu"					)					)








'''simple docstring'''





import enum
import inspect
from typing import Iterable, List, Optional, Tuple, Union

class      _a       (		enum.Enum							):
     SCREAMING_SNAKE_CASE_							:					Optional[int]					       =		"""timm"""
     SCREAMING_SNAKE_CASE_							:					int					       =		"""transformers"""







def 			__a			(     _UpperCamelCase:					Optional[Iterable[str]]      ,		_UpperCamelCase:					Optional[Iterable[int]]      ,		_UpperCamelCase:					Optional[Iterable[str]]					)		->				Union[str, Any]:

     """simple docstring"""




     if stage_names is None:
          raise ValueError("Stage_names must be set for transformers backbones"					)

     if out_features is not None:
          if not isinstance(_UpperCamelCase      ,		(list,)					):
               raise ValueError(F"""out_features must be a list {type(_UpperCamelCase					)}"""					)
          if any(feat not in stage_names for feat in out_features					):
               raise ValueError(F"""out_features must be a subset of stage_names: {stage_names} got {out_features}"""					)

     if out_indices is not None:
          if not isinstance(_UpperCamelCase      ,		(list, tuple)					):
               raise ValueError(F"""out_indices must be a list or tuple, got {type(_UpperCamelCase					)}"""					)
          if any(idx >= len(_UpperCamelCase					) for idx in out_indices					):
               raise ValueError("out_indices must be valid indices for stage_names {stage_names}, got {out_indices}"					)

     if out_features is not None and out_indices is not None:
          if len(_UpperCamelCase					) != len(_UpperCamelCase					):
               raise ValueError("out_features and out_indices should have the same length if both are set"					)
          if out_features != [stage_names[idx] for idx in out_indices]:
               raise ValueError("out_features and out_indices should correspond to the same stages if both are set"					)





def 			__a			(     _UpperCamelCase:					Optional[List[str]]      ,		_UpperCamelCase:					Optional[Union[List[int], Tuple[int]]]      ,		_UpperCamelCase:					List[str]      ,		)		->				Any:

     """simple docstring"""




     if out_indices is None and out_features is None:
          _snake_case							=   [len(_UpperCamelCase					) - 1]
          _snake_case							=   [stage_names[-1]]
     elif out_indices is None and out_features is not None:
          _snake_case							=   [stage_names.index(_UpperCamelCase					) for layer in out_features]
     elif out_features is None and out_indices is not None:
          _snake_case							=   [stage_names[idx] for idx in out_indices]
     return out_features, out_indices





def 			__a			(     _UpperCamelCase:					Optional[List[str]]      ,		_UpperCamelCase:					Optional[Union[List[int], Tuple[int]]]      ,		_UpperCamelCase:					List[str]      ,		)		->				Tuple[List[str], List[int]]:

     """simple docstring"""




     verify_out_features_out_indices(out_features=_UpperCamelCase      ,		out_indices=_UpperCamelCase      ,		stage_names=_UpperCamelCase					)
     _snake_case							,      _snake_case							=   _align_output_features_output_indices(
         out_features=_UpperCamelCase      ,		out_indices=_UpperCamelCase      ,		stage_names=_UpperCamelCase					)
     # Verify that the aligned out_features and out_indices are valid
     verify_out_features_out_indices(out_features=_UpperCamelCase      ,		out_indices=_UpperCamelCase      ,		stage_names=_UpperCamelCase					)
     return output_features, output_indices

class      _a       :
     SCREAMING_SNAKE_CASE_							:					Optional[BackboneType]					       =		None






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    None:
          if getattr(self      ,"_backbone"      ,_SCREAMING_SNAKE_CASE					) is None:
               raise ValueError("self._backbone must be set before calling _init_timm_backbone"					)

          # These will diagree with the defaults for the transformers models e.g. for resnet50
          # the transformer model has out_features = ['stem', 'stage1', 'stage2', 'stage3', 'stage4']
          # the timm model has out_features = ['act', 'layer1', 'layer2', 'layer3', 'layer4']
          _snake_case							=   [stage["module"] for stage in self._backbone.feature_info.info]
          _snake_case							=   [stage["num_chs"] for stage in self._backbone.feature_info.info]
          _snake_case							=   self._backbone.feature_info.out_indices
          _snake_case							=   self._backbone.feature_info.module_name()

          # We verify the out indices and out features are valid
          verify_out_features_out_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)
          _snake_case							,      _snake_case							=   out_features, out_indices






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    None:
          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"stage_names"					)
          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"out_features"      ,_SCREAMING_SNAKE_CASE					)
          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"out_indices"      ,_SCREAMING_SNAKE_CASE					)

          _snake_case							=   stage_names
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=_SCREAMING_SNAKE_CASE					)
          # Number of channels for each stage. This is set in the transformer backbone model init
          _snake_case							=   None






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    None:
          _snake_case							=   config

          _snake_case							=   getattr(_SCREAMING_SNAKE_CASE      ,"use_timm_backbone"      ,_SCREAMING_SNAKE_CASE					)
          _snake_case							=   BackboneType.TIMM if self.use_timm_backbone else BackboneType.TRANSFORMERS

          if self.backbone_type == BackboneType.TIMM:
               self._init_timm_backbone(_SCREAMING_SNAKE_CASE					)
          elif self.backbone_type == BackboneType.TRANSFORMERS:
               self._init_transformers_backbone(_SCREAMING_SNAKE_CASE					)
          else:
               raise ValueError(f"""backbone_type {self.backbone_type} not supported."""					)






     @property
     def        _lowercase			(				self					)							->    int:
          return self._out_features






     @out_features.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    str:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     @property
     def        _lowercase			(				self					)							->    Any:
          return self._out_indices






     @out_indices.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    int:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     @property
     def        _lowercase			(				self					)							->    List[str]:
          # the current backbones will output the number of channels for each stage
          # even if that stage is not in the out_features list.
          return {stage: self.num_features[i] for i, stage in enumerate(self.stage_names					)}






     @property
     def        _lowercase			(				self					)							->    Any:
          return [self.out_feature_channels[name] for name in self.out_features]






     def        _lowercase			(				self      ,*_SCREAMING_SNAKE_CASE      ,**_SCREAMING_SNAKE_CASE					)							->    str:
          _snake_case							=   dict(inspect.signature(self.forward					).parameters					)
          _snake_case							=   {k: v for k, v in kwargs.items() if k in signature}
          return self(*_SCREAMING_SNAKE_CASE      ,**_SCREAMING_SNAKE_CASE					)






     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE      ,_SCREAMING_SNAKE_CASE = None      ,_SCREAMING_SNAKE_CASE = None      ,_SCREAMING_SNAKE_CASE = None      ,)							->    Any:
          raise NotImplementedError("This method should be implemented by the derived class."					)






     def        _lowercase			(				self					)							->    Tuple:
          _snake_case							=   super().to_dict()
          _snake_case							=   output.pop("_out_features"					)
          _snake_case							=   output.pop("_out_indices"					)
          return output







class      _a       :






     @property
     def        _lowercase			(				self					)							->    List[str]:
          return self._out_features






     @out_features.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    str:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     @property
     def        _lowercase			(				self					)							->    Any:
          return self._out_indices






     @out_indices.setter
     def        _lowercase			(				self      ,_SCREAMING_SNAKE_CASE					)							->    Optional[Any]:
          _snake_case							,      _snake_case							=   get_aligned_output_features_output_indices(
              out_features=_SCREAMING_SNAKE_CASE      ,out_indices=_SCREAMING_SNAKE_CASE      ,stage_names=self.stage_names					)






     def        _lowercase			(				self					)							->    List[str]:
          _snake_case							=   super().to_dict()
          _snake_case							=   output.pop("_out_features"					)
          _snake_case							=   output.pop("_out_indices"					)
          return output


