

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  ):


    '''simple docstring'''






    a__       							=    UNetaDModel
    a__       							=    "sample"


    @property
    def        lowerCAmelCase_  (self					)       -> str:
        __UpperCAmelCase			   =     4
        __UpperCAmelCase			   =     14
        __UpperCAmelCase			   =     16

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

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


    @property
    def        lowerCAmelCase_  (self					)       -> int:
        return (4, 14, 16)


    @property
    def        lowerCAmelCase_  (self					)       -> Union[str, Any]:
        return (4, 14, 16)


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


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


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


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


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


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


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


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


    def        lowerCAmelCase_  (self					)       -> Any:
        __UpperCAmelCase			   =     {
            '''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''',
        }
        __UpperCAmelCase			   =     self.dummy_input
        return init_dict, inputs_dict


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

        model.to(__UpperCamelCase					)
        __UpperCAmelCase			   =     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					)       -> List[Any]:
        __UpperCAmelCase			   =     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					)

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

        with torch.no_grad():
            __UpperCAmelCase			   =     model(__UpperCamelCase       , __UpperCamelCase					).sample.permute(0       , 2       , 1					)

        __UpperCAmelCase			   =     output[0, -3:, -3:].flatten()
        # fmt: off
        __UpperCAmelCase			   =     torch.tensor([-2.137172, 1.1426016, 0.3688687, -0.766922, 0.7303146, 0.11038864, -0.4760633, 0.13270172, 0.02591348]					)
        # fmt: on
        self.assertTrue(torch.allclose(__UpperCamelCase       , __UpperCamelCase       , rtol=1E-3					)					)


    def        lowerCAmelCase_  (self					)       -> Tuple:
        # Not implemented yet for this UNet
        pass






    @slow
    def        lowerCAmelCase_  (self					)       -> Dict:
        __UpperCAmelCase			   =     '''harmonai/maestro-150k'''
        __UpperCAmelCase			   =     UNetaDModel.from_pretrained(__UpperCamelCase       , subfolder='''unet'''					)
        model.to(__UpperCamelCase					)

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

        with torch.no_grad():
            __UpperCAmelCase			   =     model(__UpperCamelCase       , __UpperCamelCase					).sample

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

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






class    A_				( lowercase_	,       lowercase_	,       unittest.TestCase  ):


    '''simple docstring'''






    a__       							=    UNetaDModel
    a__       							=    "sample"


    @property
    def        lowerCAmelCase_  (self					)       -> Dict:
        __UpperCAmelCase			   =     4
        __UpperCAmelCase			   =     14
        __UpperCAmelCase			   =     16

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

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


    @property
    def        lowerCAmelCase_  (self					)       -> Any:
        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					)       -> Union[str, Any]:
        super().test_determinism()


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


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


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


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


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

        with torch.no_grad():
            __UpperCAmelCase			   =     model(**__UpperCamelCase					)

            if isinstance(__UpperCamelCase       , __UpperCamelCase					):
                __UpperCAmelCase			   =     output.sample

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


    def        lowerCAmelCase_  (self					)       -> int:
        pass


    def        lowerCAmelCase_  (self					)       -> List[str]:
        pass


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


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

        value_function.to(__UpperCamelCase					)
        __UpperCAmelCase			   =     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					)       -> int:
        __UpperCAmelCase   ,       __UpperCAmelCase			   =     UNetaDModel.from_pretrained(
            '''bglick13/hopper-medium-v2-value-function-hor32'''       , output_loading_info=__UpperCamelCase       , subfolder='''value_function'''					)
        torch.manual_seed(0					)
        if torch.cuda.is_available():
            torch.cuda.manual_seed_all(0					)

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

        with torch.no_grad():
            __UpperCAmelCase			   =     value_function(__UpperCamelCase       , __UpperCamelCase					).sample

        # fmt: off
        __UpperCAmelCase			   =     torch.tensor([165.25] * seq_len					)
        # fmt: on
        self.assertTrue(torch.allclose(__UpperCamelCase       , __UpperCamelCase       , rtol=1E-3					)					)






    def        lowerCAmelCase_  (self					)       -> Any:
        # Not implemented yet for this UNet
        pass







# this script builds a small sample spm file tests/fixtures/test_sentencepiece_no_bos.model, with features needed by pegasus

# 1. pip install sentencepiece
#
# 2. wget https://raw.githubusercontent.com/google/sentencepiece/master/data/botchan.txt

# 3. build
import sentencepiece as spm

# pegasus:
# 1. no bos
# 2. eos_id is 1
# 3. unk_id is 2
# build a sample spm file accordingly
spm.SentencePieceTrainer.train(
    '--input=botchan.txt --model_prefix=test_sentencepiece_no_bos --bos_id=-1 --unk_id=2  --eos_id=1  --vocab_size=1000'
)

# 4. now update the fixture
# mv test_sentencepiece_no_bos.model ../../tests/fixtures/




