
"""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 	lowerCamelCase__      (      SCREAMING_SNAKE_CASE__				,			SCREAMING_SNAKE_CASE__				,			unittest.TestCase       ):





			'''simple docstring'''
			lowerCamelCase						      =	UNetaDModel
			lowerCamelCase						      =	'sample'



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

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

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



			@property
			def       _lowerCAmelCase	(							self						)				->  List[Any]:
				return (4, 14, 16)



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



			def       _lowerCAmelCase	(							self						)				->  List[Any]:
				pass



			def       _lowerCAmelCase	(							self						)				->  Dict:
				pass



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



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



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



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



			@unittest.skipIf(torch_device == """mps"""  ,  """mish op not supported in MPS"""						)
			def       _lowerCAmelCase	(							self						)				->  Dict:
				super().test_model_from_pretrained()



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



			def       _lowerCAmelCase	(							self						)				->  Optional[int]:
				_lowerCAmelCase			={
				    """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			=self.dummy_input
				return init_dict, inputs_dict



			@unittest.skipIf(torch_device == """mps"""  ,  """mish op not supported in MPS"""						)
			def       _lowerCAmelCase	(							self						)				->  Optional[int]:
				_lowerCAmelCase		,							_lowerCAmelCase			=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			=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       _lowerCAmelCase	(							self						)				->  Union[str, Any]:
				_lowerCAmelCase			=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			=model.config.in_channels
				_lowerCAmelCase			=16
				_lowerCAmelCase			=torch.randn((1, seq_len, num_features)						).permute(
				    0  ,  2  ,  1						)  # match original, we can update values and remove
				_lowerCAmelCase			=torch.full((num_features,)  ,  0						)

				with torch.no_grad():
					_lowerCAmelCase			=model(a_  ,  a_						).sample.permute(0  ,  2  ,  1						)

				_lowerCAmelCase			=output[0, -3:, -3:].flatten()
				# fmt: off
				_lowerCAmelCase			=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       _lowerCAmelCase	(							self						)				->  Any:
				pass



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

				_lowerCAmelCase			=6_55_36
				_lowerCAmelCase			=torch.sin(torch.arange(a_						)[None, None, :].repeat(1  ,  2  ,  1						)						).to(a_						)
				_lowerCAmelCase			=torch.tensor([1]						).to(a_						)

				with torch.no_grad():
					_lowerCAmelCase			=model(a_  ,  a_						).sample

				_lowerCAmelCase			=output.abs().sum()
				_lowerCAmelCase			=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 	lowerCamelCase__      (      SCREAMING_SNAKE_CASE__				,			SCREAMING_SNAKE_CASE__				,			unittest.TestCase       ):





			'''simple docstring'''
			lowerCamelCase						      =	UNetaDModel
			lowerCamelCase						      =	'sample'



			@property
			def       _lowerCAmelCase	(							self						)				->  Any:
				_lowerCAmelCase			=4
				_lowerCAmelCase			=14
				_lowerCAmelCase			=16

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

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



			@property
			def       _lowerCAmelCase	(							self						)				->  Tuple:
				return (4, 14, 16)



			@property
			def       _lowerCAmelCase	(							self						)				->  Dict:
				return (4, 14, 1)



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



			@unittest.skipIf(torch_device == """mps"""  ,  """mish op not supported in MPS"""						)
			def       _lowerCAmelCase	(							self						)				->  Dict:
				super().test_outputs_equivalence()



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



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



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



			@unittest.skipIf(torch_device == """mps"""  ,  """mish op not supported in MPS"""						)
			def       _lowerCAmelCase	(							self						)				->  Optional[Any]:
				_lowerCAmelCase		,							_lowerCAmelCase			=self.prepare_init_args_and_inputs_for_common()
				_lowerCAmelCase			=self.model_class(**a_						)
				model.to(a_						)
				model.eval()

				with torch.no_grad():
					_lowerCAmelCase			=model(**a_						)

					if isinstance(a_  ,  a_						):
						_lowerCAmelCase			=output.sample

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



			def       _lowerCAmelCase	(							self						)				->  Any:
				pass



			def       _lowerCAmelCase	(							self						)				->  int:
				pass



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



			@unittest.skipIf(torch_device == """mps"""  ,  """mish op not supported in MPS"""						)
			def       _lowerCAmelCase	(							self						)				->  Any:
				_lowerCAmelCase		,							_lowerCAmelCase			=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			=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       _lowerCAmelCase	(							self						)				->  Optional[int]:
				_lowerCAmelCase		,							_lowerCAmelCase			=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			=value_function.config.in_channels
				_lowerCAmelCase			=14
				_lowerCAmelCase			=torch.randn((1, seq_len, num_features)						).permute(
				    0  ,  2  ,  1						)  # match original, we can update values and remove
				_lowerCAmelCase			=torch.full((num_features,)  ,  0						)

				with torch.no_grad():
					_lowerCAmelCase			=value_function(a_  ,  a_						).sample

				# fmt: off
				_lowerCAmelCase			=torch.tensor([1_6_5.2_5] * seq_len						)
				# fmt: on
				self.assertTrue(torch.allclose(a_  ,  a_  ,  rtol=1e-3						)						)



			def       _lowerCAmelCase	(							self						)				->  Any:
				pass

"""simple docstring"""


from typing import Dict, Optional, Tuple

import torch



def 					_lowerCamelCase(__UpperCamelCase )       ->				torch.Tensor:
	_lowerCAmelCase			=boundaries[1] - boundaries[0]
	_lowerCAmelCase			=boundaries + step / 2
	_lowerCAmelCase			=torch.cat([bin_centers, (bin_centers[-1] + step).unsqueeze(-1 )]					,				dim=0 )
	return bin_centers



def 					_lowerCamelCase(__UpperCamelCase					,				__UpperCamelCase					,				)       ->				Tuple[torch.Tensor, torch.Tensor]:
	_lowerCAmelCase			=_calculate_bin_centers(__UpperCamelCase )
	return (
	    torch.sum(aligned_distance_error_probs * bin_centers					,				dim=-1 ),
	    bin_centers[-1],
	)



def 					_lowerCamelCase(__UpperCamelCase					,				__UpperCamelCase = 31					,				__UpperCamelCase = 64					,				**__UpperCamelCase					,				)       ->				Dict[str, torch.Tensor]:
	_lowerCAmelCase			=torch.linspace(0					,				__UpperCamelCase					,				steps=(no_bins - 1)					,				device=logits.device )

	_lowerCAmelCase			=torch.nn.functional.softmax(__UpperCamelCase					,				dim=-1 )
	_lowerCAmelCase		,							_lowerCAmelCase			=_calculate_expected_aligned_error(
	    alignment_confidence_breaks=__UpperCamelCase					,				aligned_distance_error_probs=__UpperCamelCase					,				)

	return {
	    "aligned_confidence_probs": aligned_confidence_probs,
	    "predicted_aligned_error": predicted_aligned_error,
	    "max_predicted_aligned_error": max_predicted_aligned_error,
	}



def 					_lowerCamelCase(__UpperCamelCase					,				__UpperCamelCase = None					,				__UpperCamelCase = 31					,				__UpperCamelCase = 64					,				__UpperCamelCase = 1E-8					,				**__UpperCamelCase					,				)       ->				torch.Tensor:
	if residue_weights is None:
		_lowerCAmelCase			=logits.new_ones(logits.shape[-2] )

	_lowerCAmelCase			=torch.linspace(0					,				__UpperCamelCase					,				steps=(no_bins - 1)					,				device=logits.device )

	_lowerCAmelCase			=_calculate_bin_centers(__UpperCamelCase )
	torch.sum(__UpperCamelCase )
	_lowerCAmelCase			=logits.shape[-2]
	_lowerCAmelCase			=max(__UpperCamelCase					,				19 )

	_lowerCAmelCase			=1.24 * (clipped_n - 15) ** (1.0 / 3) - 1.8

	_lowerCAmelCase			=torch.nn.functional.softmax(__UpperCamelCase					,				dim=-1 )

	_lowerCAmelCase			=1.0 / (1 + (bin_centers**2) / (da**2))
	_lowerCAmelCase			=torch.sum(probs * tm_per_bin					,				dim=-1 )

	_lowerCAmelCase			=residue_weights / (eps + residue_weights.sum())
	_lowerCAmelCase			=torch.sum(predicted_tm_term * normed_residue_mask					,				dim=-1 )

	_lowerCAmelCase			=per_alignment * residue_weights

	_lowerCAmelCase			=(weighted == torch.max(__UpperCamelCase )).nonzero()[0]
	return per_alignment[tuple(__UpperCamelCase )]



