

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     lowercase_       (    __lowercase  ,				__lowercase  ,				unittest.TestCase	):
      UpperCamelCase_     :						Optional[Any]		 =							UNetaDModel
      UpperCamelCase_     :						List[Any]		 =							"sample"



      @property
      def 		UpperCamelCase_						(     self :							str						)       ->	Optional[int]:
        _snake_case		=				4
        _snake_case		=				14
        _snake_case		=				16

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

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



      @property
      def 		UpperCamelCase_						(     self :							Any						)       ->	str:
        return (4, 14, 16)



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



      def 		UpperCamelCase_						(     self :							Any						)       ->	Tuple:
        pass



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



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



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



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



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



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



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



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



      @unittest.skipIf(torch_device == '''mps'''     ,   '''mish op not supported in MPS'''						)
      def 		UpperCamelCase_						(     self :							List[str]						)       ->	Optional[int]:
        _snake_case,	_snake_case		=				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__						)
        _snake_case		=				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 		UpperCamelCase_						(     self :							List[Any]						)       ->	Optional[Any]:
        _snake_case		=				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						)

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

        with torch.no_grad():
          _snake_case		=				model(A__     ,   A__						).sample.permute(0     ,   2     ,   1						)

        _snake_case		=				output[0, -3:, -3:].flatten()
        # fmt: off
        _snake_case		=				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(A__     ,   A__     ,   rtol=1e-3						)						)



      def 		UpperCamelCase_						(     self :							Any						)       ->	Union[str, Any]:
        # Not implemented yet for this UNet
        pass







      @slow
      def 		UpperCamelCase_						(     self :							int						)       ->	Optional[Any]:
        _snake_case		=				'''harmonai/maestro-150k'''
        _snake_case		=				UNetaDModel.from_pretrained(A__     ,   subfolder='''unet'''						)
        model.to(A__						)

        _snake_case		=				65536
        _snake_case		=				torch.sin(torch.arange(A__						)[None, None, :].repeat(1     ,   2     ,   1						)						).to(A__						)
        _snake_case		=				torch.tensor([1]						).to(A__						)

        with torch.no_grad():
          _snake_case		=				model(A__     ,   A__						).sample

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

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



class     lowercase_       (    __lowercase  ,				__lowercase  ,				unittest.TestCase	):
      UpperCamelCase_     :						int		 =							UNetaDModel
      UpperCamelCase_     :						List[Any]		 =							"sample"



      @property
      def 		UpperCamelCase_						(     self :							Union[str, Any]						)       ->	Optional[int]:
        _snake_case		=				4
        _snake_case		=				14
        _snake_case		=				16

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

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



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



      @property
      def 		UpperCamelCase_						(     self :							str						)       ->	List[str]:
        return (4, 14, 1)



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



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



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



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



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



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

        with torch.no_grad():
          _snake_case		=				model(**A__						)

          if isinstance(A__     ,   A__						):
            _snake_case		=				output.sample

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



      def 		UpperCamelCase_						(     self :							Tuple						)       ->	Dict:
        pass



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



      def 		UpperCamelCase_						(     self :							Union[str, Any]						)       ->	Union[str, Any]:
        _snake_case		=				{
            '''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''',
        }
        _snake_case		=				self.dummy_input
        return init_dict, inputs_dict



      @unittest.skipIf(torch_device == '''mps'''     ,   '''mish op not supported in MPS'''						)
      def 		UpperCamelCase_						(     self :							List[Any]						)       ->	Dict:
        _snake_case,	_snake_case		=				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__						)
        _snake_case		=				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 		UpperCamelCase_						(     self :							Optional[int]						)       ->	List[str]:
        _snake_case,	_snake_case		=				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						)

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

        with torch.no_grad():
          _snake_case		=				value_function(A__     ,   A__						).sample

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







      def 		UpperCamelCase_						(     self :							List[Any]						)       ->	Union[str, Any]:
        # Not implemented yet for this UNet
        pass




import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter



class     lowercase_       (    TensorFormatter[Mapping, np.ndarray, Mapping]	):



      def __init__(     self :							Union[str, Any]     ,   A__ :							Dict=None     ,   **A__ :							str						)       ->	Any:
        super().__init__(features=A__						)
        _snake_case		=				np_array_kwargs



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							Dict						)       ->	Dict:
        if isinstance(A__     ,   A__						):
          if column and all(
              isinstance(A__     ,   np.ndarray						) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column						):
            return np.stack(A__						)
          else:
            # don't use np.array(column, dtype=object)
            # since it fails in certain cases
            # see https://stackoverflow.com/q/51005699
            _snake_case		=				np.empty(len(A__						)     ,   dtype=A__						)
            _snake_case		=				column
            return out
        return column



      def 		UpperCamelCase_						(     self :							Tuple     ,   A__ :							Optional[int]						)       ->	Optional[Any]:
        if isinstance(A__     ,   (str, bytes, type(A__						))						):
          return value
        elif isinstance(A__     ,   (np.character, np.ndarray)						) and np.issubdtype(value.dtype     ,   np.character						):
          return value
        elif isinstance(A__     ,   np.number						):
          return value

        _snake_case		=				{}

        if isinstance(A__     ,   np.ndarray						) and np.issubdtype(value.dtype     ,   np.integer						):
          _snake_case		=				{'''dtype''': np.intaa}
        elif isinstance(A__     ,   np.ndarray						) and np.issubdtype(value.dtype     ,   np.floating						):
          _snake_case		=				{'''dtype''': np.floataa}
        elif config.PIL_AVAILABLE and "PIL" in sys.modules:
          import PIL.Image

          if isinstance(A__     ,   PIL.Image.Image						):
            return np.asarray(A__     ,   **self.np_array_kwargs						)

        return np.asarray(A__     ,   **{**default_dtype, **self.np_array_kwargs}						)



      def 		UpperCamelCase_						(     self :							Union[str, Any]     ,   A__ :							Union[str, Any]						)       ->	int:
        # support for torch, tf, jax etc.
        if config.TORCH_AVAILABLE and "torch" in sys.modules:
          import torch

          if isinstance(A__     ,   torch.Tensor						):
            return self._tensorize(data_struct.detach().cpu().numpy()[()]						)
        if hasattr(A__     ,   '''__array__'''						) and not isinstance(A__     ,   (np.ndarray, np.character, np.number)						):
          _snake_case		=				data_struct.__array__()
        # support for nested types like struct of list of struct
        if isinstance(A__     ,   np.ndarray						):
          if data_struct.dtype == object:
            return self._consolidate([self.recursive_tensorize(A__						) for substruct in data_struct]						)
        if isinstance(A__     ,   (list, tuple)						):
          return self._consolidate([self.recursive_tensorize(A__						) for substruct in data_struct]						)
        return self._tensorize(A__						)



      def 		UpperCamelCase_						(     self :							List[Any]     ,   A__ :							dict						)       ->	Tuple:
        return map_nested(self._recursive_tensorize     ,   A__     ,   map_list=A__						)



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							pa.Table						)       ->	Mapping:
        _snake_case		=				self.numpy_arrow_extractor().extract_row(A__						)
        _snake_case		=				self.python_features_decoder.decode_row(A__						)
        return self.recursive_tensorize(A__						)



      def 		UpperCamelCase_						(     self :							Optional[Any]     ,   A__ :							pa.Table						)       ->	np.ndarray:
        _snake_case		=				self.numpy_arrow_extractor().extract_column(A__						)
        _snake_case		=				self.python_features_decoder.decode_column(A__     ,   pa_table.column_names[0]						)
        _snake_case		=				self.recursive_tensorize(A__						)
        _snake_case		=				self._consolidate(A__						)
        return column







      def 		UpperCamelCase_						(     self :							Dict     ,   A__ :							pa.Table						)       ->	Mapping:
        _snake_case		=				self.numpy_arrow_extractor().extract_batch(A__						)
        _snake_case		=				self.python_features_decoder.decode_batch(A__						)
        _snake_case		=				self.recursive_tensorize(A__						)
        for column_name in batch:
          _snake_case		=				self._consolidate(batch[column_name]						)
        return batch

