





# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

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


if TYPE_CHECKING:
     import torch
class      UpperCamelCase_						(					TensorFormatter[Mapping, '''torch.Tensor''', Mapping]   ):




     def __init__(						self			,  lowerCAmelCase_=None			,  **lowerCAmelCase_					)						->							List[str]:
          super().__init__(features=lowerCamelCase__					)
          _snake_case													=			torch_tensor_kwargs
          import torch  # noqa import torch at initialization




     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Optional[Any]:
          import torch

          if isinstance(lowerCamelCase__			,  lowerCamelCase__					) and column:
               if all(
                   isinstance(lowerCamelCase__			,  torch.Tensor					) and x.shape == column[0].shape and x.dtype == column[0].dtype
                   for x in column					):
                    return torch.stack(lowerCamelCase__					)
          return column




     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							int:
          import torch

          if isinstance(lowerCamelCase__			,  (str, bytes, type(lowerCamelCase__					))					):
               return value
          elif isinstance(lowerCamelCase__			,  (np.character, np.ndarray)					) and np.issubdtype(value.dtype			,  np.character					):
               return value.tolist()

          _snake_case													=			{}

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

               if isinstance(lowerCamelCase__			,  PIL.Image.Image					):
                    _snake_case													=			np.asarray(lowerCamelCase__					)
          return torch.tensor(lowerCamelCase__			,  **{**default_dtype, **self.torch_tensor_kwargs}					)




     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Dict:
          import torch

          # support for torch, tf, jax etc.
          if hasattr(lowerCamelCase__			,  '__array__'					) and not isinstance(lowerCamelCase__			,  torch.Tensor					):
               _snake_case													=			data_struct.__array__()
          # support for nested types like struct of list of struct
          if isinstance(lowerCamelCase__			,  np.ndarray					):
               if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
                    return self._consolidate([self.recursive_tensorize(lowerCamelCase__					) for substruct in data_struct]					)
          elif isinstance(lowerCamelCase__			,  (list, tuple)					):
               return self._consolidate([self.recursive_tensorize(lowerCamelCase__					) for substruct in data_struct]					)
          return self._tensorize(lowerCamelCase__					)




     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Optional[int]:
          return map_nested(self._recursive_tensorize			,  lowerCamelCase__			,  map_list=lowerCamelCase__					)




     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Mapping:
          _snake_case													=			self.numpy_arrow_extractor().extract_row(lowerCamelCase__					)
          _snake_case													=			self.python_features_decoder.decode_row(lowerCamelCase__					)
          return self.recursive_tensorize(lowerCamelCase__					)




     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							"torch.Tensor":
          _snake_case													=			self.numpy_arrow_extractor().extract_column(lowerCamelCase__					)
          _snake_case													=			self.python_features_decoder.decode_column(lowerCamelCase__			,  pa_table.column_names[0]					)
          _snake_case													=			self.recursive_tensorize(lowerCamelCase__					)
          _snake_case													=			self._consolidate(lowerCamelCase__					)
          return column






     def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Mapping:
          _snake_case													=			self.numpy_arrow_extractor().extract_batch(lowerCamelCase__					)
          _snake_case													=			self.python_features_decoder.decode_batch(lowerCamelCase__					)
          _snake_case													=			self.recursive_tensorize(lowerCamelCase__					)
          for column_name in batch:
               _snake_case													=			self._consolidate(batch[column_name]					)
          return batch










import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch
@require_torch
class      UpperCamelCase_						:




					@property
					def 		lowerCAmelCase     (						self					)						->							int:
										return self.get_dummy_input()




					@property
					def 		lowerCAmelCase     (						self					)						->							Optional[Any]:
										if self.block_type == "down":
															return (4, 32, 16, 16)
										elif self.block_type == "mid":
															return (4, 32, 32, 32)
										elif self.block_type == "up":
															return (4, 32, 64, 64)

										raise ValueError(F'''\'{self.block_type}\' is not a supported block_type. Set it to \'up\', \'mid\', or \'down\'.'''					)




					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_=True			,  lowerCAmelCase_=False			,  lowerCAmelCase_=False			,  lowerCAmelCase_=False			,  )						->							List[str]:
										_snake_case													=			4
										_snake_case													=			32
										_snake_case													=			(32, 32)

										_snake_case													=			torch.manual_seed(0					)
										_snake_case													=			torch.device(lowerCAmelCase_					)
										_snake_case													=			(batch_size, num_channels) + sizes
										_snake_case													=			randn_tensor(lowerCAmelCase_			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					)
										_snake_case													=			{'hidden_states': hidden_states}

										if include_temb:
															_snake_case													=			128
															_snake_case													=			randn_tensor((batch_size, temb_channels)			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					)

										if include_res_hidden_states_tuple:
															_snake_case													=			torch.manual_seed(1					)
															_snake_case													=			(randn_tensor(lowerCAmelCase_			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					),)

										if include_encoder_hidden_states:
															_snake_case													=			floats_tensor((batch_size, 32, 32)					).to(lowerCAmelCase_					)

										if include_skip_sample:
															_snake_case													=			randn_tensor(((batch_size, 3) + sizes)			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					)

										return dummy_input




					def 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case													=			{
										    'in_channels': 32,
										    'out_channels': 32,
										    'temb_channels': 128,
										}
										if self.block_type == "up":
															_snake_case													=			32

										if self.block_type == "mid":
															init_dict.pop('out_channels'					)

										_snake_case													=			self.dummy_input
										return init_dict, inputs_dict




					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Optional[int]:
										_snake_case							,       _snake_case													=			self.prepare_init_args_and_inputs_for_common()
										_snake_case													=			self.block_class(**lowerCAmelCase_					)
										unet_block.to(lowerCAmelCase_					)
										unet_block.eval()

										with torch.no_grad():
															_snake_case													=			unet_block(**lowerCAmelCase_					)

										if isinstance(lowerCAmelCase_			,  lowerCAmelCase_					):
															_snake_case													=			output[0]

										self.assertEqual(output.shape			,  self.output_shape					)

										_snake_case													=			output[0, -1, -3:, -3:]
										_snake_case													=			torch.tensor(lowerCAmelCase_					).to(lowerCAmelCase_					)
										assert torch_all_close(output_slice.flatten()			,  lowerCAmelCase_			,  atol=5E-3					)






					@unittest.skipIf(torch_device == 'mps'			,  'Training is not supported in mps'					)
					def 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case							,       _snake_case													=			self.prepare_init_args_and_inputs_for_common()
										_snake_case													=			self.block_class(**lowerCAmelCase_					)
										model.to(lowerCAmelCase_					)
										model.train()
										_snake_case													=			model(**lowerCAmelCase_					)

										if isinstance(lowerCAmelCase_			,  lowerCAmelCase_					):
															_snake_case													=			output[0]

										_snake_case													=			torch.device(lowerCAmelCase_					)
										_snake_case													=			randn_tensor(output.shape			,  device=lowerCAmelCase_					)
										_snake_case													=			torch.nn.functional.mse_loss(lowerCAmelCase_			,  lowerCAmelCase_					)
										loss.backward()



