code: 
  
  
  
  
  
  
  """simple docstring"""
  import os
  import tempfile
  import unittest
  
  import numpy as np
  import torch
  
  from diffusers.models import ModelMixin, UNetaDConditionModel
  from diffusers.models.attention_processor import AttnProcessor, LoRAAttnProcessor
  from diffusers.utils import (
      floats_tensor,
      logging,
      skip_mps,
      torch_device,
  )
  from diffusers.utils.import_utils import is_xformers_available
  from diffusers.utils.testing_utils import enable_full_determinism
  
  from .test_modeling_common import ModelTesterMixin, UNetTesterMixin
  
  
  enable_full_determinism()
  
  snake_case   :   List[Any]    		=		logging.get_logger(__name__)
  
  
  def        A	(							__snake_case:   str   ,    __snake_case:   Union[str, Any] = True     )						->		List[str]:
  
         """simple docstring"""
  
  
  
  
  
         __magic_name__										=						{}
         for name in model.attn_processors.keys():
                __magic_name__										=						name.endswith('attn2.processor'     ) and not (
                    name.startswith('transformer_in'     ) or "temp_attentions" in name.split('.'     )
                )
                __magic_name__										=						model.config.cross_attention_dim if has_cross_attention else None
                if name.startswith('mid_block'     ):
                       __magic_name__										=						model.config.block_out_channels[-1]
                elif name.startswith('up_blocks'     ):
                       __magic_name__										=						int(name[len('up_blocks.'     )]     )
                       __magic_name__										=						list(reversed(model.config.block_out_channels     )     )[block_id]
                elif name.startswith('down_blocks'     ):
                       __magic_name__										=						int(name[len('down_blocks.'     )]     )
                       __magic_name__										=						model.config.block_out_channels[block_id]
                elif name.startswith('transformer_in'     ):
                       # Note that the `8 * ...` comes from: https://github.com/huggingface/diffusers/blob/7139f0e874f10b2463caa8cbd585762a309d12d6/src/diffusers/models/unet_3d_condition.py#L148
                       __magic_name__										=						8 * model.config.attention_head_dim
  
                __magic_name__										=						LoRAAttnProcessor(hidden_size=__snake_case   ,    cross_attention_dim=__snake_case     )
                __magic_name__										=						lora_attn_procs[name].to(model.device     )
  
                if mock_weights:
                       # add 1 to weights to mock trained weights
                       with torch.no_grad():
                              lora_attn_procs[name].to_q_lora.up.weight += 1
                              lora_attn_procs[name].to_k_lora.up.weight += 1
                              lora_attn_procs[name].to_v_lora.up.weight += 1
                              lora_attn_procs[name].to_out_lora.up.weight += 1
  
         return lora_attn_procs
  
  
  
  
  
  
  @skip_mps
  class 					UpperCamelCase__   (							a__       ,  a__       ,  unittest.TestCase):
  
  
  
  
  
  
  
      """simple docstring"""
      __UpperCAmelCase											=   UNetaDConditionModel
      __UpperCAmelCase											=   """sample"""
  
  
  
  
  
  
  
      @property
      def 							a__ (		self   :  str   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						4
             __magic_name__										=						4
             __magic_name__										=						4
             __magic_name__										=						(3_2, 3_2)
  
             __magic_name__										=						floats_tensor((batch_size, num_channels, num_frames) + sizes   ).to(__A   )
             __magic_name__										=						torch.tensor([1_0]   ).to(__A   )
             __magic_name__										=						floats_tensor((batch_size, 4, 3_2)   ).to(__A   )
  
             return {"sample": noise, "timestep": time_step, "encoder_hidden_states": encoder_hidden_states}
  
  
  
  
  
  
  
      @property
      def 							a__ (		self   :  List[str]   ):
  
  
  
  
  
             '''simple docstring'''
  
             return (4, 4, 3_2, 3_2)
  
  
  
  
  
  
  
      @property
      def 							a__ (		self   :  List[Any]   ):
  
  
  
  
  
             '''simple docstring'''
  
             return (4, 4, 3_2, 3_2)
  
  
  
  
  
  
  
      def 							a__ (		self   :  Union[str, Any]   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						{
                 "block_out_channels": (3_2, 6_4),
                 "down_block_types": (
                     "CrossAttnDownBlock3D",
                     "DownBlock3D",
                 ),
                 "up_block_types": ("UpBlock3D", "CrossAttnUpBlock3D"),
                 "cross_attention_dim": 3_2,
                 "attention_head_dim": 8,
                 "out_channels": 4,
                 "in_channels": 4,
                 "layers_per_block": 1,
                 "sample_size": 3_2,
             }
             __magic_name__										=						self.dummy_input
             return init_dict, inputs_dict
  
  
  
  
  
  
  
      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()							,						reason='XFormers attention is only available with CUDA and `xformers` installed'							,						)
      def 							a__ (		self   :  List[str]   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
             __magic_name__										=						self.model_class(**__A   )
  
             model.enable_xformers_memory_efficient_attention()
  
             assert (
                 model.mid_block.attentions[0].transformer_blocks[0].attna.processor.__class__.__name__
                 == "XFormersAttnProcessor"
             ), "xformers is not enabled"
  
  
  
  
  
  
  
      def 							a__ (		self   :  Dict   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						3_2
  
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
             model.eval()
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   )
  
                    if isinstance(__A							,						__A   ):
                           __magic_name__										=						output.sample
  
             self.assertIsNotNone(__A   )
             __magic_name__										=						inputs_dict["sample"].shape
             self.assertEqual(output.shape							,						__A							,						'Input and output shapes do not match'   )
  
  
  
  
  
  
  
      def 							a__ (		self   :  str   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
             model.eval()
  
             with torch.no_grad():
                    # Warmup pass when using mps (see #372)
                    if torch_device == "mps" and isinstance(__A							,						__A   ):
                           model(**self.dummy_input   )
  
                    __magic_name__										=						model(**__A   )
                    if isinstance(__A							,						__A   ):
                           __magic_name__										=						first.sample
  
                    __magic_name__										=						model(**__A   )
                    if isinstance(__A							,						__A   ):
                           __magic_name__										=						second.sample
  
             __magic_name__										=						first.cpu().numpy()
             __magic_name__										=						second.cpu().numpy()
             __magic_name__										=						out_a[~np.isnan(__A   )]
             __magic_name__										=						out_a[~np.isnan(__A   )]
             __magic_name__										=						np.amax(np.abs(out_a - out_a   )   )
             self.assertLessEqual(__A							,						1e-5   )
  
  
  
  
  
  
  
      def 							a__ (		self   :  Union[str, Any]   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
             model.eval()
  
             model.set_attention_slice('auto'   )
             with torch.no_grad():
                    __magic_name__										=						model(**__A   )
             assert output is not None
  
             model.set_attention_slice('max'   )
             with torch.no_grad():
                    __magic_name__										=						model(**__A   )
             assert output is not None
  
             model.set_attention_slice(2   )
             with torch.no_grad():
                    __magic_name__										=						model(**__A   )
             assert output is not None
  
  
  
  
  
  
  
      def 							a__ (		self   :  Optional[Any]   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   ).sample
  
             __magic_name__										=						create_lora_layers(__A   )
  
             # make sure we can set a list of attention processors
             model.set_attn_processor(__A   )
             model.to(__A   )
  
             # test that attn processors can be set to itself
             model.set_attn_processor(model.attn_processors   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A							,						cross_attention_kwargs={'scale': 0.0}   ).sample
                    __magic_name__										=						model(**__A							,						cross_attention_kwargs={'scale': 0.5}   ).sample
                    __magic_name__										=						model(**__A							,						cross_attention_kwargs={'scale': 0.5}   ).sample
  
             assert (samplea - samplea).abs().max() < 3e-3
             assert (samplea - samplea).abs().max() < 3e-3
  
             # sample 2 and sample 3 should be different
             assert (samplea - samplea).abs().max() > 3e-3
  
  
  
  
  
  
  
      def 							a__ (		self   :  Dict   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             torch.manual_seed(0   )
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   ).sample
  
             __magic_name__										=						create_lora_layers(__A   )
             model.set_attn_processor(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A							,						cross_attention_kwargs={'scale': 0.5}   ).sample
  
             with tempfile.TemporaryDirectory() as tmpdirname:
                    model.save_attn_procs(__A   )
                    self.assertTrue(os.path.isfile(os.path.join(__A							,						'pytorch_lora_weights.bin'   )   )   )
                    torch.manual_seed(0   )
                    __magic_name__										=						self.model_class(**__A   )
                    new_model.to(__A   )
                    new_model.load_attn_procs(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						new_model(**__A							,						cross_attention_kwargs={'scale': 0.5}   ).sample
  
             assert (sample - new_sample).abs().max() < 1e-3
  
             # LoRA and no LoRA should NOT be the same
             assert (sample - old_sample).abs().max() > 1e-4
  
  
  
  
  
  
  
      def 							a__ (		self   :  str   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             torch.manual_seed(0   )
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   ).sample
  
             __magic_name__										=						create_lora_layers(__A   )
             model.set_attn_processor(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A							,						cross_attention_kwargs={'scale': 0.5}   ).sample
  
             with tempfile.TemporaryDirectory() as tmpdirname:
                    model.save_attn_procs(__A							,						safe_serialization=__A   )
                    self.assertTrue(os.path.isfile(os.path.join(__A							,						'pytorch_lora_weights.safetensors'   )   )   )
                    torch.manual_seed(0   )
                    __magic_name__										=						self.model_class(**__A   )
                    new_model.to(__A   )
                    new_model.load_attn_procs(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						new_model(**__A							,						cross_attention_kwargs={'scale': 0.5}   ).sample
  
             assert (sample - new_sample).abs().max() < 3e-3
  
             # LoRA and no LoRA should NOT be the same
             assert (sample - old_sample).abs().max() > 1e-4
  
  
  
  
  
  
  
      def 							a__ (		self   :  Optional[int]   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             torch.manual_seed(0   )
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
  
             __magic_name__										=						create_lora_layers(__A							,						mock_weights=__A   )
             model.set_attn_processor(__A   )
             # Saving as torch, properly reloads with directly filename
             with tempfile.TemporaryDirectory() as tmpdirname:
                    model.save_attn_procs(__A   )
                    self.assertTrue(os.path.isfile(os.path.join(__A							,						'pytorch_lora_weights.bin'   )   )   )
                    torch.manual_seed(0   )
                    __magic_name__										=						self.model_class(**__A   )
                    new_model.to(__A   )
                    new_model.load_attn_procs(__A							,						weight_name='pytorch_lora_weights.bin'   )
  
  
  
  
  
  
  
      def 							a__ (		self   :  Dict   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             torch.manual_seed(0   )
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
  
             __magic_name__										=						create_lora_layers(__A							,						mock_weights=__A   )
             model.set_attn_processor(__A   )
             # Saving as torch, properly reloads with directly filename
             with tempfile.TemporaryDirectory() as tmpdirname:
                    model.save_attn_procs(__A   )
                    self.assertTrue(os.path.isfile(os.path.join(__A							,						'pytorch_lora_weights.bin'   )   )   )
                    torch.manual_seed(0   )
                    __magic_name__										=						self.model_class(**__A   )
                    new_model.to(__A   )
                    with self.assertRaises(__A   ) as e:
                           new_model.load_attn_procs(__A							,						use_safetensors=__A   )
                    self.assertIn('Error no file named pytorch_lora_weights.safetensors'							,						str(e.exception   )   )
  
  
  
  
  
  
  
      def 							a__ (		self   :  str   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						8
  
             torch.manual_seed(0   )
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   ).sample
  
             __magic_name__										=						create_lora_layers(__A   )
             model.set_attn_processor(__A   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A							,						cross_attention_kwargs={'scale': 0.0}   ).sample
  
             model.set_attn_processor(AttnProcessor()   )
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   ).sample
  
             assert (sample - new_sample).abs().max() < 1e-4
             assert (sample - old_sample).abs().max() < 3e-3
  
  
  
  
  
  
  
      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()							,						reason='XFormers attention is only available with CUDA and `xformers` installed'							,						)
      def 							a__ (		self   :  Any   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
  
             __magic_name__										=						4
  
             torch.manual_seed(0   )
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
             __magic_name__										=						create_lora_layers(__A   )
             model.set_attn_processor(__A   )
  
             # default
             with torch.no_grad():
                    __magic_name__										=						model(**__A   ).sample
  
                    model.enable_xformers_memory_efficient_attention()
                    __magic_name__										=						model(**__A   ).sample
  
                    model.disable_xformers_memory_efficient_attention()
                    __magic_name__										=						model(**__A   ).sample
  
             assert (sample - on_sample).abs().max() < 1e-4
             assert (sample - off_sample).abs().max() < 1e-4
  
  
  
  
  
  
  
      def 							a__ (		self   :  str   ):
  
  
  
  
  
             '''simple docstring'''
  
             __magic_name__										=						self.prepare_init_args_and_inputs_for_common()
             __magic_name__										=						3_2
  
             __magic_name__										=						self.model_class(**__A   )
             model.to(__A   )
             model.eval()
  
             with torch.no_grad():
                    __magic_name__										=						model(**__A   )[0]
  
             model.enable_forward_chunking()
             with torch.no_grad():
                    __magic_name__										=						model(**__A   )[0]
  
             self.assertEqual(output.shape							,						output_a.shape							,						'Shape doesn\'t match'   )
             assert np.abs(output.cpu() - output_a.cpu()   ).max() < 1e-2
  
  
  # (todo: sayakpaul) implement SLOW tests.
  
  
style_context: 
  
  
  
  
  
  
  """simple docstring"""
  from typing import List, Optional, Union
  
  import numpy as np
  
  from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
  from ...feature_extraction_utils import BatchFeature
  from ...utils import PaddingStrategy, TensorType, logging
  
  
  snake_case   :   Tuple    		=		logging.get_logger(__name__)
  
  
  
  class 					UpperCamelCase__   (							a_):
  
  
  
  
  
  
  
      """simple docstring"""
      __UpperCAmelCase											=   ["""input_values""", """attention_mask"""]
  
  
  
  
  
  
  
      def __init__(		self   :  Union[str, Any]							,						UpperCamelCase_   :  int=1							,						UpperCamelCase_   :  Optional[int]=1_6_0_0_0							,						UpperCamelCase_   :  List[str]=0.0							,						UpperCamelCase_   :  Optional[int]=False							,						UpperCamelCase_   :  Tuple=True							,						**UpperCamelCase_   :  Tuple							,						):
  
  
  
  
  
             '''simple docstring'''
  
             super().__init__(feature_size=UpperCamelCase_							,						sampling_rate=UpperCamelCase_							,						padding_value=UpperCamelCase_							,						**UpperCamelCase_   )
             __magic_name__										=						return_attention_mask
             __magic_name__										=						do_normalize
  
  
  
  
  
  
  
      @staticmethod
      def 							a__ (		UpperCamelCase_   :  List[np.ndarray]							,						UpperCamelCase_   :  List[np.ndarray]							,						UpperCamelCase_   :  float = 0.0   ):
  
  
  
  
  
             '''simple docstring'''
  
             if attention_mask is not None:
                    __magic_name__										=						np.array(UpperCamelCase_							,						np.intaa   )
                    __magic_name__										=						[]
  
                    for vector, length in zip(UpperCamelCase_							,						attention_mask.sum(-1   )   ):
                           __magic_name__										=						(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7   )
                           if length < normed_slice.shape[0]:
                                  __magic_name__										=						padding_value
  
                           normed_input_values.append(UpperCamelCase_   )
             else:
                    __magic_name__										=						[(x - x.mean()) / np.sqrt(x.var() + 1e-7   ) for x in input_values]
  
             return normed_input_values
  
  
  
  
  
  
  
      def __call__(		self   :  Any							,						UpperCamelCase_   :  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,						UpperCamelCase_   :  Union[bool, str, PaddingStrategy] = False							,						UpperCamelCase_   :  Optional[int] = None							,						UpperCamelCase_   :  bool = False							,						UpperCamelCase_   :  Optional[int] = None							,						UpperCamelCase_   :  Optional[bool] = None							,						UpperCamelCase_   :  Optional[Union[str, TensorType]] = None							,						UpperCamelCase_   :  Optional[int] = None							,						**UpperCamelCase_   :  Any							,						):
  
  
  
  
  
             '''simple docstring'''
  
             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.'   )
  
             __magic_name__										=						isinstance(UpperCamelCase_							,						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}"""   )
             __magic_name__										=						is_batched_numpy or (
                 isinstance(UpperCamelCase_							,						(list, tuple)   ) and (isinstance(raw_speech[0]							,						(np.ndarray, tuple, list)   ))
             )
  
             # always return batch
             if not is_batched:
                    __magic_name__										=						[raw_speech]
  
             # convert into correct format for padding
             __magic_name__										=						BatchFeature({'input_values': raw_speech}   )
  
             __magic_name__										=						self.pad(
                 UpperCamelCase_							,						padding=UpperCamelCase_							,						max_length=UpperCamelCase_							,						truncation=UpperCamelCase_							,						pad_to_multiple_of=UpperCamelCase_							,						return_attention_mask=UpperCamelCase_							,						)
  
             # convert input values to correct format
             __magic_name__										=						padded_inputs['input_values']
             if not isinstance(input_values[0]							,						np.ndarray   ):
                    __magic_name__										=						[np.asarray(UpperCamelCase_							,						dtype=np.floataa   ) for array in input_values]
             elif (
                 not isinstance(UpperCamelCase_							,						np.ndarray   )
                 and isinstance(input_values[0]							,						np.ndarray   )
                 and input_values[0].dtype is np.dtype(np.floataa   )
             ):
                    __magic_name__										=						[array.astype(np.floataa   ) for array in input_values]
             elif isinstance(UpperCamelCase_							,						np.ndarray   ) and input_values.dtype is np.dtype(np.floataa   ):
                    __magic_name__										=						input_values.astype(np.floataa   )
  
             # convert attention_mask to correct format
             __magic_name__										=						padded_inputs.get('attention_mask'   )
             if attention_mask is not None:
                    __magic_name__										=						[np.asarray(UpperCamelCase_							,						dtype=np.intaa   ) for array in attention_mask]
  
             # zero-mean and unit-variance normalization
             if self.do_normalize:
                    __magic_name__										=						(
                        attention_mask
                        if self._get_padding_strategies(UpperCamelCase_							,						max_length=UpperCamelCase_   ) is not PaddingStrategy.DO_NOT_PAD
                        else None
                    )
                    __magic_name__										=						self.zero_mean_unit_var_norm(
                        padded_inputs['input_values']							,						attention_mask=UpperCamelCase_							,						padding_value=self.padding_value   )
  
             if return_tensors is not None:
                    __magic_name__										=						padded_inputs.convert_to_tensors(UpperCamelCase_   )
  
             return padded_inputs