


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						(   _snake_case		,					_snake_case		,					unittest.TestCase							):
 SCREAMING_SNAKE_CASE__   = UNetaDModel
 SCREAMING_SNAKE_CASE__   = 'sample'




 @property
 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:int											= 4
       a		:List[Any]											= 14
       a		:str											= 16

       a		:List[Any]											= floats_tensor((batch_size, num_features, seq_len)					).to(_lowerCamelCase					)
       a		:Optional[int]											= torch.tensor([10] * batch_size					).to(_lowerCamelCase					)

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




 @property
 def  SCREAMING_SNAKE_CASE__       (	self					):
       return (4, 14, 16)




 @property
 def  SCREAMING_SNAKE_CASE__       (	self					):
       return (4, 14, 16)




 def  SCREAMING_SNAKE_CASE__       (	self					):
       pass




 def  SCREAMING_SNAKE_CASE__       (	self					):
       pass




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




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




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




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




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




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




 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:int											= {
           '''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''',
       }
       a		: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					):
       a							,       a		:Optional[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					)
       a		: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					):
       a		: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					)

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

       with torch.no_grad():
             a		:Optional[Any]											= model(_lowerCamelCase      ,						_lowerCamelCase					).sample.permute(0      ,						2      ,						1					)

       a		:str											= output[0, -3:, -3:].flatten()
       # fmt: off
       a		: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					):
       # Not implemented yet for this UNet
       pass




 @slow
 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Optional[int]											= '''harmonai/maestro-150k'''
       a		:str											= UNetaDModel.from_pretrained(_lowerCamelCase      ,						subfolder='''unet'''					)
       model.to(_lowerCamelCase					)

       a		:Optional[int]											= 6_5536
       a		:Optional[Any]											= torch.sin(torch.arange(_lowerCamelCase					)[None, None, :].repeat(1      ,						2      ,						1					)					).to(_lowerCamelCase					)
       a		:str											= torch.tensor([1]					).to(_lowerCamelCase					)

       with torch.no_grad():
             a		:int											= model(_lowerCamelCase      ,						_lowerCamelCase					).sample

       a		:Tuple											= output.abs().sum()
       a		:str											= output.abs().max()

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







class 	_snake_case						(   _snake_case		,					_snake_case		,					unittest.TestCase							):
 SCREAMING_SNAKE_CASE__   = UNetaDModel
 SCREAMING_SNAKE_CASE__   = 'sample'




 @property
 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Tuple											= 4
       a		:List[Any]											= 14
       a		:int											= 16

       a		:Optional[Any]											= floats_tensor((batch_size, num_features, seq_len)					).to(_lowerCamelCase					)
       a		:List[str]											= torch.tensor([10] * batch_size					).to(_lowerCamelCase					)

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




 @property
 def  SCREAMING_SNAKE_CASE__       (	self					):
       return (4, 14, 16)




 @property
 def  SCREAMING_SNAKE_CASE__       (	self					):
       return (4, 14, 1)




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




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




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




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




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




 @unittest.skipIf(torch_device == '''mps'''      ,						'''mish op not supported in MPS'''					)
 def  SCREAMING_SNAKE_CASE__       (	self					):
       # UNetRL is a value-function is different output shape
       a							,       a		:List[str]											= self.prepare_init_args_and_inputs_for_common()
       a		:Tuple											= self.model_class(**_lowerCamelCase					)
       model.to(_lowerCamelCase					)
       model.eval()

       with torch.no_grad():
             a		:Optional[Any]											= model(**_lowerCamelCase					)

             if isinstance(_lowerCamelCase      ,						_lowerCamelCase					):
                   a		:str											= output.sample

       self.assertIsNotNone(_lowerCamelCase					)
       a		:Dict											= 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					):
       pass




 def  SCREAMING_SNAKE_CASE__       (	self					):
       pass




 def  SCREAMING_SNAKE_CASE__       (	self					):
       a		:Optional[int]											= {
           '''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''',
       }
       a		:List[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					):
       a							,       a		:List[str]											= 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					)
       a		:Optional[int]											= 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					):
       a							,       a		:List[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					)

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

       with torch.no_grad():
             a		:List[Any]											= value_function(_lowerCamelCase      ,						_lowerCamelCase					).sample

       # fmt: off
       a		:Optional[int]											= torch.tensor([165.25] * seq_len					)
       # fmt: on
       self.assertTrue(torch.allclose(_lowerCamelCase      ,						_lowerCamelCase      ,						rtol=1e-3					)					)




 def  SCREAMING_SNAKE_CASE__       (	self					):
       # Not implemented yet for this UNet
       pass









import warnings

from ...utils import logging
from .image_processing_convnext import ConvNextImageProcessor


snake_case       :					Tuple        =  logging.get_logger(__name__)







class 	_snake_case						(   _snake_case							):




 def __init__(	self      ,						*_lowerCamelCase      ,						**_lowerCamelCase					):
       warnings.warn(
           '''The class ConvNextFeatureExtractor is deprecated and will be removed in version 5 of Transformers.'''
           ''' Please use ConvNextImageProcessor instead.'''      ,						_lowerCamelCase      ,						)
       super().__init__(*_lowerCamelCase      ,						**_lowerCamelCase					)





