


'''simple docstring'''






import gc
import unittest

import numpy as np
import torch

from diffusers import (
    AudioDiffusionPipeline,
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    DiffusionPipeline,
    Mel,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()




class 	a       (       unittest.TestCase    ):
 def 							__UpperCAmelCase						(	self  )   ->							List[Any]:
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()
 @property
 def 							__UpperCAmelCase						(	self  )   ->							Any:
      torch.manual_seed(0  )
      _a      =							UNetaDModel(
          sample_size=(32, 64) ,			in_channels=1 ,			out_channels=1 ,			layers_per_block=2 ,			block_out_channels=(1_28, 1_28) ,			down_block_types=('AttnDownBlock2D', 'DownBlock2D') ,			up_block_types=('UpBlock2D', 'AttnUpBlock2D') ,			)
      return model
 @property
 def 							__UpperCAmelCase						(	self  )   ->							Tuple:
      torch.manual_seed(0  )
      _a      =							UNetaDConditionModel(
          sample_size=(64, 32) ,			in_channels=1 ,			out_channels=1 ,			layers_per_block=2 ,			block_out_channels=(1_28, 1_28) ,			down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D') ,			up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D') ,			cross_attention_dim=10 ,			)
      return model
 @property
 def 							__UpperCAmelCase						(	self  )   ->							Optional[Any]:
      torch.manual_seed(0  )
      _a      =							AutoencoderKL(
          sample_size=(1_28, 64) ,			in_channels=1 ,			out_channels=1 ,			latent_channels=1 ,			layers_per_block=2 ,			block_out_channels=(1_28, 1_28) ,			down_block_types=('DownEncoderBlock2D', 'DownEncoderBlock2D') ,			up_block_types=('UpDecoderBlock2D', 'UpDecoderBlock2D') ,			)
      _a      =							UNetaDModel(
          sample_size=(64, 32) ,			in_channels=1 ,			out_channels=1 ,			layers_per_block=2 ,			block_out_channels=(1_28, 1_28) ,			down_block_types=('AttnDownBlock2D', 'DownBlock2D') ,			up_block_types=('UpBlock2D', 'AttnUpBlock2D') ,			)
      return vqvae, unet




 @slow
 def 							__UpperCAmelCase						(	self  )   ->							Tuple:
      _a      =							'cpu'  # ensure determinism for the device-dependent torch.Generator
      _a      =							Mel(
          x_res=self.dummy_unet.config.sample_size[1] ,			y_res=self.dummy_unet.config.sample_size[0] ,			)

      _a      =							DDPMScheduler()
      _a      =							AudioDiffusionPipeline(vqvae=__magic_name__ ,			unet=self.dummy_unet ,			mel=__magic_name__ ,			scheduler=__magic_name__  )
      _a      =							pipe.to(__magic_name__  )
      pipe.set_progress_bar_config(disable=__magic_name__  )

      _a      =							torch.Generator(device=__magic_name__  ).manual_seed(42  )
      _a      =							pipe(generator=__magic_name__ ,			steps=4  )
      _a      =							output.audios[0]
      _a      =							output.images[0]

      _a      =							torch.Generator(device=__magic_name__  ).manual_seed(42  )
      _a      =							pipe(generator=__magic_name__ ,			steps=4 ,			return_dict=__magic_name__  )
      _a      =							output[0][0]

      assert audio.shape == (1, (self.dummy_unet.config.sample_size[1] - 1) * mel.hop_length)
      assert (
          image.height == self.dummy_unet.config.sample_size[0]
          and image.width == self.dummy_unet.config.sample_size[1]
      )
      _a      =							np.frombuffer(image.tobytes() ,			dtype='uint8'  )[:10]
      _a      =							np.frombuffer(image_from_tuple.tobytes() ,			dtype='uint8'  )[:10]
      _a      =							np.array([69, 2_55, 2_55, 2_55, 0, 0, 77, 1_81, 12, 1_27]  )

      assert np.abs(image_slice.flatten() - expected_slice  ).max() == 0
      assert np.abs(image_from_tuple_slice.flatten() - expected_slice  ).max() == 0

      _a      =							Mel(
          x_res=self.dummy_vqvae_and_unet[0].config.sample_size[1] ,			y_res=self.dummy_vqvae_and_unet[0].config.sample_size[0] ,			)

      _a      =							DDIMScheduler()
      _a      =							self.dummy_vqvae_and_unet
      _a      =							AudioDiffusionPipeline(
          vqvae=self.dummy_vqvae_and_unet[0] ,			unet=dummy_vqvae_and_unet[1] ,			mel=__magic_name__ ,			scheduler=__magic_name__  )
      _a      =							pipe.to(__magic_name__  )
      pipe.set_progress_bar_config(disable=__magic_name__  )

      np.random.seed(0  )
      _a      =							np.random.uniform(-1 ,			1 ,			((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)  )
      _a      =							torch.Generator(device=__magic_name__  ).manual_seed(42  )
      _a      =							pipe(raw_audio=__magic_name__ ,			generator=__magic_name__ ,			start_step=5 ,			steps=10  )
      _a      =							output.images[0]

      assert (
          image.height == self.dummy_vqvae_and_unet[0].config.sample_size[0]
          and image.width == self.dummy_vqvae_and_unet[0].config.sample_size[1]
      )
      _a      =							np.frombuffer(image.tobytes() ,			dtype='uint8'  )[:10]
      _a      =							np.array([1_20, 1_17, 1_10, 1_09, 1_38, 1_67, 1_38, 1_48, 1_32, 1_21]  )

      assert np.abs(image_slice.flatten() - expected_slice  ).max() == 0

      _a      =							self.dummy_unet_condition
      _a      =							AudioDiffusionPipeline(
          vqvae=self.dummy_vqvae_and_unet[0] ,			unet=__magic_name__ ,			mel=__magic_name__ ,			scheduler=__magic_name__  )
      _a      =							pipe.to(__magic_name__  )
      pipe.set_progress_bar_config(disable=__magic_name__  )

      np.random.seed(0  )
      _a      =							torch.rand((1, 1, 10)  )
      _a      =							pipe(generator=__magic_name__ ,			encoding=__magic_name__  )
      _a      =							output.images[0]
      _a      =							np.frombuffer(image.tobytes() ,			dtype='uint8'  )[:10]
      _a      =							np.array([1_07, 1_03, 1_20, 1_27, 1_42, 1_22, 1_13, 1_22, 97, 1_11]  )

      assert np.abs(image_slice.flatten() - expected_slice  ).max() == 0





@slow
@require_torch_gpu
class 	a       (       unittest.TestCase    ):
 def 							__UpperCAmelCase						(	self  )   ->							Any:
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()




 def 							__UpperCAmelCase						(	self  )   ->							Dict:
      _a      =							torch_device

      _a      =							DiffusionPipeline.from_pretrained('teticio/audio-diffusion-ddim-256'  )
      _a      =							pipe.to(__magic_name__  )
      pipe.set_progress_bar_config(disable=__magic_name__  )

      _a      =							torch.Generator(device=__magic_name__  ).manual_seed(42  )
      _a      =							pipe(generator=__magic_name__  )
      _a      =							output.audios[0]
      _a      =							output.images[0]

      assert audio.shape == (1, (pipe.unet.config.sample_size[1] - 1) * pipe.mel.hop_length)
      assert image.height == pipe.unet.config.sample_size[0] and image.width == pipe.unet.config.sample_size[1]
      _a      =							np.frombuffer(image.tobytes() ,			dtype='uint8'  )[:10]
      _a      =							np.array([1_51, 1_67, 1_54, 1_44, 1_22, 1_34, 1_21, 1_05, 70, 26]  )

      assert np.abs(image_slice.flatten() - expected_slice  ).max() == 0





'''simple docstring'''






import unicodedata
from dataclasses import dataclass
from typing import Optional, Union

import numpy as np

from transformers.data.data_collator import DataCollatorMixin
from transformers.file_utils import PaddingStrategy
from transformers.tokenization_utils_base import PreTrainedTokenizerBase






def 		_A			(lowerCAmelCase__					:int			,  lowerCAmelCase__					:Any			,  lowerCAmelCase__					:Optional[Any]			,  lowerCAmelCase__					:Dict						)				->					Optional[int]:




     '''simple docstring'''



     if isinstance(lowerCAmelCase__			,  lowerCAmelCase__						):
          _a      =							np.full((len(lowerCAmelCase__						), sequence_length, 2)			,  lowerCAmelCase__						)
     else:
          _a      =							np.full((len(lowerCAmelCase__						), sequence_length)			,  lowerCAmelCase__						)

     for i, tensor in enumerate(lowerCAmelCase__						):
          if padding_side == "right":
               if isinstance(lowerCAmelCase__			,  lowerCAmelCase__						):
                    _a      =							tensor[:sequence_length]
               else:
                    _a      =							tensor[:sequence_length]
          else:
               if isinstance(lowerCAmelCase__			,  lowerCAmelCase__						):
                    _a      =							tensor[:sequence_length]
               else:
                    _a      =							tensor[:sequence_length]

     return out_tensor.tolist()






def 		_A			(lowerCAmelCase__					:Any						)				->					Union[str, Any]:




     '''simple docstring'''



     _a      =							ord(lowerCAmelCase__						)
     if (cp >= 33 and cp <= 47) or (cp >= 58 and cp <= 64) or (cp >= 91 and cp <= 96) or (cp >= 1_23 and cp <= 1_26):
          return True
     _a      =							unicodedata.category(lowerCAmelCase__						)
     if cat.startswith('P'						):
          return True
     return False






@dataclass
class 	a       (       _SCREAMING_SNAKE_CASE    ):
 _lowerCAmelCase	   =		42
 _lowerCAmelCase	   =		True
 _lowerCAmelCase	   =		None
 _lowerCAmelCase	   =		None
 _lowerCAmelCase	   =		-1_0_0
 _lowerCAmelCase	   =		"pt"
 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							Any:
      import torch

      _a      =							'label' if 'label' in features[0].keys() else 'labels'
      _a      =							[feature[label_name] for feature in features] if label_name in features[0].keys() else None
      _a      =							self.tokenizer.pad(
          __magic_name__ ,			padding=self.padding ,			max_length=self.max_length ,			pad_to_multiple_of=self.pad_to_multiple_of ,			return_tensors='pt' if labels is None else None ,			)

      if labels is None:
           return batch

      _a      =							torch.tensor(batch['entity_ids']  ).shape[1]
      _a      =							self.tokenizer.padding_side
      if padding_side == "right":
           _a      =							[
               list(__magic_name__  ) + [self.label_pad_token_id] * (sequence_length - len(__magic_name__  )) for label in labels
           ]
      else:
           _a      =							[
               [self.label_pad_token_id] * (sequence_length - len(__magic_name__  )) + list(__magic_name__  ) for label in labels
           ]

      _a      =							[feature['ner_tags'] for feature in features]
      _a      =							padding_tensor(__magic_name__ ,			-1 ,			__magic_name__ ,			__magic_name__  )
      _a      =							[feature['original_entity_spans'] for feature in features]
      _a      =							padding_tensor(__magic_name__ ,			(-1, -1) ,			__magic_name__ ,			__magic_name__  )
      _a      =							{k: torch.tensor(__magic_name__ ,			dtype=torch.intaa  ) for k, v in batch.items()}

      return batch

