

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 SCREAMING_SNAKE_CASE	(  a_   ,							a_   ,							unittest.TestCase ):
  """simple docstring"""






  lowerCamelCase    :			Any						      =UNetaDModel
  lowerCamelCase    :			Dict						      ="sample"







  @property
  def        SCREAMING_SNAKE_CASE    (			self :    Dict    )							->	int:






     """simple docstring"""







     __lowerCAmelCase  :			Union[str, Any]					    =						4
     __lowerCAmelCase  :			Optional[int]					    =						14
     __lowerCAmelCase  :			Optional[int]					    =						16

     __lowerCAmelCase  :			Tuple					    =						floats_tensor((batch_size, num_features, seq_len)    ).to(lowerCAmelCase    )
     __lowerCAmelCase  :			Optional[Any]					    =						torch.tensor([10] * batch_size    ).to(lowerCAmelCase    )

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







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






     """simple docstring"""







     return (4, 14, 16)







  @property
  def        SCREAMING_SNAKE_CASE    (			self :    Optional[int]    )							->	Any:






     """simple docstring"""







     return (4, 14, 16)







  def        SCREAMING_SNAKE_CASE    (			self :    List[str]    )							->	int:






     """simple docstring"""







     pass







  def        SCREAMING_SNAKE_CASE    (			self :    Dict    )							->	Optional[Any]:






     """simple docstring"""







     pass







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






     """simple docstring"""







     super().test_determinism()







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






     """simple docstring"""







     super().test_outputs_equivalence()







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






     """simple docstring"""







     super().test_from_save_pretrained()







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






     """simple docstring"""







     super().test_from_save_pretrained_variant()







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






     """simple docstring"""







     super().test_model_from_pretrained()







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






     """simple docstring"""







     super().test_output()







  def        SCREAMING_SNAKE_CASE    (			self :    str    )							->	str:






     """simple docstring"""







     __lowerCAmelCase  :			Optional[Any]					    =						{
         """block_out_channels""": (32, 64, 1_28, 2_56),
         """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""",
     }
     __lowerCAmelCase  :			List[Any]					    =						self.dummy_input
     return init_dict, inputs_dict







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






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			int					    =						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    )
     __lowerCAmelCase  :			List[Any]					    =						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        SCREAMING_SNAKE_CASE    (			self :    str    )							->	Optional[Any]:






     """simple docstring"""







     __lowerCAmelCase  :			Optional[int]					    =						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  :			Union[str, Any]					    =						model.config.in_channels
     __lowerCAmelCase  :			int					    =						16
     __lowerCAmelCase  :			List[Any]					    =						torch.randn((1, seq_len, num_features)    ).permute(
         0       ,							2       ,							1    )  # match original, we can update values and remove
     __lowerCAmelCase  :			str					    =						torch.full((num_features,)       ,							0    )

     with torch.no_grad():
        __lowerCAmelCase  :			List[str]					    =						model(lowerCAmelCase       ,							lowerCAmelCase    ).sample.permute(0       ,							2       ,							1    )

     __lowerCAmelCase  :			List[Any]					    =						output[0, -3:, -3:].flatten()
     # fmt: off
     __lowerCAmelCase  :			Dict					    =						torch.tensor([-2.13_7172, 1.142_6016, 0.368_8687, -0.76_6922, 0.730_3146, 0.1103_8864, -0.476_0633, 0.1327_0172, 0.0259_1348]    )
     # fmt: on
     self.assertTrue(torch.allclose(lowerCAmelCase       ,							lowerCAmelCase       ,							rtol=1e-3    )    )







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






     """simple docstring"""







     pass







  @slow
  def        SCREAMING_SNAKE_CASE    (			self :    int    )							->	str:






     """simple docstring"""







     __lowerCAmelCase  :			str					    =						"""harmonai/maestro-150k"""
     __lowerCAmelCase  :			List[Any]					    =						UNetaDModel.from_pretrained(lowerCAmelCase       ,							subfolder="""unet"""    )
     model.to(lowerCAmelCase    )

     __lowerCAmelCase  :			Optional[Any]					    =						6_55_36
     __lowerCAmelCase  :			Optional[Any]					    =						torch.sin(torch.arange(lowerCAmelCase    )[None, None, :].repeat(1       ,							2       ,							1    )    ).to(lowerCAmelCase    )
     __lowerCAmelCase  :			List[str]					    =						torch.tensor([1]    ).to(lowerCAmelCase    )

     with torch.no_grad():
        __lowerCAmelCase  :			List[Any]					    =						model(lowerCAmelCase       ,							lowerCAmelCase    ).sample

     __lowerCAmelCase  :			str					    =						output.abs().sum()
     __lowerCAmelCase  :			Any					    =						output.abs().max()

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






class SCREAMING_SNAKE_CASE	(  a_   ,							a_   ,							unittest.TestCase ):
  """simple docstring"""






  lowerCamelCase    :			Optional[Any]						      =UNetaDModel
  lowerCamelCase    :			Tuple						      ="sample"







  @property
  def        SCREAMING_SNAKE_CASE    (			self :    List[Any]    )							->	int:






     """simple docstring"""







     __lowerCAmelCase  :			Dict					    =						4
     __lowerCAmelCase  :			Dict					    =						14
     __lowerCAmelCase  :			str					    =						16

     __lowerCAmelCase  :			str					    =						floats_tensor((batch_size, num_features, seq_len)    ).to(lowerCAmelCase    )
     __lowerCAmelCase  :			Dict					    =						torch.tensor([10] * batch_size    ).to(lowerCAmelCase    )

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







  @property
  def        SCREAMING_SNAKE_CASE    (			self :    Optional[Any]    )							->	Optional[Any]:






     """simple docstring"""







     return (4, 14, 16)







  @property
  def        SCREAMING_SNAKE_CASE    (			self :    Optional[int]    )							->	Any:






     """simple docstring"""







     return (4, 14, 1)







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






     """simple docstring"""







     super().test_determinism()







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






     """simple docstring"""







     super().test_outputs_equivalence()







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






     """simple docstring"""







     super().test_from_save_pretrained()







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






     """simple docstring"""







     super().test_from_save_pretrained_variant()







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






     """simple docstring"""







     super().test_model_from_pretrained()







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






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			Tuple					    =						self.prepare_init_args_and_inputs_for_common()
     __lowerCAmelCase  :			Any					    =						self.model_class(**lowerCAmelCase    )
     model.to(lowerCAmelCase    )
     model.eval()

     with torch.no_grad():
        __lowerCAmelCase  :			int					    =						model(**lowerCAmelCase    )

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

     self.assertIsNotNone(lowerCAmelCase    )
     __lowerCAmelCase  :			Optional[Any]					    =						torch.Size((inputs_dict["""sample"""].shape[0], 1)    )
     self.assertEqual(output.shape       ,							lowerCAmelCase       ,							"""Input and output shapes do not match"""    )







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






     """simple docstring"""







     pass







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






     """simple docstring"""







     pass







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






     """simple docstring"""







     __lowerCAmelCase  :			str					    =						{
         """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, 1_28, 2_56],
         """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  :			str					    =						self.dummy_input
     return init_dict, inputs_dict







  @unittest.skipIf(torch_device == """mps"""       ,							"""mish op not supported in MPS"""    )
  def        SCREAMING_SNAKE_CASE    (			self :    Tuple    )							->	str:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			Union[str, Any]					    =						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    )
     __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        SCREAMING_SNAKE_CASE    (			self :    int    )							->	List[str]:






     """simple docstring"""







     __lowerCAmelCase    ,__lowerCAmelCase  :			Optional[Any]					    =						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    )

     __lowerCAmelCase  :			Any					    =						value_function.config.in_channels
     __lowerCAmelCase  :			Optional[Any]					    =						14
     __lowerCAmelCase  :			Union[str, Any]					    =						torch.randn((1, seq_len, num_features)    ).permute(
         0       ,							2       ,							1    )  # match original, we can update values and remove
     __lowerCAmelCase  :			Optional[int]					    =						torch.full((num_features,)       ,							0    )

     with torch.no_grad():
        __lowerCAmelCase  :			int					    =						value_function(lowerCAmelCase       ,							lowerCAmelCase    ).sample

     # fmt: off
     __lowerCAmelCase  :			List[Any]					    =						torch.tensor([165.25] * seq_len    )
     # fmt: on
     self.assertTrue(torch.allclose(lowerCAmelCase       ,							lowerCAmelCase       ,							rtol=1e-3    )    )







  def        SCREAMING_SNAKE_CASE    (			self :    List[str]    )							->	List[str]:






     """simple docstring"""







     pass





import importlib.metadata
import subprocess
import sys





def 	snake_case_      (__A     :						bool = False     )			->	str:
   __lowerCAmelCase  :			Any					    =						False
   if "IPython" in sys.modules:
      __lowerCAmelCase  :			Optional[int]					    =						"""google.colab""" in str(sys.modules["""IPython"""].get_ipython()     )

   if in_colab:
      if upgrade:
         __lowerCAmelCase  :			int					    =						["""pip""", """install""", """-U""", """torch"""]
         subprocess.run(__A					,   check=__A     )
      # get the current version of torch
      __lowerCAmelCase  :			List[Any]					    =						importlib.metadata.version("""torch"""     )
      __lowerCAmelCase  :			List[Any]					    =						torch_version[: torch_version.rindex("""."""     )]
      __lowerCAmelCase  :			int					    =						f'''https://storage.googleapis.com/tpu-pytorch/wheels/colab/torch_xla-{torch_version_trunc}-cp37-cp37m-linux_x86_64.whl'''
      __lowerCAmelCase  :			List[Any]					    =						["""pip""", """install""", xla_wheel]
      subprocess.run(__A					,   check=__A     )
   else:
      raise RuntimeError("""`install_xla` utility works only on google colab."""     )


