



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  _lowerCAmelCase     (_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase=None):
					UpperCamelCase_			= 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_			= weight.view(torch.intaa)
										UpperCamelCase_			= "bfloat16"
					UpperCamelCase_			= weight.cpu().numpy()
					UpperCamelCase_			= os.path.join(_lowerCAmelCase ,						f"""{weight_name}.dat""")
					if index is not None:
										if dtype is None:
															UpperCamelCase_			= str(array.dtype)
										UpperCamelCase_			= {"dtype": dtype, "shape": list(array.shape)}
					if array.ndim == 0:
										UpperCamelCase_			= array[None]
					UpperCamelCase_			= np.memmap(_lowerCAmelCase ,						dtype=array.dtype ,						mode="w+" ,						shape=array.shape)
					UpperCamelCase_			= array[:]
					file_array.flush()
					return index



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

					UpperCamelCase_			= weight_info["dtype"]
					if dtype == "bfloat16":
										# NumPy does not support bfloat16 so this was saved as a int16
										UpperCamelCase_			= "int16"

					UpperCamelCase_			= np.memmap(_lowerCAmelCase ,						dtype=_lowerCAmelCase ,						shape=_lowerCAmelCase ,						mode="r")

					if len(weight_info["shape"]) == 0:
										UpperCamelCase_			= weight[0]
					UpperCamelCase_			= torch.tensor(_lowerCAmelCase)
					if weight_info["dtype"] == "bfloat16":
										UpperCamelCase_			= weight.view(torch.bfloataa)

					return weight



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

					UpperCamelCase_			= os.path.join(_lowerCAmelCase ,						"index.json")
					if os.path.isfile(_lowerCAmelCase):
										with open(_lowerCAmelCase ,						"r" ,						encoding="utf-8") as f:
															UpperCamelCase_			= json.load(_lowerCAmelCase)
					else:
										UpperCamelCase_			= {}
					current_index.update(_lowerCAmelCase)

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



def  _lowerCAmelCase     (_lowerCAmelCase ,						_lowerCAmelCase):
					os.makedirs(_lowerCAmelCase ,						exist_ok=_lowerCAmelCase)
					UpperCamelCase_			= {}
					for name, parameter in state_dict.items():
										UpperCamelCase_			= offload_weight(_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase ,						index=_lowerCAmelCase)

					# Update index
					save_offload_index(_lowerCAmelCase ,						_lowerCAmelCase)


class    _lowercase (a_   ):
				'''simple docstring'''







				def __init__(			self  ,			snake_case__  ,			snake_case__     ):

									'''simple docstring'''


									UpperCamelCase_			= dataset
									UpperCamelCase_			= prefix

				def __getitem__(			self  ,			snake_case__     ):

									'''simple docstring'''


									return self.dataset[F"""{self.prefix}{key}"""]

				def __iter__(			self     ):

									'''simple docstring'''


									return iter([key for key in self.dataset if key.startswith(self.prefix     )]     )

				def __len__(			self     ):

									'''simple docstring'''


									return len(self.dataset     )




class    _lowercase (a_   ):
				'''simple docstring'''







				def __init__(			self  ,			snake_case__ = None  ,			snake_case__ = None  ,			snake_case__ = None  ,			snake_case__=None  ,			):

									'''simple docstring'''


									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_			= {} if state_dict is None else state_dict
									UpperCamelCase_			= save_folder
									if index is None and save_folder is not None:
														with open(os.path.join(snake_case__  ,			"index.json"     )     ) as f:
																			UpperCamelCase_			= json.load(snake_case__     )
									UpperCamelCase_			= {} if index is None else index
									UpperCamelCase_			= list(self.state_dict.keys()     )
									self.all_keys.extend([key for key in self.index if key not in self.all_keys]     )
									UpperCamelCase_			= device

				def __getitem__(			self  ,			snake_case__     ):

									'''simple docstring'''


									if key in self.state_dict:
														return self.state_dict[key]
									UpperCamelCase_			= 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_			= "cpu" if self.device is None else self.device
														with safe_open(weight_info["safetensors_file"]  ,			framework="pt"  ,			device=snake_case__     ) as f:
																			UpperCamelCase_			= f.get_tensor(weight_info.get("weight_name"  ,			snake_case__     )     )

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

									UpperCamelCase_			= os.path.join(self.save_folder  ,			F"""{key}.dat"""     )
									return load_offloaded_weight(snake_case__  ,			snake_case__     )

				def __iter__(			self     ):

									'''simple docstring'''


									return iter(self.all_keys     )

				def __len__(			self     ):

									'''simple docstring'''


									return len(self.all_keys     )




def  _lowerCAmelCase     (_lowerCAmelCase ,						_lowerCAmelCase):
					UpperCamelCase_			= {}
					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 os
import pathlib
import tempfile
import uuid

import numpy as np

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


__lowercase				    =      logging.get_logger(__name__)

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

if is_torch_available():
      import torch

if is_soundfile_availble():
      import soundfile as sf

class 	__snake_case					:






    '''simple docstring'''





    def __init__( self    :		Optional[int]      ,    snake_case    :		int				):

      """simple docstring"""





      __lowerCAmelCase								=					value





    def __str__( self    :		List[Any]				):

      """simple docstring"""





      return self.to_string()





    def 						A		( self    :		int				):

      """simple docstring"""





      logger.error(
          '''This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'''				)
      return self._value





    def 						A		( self    :		Optional[int]				):

      """simple docstring"""





      logger.error(
          '''This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'''				)
      return str(self._value				)


class 	__snake_case					(      _a							,					_a ):






    '''simple docstring'''





    def 						A		( self    :		int				):

      """simple docstring"""





      return self._value





    def 						A		( self    :		List[str]				):

      """simple docstring"""





      return self._value


class 	__snake_case					(      _a							,					_a ):






    '''simple docstring'''





    def __init__( self    :		int      ,    snake_case    :		Optional[int]				):

      """simple docstring"""





      super().__init__(snake_case				)

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

      __lowerCAmelCase								=					None
      __lowerCAmelCase								=					None
      __lowerCAmelCase								=					None

      if isinstance(snake_case      ,    snake_case				):
        __lowerCAmelCase								=					value
      elif isinstance(snake_case      ,    (str, pathlib.Path)				):
        __lowerCAmelCase								=					value
      elif isinstance(snake_case      ,    torch.Tensor				):
        __lowerCAmelCase								=					value
      else:
        raise ValueError(f'''Unsupported type for {self.__class__.__name__}: {type(snake_case				)}'''				)





    def 						A		( self    :		Tuple      ,    snake_case    :		Union[str, Any]=None      ,    snake_case    :		int=None				):

      """simple docstring"""





      from IPython.display import Image, display

      display(Image(self.to_string()				)				)





    def 						A		( self    :		Optional[Any]				):

      """simple docstring"""





      if self._raw is not None:
        return self._raw

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





    def 						A		( self    :		List[Any]				):

      """simple docstring"""





      if self._path is not None:
        return self._path

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

        return self._path

      if self._tensor is not None:
        __lowerCAmelCase								=					self._tensor.cpu().detach().numpy()

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

        __lowerCAmelCase								=					tempfile.mkdtemp()
        __lowerCAmelCase								=					os.path.join(snake_case      ,    str(uuid.uuida()				) + '''.png'''				)

        img.save(self._path				)

        return self._path


class 	__snake_case					(      _a ):






    '''simple docstring'''





    def __init__( self    :		List[Any]      ,    snake_case    :		int      ,    snake_case    :		Any=16_000				):

      """simple docstring"""





      super().__init__(snake_case				)

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

      __lowerCAmelCase								=					None
      __lowerCAmelCase								=					None

      __lowerCAmelCase								=					samplerate

      if isinstance(snake_case      ,    (str, pathlib.Path)				):
        __lowerCAmelCase								=					value
      elif isinstance(snake_case      ,    torch.Tensor				):
        __lowerCAmelCase								=					value
      else:
        raise ValueError(f'''Unsupported audio type: {type(snake_case				)}'''				)





    def 						A		( self    :		Optional[int]      ,    snake_case    :		Tuple=None      ,    snake_case    :		Tuple=None				):

      """simple docstring"""





      from IPython.display import Audio, display

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





    def 						A		( self    :		Optional[int]				):

      """simple docstring"""





      if self._tensor is not None:
        return self._tensor

      if self._path is not None:
        __lowerCAmelCase						,    __lowerCAmelCase								=					sf.read(self._path				)
        __lowerCAmelCase								=					torch.tensor(snake_case				)
        return self._tensor





    def 						A		( self    :		Optional[Any]				):

      """simple docstring"""





      if self._path is not None:
        return self._path

      if self._tensor is not None:
        __lowerCAmelCase								=					tempfile.mkdtemp()
        __lowerCAmelCase								=					os.path.join(snake_case      ,    str(uuid.uuida()				) + '''.wav'''				)
        sf.write(self._path      ,    self._tensor      ,    samplerate=self.samplerate				)
        return self._path


__lowercase				    =      {"""text""": AgentText, """image""": AgentImage, """audio""": AgentAudio}
__lowercase				    =      {str: AgentText}

if is_vision_available():
      __lowercase				    =      AgentImage


def        __SCREAMING_SNAKE_CASE	(				*__lowercase:		List[Any]  ,				**__lowercase:		Tuple):



  '''simple docstring'''






  __lowerCAmelCase								=					[(arg.to_raw() if isinstance(__lowercase  ,				__lowercase) else arg) for arg in args]
  __lowerCAmelCase								=					{k: (v.to_raw() if isinstance(__lowercase  ,				__lowercase) else v) for k, v in kwargs.items()}
  return args, kwargs




def        __SCREAMING_SNAKE_CASE	(				__lowercase:		List[Any]  ,				__lowercase:		List[Any]=None):



  '''simple docstring'''






  if isinstance(__lowercase  ,				__lowercase):
    __lowerCAmelCase								=					{}
    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:
          __lowerCAmelCase								=					AGENT_TYPE_MAPPING[output_types[i]](__lowercase)
        else:
          __lowerCAmelCase								=					AgentType(__lowercase)

      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(__lowercase  ,				_k):
            __lowerCAmelCase								=					_v(__lowercase)
        if k not in decoded_outputs:
          __lowerCAmelCase								=					AgentType[v]

  elif isinstance(__lowercase  ,				(list, tuple)):
    __lowerCAmelCase								=					type(__lowercase)()
    for i, v in enumerate(__lowercase):
      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]](__lowercase))
        else:
          decoded_outputs.append(AgentType(__lowercase))
      else:
        # If the class does not have defined output, then we map according to the type
        __lowerCAmelCase								=					False
        for _k, _v in INSTANCE_TYPE_MAPPING.items():
          if isinstance(__lowercase  ,				_k):
            decoded_outputs.append(_v(__lowercase))
            __lowerCAmelCase								=					True

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

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

    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(__lowercase  ,				_k):
          return _v(__lowercase)
      return AgentType(__lowercase)

  return decoded_outputs
