



"""simple docstring"""




import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

from tqdm import auto as tqdm_lib


_UpperCamelCase:				Union[str, Any]    							=  threading.Lock()
_UpperCamelCase:				Optional[logging.Handler]    							=  None

_UpperCamelCase:				Optional[int]    							=  {
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

_UpperCamelCase:				Optional[int]    							=  logging.WARNING

_UpperCamelCase:				Union[str, Any]    							=  True


def   lowercase__     (			)     ->			Optional[Any]:






      '''simple docstring'''


      lowercase  :      Tuple					=   os.getenv('DIFFUSERS_VERBOSITY'	,      _UpperCAmelCase				)
      if env_level_str:
            if env_level_str in log_levels:
                  return log_levels[env_level_str]
            else:
                  logging.getLogger().warning(
                      f'''Unknown option DIFFUSERS_VERBOSITY={env_level_str}, '''
                      f'''has to be one of: { ', '.join(log_levels.keys()				) }'''				)
      return _default_log_level


def   lowercase__     (			)     ->			str:






      '''simple docstring'''


      return __name__.split('.'				)[0]


def   lowercase__     (			)     ->			logging.Logger:






      '''simple docstring'''


      return logging.getLogger(_get_library_name()				)


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      global _default_handler

      with _lock:
            if _default_handler:
                  # This library has already configured the library root logger.
                  return
            lowercase  :      Dict					=   logging.StreamHandler()  # Set sys.stderr as stream.
            lowercase  :      str					=   sys.stderr.flush

            # Apply our default configuration to the library root logger.
            lowercase  :      Tuple					=   _get_library_root_logger()
            library_root_logger.addHandler(_default_handler				)
            library_root_logger.setLevel(_get_default_logging_level()				)
            lowercase  :      List[Any]					=   False


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      global _default_handler

      with _lock:
            if not _default_handler:
                  return

            lowercase  :      Tuple					=   _get_library_root_logger()
            library_root_logger.removeHandler(_default_handler				)
            library_root_logger.setLevel(logging.NOTSET				)
            lowercase  :      Union[str, Any]					=   None


def   lowercase__     (			)     ->			Tuple:






      '''simple docstring'''


      return log_levels


def   lowercase__     (			_UpperCAmelCase = None				)     ->			logging.Logger:






      '''simple docstring'''


      if name is None:
            lowercase  :      int					=   _get_library_name()

      _configure_library_root_logger()
      return logging.getLogger(_UpperCAmelCase				)


def   lowercase__     (			)     ->			int:






      '''simple docstring'''


      _configure_library_root_logger()
      return _get_library_root_logger().getEffectiveLevel()


def   lowercase__     (			_UpperCAmelCase				)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()
      _get_library_root_logger().setLevel(_UpperCAmelCase				)


def   lowercase__     (			)     ->			List[Any]:






      '''simple docstring'''


      return set_verbosity(_UpperCAmelCase				)


def   lowercase__     (			)     ->			List[Any]:






      '''simple docstring'''


      return set_verbosity(_UpperCAmelCase				)


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






      '''simple docstring'''


      return set_verbosity(_UpperCAmelCase				)


def   lowercase__     (			)     ->			List[Any]:






      '''simple docstring'''


      return set_verbosity(_UpperCAmelCase				)


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()

      assert _default_handler is not None
      _get_library_root_logger().removeHandler(_default_handler				)


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()

      assert _default_handler is not None
      _get_library_root_logger().addHandler(_default_handler				)


def   lowercase__     (			_UpperCAmelCase				)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()

      assert handler is not None
      _get_library_root_logger().addHandler(_UpperCAmelCase				)


def   lowercase__     (			_UpperCAmelCase				)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()

      assert handler is not None and handler not in _get_library_root_logger().handlers
      _get_library_root_logger().removeHandler(_UpperCAmelCase				)


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()
      lowercase  :      Optional[int]					=   False


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      _configure_library_root_logger()
      lowercase  :      str					=   True


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      lowercase  :      Tuple					=   _get_library_root_logger().handlers

      for handler in handlers:
            lowercase  :      Tuple					=   logging.Formatter('[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s'				)
            handler.setFormatter(_UpperCAmelCase				)


def   lowercase__     (			)     ->			None:






      '''simple docstring'''


      lowercase  :      Optional[int]					=   _get_library_root_logger().handlers

      for handler in handlers:
            handler.setFormatter(_UpperCAmelCase				)


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






      '''simple docstring'''


      lowercase  :      Optional[Any]					=   os.getenv('DIFFUSERS_NO_ADVISORY_WARNINGS'	,      _UpperCAmelCase				)
      if no_advisory_warnings:
            return
      self.warning(*_UpperCAmelCase	,      **_UpperCAmelCase				)


_UpperCamelCase:				Union[str, Any]    							=  warning_advice


class   a__		:


   def __init__(							self      :							Optional[Any],     *lowerCAmelCase      :							Union[str, Any],     **lowerCAmelCase      :							Any			)    ->       Union[str, Any]:  # pylint: disable=unused-argument
         lowercase  :      List[str]					=   args[0] if args else None


   def __iter__(							self      :							Optional[int]			)    ->       int:
         return iter(self._iterator			)


   def __getattr__(							self      :							Optional[int],     lowerCAmelCase      :							Union[str, Any]			)    ->       Optional[int]:

         def empty_fn(*lowerCAmelCase      :							List[str],     **lowerCAmelCase      :							int			):  # pylint: disable=unused-argument
               return

         return empty_fn


   def __enter__(							self      :							Dict			)    ->       List[Any]:
         return self






   def __exit__(							self      :							Any,     lowerCAmelCase      :							Optional[Any],     lowerCAmelCase      :							Optional[int],     lowerCAmelCase      :							str			)    ->       int:
         return




class   a__		:


   def __call__(							self      :							Any,     *lowerCAmelCase      :							Any,     **lowerCAmelCase      :							Tuple			)    ->       Dict:
         if _tqdm_active:
               return tqdm_lib.tqdm(*lowerCAmelCase,     **lowerCAmelCase			)
         else:
               return EmptyTqdm(*lowerCAmelCase,     **lowerCAmelCase			)


   def        lowercase				(							self      :							Optional[Any],     *lowerCAmelCase      :							Any,     **lowerCAmelCase      :							List[Any]			)    ->       Optional[Any]:
         lowercase  :      Optional[int]					=   None
         if _tqdm_active:
               return tqdm_lib.tqdm.set_lock(*lowerCAmelCase,     **lowerCAmelCase			)






   def        lowercase				(							self      :							List[Any]			)    ->       List[str]:
         if _tqdm_active:
               return tqdm_lib.tqdm.get_lock()




_UpperCamelCase:				Dict    							=  _tqdm_cls()


def   lowercase__     (			)     ->			bool:






      '''simple docstring'''


      global _tqdm_active
      return bool(_tqdm_active				)


def   lowercase__     (			)     ->			List[str]:






      '''simple docstring'''


      global _tqdm_active
      lowercase  :      Optional[int]					=   True





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






      '''simple docstring'''


      global _tqdm_active
      lowercase  :      int					=   False





"""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
