





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


SCREAMING_SNAKE_CASE_      :			Dict		  =      logging.get_logger(__name__)

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

if is_torch_available():
		import torch

if is_soundfile_availble():
		import soundfile as sf

class 		snake_case_      :







	'''simple docstring'''




	def __init__(	self       :    Union[str, Any]				,			__lowerCamelCase       :    Tuple      )					->   List[str]:

			'''simple docstring'''




			__lowercase					=    value




	def __str__(	self       :    Any      )					->   List[str]:

			'''simple docstring'''




			return self.to_string()




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

			'''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       UpperCAmelCase (	self       :    List[str]      )					->   str:

			'''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_      (       UpperCAmelCase_    ,    UpperCAmelCase_	):







	'''simple docstring'''




	def       UpperCAmelCase (	self       :    Any      )					->   str:

			'''simple docstring'''




			return self._value






	def       UpperCAmelCase (	self       :    List[str]      )					->   Dict:

			'''simple docstring'''




			return self._value





class 		snake_case_      (       UpperCAmelCase_    ,    UpperCAmelCase_	):







	'''simple docstring'''




	def __init__(	self       :    Any				,			__lowerCamelCase       :    Tuple      )					->   List[Any]:

			'''simple docstring'''




			super().__init__(__lowerCamelCase      )

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

			__lowercase					=    None
			__lowercase					=    None
			__lowercase					=    None

			if isinstance(__lowerCamelCase				,			__lowerCamelCase      ):
					__lowercase					=    value
			elif isinstance(__lowerCamelCase				,			(str, pathlib.Path)      ):
					__lowercase					=    value
			elif isinstance(__lowerCamelCase				,			torch.Tensor      ):
					__lowercase					=    value
			else:
					raise ValueError(F"Unsupported type for {self.__class__.__name__}: {type(__lowerCamelCase      )}"      )




	def       UpperCAmelCase (	self       :    Any				,			__lowerCamelCase       :    Optional[int]=None				,			__lowerCamelCase       :    str=None      )					->   List[str]:

			'''simple docstring'''




			from IPython.display import Image, display

			display(Image(self.to_string()      )      )




	def       UpperCAmelCase (	self       :    List[str]      )					->   Tuple:

			'''simple docstring'''




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

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






	def       UpperCAmelCase (	self       :    str      )					->   Tuple:

			'''simple docstring'''




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

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

					return self._path

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

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

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

					img.save(self._path      )

					return self._path





class 		snake_case_      (       UpperCAmelCase_	):







	'''simple docstring'''




	def __init__(	self       :    Optional[int]				,			__lowerCamelCase       :    Dict				,			__lowerCamelCase       :    Union[str, Any]=16_000      )					->   Tuple:

			'''simple docstring'''




			super().__init__(__lowerCamelCase      )

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

			__lowercase					=    None
			__lowercase					=    None

			__lowercase					=    samplerate

			if isinstance(__lowerCamelCase				,			(str, pathlib.Path)      ):
					__lowercase					=    value
			elif isinstance(__lowerCamelCase				,			torch.Tensor      ):
					__lowercase					=    value
			else:
					raise ValueError(F"Unsupported audio type: {type(__lowerCamelCase      )}"      )




	def       UpperCAmelCase (	self       :    Union[str, Any]				,			__lowerCamelCase       :    Optional[Any]=None				,			__lowerCamelCase       :    Any=None      )					->   int:

			'''simple docstring'''




			from IPython.display import Audio, display

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




	def       UpperCAmelCase (	self       :    str      )					->   List[Any]:

			'''simple docstring'''




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

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






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

			'''simple docstring'''




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

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





SCREAMING_SNAKE_CASE_      :			Optional[Any]		  =      {'''text''': AgentText, '''image''': AgentImage, '''audio''': AgentAudio}
SCREAMING_SNAKE_CASE_      :			List[Any]		  =      {str: AgentText}

if is_vision_available():
		SCREAMING_SNAKE_CASE_      :			Any		  =      AgentImage



def    SCREAMING_SNAKE_CASE							(			*snake_case	,      **snake_case       )		->						int:
		__lowercase					=    [(arg.to_raw() if isinstance(snake_case	,      snake_case       ) else arg) for arg in args]
		__lowercase					=    {k: (v.to_raw() if isinstance(snake_case	,      snake_case       ) else v) for k, v in kwargs.items()}
		return args, kwargs



def    SCREAMING_SNAKE_CASE							(			snake_case	,      snake_case=None       )		->						Union[str, Any]:
		if isinstance(snake_case	,      snake_case       ):
				__lowercase					=    {}
				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					=    AGENT_TYPE_MAPPING[output_types[i]](snake_case       )
								else:
										__lowercase					=    AgentType(snake_case       )

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

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

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

		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					=    AGENT_TYPE_MAPPING[output_types[0]](snake_case       )

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

		return decoded_outputs











import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, slow, torch_device

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin

class 		snake_case_      (       UpperCAmelCase_    ,    UpperCAmelCase_    ,    unittest.TestCase	):







	'''simple docstring'''
	__UpperCamelCase        =		UNetaDModel
	__UpperCamelCase        =		'''sample'''




	@property
	def       UpperCAmelCase (	self       :    str      )					->   List[str]:

			'''simple docstring'''




			__lowercase					=    4
			__lowercase					=    14
			__lowercase					=    16

			__lowercase					=    floats_tensor((batch_size, num_features, seq_len)      ).to(__lowerCamelCase      )
			__lowercase					=    torch.tensor([10] * batch_size      ).to(__lowerCamelCase      )

			return {"sample": noise, "timestep": time_step}




	@property
	def       UpperCAmelCase (	self       :    List[str]      )					->   Union[str, Any]:

			'''simple docstring'''




			return (4, 14, 16)




	@property
	def       UpperCAmelCase (	self       :    Dict      )					->   List[Any]:

			'''simple docstring'''




			return (4, 14, 16)




	def       UpperCAmelCase (	self       :    Optional[int]      )					->   Dict:

			'''simple docstring'''




			pass




	def       UpperCAmelCase (	self       :    Union[str, Any]      )					->   Dict:

			'''simple docstring'''




			pass




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    List[Any]      )					->   Optional[Any]:

			'''simple docstring'''




			super().test_determinism()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    List[str]      )					->   Dict:

			'''simple docstring'''




			super().test_outputs_equivalence()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Optional[Any]      )					->   List[Any]:

			'''simple docstring'''




			super().test_from_save_pretrained()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Union[str, Any]      )					->   Any:

			'''simple docstring'''




			super().test_from_save_pretrained_variant()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Optional[Any]      )					->   int:

			'''simple docstring'''




			super().test_model_from_pretrained()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Optional[int]      )					->   Tuple:

			'''simple docstring'''




			super().test_output()




	def       UpperCAmelCase (	self       :    List[Any]      )					->   List[str]:

			'''simple docstring'''




			__lowercase					=    {
			    'block_out_channels': (32, 64, 128, 256),
			    'in_channels': 14,
			    'out_channels': 14,
			    'time_embedding_type': 'positional',
			    'use_timestep_embedding': True,
			    'flip_sin_to_cos': False,
			    'freq_shift': 1.0,
			    'out_block_type': 'OutConv1DBlock',
			    'mid_block_type': 'MidResTemporalBlock1D',
			    'down_block_types': ('DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D'),
			    'up_block_types': ('UpResnetBlock1D', 'UpResnetBlock1D', 'UpResnetBlock1D'),
			    'act_fn': 'mish',
			}
			__lowercase					=    self.dummy_input
			return init_dict, inputs_dict




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Optional[int]      )					->   Any:

			'''simple docstring'''




			__lowercase	,			__lowercase					=    UNetaDModel.from_pretrained(
			    'bglick13/hopper-medium-v2-value-function-hor32'				,			output_loading_info=__lowerCamelCase				,			subfolder='unet'      )
			self.assertIsNotNone(__lowerCamelCase      )
			self.assertEqual(len(loading_info['missing_keys']      )				,			0      )

			model.to(__lowerCamelCase      )
			__lowercase					=    model(**self.dummy_input      )

			assert image is not None, "Make sure output is not None"




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    List[str]      )					->   Dict:

			'''simple docstring'''




			__lowercase					=    UNetaDModel.from_pretrained('bglick13/hopper-medium-v2-value-function-hor32'				,			subfolder='unet'      )
			torch.manual_seed(0      )
			if torch.cuda.is_available():
					torch.cuda.manual_seed_all(0      )

			__lowercase					=    model.config.in_channels
			__lowercase					=    16
			__lowercase					=    torch.randn((1, seq_len, num_features)      ).permute(
			    0				,			2				,			1      )  # match original, we can update values and remove
			__lowercase					=    torch.full((num_features,)				,			0      )

			with torch.no_grad():
					__lowercase					=    model(__lowerCamelCase				,			__lowerCamelCase      ).sample.permute(0				,			2				,			1      )

			__lowercase					=    output[0, -3:, -3:].flatten()
			# fmt: off
			__lowercase					=    torch.tensor([-2.137172, 1.1426016, 0.3688687, -0.766922, 0.7303146, 0.11038864, -0.4760633, 0.13270172, 0.02591348]      )
			# fmt: on
			self.assertTrue(torch.allclose(__lowerCamelCase				,			__lowerCamelCase				,			rtol=1E-3      )      )




	def       UpperCAmelCase (	self       :    List[Any]      )					->   int:

			'''simple docstring'''




			pass






	@slow
	def       UpperCAmelCase (	self       :    List[str]      )					->   Dict:

			'''simple docstring'''




			__lowercase					=    'harmonai/maestro-150k'
			__lowercase					=    UNetaDModel.from_pretrained(__lowerCamelCase				,			subfolder='unet'      )
			model.to(__lowerCamelCase      )

			__lowercase					=    65_536
			__lowercase					=    torch.sin(torch.arange(__lowerCamelCase      )[None, None, :].repeat(1				,			2				,			1      )      ).to(__lowerCamelCase      )
			__lowercase					=    torch.tensor([1]      ).to(__lowerCamelCase      )

			with torch.no_grad():
					__lowercase					=    model(__lowerCamelCase				,			__lowerCamelCase      ).sample

			__lowercase					=    output.abs().sum()
			__lowercase					=    output.abs().max()

			assert (output_sum - 224.0896).abs() < 0.5
			assert (output_max - 0.0607).abs() < 4E-4





class 		snake_case_      (       UpperCAmelCase_    ,    UpperCAmelCase_    ,    unittest.TestCase	):







	'''simple docstring'''
	__UpperCamelCase        =		UNetaDModel
	__UpperCamelCase        =		'''sample'''




	@property
	def       UpperCAmelCase (	self       :    int      )					->   Union[str, Any]:

			'''simple docstring'''




			__lowercase					=    4
			__lowercase					=    14
			__lowercase					=    16

			__lowercase					=    floats_tensor((batch_size, num_features, seq_len)      ).to(__lowerCamelCase      )
			__lowercase					=    torch.tensor([10] * batch_size      ).to(__lowerCamelCase      )

			return {"sample": noise, "timestep": time_step}




	@property
	def       UpperCAmelCase (	self       :    List[str]      )					->   Union[str, Any]:

			'''simple docstring'''




			return (4, 14, 16)




	@property
	def       UpperCAmelCase (	self       :    Tuple      )					->   Union[str, Any]:

			'''simple docstring'''




			return (4, 14, 1)




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Dict      )					->   str:

			'''simple docstring'''




			super().test_determinism()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    str      )					->   int:

			'''simple docstring'''




			super().test_outputs_equivalence()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    str      )					->   List[Any]:

			'''simple docstring'''




			super().test_from_save_pretrained()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    List[Any]      )					->   int:

			'''simple docstring'''




			super().test_from_save_pretrained_variant()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    List[str]      )					->   Dict:

			'''simple docstring'''




			super().test_model_from_pretrained()




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    List[str]      )					->   Tuple:

			'''simple docstring'''




			__lowercase	,			__lowercase					=    self.prepare_init_args_and_inputs_for_common()
			__lowercase					=    self.model_class(**__lowerCamelCase      )
			model.to(__lowerCamelCase      )
			model.eval()

			with torch.no_grad():
					__lowercase					=    model(**__lowerCamelCase      )

					if isinstance(__lowerCamelCase				,			__lowerCamelCase      ):
							__lowercase					=    output.sample

			self.assertIsNotNone(__lowerCamelCase      )
			__lowercase					=    torch.Size((inputs_dict['sample'].shape[0], 1)      )
			self.assertEqual(output.shape				,			__lowerCamelCase				,			'Input and output shapes do not match'      )




	def       UpperCAmelCase (	self       :    Optional[int]      )					->   str:

			'''simple docstring'''




			pass




	def       UpperCAmelCase (	self       :    Optional[int]      )					->   Optional[int]:

			'''simple docstring'''




			pass




	def       UpperCAmelCase (	self       :    Optional[Any]      )					->   Optional[Any]:

			'''simple docstring'''




			__lowercase					=    {
			    'in_channels': 14,
			    'out_channels': 14,
			    'down_block_types': ['DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D'],
			    'up_block_types': [],
			    'out_block_type': 'ValueFunction',
			    'mid_block_type': 'ValueFunctionMidBlock1D',
			    'block_out_channels': [32, 64, 128, 256],
			    'layers_per_block': 1,
			    'downsample_each_block': True,
			    'use_timestep_embedding': True,
			    'freq_shift': 1.0,
			    'flip_sin_to_cos': False,
			    'time_embedding_type': 'positional',
			    'act_fn': 'mish',
			}
			__lowercase					=    self.dummy_input
			return init_dict, inputs_dict




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Any      )					->   Union[str, Any]:

			'''simple docstring'''




			__lowercase	,			__lowercase					=    UNetaDModel.from_pretrained(
			    'bglick13/hopper-medium-v2-value-function-hor32'				,			output_loading_info=__lowerCamelCase				,			subfolder='value_function'      )
			self.assertIsNotNone(__lowerCamelCase      )
			self.assertEqual(len(vf_loading_info['missing_keys']      )				,			0      )

			value_function.to(__lowerCamelCase      )
			__lowercase					=    value_function(**self.dummy_input      )

			assert image is not None, "Make sure output is not None"




	@unittest.skipIf(torch_device == 'mps'				,			'mish op not supported in MPS'      )
	def       UpperCAmelCase (	self       :    Union[str, Any]      )					->   Optional[Any]:

			'''simple docstring'''




			__lowercase	,			__lowercase					=    UNetaDModel.from_pretrained(
			    'bglick13/hopper-medium-v2-value-function-hor32'				,			output_loading_info=__lowerCamelCase				,			subfolder='value_function'      )
			torch.manual_seed(0      )
			if torch.cuda.is_available():
					torch.cuda.manual_seed_all(0      )

			__lowercase					=    value_function.config.in_channels
			__lowercase					=    14
			__lowercase					=    torch.randn((1, seq_len, num_features)      ).permute(
			    0				,			2				,			1      )  # match original, we can update values and remove
			__lowercase					=    torch.full((num_features,)				,			0      )

			with torch.no_grad():
					__lowercase					=    value_function(__lowerCamelCase				,			__lowerCamelCase      ).sample

			# fmt: off
			__lowercase					=    torch.tensor([165.25] * seq_len      )
			# fmt: on
			self.assertTrue(torch.allclose(__lowerCamelCase				,			__lowerCamelCase				,			rtol=1E-3      )      )






	def       UpperCAmelCase (	self       :    Dict      )					->   str:

			'''simple docstring'''




			pass




