
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 							UpperCAmelCase	(   __lowerCamelCase						, __lowerCamelCase						, unittest.TestCase					):
					a__: Optional[Any]			  =				UNetaDModel
					a__: str			  =				"""sample"""

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

											lowercase       :    List[Any]											=  floats_tensor((batch_size, num_features, seq_len)					).to(lowerCAmelCase					)
											lowercase       :    Any											=  torch.tensor([10] * batch_size					).to(lowerCAmelCase					)

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

					@property
					def        _lowerCAmelCase	(			self      :					List[str]					):
											return (4, 14, 16)

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

					def        _lowerCAmelCase	(			self      :					List[str]					):
											pass

					def        _lowerCAmelCase	(			self      :					Optional[Any]					):
											pass

					@unittest.skipIf(torch_device == '''mps''' ,      '''mish op not supported in MPS'''					)
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											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      :					int					):
											super().test_from_save_pretrained()

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

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

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

					def        _lowerCAmelCase	(			self      :					Dict					):
											lowercase       :    List[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''',
											}
											lowercase       :    List[str]											=  self.dummy_input
											return init_dict, inputs_dict

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

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

											with torch.no_grad():
																	lowercase       :    Any											=  model(lowerCAmelCase ,      lowerCAmelCase					).sample.permute(0 ,      2 ,      1					)

											lowercase       :    Any											=  output[0, -3:, -3:].flatten()
											# fmt: off
											lowercase       :    Optional[Any]											=  torch.tensor([-2.13_71_72, 1.1_42_60_16, 0.3_68_86_87, -0.76_69_22, 0.7_30_31_46, 0.11_03_88_64, -0.4_76_06_33, 0.13_27_01_72, 0.02_59_13_48]					)
											# fmt: on
											self.assertTrue(torch.allclose(lowerCAmelCase ,      lowerCAmelCase ,      rtol=1E-3					)					)

					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											# Not implemented yet for this UNet
											pass




					@slow
					def        _lowerCAmelCase	(			self      :					str					):
											lowercase       :    Tuple											=  '''harmonai/maestro-150k'''
											lowercase       :    List[Any]											=  UNetaDModel.from_pretrained(lowerCAmelCase ,      subfolder='''unet'''					)
											model.to(lowerCAmelCase					)

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

											with torch.no_grad():
																	lowercase       :    List[str]											=  model(lowerCAmelCase ,      lowerCAmelCase					).sample

											lowercase       :    str											=  output.abs().sum()
											lowercase       :    int											=  output.abs().max()

											assert (output_sum - 2_24.08_96).abs() < 0.5
											assert (output_max - 0.06_07).abs() < 4E-4







class 							UpperCAmelCase	(   __lowerCamelCase						, __lowerCamelCase						, unittest.TestCase					):
					a__: Union[str, Any]			  =				UNetaDModel
					a__: Optional[Any]			  =				"""sample"""

					@property
					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											lowercase       :    int											=  4
											lowercase       :    Tuple											=  14
											lowercase       :    str											=  16

											lowercase       :    Tuple											=  floats_tensor((batch_size, num_features, seq_len)					).to(lowerCAmelCase					)
											lowercase       :    Optional[int]											=  torch.tensor([10] * batch_size					).to(lowerCAmelCase					)

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

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

					@property
					def        _lowerCAmelCase	(			self      :					List[Any]					):
											return (4, 14, 1)

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

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

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

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

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

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

											with torch.no_grad():
																	lowercase       :    str											=  model(**lowerCAmelCase					)

																	if isinstance(lowerCAmelCase ,      lowerCAmelCase					):
																							lowercase       :    Tuple											=  output.sample

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

					def        _lowerCAmelCase	(			self      :					Optional[int]					):
											pass

					def        _lowerCAmelCase	(			self      :					List[Any]					):
											pass

					def        _lowerCAmelCase	(			self      :					Tuple					):
											lowercase       :    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''',
											}
											lowercase       :    List[Any]											=  self.dummy_input
											return init_dict, inputs_dict

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

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

											with torch.no_grad():
																	lowercase       :    List[str]											=  value_function(lowerCAmelCase ,      lowerCAmelCase					).sample

											# fmt: off
											lowercase       :    List[str]											=  torch.tensor([1_65.25] * seq_len					)
											# fmt: on
											self.assertTrue(torch.allclose(lowerCAmelCase ,      lowerCAmelCase ,      rtol=1E-3					)					)




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




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


snake_case__  							=							logging.get_logger(__name__)







class 							UpperCAmelCase	(   __lowerCamelCase					):
					a__: str			  =				["""input_values""", """attention_mask"""]

					def __init__(			self      :					Union[str, Any] ,      lowerCAmelCase      :					Tuple=1 ,      lowerCAmelCase      :					Dict=1_6000 ,      lowerCAmelCase      :					Optional[Any]=128 ,      lowerCAmelCase      :					Dict=1024 ,      lowerCAmelCase      :					List[Any]=0.0 ,      lowerCAmelCase      :					Any=True ,      lowerCAmelCase      :					Tuple=-4.2_67_73_93 ,      lowerCAmelCase      :					Dict=4.5_68_99_74 ,      lowerCAmelCase      :					int=False ,      **lowerCAmelCase      :					Optional[Any] ,      ):
											super().__init__(feature_size=lowerCAmelCase ,      sampling_rate=lowerCAmelCase ,      padding_value=lowerCAmelCase ,      **lowerCAmelCase					)
											lowercase       :    Optional[int]											=  num_mel_bins
											lowercase       :    List[Any]											=  max_length
											lowercase       :    Any											=  do_normalize
											lowercase       :    Any											=  mean
											lowercase       :    Union[str, Any]											=  std
											lowercase       :    int											=  return_attention_mask

					def        _lowerCAmelCase	(			self      :					int ,      lowerCAmelCase      :					np.ndarray ,      lowerCAmelCase      :					int ,      ):
											lowercase       :    List[Any]											=  torch.from_numpy(lowerCAmelCase					).unsqueeze(0					)
											lowercase       :    List[str]											=  ta_kaldi.fbank(
											    lowerCAmelCase ,      htk_compat=lowerCAmelCase ,      sample_frequency=self.sampling_rate ,      use_energy=lowerCAmelCase ,      window_type='''hanning''' ,      num_mel_bins=self.num_mel_bins ,      dither=0.0 ,      frame_shift=10 ,      )

											lowercase       :    Tuple											=  fbank.shape[0]
											lowercase       :    Optional[Any]											=  max_length - n_frames

											# pad or truncate, depending on difference
											if difference > 0:
																	lowercase       :    int											=  torch.nn.ZeroPadad((0, 0, 0, difference)					)
																	lowercase       :    str											=  pad_module(lowerCAmelCase					)
											elif difference < 0:
																	lowercase       :    Optional[int]											=  fbank[0:max_length, :]

											lowercase       :    List[str]											=  fbank.numpy()

											return fbank

					def        _lowerCAmelCase	(			self      :					Tuple ,      lowerCAmelCase      :					np.ndarray					):
											return (input_values - (self.mean)) / (self.std * 2)




					def __call__(			self      :					Any ,      lowerCAmelCase      :					Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]] ,      lowerCAmelCase      :					Optional[int] = None ,      lowerCAmelCase      :					Optional[Union[str, TensorType]] = None ,      **lowerCAmelCase      :					Tuple ,      ):

											if sampling_rate is not None:
																	if sampling_rate != self.sampling_rate:
																							raise ValueError(
																							    f'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
																							    f''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
																							    f''' {self.sampling_rate} and not {sampling_rate}.'''					)
											else:
																	logger.warning(
																	    '''It is strongly recommended to pass the `sampling_rate` argument to this function. '''
																	    '''Failing to do so can result in silent errors that might be hard to debug.'''					)

											lowercase       :    Optional[Any]											=  isinstance(lowerCAmelCase ,      np.ndarray					) and len(raw_speech.shape					) > 1
											if is_batched_numpy and len(raw_speech.shape					) > 2:
																	raise ValueError(f'''Only mono-channel audio is supported for input to {self}'''					)
											lowercase       :    Tuple											=  is_batched_numpy or (
											    isinstance(lowerCAmelCase ,      (list, tuple)					) and (isinstance(raw_speech[0] ,      (np.ndarray, tuple, list)					))
											)

											if is_batched:
																	lowercase       :    Optional[int]											=  [np.asarray(lowerCAmelCase ,      dtype=np.floataa					) for speech in raw_speech]
											elif not is_batched and not isinstance(lowerCAmelCase ,      np.ndarray					):
																	lowercase       :    Dict											=  np.asarray(lowerCAmelCase ,      dtype=np.floataa					)
											elif isinstance(lowerCAmelCase ,      np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
																	lowercase       :    int											=  raw_speech.astype(np.floataa					)

											# always return batch
											if not is_batched:
																	lowercase       :    Dict											=  [raw_speech]

											# extract fbank features and pad/truncate to max_length
											lowercase       :    Dict											=  [self._extract_fbank_features(lowerCAmelCase ,      max_length=self.max_length					) for waveform in raw_speech]

											# convert into BatchFeature
											lowercase       :    Any											=  BatchFeature({'''input_values''': features}					)

											# make sure list is in array format
											lowercase       :    str											=  padded_inputs.get('''input_values'''					)
											if isinstance(input_values[0] ,      lowerCAmelCase					):
																	lowercase       :    Any											=  [np.asarray(lowerCAmelCase ,      dtype=np.floataa					) for feature in input_values]

											# normalization
											if self.do_normalize:
																	lowercase       :    List[str]											=  [self.normalize(lowerCAmelCase					) for feature in input_values]

											if return_tensors is not None:
																	lowercase       :    str											=  padded_inputs.convert_to_tensors(lowerCAmelCase					)

											return padded_inputs


