



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__     (			_UpperCAmelCase	,      _UpperCAmelCase	,      _UpperCAmelCase	,      _UpperCAmelCase=None				)     ->			List[Any]:






      '''simple docstring'''


      lowercase  :      List[str]					=   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.
            lowercase  :      Optional[int]					=   weight.view(torch.intaa				)
            lowercase  :      Dict					=   'bfloat16'
      lowercase  :      str					=   weight.cpu().numpy()
      lowercase  :      Any					=   os.path.join(_UpperCAmelCase	,      f'''{weight_name}.dat'''				)
      if index is not None:
            if dtype is None:
                  lowercase  :      Union[str, Any]					=   str(array.dtype				)
            lowercase  :      Dict					=   {'dtype': dtype, 'shape': list(array.shape				)}
      if array.ndim == 0:
            lowercase  :      Optional[int]					=   array[None]
      lowercase  :      Dict					=   np.memmap(_UpperCAmelCase	,      dtype=array.dtype	,      mode='w+'	,      shape=array.shape				)
      lowercase  :      Union[str, Any]					=   array[:]
      file_array.flush()
      return index


def   lowercase__     (			_UpperCAmelCase	,      _UpperCAmelCase				)     ->			Optional[Any]:






      '''simple docstring'''


      lowercase  :      Tuple					=   tuple(weight_info['shape']				)
      if shape == ():
            # NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
            lowercase  :      int					=   (1,)

      lowercase  :      List[Any]					=   weight_info['dtype']
      if dtype == "bfloat16":
            # NumPy does not support bfloat16 so this was saved as a int16
            lowercase  :      str					=   'int16'

      lowercase  :      Tuple					=   np.memmap(_UpperCAmelCase	,      dtype=_UpperCAmelCase	,      shape=_UpperCAmelCase	,      mode='r'				)

      if len(weight_info['shape']				) == 0:
            lowercase  :      List[str]					=   weight[0]
      lowercase  :      Optional[int]					=   torch.tensor(_UpperCAmelCase				)
      if weight_info["dtype"] == "bfloat16":
            lowercase  :      Tuple					=   weight.view(torch.bfloataa				)

      return weight


def   lowercase__     (			_UpperCAmelCase	,      _UpperCAmelCase				)     ->			List[Any]:






      '''simple docstring'''


      if index is None or len(_UpperCAmelCase				) == 0:
            # Nothing to save
            return

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

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


def   lowercase__     (			_UpperCAmelCase	,      _UpperCAmelCase				)     ->			Union[str, Any]:






      '''simple docstring'''


      os.makedirs(_UpperCAmelCase	,      exist_ok=_UpperCAmelCase				)
      lowercase  :      int					=   {}
      for name, parameter in state_dict.items():
            lowercase  :      List[str]					=   offload_weight(_UpperCAmelCase	,      _UpperCAmelCase	,      _UpperCAmelCase	,      index=_UpperCAmelCase				)

      # Update index
      save_offload_index(_UpperCAmelCase	,      _UpperCAmelCase				)


class   a__		(						SCREAMING_SNAKE_CASE__     ):


   def __init__(							self      :							Dict,     lowerCAmelCase      :							Mapping,     lowerCAmelCase      :							str			)    ->       Union[str, Any]:
         lowercase  :      Optional[Any]					=   dataset
         lowercase  :      List[Any]					=   prefix


   def __getitem__(							self      :							str,     lowerCAmelCase      :							int			)    ->       List[str]:
         return self.dataset[f'''{self.prefix}{key}''']


   def __iter__(							self      :							int			)    ->       str:
         return iter([key for key in self.dataset if key.startswith(self.prefix			)]			)






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




class   a__		(						SCREAMING_SNAKE_CASE__     ):


   def __init__(							self      :							Dict,     lowerCAmelCase      :							Dict[str, torch.Tensor] = None,     lowerCAmelCase      :							Optional[Union[str, os.PathLike]] = None,     lowerCAmelCase      :							Mapping = None,     lowerCAmelCase      :							int=None,     )    ->       Union[str, Any]:
         if state_dict is None and save_folder is None:
               raise ValueError('Need either a `state_dict` or a `save_folder` containing offloaded weights.'			)

         lowercase  :      List[Any]					=   {} if state_dict is None else state_dict
         lowercase  :      Tuple					=   save_folder
         if index is None and save_folder is not None:
               with open(os.path.join(lowerCAmelCase,     'index.json'			)			) as f:
                     lowercase  :      Dict					=   json.load(lowerCAmelCase			)
         lowercase  :      Optional[Any]					=   {} if index is None else index
         lowercase  :      str					=   list(self.state_dict.keys()			)
         self.all_keys.extend([key for key in self.index if key not in self.all_keys]			)
         lowercase  :      Any					=   device


   def __getitem__(							self      :							List[Any],     lowerCAmelCase      :							str			)    ->       List[Any]:
         # State dict gets priority
         if key in self.state_dict:
               return self.state_dict[key]
         lowercase  :      str					=   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

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

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

         lowercase  :      List[str]					=   os.path.join(self.save_folder,     f'''{key}.dat'''			)
         return load_offloaded_weight(lowerCAmelCase,     lowerCAmelCase			)


   def __iter__(							self      :							List[str]			)    ->       List[Any]:
         return iter(self.all_keys			)






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




def   lowercase__     (			_UpperCAmelCase	,      _UpperCAmelCase				)     ->			Union[str, Any]:






      '''simple docstring'''


      lowercase  :      List[str]					=   {}
      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





"""simple docstring"""




import os
import pathlib
import tempfile
import uuid

import numpy as np

from ..utils import is_soundfile_availble, is_torch_available, is_vision_available, logging


_UpperCamelCase:				List[str]    							=  logging.get_logger(__name__)

if is_vision_available():
     import PIL.Image
     from PIL import Image
     from PIL.Image import Image as ImageType
else:
     _UpperCamelCase:				Tuple    							=  object

if is_torch_available():
     import torch

if is_soundfile_availble():
     import soundfile as sf


class   a__		:


   def __init__(							self      :							Union[str, Any],     lowerCAmelCase      :							Optional[int]			)    ->       Any:
         lowercase  :      Dict					=   value


   def __str__(							self      :							List[str]			)    ->       Union[str, Any]:
         return self.to_string()


   def        lowercase				(							self      :							Dict			)    ->       Optional[Any]:
         logger.error(
             'This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'			)
         return self._value






   def        lowercase				(							self      :							List[Any]			)    ->       str:
         logger.error(
             'This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'			)
         return str(self._value			)




class   a__		(						SCREAMING_SNAKE_CASE__,  SCREAMING_SNAKE_CASE__     ):


   def        lowercase				(							self      :							List[Any]			)    ->       List[Any]:
         return self._value






   def        lowercase				(							self      :							str			)    ->       Optional[int]:
         return self._value




class   a__		(						SCREAMING_SNAKE_CASE__,  SCREAMING_SNAKE_CASE__     ):


   def __init__(							self      :							int,     lowerCAmelCase      :							Optional[Any]			)    ->       Any:
         super().__init__(lowerCAmelCase			)

         if not is_vision_available():
               raise ImportError('PIL must be installed in order to handle images.'			)

         lowercase  :      Union[str, Any]					=   None
         lowercase  :      List[str]					=   None
         lowercase  :      Optional[Any]					=   None

         if isinstance(lowerCAmelCase,     lowerCAmelCase			):
               lowercase  :      List[Any]					=   value
         elif isinstance(lowerCAmelCase,     (str, pathlib.Path)			):
               lowercase  :      Any					=   value
         elif isinstance(lowerCAmelCase,     torch.Tensor			):
               lowercase  :      str					=   value
         else:
               raise ValueError(f'''Unsupported type for {self.__class__.__name__}: {type(lowerCAmelCase			)}'''			)


   def        lowercase				(							self      :							int,     lowerCAmelCase      :							Union[str, Any]=None,     lowerCAmelCase      :							Optional[Any]=None			)    ->       List[Any]:
         from IPython.display import Image, display

         display(Image(self.to_string()			)			)


   def        lowercase				(							self      :							int			)    ->       List[str]:
         if self._raw is not None:
               return self._raw

         if self._path is not None:
               lowercase  :      Any					=   Image.open(self._path			)
               return self._raw






   def        lowercase				(							self      :							List[Any]			)    ->       List[Any]:
         if self._path is not None:
               return self._path

         if self._raw is not None:
               lowercase  :      Tuple					=   tempfile.mkdtemp()
               lowercase  :      Dict					=   os.path.join(lowerCAmelCase,     str(uuid.uuida()			) + '.png'			)
               self._raw.save(self._path			)

               return self._path

         if self._tensor is not None:
               lowercase  :      Optional[int]					=   self._tensor.cpu().detach().numpy()

               # There is likely simpler than load into image into save
               lowercase  :      Dict					=   Image.fromarray((array * 255).astype(np.uinta			)			)

               lowercase  :      Optional[int]					=   tempfile.mkdtemp()
               lowercase  :      Any					=   os.path.join(lowerCAmelCase,     str(uuid.uuida()			) + '.png'			)

               img.save(self._path			)

               return self._path




class   a__		(						SCREAMING_SNAKE_CASE__     ):


   def __init__(							self      :							Dict,     lowerCAmelCase      :							Any,     lowerCAmelCase      :							Any=16000			)    ->       Tuple:
         super().__init__(lowerCAmelCase			)

         if not is_soundfile_availble():
               raise ImportError('soundfile must be installed in order to handle audio.'			)

         lowercase  :      List[str]					=   None
         lowercase  :      List[Any]					=   None

         lowercase  :      Optional[Any]					=   samplerate

         if isinstance(lowerCAmelCase,     (str, pathlib.Path)			):
               lowercase  :      List[Any]					=   value
         elif isinstance(lowerCAmelCase,     torch.Tensor			):
               lowercase  :      List[str]					=   value
         else:
               raise ValueError(f'''Unsupported audio type: {type(lowerCAmelCase			)}'''			)


   def        lowercase				(							self      :							Optional[int],     lowerCAmelCase      :							Optional[Any]=None,     lowerCAmelCase      :							int=None			)    ->       Optional[Any]:
         from IPython.display import Audio, display

         display(Audio(self.to_string(),     rate=self.samplerate			)			)


   def        lowercase				(							self      :							Any			)    ->       Union[str, Any]:
         if self._tensor is not None:
               return self._tensor

         if self._path is not None:
               lowercase	,    lowercase  :      List[Any]					=   sf.read(self._path			)
               lowercase  :      int					=   torch.tensor(lowerCAmelCase			)
               return self._tensor






   def        lowercase				(							self      :							Tuple			)    ->       Tuple:
         if self._path is not None:
               return self._path

         if self._tensor is not None:
               lowercase  :      str					=   tempfile.mkdtemp()
               lowercase  :      Union[str, Any]					=   os.path.join(lowerCAmelCase,     str(uuid.uuida()			) + '.wav'			)
               sf.write(self._path,     self._tensor,     samplerate=self.samplerate			)
               return self._path




_UpperCamelCase:				Optional[int]    							=  {'text': AgentText, 'image': AgentImage, 'audio': AgentAudio}
_UpperCamelCase:				Optional[int]    							=  {str: AgentText}

if is_vision_available():
     _UpperCamelCase:				str    							=  AgentImage


def   lowercase__     (			*_UpperCAmelCase	,      **_UpperCAmelCase				)     ->			Any:






      '''simple docstring'''


      lowercase  :      str					=   [(arg.to_raw() if isinstance(_UpperCAmelCase	,      _UpperCAmelCase				) else arg) for arg in args]
      lowercase  :      Optional[Any]					=   {k: (v.to_raw() if isinstance(_UpperCAmelCase	,      _UpperCAmelCase				) else v) for k, v in kwargs.items()}
      return args, kwargs





def   lowercase__     (			_UpperCAmelCase	,      _UpperCAmelCase=None				)     ->			Tuple:






      '''simple docstring'''


      if isinstance(_UpperCAmelCase	,      _UpperCAmelCase				):
            lowercase  :      List[Any]					=   {}
            for i, (k, v) in enumerate(outputs.items()				):
                  if output_types is not None:
                        # If the class has defined outputs, we can map directly according to the class definition
                        if output_types[i] in AGENT_TYPE_MAPPING:
                              lowercase  :      int					=   AGENT_TYPE_MAPPING[output_types[i]](_UpperCAmelCase				)
                        else:
                              lowercase  :      Union[str, Any]					=   AgentType(_UpperCAmelCase				)

                  else:
                        # If the class does not have defined output, then we map according to the type
                        for _k, _v in INSTANCE_TYPE_MAPPING.items():
                              if isinstance(_UpperCAmelCase	,      _k				):
                                    lowercase  :      Optional[int]					=   _v(_UpperCAmelCase				)
                        if k not in decoded_outputs:
                              lowercase  :      str					=   AgentType[v]

      elif isinstance(_UpperCAmelCase	,      (list, tuple)				):
            lowercase  :      Union[str, Any]					=   type(_UpperCAmelCase				)()
            for i, v in enumerate(_UpperCAmelCase				):
                  if output_types is not None:
                        # If the class has defined outputs, we can map directly according to the class definition
                        if output_types[i] in AGENT_TYPE_MAPPING:
                              decoded_outputs.append(AGENT_TYPE_MAPPING[output_types[i]](_UpperCAmelCase				)				)
                        else:
                              decoded_outputs.append(AgentType(_UpperCAmelCase				)				)
                  else:
                        # If the class does not have defined output, then we map according to the type
                        lowercase  :      Any					=   False
                        for _k, _v in INSTANCE_TYPE_MAPPING.items():
                              if isinstance(_UpperCAmelCase	,      _k				):
                                    decoded_outputs.append(_v(_UpperCAmelCase				)				)
                                    lowercase  :      Optional[Any]					=   True

                        if not found:
                              decoded_outputs.append(AgentType(_UpperCAmelCase				)				)

      else:
            if output_types[0] in AGENT_TYPE_MAPPING:
                  # If the class has defined outputs, we can map directly according to the class definition
                  lowercase  :      List[Any]					=   AGENT_TYPE_MAPPING[output_types[0]](_UpperCAmelCase				)

            else:
                  # If the class does not have defined output, then we map according to the type
                  for _k, _v in INSTANCE_TYPE_MAPPING.items():
                        if isinstance(_UpperCAmelCase	,      _k				):
                              return _v(_UpperCAmelCase				)
                  return AgentType(_UpperCAmelCase				)

      return decoded_outputs
