
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 		__magic_name__   (						lowerCAmelCase_,				lowerCAmelCase_,				unittest.TestCase    ):
  _lowerCAmelCase		=	UNetaDModel
  _lowerCAmelCase		=	"sample"





  @property
  def       _A							(							self    :	Optional[Any]				):
       lowerCAmelCase		:					List[str]  	=     4
       lowerCAmelCase		:					str  	=     1_4
       lowerCAmelCase		:					int  	=     1_6

       lowerCAmelCase		:					List[str]  	=     floats_tensor((batch_size, num_features, seq_len)				).to(__a				)
       lowerCAmelCase		:					Optional[Any]  	=     torch.tensor([1_0] * batch_size				).to(__a				)

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





  @property
  def       _A							(							self    :	Dict				):
       return (4, 1_4, 1_6)





  @property
  def       _A							(							self    :	Tuple				):
       return (4, 1_4, 1_6)





  def       _A							(							self    :	int				):
       pass





  def       _A							(							self    :	Optional[Any]				):
       pass





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	Tuple				):
       super().test_determinism()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	List[str]				):
       super().test_outputs_equivalence()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	str				):
       super().test_from_save_pretrained()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	int				):
       super().test_from_save_pretrained_variant()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	Optional[int]				):
       super().test_model_from_pretrained()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	Tuple				):
       super().test_output()





  def       _A							(							self    :	int				):
       lowerCAmelCase		:					Union[str, Any]  	=     {
           'block_out_channels': (3_2, 6_4, 1_2_8, 2_5_6),
           'in_channels': 1_4,
           'out_channels': 1_4,
           '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',
       }
       lowerCAmelCase		:					str  	=     self.dummy_input
       return init_dict, inputs_dict





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

       model.to(__a				)
       lowerCAmelCase		:					Optional[int]  	=     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    :	str				):
       lowerCAmelCase		:					List[Any]  	=     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				)

       lowerCAmelCase		:					str  	=     model.config.in_channels
       lowerCAmelCase		:					int  	=     1_6
       lowerCAmelCase		:					List[str]  	=     torch.randn((1, seq_len, num_features)				).permute(
           0						,  2						,  1				)  # match original, we can update values and remove
       lowerCAmelCase		:					List[Any]  	=     torch.full((num_features,)						,  0				)

       with torch.no_grad():
            lowerCAmelCase		:					Tuple  	=     model(__a						,  __a				).sample.permute(0						,  2						,  1				)

       lowerCAmelCase		:					int  	=     output[0, -3:, -3:].flatten()
       # fmt: off
       lowerCAmelCase		:					List[str]  	=     torch.tensor([-2.1_3_7_1_7_2, 1.1_4_2_6_0_1_6, 0.3_6_8_8_6_8_7, -0.7_6_6_9_2_2, 0.7_3_0_3_1_4_6, 0.1_1_0_3_8_8_6_4, -0.4_7_6_0_6_3_3, 0.1_3_2_7_0_1_7_2, 0.0_2_5_9_1_3_4_8]				)
       # fmt: on
       self.assertTrue(torch.allclose(__a						,  __a						,  rtol=1E-3				)				)





  def       _A							(							self    :	Dict				):
       pass





  @slow
  def       _A							(							self    :	Dict				):
       lowerCAmelCase		:					Tuple  	=     'harmonai/maestro-150k'
       lowerCAmelCase		:					Any  	=     UNetaDModel.from_pretrained(__a						,  subfolder='''unet'''				)
       model.to(__a				)

       lowerCAmelCase		:					Optional[Any]  	=     6_5_5_3_6
       lowerCAmelCase		:					str  	=     torch.sin(torch.arange(__a				)[None, None, :].repeat(1						,  2						,  1				)				).to(__a				)
       lowerCAmelCase		:					int  	=     torch.tensor([1]				).to(__a				)

       with torch.no_grad():
            lowerCAmelCase		:					Optional[int]  	=     model(__a						,  __a				).sample

       lowerCAmelCase		:					Union[str, Any]  	=     output.abs().sum()
       lowerCAmelCase		:					Dict  	=     output.abs().max()

       assert (output_sum - 2_2_4.0_8_9_6).abs() < 0.5
       assert (output_max - 0.0_6_0_7).abs() < 4E-4



class 		__magic_name__   (						lowerCAmelCase_,				lowerCAmelCase_,				unittest.TestCase    ):
  _lowerCAmelCase		=	UNetaDModel
  _lowerCAmelCase		=	"sample"





  @property
  def       _A							(							self    :	Optional[Any]				):
       lowerCAmelCase		:					str  	=     4
       lowerCAmelCase		:					str  	=     1_4
       lowerCAmelCase		:					Optional[int]  	=     1_6

       lowerCAmelCase		:					str  	=     floats_tensor((batch_size, num_features, seq_len)				).to(__a				)
       lowerCAmelCase		:					Optional[Any]  	=     torch.tensor([1_0] * batch_size				).to(__a				)

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





  @property
  def       _A							(							self    :	Dict				):
       return (4, 1_4, 1_6)





  @property
  def       _A							(							self    :	Union[str, Any]				):
       return (4, 1_4, 1)





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	str				):
       super().test_determinism()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	Any				):
       super().test_outputs_equivalence()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	List[Any]				):
       super().test_from_save_pretrained()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	str				):
       super().test_from_save_pretrained_variant()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	Optional[int]				):
       super().test_model_from_pretrained()





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	List[Any]				):
       lowerCAmelCase		:					List[Any]  	=     self.prepare_init_args_and_inputs_for_common()
       lowerCAmelCase		:					int  	=     self.model_class(**__a				)
       model.to(__a				)
       model.eval()

       with torch.no_grad():
            lowerCAmelCase		:					Optional[Any]  	=     model(**__a				)

            if isinstance(__a						,  __a				):
                 lowerCAmelCase		:					Any  	=     output.sample

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





  def       _A							(							self    :	Tuple				):
       pass





  def       _A							(							self    :	Dict				):
       pass





  def       _A							(							self    :	Dict				):
       lowerCAmelCase		:					int  	=     {
           'in_channels': 1_4,
           'out_channels': 1_4,
           'down_block_types': ['DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D'],
           'up_block_types': [],
           'out_block_type': 'ValueFunction',
           'mid_block_type': 'ValueFunctionMidBlock1D',
           'block_out_channels': [3_2, 6_4, 1_2_8, 2_5_6],
           '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',
       }
       lowerCAmelCase		:					Union[str, Any]  	=     self.dummy_input
       return init_dict, inputs_dict





  @unittest.skipIf(torch_device == '''mps'''						,  '''mish op not supported in MPS'''				)
  def       _A							(							self    :	Any				):
       lowerCAmelCase		:					List[str]  	=     UNetaDModel.from_pretrained(
           '''bglick13/hopper-medium-v2-value-function-hor32'''						,  output_loading_info=__a						,  subfolder='''value_function'''				)
       self.assertIsNotNone(__a				)
       self.assertEqual(len(vf_loading_info['''missing_keys''']				)						,  0				)

       value_function.to(__a				)
       lowerCAmelCase		:					Optional[Any]  	=     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]				):
       lowerCAmelCase		:					Dict  	=     UNetaDModel.from_pretrained(
           '''bglick13/hopper-medium-v2-value-function-hor32'''						,  output_loading_info=__a						,  subfolder='''value_function'''				)
       torch.manual_seed(0				)
       if torch.cuda.is_available():
            torch.cuda.manual_seed_all(0				)

       lowerCAmelCase		:					str  	=     value_function.config.in_channels
       lowerCAmelCase		:					Dict  	=     1_4
       lowerCAmelCase		:					List[str]  	=     torch.randn((1, seq_len, num_features)				).permute(
           0						,  2						,  1				)  # match original, we can update values and remove
       lowerCAmelCase		:					List[Any]  	=     torch.full((num_features,)						,  0				)

       with torch.no_grad():
            lowerCAmelCase		:					Any  	=     value_function(__a						,  __a				).sample

       # fmt: off
       lowerCAmelCase		:					str  	=     torch.tensor([1_6_5.2_5] * seq_len				)
       # fmt: on
       self.assertTrue(torch.allclose(__a						,  __a						,  rtol=1E-3				)				)





  def       _A							(							self    :	Optional[Any]				):
       pass


from ..utils import requires_backends


__SCREAMING_SNAKE_CASE			:   Union[str, Any]         =     None


class 		__magic_name__   :





		def __init__(							self    :	Optional[int]						,  *lowerCamelCase__    :	List[str]						,  **lowerCamelCase__    :	Optional[Any]				):
							requires_backends(self						,  ['''detectron2''']				)





		@classmethod
		def       _A							(							cls    :	Dict						,  *lowerCamelCase__    :	List[str]						,  **lowerCamelCase__    :	Optional[Any]				):
							requires_backends(cls						,  ['''detectron2''']				)
