
import json
import os
from collections.abc import Mapping
from typing import Dict, List, Optional, Union

import numpy as np
import torch

from .imports import is_safetensors_available



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE=None )	->     int:
			UpperCamelCase__    :		List[Any]     =				None
			# Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
			if str(weight.dtype ) == "torch.bfloat16":
						# Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
						UpperCamelCase__    :		List[Any]     =				weight.view(torch.intaa )
						UpperCamelCase__    :		Any     =				'bfloat16'
			UpperCamelCase__    :		Union[str, Any]     =				weight.cpu().numpy()
			UpperCamelCase__    :		List[Any]     =				os.path.join(__SCREAMING_SNAKE_CASE				,    F"""{weight_name}.dat""" )
			if index is not None:
						if dtype is None:
									UpperCamelCase__    :		Any     =				str(array.dtype )
						UpperCamelCase__    :		Any     =				{'dtype': dtype, 'shape': list(array.shape )}
			if array.ndim == 0:
						UpperCamelCase__    :		Optional[Any]     =				array[None]
			UpperCamelCase__    :		Optional[Any]     =				np.memmap(__SCREAMING_SNAKE_CASE				,    dtype=array.dtype				,    mode='w+'				,    shape=array.shape )
			UpperCamelCase__    :		Any     =				array[:]
			file_array.flush()
			return index



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )	->     int:
			UpperCamelCase__    :		List[Any]     =				tuple(weight_info['shape'] )
			if shape == ():
						# NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
						UpperCamelCase__    :		Optional[Any]     =				(1,)

			UpperCamelCase__    :		Dict     =				weight_info['dtype']
			if dtype == "bfloat16":
						# NumPy does not support bfloat16 so this was saved as a int16
						UpperCamelCase__    :		Optional[int]     =				'int16'

			UpperCamelCase__    :		Optional[Any]     =				np.memmap(__SCREAMING_SNAKE_CASE				,    dtype=__SCREAMING_SNAKE_CASE				,    shape=__SCREAMING_SNAKE_CASE				,    mode='r' )

			if len(weight_info['shape'] ) == 0:
						UpperCamelCase__    :		Tuple     =				weight[0]
			UpperCamelCase__    :		int     =				torch.tensor(__SCREAMING_SNAKE_CASE )
			if weight_info["dtype"] == "bfloat16":
						UpperCamelCase__    :		int     =				weight.view(torch.bfloataa )

			return weight



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )	->     List[str]:
			if index is None or len(__SCREAMING_SNAKE_CASE ) == 0:
						# Nothing to save
						return

			UpperCamelCase__    :		str     =				os.path.join(__SCREAMING_SNAKE_CASE				,    'index.json' )
			if os.path.isfile(__SCREAMING_SNAKE_CASE ):
						with open(__SCREAMING_SNAKE_CASE				,    'r'				,    encoding='utf-8' ) as f:
									UpperCamelCase__    :		Optional[Any]     =				json.load(__SCREAMING_SNAKE_CASE )
			else:
						UpperCamelCase__    :		int     =				{}
			current_index.update(__SCREAMING_SNAKE_CASE )

			with open(__SCREAMING_SNAKE_CASE				,    'w'				,    encoding='utf-8' ) as f:
						json.dump(__SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    indent=2 )



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )	->     Tuple:
			os.makedirs(__SCREAMING_SNAKE_CASE				,    exist_ok=__SCREAMING_SNAKE_CASE )
			UpperCamelCase__    :		Tuple     =				{}
			for name, parameter in state_dict.items():
						UpperCamelCase__    :		Optional[int]     =				offload_weight(__SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    index=__SCREAMING_SNAKE_CASE )

			# Update index
			save_offload_index(__SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )



class 			UpperCamelCase_			(						__UpperCamelCase   ):




							'''simple docstring'''

							def __init__(				self     ,						UpperCamelCase     ,						UpperCamelCase) ->	List[Any]:
										UpperCamelCase__    :		Dict     =				dataset
										UpperCamelCase__    :		str     =				prefix
							def __getitem__(				self     ,						UpperCamelCase) ->	int:
										return self.dataset[F"""{self.prefix}{key}"""]
							def __iter__(				self) ->	List[str]:
										return iter([key for key in self.dataset if key.startswith(self.prefix)])






							def __len__(				self) ->	Union[str, Any]:
										return len(self.dataset)



class 			UpperCamelCase_			(						__UpperCamelCase   ):




							'''simple docstring'''

							def __init__(				self     ,						UpperCamelCase = None     ,						UpperCamelCase = None     ,						UpperCamelCase = None     ,						UpperCamelCase=None     ,						) ->	Optional[int]:
										if state_dict is None and save_folder is None:
													raise ValueError('Need either a `state_dict` or a `save_folder` containing offloaded weights.')

										UpperCamelCase__    :		int     =				{} if state_dict is None else state_dict
										UpperCamelCase__    :		Tuple     =				save_folder
										if index is None and save_folder is not None:
													with open(os.path.join(UpperCamelCase     ,						'index.json')) as f:
																UpperCamelCase__    :		Tuple     =				json.load(UpperCamelCase)
										UpperCamelCase__    :		int     =				{} if index is None else index
										UpperCamelCase__    :		Any     =				list(self.state_dict.keys())
										self.all_keys.extend([key for key in self.index if key not in self.all_keys])
										UpperCamelCase__    :		List[str]     =				device
							def __getitem__(				self     ,						UpperCamelCase) ->	Optional[int]:
										# State dict gets priority
										if key in self.state_dict:
													return self.state_dict[key]
										UpperCamelCase__    :		Dict     =				self.index[key]
										if weight_info.get('safetensors_file') is not None:
													if not is_safetensors_available():
																raise ImportError('These offloaded weights require the use of safetensors: `pip install safetensors`.')

													from safetensors import safe_open

													UpperCamelCase__    :		Optional[Any]     =				'cpu' if self.device is None else self.device
													with safe_open(weight_info['safetensors_file']     ,						framework='pt'     ,						device=UpperCamelCase) as f:
																UpperCamelCase__    :		Dict     =				f.get_tensor(weight_info.get('weight_name'     ,						UpperCamelCase))

													if "dtype" in weight_info:
																return tensor.to(getattr(UpperCamelCase     ,						weight_info['dtype']))
													else:
																return tensor

										UpperCamelCase__    :		Any     =				os.path.join(self.save_folder     ,						F"""{key}.dat""")
										return load_offloaded_weight(UpperCamelCase     ,						UpperCamelCase)
							def __iter__(				self) ->	str:
										return iter(self.all_keys)






							def __len__(				self) ->	int:
										return len(self.all_keys)



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )	->     Optional[Any]:
			UpperCamelCase__    :		Any     =				{}
			for module_name in submodule_names:
						# We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
						# submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
						result.update(
						    {
						        key: param
						        for key, param in state_dict.items()
						        if key == module_name or key.startswith(module_name + '.' )
						    } )
			return result





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



class 			UpperCamelCase_			(						enum.Enum   ):




							'''simple docstring'''

							UpperCamelCase_					   =							"""timm"""
							UpperCamelCase_					   =							"""transformers"""



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE )	->     Dict:
			if stage_names is None:
						raise ValueError('Stage_names must be set for transformers backbones' )

			if out_features is not None:
						if not isinstance(__SCREAMING_SNAKE_CASE				,    (list,) ):
									raise ValueError(F"""out_features must be a list {type(__SCREAMING_SNAKE_CASE )}""" )
						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(__SCREAMING_SNAKE_CASE				,    (list, tuple) ):
									raise ValueError(F"""out_indices must be a list or tuple, got {type(__SCREAMING_SNAKE_CASE )}""" )
						if any(idx >= len(__SCREAMING_SNAKE_CASE ) 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(__SCREAMING_SNAKE_CASE ) != len(__SCREAMING_SNAKE_CASE ):
									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       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    )	->     List[Any]:
			if out_indices is None and out_features is None:
						UpperCamelCase__    :		List[Any]     =				[len(__SCREAMING_SNAKE_CASE ) - 1]
						UpperCamelCase__    :		Tuple     =				[stage_names[-1]]
			elif out_indices is None and out_features is not None:
						UpperCamelCase__    :		str     =				[stage_names.index(__SCREAMING_SNAKE_CASE ) for layer in out_features]
			elif out_features is None and out_indices is not None:
						UpperCamelCase__    :		int     =				[stage_names[idx] for idx in out_indices]
			return out_features, out_indices



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    )	->     Tuple[List[str], List[int]]:
			verify_out_features_out_indices(out_features=__SCREAMING_SNAKE_CASE				,    out_indices=__SCREAMING_SNAKE_CASE				,    stage_names=__SCREAMING_SNAKE_CASE )
			UpperCamelCase__   ,			UpperCamelCase__    :		str     =				_align_output_features_output_indices(
			    out_features=__SCREAMING_SNAKE_CASE				,    out_indices=__SCREAMING_SNAKE_CASE				,    stage_names=__SCREAMING_SNAKE_CASE )
			# Verify that the aligned out_features and out_indices are valid
			verify_out_features_out_indices(out_features=__SCREAMING_SNAKE_CASE				,    out_indices=__SCREAMING_SNAKE_CASE				,    stage_names=__SCREAMING_SNAKE_CASE )
			return output_features, output_indices



class 			UpperCamelCase_			:




							'''simple docstring'''

							UpperCamelCase_					   =							None
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	None:
										if getattr(self     ,						'_backbone'     ,						UpperCamelCase) 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']
										UpperCamelCase__    :		int     =				[stage['module'] for stage in self._backbone.feature_info.info]
										UpperCamelCase__    :		str     =				[stage['num_chs'] for stage in self._backbone.feature_info.info]
										UpperCamelCase__    :		Optional[int]     =				self._backbone.feature_info.out_indices
										UpperCamelCase__    :		Optional[int]     =				self._backbone.feature_info.module_name()

										# We verify the out indices and out features are valid
										verify_out_features_out_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[int]     =				out_features, out_indices
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	None:
										UpperCamelCase__    :		List[str]     =				getattr(UpperCamelCase     ,						'stage_names')
										UpperCamelCase__    :		List[str]     =				getattr(UpperCamelCase     ,						'out_features'     ,						UpperCamelCase)
										UpperCamelCase__    :		Optional[Any]     =				getattr(UpperCamelCase     ,						'out_indices'     ,						UpperCamelCase)

										UpperCamelCase__    :		Tuple     =				stage_names
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[Any]     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=UpperCamelCase)
										# Number of channels for each stage. This is set in the transformer backbone model init
										UpperCamelCase__    :		Dict     =				None
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	None:
										UpperCamelCase__    :		List[Any]     =				config

										UpperCamelCase__    :		Dict     =				getattr(UpperCamelCase     ,						'use_timm_backbone'     ,						UpperCamelCase)
										UpperCamelCase__    :		List[str]     =				BackboneType.TIMM if self.use_timm_backbone else BackboneType.TRANSFORMERS

										if self.backbone_type == BackboneType.TIMM:
													self._init_timm_backbone(UpperCamelCase)
										elif self.backbone_type == BackboneType.TRANSFORMERS:
													self._init_transformers_backbone(UpperCamelCase)
										else:
													raise ValueError(F"""backbone_type {self.backbone_type} not supported.""")
							@property
							def        lowerCAmelCase__ (				self) ->	List[str]:
										return self._out_features
							@out_features.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	int:
										UpperCamelCase__   ,			UpperCamelCase__    :		str     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
							@property
							def        lowerCAmelCase__ (				self) ->	List[Any]:
										return self._out_indices
							@out_indices.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	int:
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[int]     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
							@property
							def        lowerCAmelCase__ (				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        lowerCAmelCase__ (				self) ->	Union[str, Any]:
										return [self.out_feature_channels[name] for name in self.out_features]
							def        lowerCAmelCase__ (				self     ,						*UpperCamelCase     ,						**UpperCamelCase) ->	str:
										UpperCamelCase__    :		List[Any]     =				dict(inspect.signature(self.forward).parameters)
										UpperCamelCase__    :		Any     =				{k: v for k, v in kwargs.items() if k in signature}
										return self(*UpperCamelCase     ,						**UpperCamelCase)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase = None     ,						UpperCamelCase = None     ,						UpperCamelCase = None     ,						) ->	Union[str, Any]:
										raise NotImplementedError('This method should be implemented by the derived class.')






							def        lowerCAmelCase__ (				self) ->	int:
										UpperCamelCase__    :		Any     =				super().to_dict()
										UpperCamelCase__    :		Optional[Any]     =				output.pop('_out_features')
										UpperCamelCase__    :		Any     =				output.pop('_out_indices')
										return output



class 			UpperCamelCase_			:




							'''simple docstring'''

							@property
							def        lowerCAmelCase__ (				self) ->	Tuple:
										return self._out_features
							@out_features.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	Tuple:
										UpperCamelCase__   ,			UpperCamelCase__    :		Union[str, Any]     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)
							@property
							def        lowerCAmelCase__ (				self) ->	Optional[int]:
										return self._out_indices
							@out_indices.setter
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	int:
										UpperCamelCase__   ,			UpperCamelCase__    :		str     =				get_aligned_output_features_output_indices(
										    out_features=UpperCamelCase     ,						out_indices=UpperCamelCase     ,						stage_names=self.stage_names)






							def        lowerCAmelCase__ (				self) ->	Tuple:
										UpperCamelCase__    :		List[Any]     =				super().to_dict()
										UpperCamelCase__    :		Optional[int]     =				output.pop('_out_features')
										UpperCamelCase__    :		Tuple     =				output.pop('_out_indices')
										return output



