




"""simple docstring"""

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      A__   (   lowercase_  ,						lowercase_  ,						unittest.TestCase  ):
	A_				:					List[str]      =   UNetaDModel
	A_				:					List[Any]      =   """sample"""




	@property
	def 				a_		( self	:						List[Any]						)       ->			Any:
						__snake_case        =						4
						__snake_case        =						14
						__snake_case        =						16

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

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




	@property
	def 				a_		( self	:						Dict						)       ->			Optional[int]:
						return (4, 14, 16)




	@property
	def 				a_		( self	:						Optional[int]						)       ->			int:
						return (4, 14, 16)




	def 				a_		( self	:						Optional[Any]						)       ->			Optional[int]:
						pass




	def 				a_		( self	:						Optional[Any]						)       ->			str:
						pass




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




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Optional[Any]						)       ->			List[str]:
						super().test_outputs_equivalence()




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Union[str, Any]						)       ->			List[Any]:
						super().test_from_save_pretrained()




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




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						List[str]						)       ->			str:
						super().test_model_from_pretrained()




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




	def 				a_		( self	:						Optional[Any]						)       ->			List[Any]:
						__snake_case        =						{
						    """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""",
						}
						__snake_case        =						self.dummy_input
						return init_dict, inputs_dict




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						List[Any]						)       ->			List[str]:
						__snake_case   ,   __snake_case        =						UNetaDModel.from_pretrained(
						    """bglick13/hopper-medium-v2-value-function-hor32"""   ,     output_loading_info=lowercase   ,     subfolder="""unet"""						)
						self.assertIsNotNone(lowercase						)
						self.assertEqual(len(loading_info["""missing_keys"""]						)   ,     0						)

						model.to(lowercase						)
						__snake_case        =						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 				a_		( self	:						Dict						)       ->			Union[str, Any]:
						__snake_case        =						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						)

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

						with torch.no_grad():
											__snake_case        =						model(lowercase   ,     lowercase						).sample.permute(0   ,     2   ,     1						)

						__snake_case        =						output[0, -3:, -3:].flatten()
						# fmt: off
						__snake_case        =						torch.tensor([-2.13_71_72, 1.1_42_60_16, 0.3_68_86_87, -0.76_69_22, 0.7_30_31_46, 0.11_03_88_64, -0.4_76_06_33, 0.13_27_01_72, 0.02_59_13_48]						)
						# fmt: on
						self.assertTrue(torch.allclose(lowercase   ,     lowercase   ,     rtol=1e-3						)						)




	def 				a_		( self	:						Dict						)       ->			int:
						# Not implemented yet for this UNet
						pass




	@slow
	def 				a_		( self	:						str						)       ->			Union[str, Any]:
						__snake_case        =						"""harmonai/maestro-150k"""
						__snake_case        =						UNetaDModel.from_pretrained(lowercase   ,     subfolder="""unet"""						)
						model.to(lowercase						)

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

						with torch.no_grad():
											__snake_case        =						model(lowercase   ,     lowercase						).sample

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

						assert (output_sum - 2_24.08_96).abs() < 0.5
						assert (output_max - 0.06_07).abs() < 4e-4

class      A__   (   lowercase_  ,						lowercase_  ,						unittest.TestCase  ):
	A_				:					Any      =   UNetaDModel
	A_				:					Tuple      =   """sample"""




	@property
	def 				a_		( self	:						Optional[int]						)       ->			Optional[Any]:
						__snake_case        =						4
						__snake_case        =						14
						__snake_case        =						16

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

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




	@property
	def 				a_		( self	:						Optional[int]						)       ->			Dict:
						return (4, 14, 16)




	@property
	def 				a_		( self	:						Any						)       ->			str:
						return (4, 14, 1)




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						str						)       ->			int:
						super().test_determinism()




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Union[str, Any]						)       ->			List[str]:
						super().test_outputs_equivalence()




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Optional[int]						)       ->			Dict:
						super().test_from_save_pretrained()




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Dict						)       ->			Optional[Any]:
						super().test_from_save_pretrained_variant()




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Tuple						)       ->			Optional[Any]:
						super().test_model_from_pretrained()




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Union[str, Any]						)       ->			int:
						# UNetRL is a value-function is different output shape
						__snake_case   ,   __snake_case        =						self.prepare_init_args_and_inputs_for_common()
						__snake_case        =						self.model_class(**lowercase						)
						model.to(lowercase						)
						model.eval()

						with torch.no_grad():
											__snake_case        =						model(**lowercase						)

											if isinstance(lowercase   ,     lowercase						):
																__snake_case        =						output.sample

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




	def 				a_		( self	:						Optional[Any]						)       ->			Any:
						pass




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




	def 				a_		( self	:						int						)       ->			Any:
						__snake_case        =						{
						    """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""",
						}
						__snake_case        =						self.dummy_input
						return init_dict, inputs_dict




	@unittest.skipIf(torch_device == """mps"""   ,     """mish op not supported in MPS"""						)
	def 				a_		( self	:						Dict						)       ->			List[Any]:
						__snake_case   ,   __snake_case        =						UNetaDModel.from_pretrained(
						    """bglick13/hopper-medium-v2-value-function-hor32"""   ,     output_loading_info=lowercase   ,     subfolder="""value_function"""						)
						self.assertIsNotNone(lowercase						)
						self.assertEqual(len(vf_loading_info["""missing_keys"""]						)   ,     0						)

						value_function.to(lowercase						)
						__snake_case        =						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 				a_		( self	:						Optional[int]						)       ->			List[Any]:
						__snake_case   ,   __snake_case        =						UNetaDModel.from_pretrained(
						    """bglick13/hopper-medium-v2-value-function-hor32"""   ,     output_loading_info=lowercase   ,     subfolder="""value_function"""						)
						torch.manual_seed(0						)
						if torch.cuda.is_available():
											torch.cuda.manual_seed_all(0						)

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

						with torch.no_grad():
											__snake_case        =						value_function(lowercase   ,     lowercase						).sample

						# fmt: off
						__snake_case        =						torch.tensor([1_65.25] * seq_len						)
						# fmt: on
						self.assertTrue(torch.allclose(lowercase   ,     lowercase   ,     rtol=1e-3						)						)




	def 				a_		( self	:						Union[str, Any]						)       ->			int:
						# Not implemented yet for this UNet
						pass








"""simple docstring"""

import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig
from ..clip.configuration_clip import CLIPVisionConfig


UpperCamelCase__			:     Optional[Any]			       =				logging.get_logger(__name__)
class      A__   (   lowercase_  ):
	A_				:					List[str]      =   """vision-text-dual-encoder"""
	A_				:					Tuple      =   True




	def __init__( self	:						Tuple   ,     lowercase	:						List[Any]=512   ,     lowercase	:						Any=2.65_92   ,     **lowercase	:						int						)       ->			int:
						super().__init__(**lowercase						)

						if "vision_config" not in kwargs:
											raise ValueError("""`vision_config` can not be `None`."""						)

						if "text_config" not in kwargs:
											raise ValueError("""`text_config` can not be `None`."""						)

						__snake_case        =						kwargs.pop("""vision_config"""						)
						__snake_case        =						kwargs.pop("""text_config"""						)

						__snake_case        =						vision_config.pop("""model_type"""						)
						__snake_case        =						text_config.pop("""model_type"""						)

						if vision_model_type == "clip":
											__snake_case        =						AutoConfig.for_model(lowercase   ,     **lowercase						).vision_config
						elif vision_model_type == "clip_vision_model":
											__snake_case        =						CLIPVisionConfig(**lowercase						)
						else:
											__snake_case        =						AutoConfig.for_model(lowercase   ,     **lowercase						)

						__snake_case        =						AutoConfig.for_model(lowercase   ,     **lowercase						)

						__snake_case        =						projection_dim
						__snake_case        =						logit_scale_init_value




	@classmethod
	def 				a_		( cls	:						Union[str, Any]   ,     lowercase	:						PretrainedConfig   ,     lowercase	:						PretrainedConfig   ,     **lowercase	:						Optional[int]						)       ->			Dict:
						return cls(vision_config=vision_config.to_dict()   ,     text_config=text_config.to_dict()   ,     **lowercase						)




	def 				a_		( self	:						Dict						)       ->			Dict:
						__snake_case        =						copy.deepcopy(self.__dict__						)
						__snake_case        =						self.vision_config.to_dict()
						__snake_case        =						self.text_config.to_dict()
						__snake_case        =						self.__class__.model_type
						return output


