

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_   ( snake_case__  ,							snake_case__  ,							unittest.TestCase):
 _lowerCAmelCase       		=  UNetaDModel
 _lowerCAmelCase       		=  '''sample'''




 @property
 def 						a					( self       ):
   __UpperCAmelCase  :     Optional[Any]		=    4
   __UpperCAmelCase  :     List[str]		=    14
   __UpperCAmelCase  :     Tuple		=    16

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

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




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




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




 def 						a					( self       ):
   pass




 def 						a					( self       ):
   pass




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




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




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




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




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




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




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




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

   model.to(__snake_case       )
   __UpperCAmelCase  :     Optional[int]		=    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 						a					( self       ):
   __UpperCAmelCase  :     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       )

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

   with torch.no_grad():
     __UpperCAmelCase  :     List[Any]		=    model(__snake_case     ,     __snake_case       ).sample.permute(0     ,     2     ,     1       )

   __UpperCAmelCase  :     Union[str, Any]		=    output[0, -3:, -3:].flatten()
   # fmt: off
   __UpperCAmelCase  :     Dict		=    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(__snake_case     ,     __snake_case     ,     rtol=1E-3       )       )




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




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

   __UpperCAmelCase  :     Tuple		=    65_536
   __UpperCAmelCase  :     Optional[Any]		=    torch.sin(torch.arange(__snake_case       )[None, None, :].repeat(1     ,     2     ,     1       )       ).to(__snake_case       )
   __UpperCAmelCase  :     str		=    torch.tensor([1]       ).to(__snake_case       )

   with torch.no_grad():
     __UpperCAmelCase  :     List[str]		=    model(__snake_case     ,     __snake_case       ).sample

   __UpperCAmelCase  :     int		=    output.abs().sum()
   __UpperCAmelCase  :     List[str]		=    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_   ( snake_case__  ,							snake_case__  ,							unittest.TestCase):
 _lowerCAmelCase       		=  UNetaDModel
 _lowerCAmelCase       		=  '''sample'''




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

   __UpperCAmelCase  :     str		=    floats_tensor((batch_size, num_features, seq_len)       ).to(__snake_case       )
   __UpperCAmelCase  :     Any		=    torch.tensor([10] * batch_size       ).to(__snake_case       )

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




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




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




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




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




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




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




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




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

   with torch.no_grad():
     __UpperCAmelCase  :     Tuple		=    model(**__snake_case       )

     if isinstance(__snake_case     ,     __snake_case       ):
       __UpperCAmelCase  :     Optional[int]		=    output.sample

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




 def 						a					( self       ):
   pass




 def 						a					( self       ):
   pass




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




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

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

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

   with torch.no_grad():
     __UpperCAmelCase  :     Union[str, Any]		=    value_function(__snake_case     ,     __snake_case       ).sample

   # fmt: off
   __UpperCAmelCase  :     int		=    torch.tensor([1_6_5.2_5] * seq_len       )
   # fmt: on
   self.assertTrue(torch.allclose(__snake_case     ,     __snake_case     ,     rtol=1E-3       )       )




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








def    _snake_case		(snake_case_				:    int						,					snake_case_				:    int						,					snake_case_				:    int						,					snake_case_				:    int     )       ->		int:
  return len_board * len_board_column * row + column





def    _snake_case		(snake_case_				:    list[list[str]]						,					snake_case_				:    str						,					snake_case_				:    int						,					snake_case_				:    int						,					snake_case_				:    int						,					snake_case_				:    set[int]						,					)       ->		bool:

  if board[row][column] != word[word_index]:
    return False

  if word_index == len(snake_case_     ) - 1:
    return True

  __UpperCAmelCase  :     List[Any]		=    [(0, 1), (0, -1), (-1, 0), (1, 0)]
  __UpperCAmelCase  :     Dict		=    len(snake_case_     )
  __UpperCAmelCase  :     str		=    len(board[0]     )
  for direction in traverts_directions:
    __UpperCAmelCase  :     Dict		=    row + direction[0]
    __UpperCAmelCase  :     Optional[int]		=    column + direction[1]
    if not (0 <= next_i < len_board and 0 <= next_j < len_board_column):
      continue

    __UpperCAmelCase  :     str		=    get_point_key(snake_case_						,					snake_case_						,					snake_case_						,					snake_case_     )
    if key in visited_points_set:
      continue

    visited_points_set.add(snake_case_     )
    if exits_word(snake_case_						,					snake_case_						,					snake_case_						,					snake_case_						,					word_index + 1						,					snake_case_     ):
      return True

    visited_points_set.remove(snake_case_     )

  return False





def    _snake_case		(snake_case_				:    list[list[str]]						,					snake_case_				:    str     )       ->		bool:
  __UpperCAmelCase  :     str		=    (
      'The board should be a non empty matrix of single chars strings.'
  )

  __UpperCAmelCase  :     Any		=    len(snake_case_     )
  if not isinstance(snake_case_						,					snake_case_     ) or len(snake_case_     ) == 0:
    raise ValueError(snake_case_     )

  for row in board:
    if not isinstance(snake_case_						,					snake_case_     ) or len(snake_case_     ) == 0:
      raise ValueError(snake_case_     )

    for item in row:
      if not isinstance(snake_case_						,					snake_case_     ) or len(snake_case_     ) != 1:
        raise ValueError(snake_case_     )

    # Validate word
  if not isinstance(snake_case_						,					snake_case_     ) or len(snake_case_     ) == 0:
    raise ValueError(
        'The word parameter should be a string of length greater than 0.'     )

  __UpperCAmelCase  :     str		=    len(board[0]     )
  for i in range(snake_case_     ):
    for j in range(snake_case_     ):
      if exits_word(
          snake_case_						,					snake_case_						,					snake_case_						,					snake_case_						,					0						,					{get_point_key(snake_case_						,					snake_case_						,					snake_case_						,					snake_case_     )}     ):
        return True

  return False


if __name__ == "__main__":
      import doctest

      doctest.testmod()





