'''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       UpperCAmelCase__  (       unittest.TestCase			):






   """simple docstring"""
   def 		_lowerCamelCase (self				)    ->     Optional[int]:
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()
   @property
   def 		_lowerCamelCase (self				)    ->     Optional[Any]:
        torch.manual_seed(0				)
        lowercase_					:       List[str]     				=					UNetaDModel(
            sample_size=(32, 64)       ,				in_channels=1       ,				out_channels=1       ,				layers_per_block=2       ,				block_out_channels=(128, 128)       ,				down_block_types=('AttnDownBlock2D', 'DownBlock2D')       ,				up_block_types=('UpBlock2D', 'AttnUpBlock2D')       ,				)
        return model
   @property
   def 		_lowerCamelCase (self				)    ->     Dict:
        torch.manual_seed(0				)
        lowercase_					:       Optional[Any]     				=					UNetaDConditionModel(
            sample_size=(64, 32)       ,				in_channels=1       ,				out_channels=1       ,				layers_per_block=2       ,				block_out_channels=(128, 128)       ,				down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D')       ,				up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D')       ,				cross_attention_dim=10       ,				)
        return model
   @property
   def 		_lowerCamelCase (self				)    ->     List[Any]:
        torch.manual_seed(0				)
        lowercase_					:       int     				=					AutoencoderKL(
            sample_size=(128, 64)       ,				in_channels=1       ,				out_channels=1       ,				latent_channels=1       ,				layers_per_block=2       ,				block_out_channels=(128, 128)       ,				down_block_types=('DownEncoderBlock2D', 'DownEncoderBlock2D')       ,				up_block_types=('UpDecoderBlock2D', 'UpDecoderBlock2D')       ,				)
        lowercase_					:       Dict     				=					UNetaDModel(
            sample_size=(64, 32)       ,				in_channels=1       ,				out_channels=1       ,				layers_per_block=2       ,				block_out_channels=(128, 128)       ,				down_block_types=('AttnDownBlock2D', 'DownBlock2D')       ,				up_block_types=('UpBlock2D', 'AttnUpBlock2D')       ,				)
        return vqvae, unet





   @slow
   def 		_lowerCamelCase (self				)    ->     Union[str, Any]:
        lowercase_					:       str     				=					'cpu'  # ensure determinism for the device-dependent torch.Generator
        lowercase_					:       Optional[int]     				=					Mel(
            x_res=self.dummy_unet.config.sample_size[1]       ,				y_res=self.dummy_unet.config.sample_size[0]       ,				)

        lowercase_					:       Any     				=					DDPMScheduler()
        lowercase_					:       Optional[Any]     				=					AudioDiffusionPipeline(vqvae=_a       ,				unet=self.dummy_unet       ,				mel=_a       ,				scheduler=_a				)
        lowercase_					:       Union[str, Any]     				=					pipe.to(_a				)
        pipe.set_progress_bar_config(disable=_a				)

        lowercase_					:       Dict     				=					torch.Generator(device=_a				).manual_seed(42				)
        lowercase_					:       Optional[int]     				=					pipe(generator=_a       ,				steps=4				)
        lowercase_					:       Dict     				=					output.audios[0]
        lowercase_					:       List[Any]     				=					output.images[0]

        lowercase_					:       Dict     				=					torch.Generator(device=_a				).manual_seed(42				)
        lowercase_					:       Tuple     				=					pipe(generator=_a       ,				steps=4       ,				return_dict=_a				)
        lowercase_					:       List[Any]     				=					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]
        )
        lowercase_					:       str     				=					np.frombuffer(image.tobytes()       ,				dtype='uint8'				)[:10]
        lowercase_					:       List[Any]     				=					np.frombuffer(image_from_tuple.tobytes()       ,				dtype='uint8'				)[:10]
        lowercase_					:       int     				=					np.array([69, 255, 255, 255, 0, 0, 77, 181, 12, 127]				)

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

        lowercase_					:       List[Any]     				=					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]       ,				)

        lowercase_					:       Optional[int]     				=					DDIMScheduler()
        lowercase_					:       Any     				=					self.dummy_vqvae_and_unet
        lowercase_					:       int     				=					AudioDiffusionPipeline(
            vqvae=self.dummy_vqvae_and_unet[0]       ,				unet=dummy_vqvae_and_unet[1]       ,				mel=_a       ,				scheduler=_a				)
        lowercase_					:       str     				=					pipe.to(_a				)
        pipe.set_progress_bar_config(disable=_a				)

        np.random.seed(0				)
        lowercase_					:       Optional[int]     				=					np.random.uniform(-1       ,				1       ,				((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)				)
        lowercase_					:       Union[str, Any]     				=					torch.Generator(device=_a				).manual_seed(42				)
        lowercase_					:       int     				=					pipe(raw_audio=_a       ,				generator=_a       ,				start_step=5       ,				steps=10				)
        lowercase_					:       List[Any]     				=					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]
        )
        lowercase_					:       Union[str, Any]     				=					np.frombuffer(image.tobytes()       ,				dtype='uint8'				)[:10]
        lowercase_					:       Optional[int]     				=					np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121]				)

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

        lowercase_					:       Any     				=					self.dummy_unet_condition
        lowercase_					:       Optional[Any]     				=					AudioDiffusionPipeline(
            vqvae=self.dummy_vqvae_and_unet[0]       ,				unet=_a       ,				mel=_a       ,				scheduler=_a				)
        lowercase_					:       int     				=					pipe.to(_a				)
        pipe.set_progress_bar_config(disable=_a				)

        np.random.seed(0				)
        lowercase_					:       List[str]     				=					torch.rand((1, 1, 10)				)
        lowercase_					:       int     				=					pipe(generator=_a       ,				encoding=_a				)
        lowercase_					:       List[str]     				=					output.images[0]
        lowercase_					:       Optional[Any]     				=					np.frombuffer(image.tobytes()       ,				dtype='uint8'				)[:10]
        lowercase_					:       Optional[int]     				=					np.array([107, 103, 120, 127, 142, 122, 113, 122, 97, 111]				)

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



@slow
@require_torch_gpu
class       UpperCAmelCase__  (       unittest.TestCase			):






   """simple docstring"""
   def 		_lowerCamelCase (self				)    ->     Any:
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()





   def 		_lowerCamelCase (self				)    ->     int:
        lowercase_					:       str     				=					torch_device

        lowercase_					:       Tuple     				=					DiffusionPipeline.from_pretrained('teticio/audio-diffusion-ddim-256'				)
        lowercase_					:       List[str]     				=					pipe.to(_a				)
        pipe.set_progress_bar_config(disable=_a				)

        lowercase_					:       Dict     				=					torch.Generator(device=_a				).manual_seed(42				)
        lowercase_					:       Optional[int]     				=					pipe(generator=_a				)
        lowercase_					:       List[Any]     				=					output.audios[0]
        lowercase_					:       Tuple     				=					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]
        lowercase_					:       int     				=					np.frombuffer(image.tobytes()       ,				dtype='uint8'				)[:10]
        lowercase_					:       Tuple     				=					np.array([151, 167, 154, 144, 122, 134, 121, 105, 70, 26]				)

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





'''simple docstring'''




import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
 from PIL import Image

 from transformers import AlignProcessor, EfficientNetImageProcessor


@require_vision
class       UpperCAmelCase__  (       unittest.TestCase			):






   """simple docstring"""
   def 		_lowerCamelCase (self				)    ->     List[Any]:
        lowercase_					:       Any     				=					tempfile.mkdtemp()

        lowercase_					:       int     				=					[
            '[UNK]',
            '[CLS]',
            '[SEP]',
            '[PAD]',
            '[MASK]',
            'want',
            '##want',
            '##ed',
            'wa',
            'un',
            'runn',
            '##ing',
            ',',
            'low',
            'lowest',
        ]
        lowercase_					:       Union[str, Any]     				=					os.path.join(self.tmpdirname       ,				VOCAB_FILES_NAMES['vocab_file']				)
        with open(self.vocab_file       ,				'w'       ,				encoding='utf-8'				) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]				)				)

        lowercase_					:       Union[str, Any]     				=					{
            'do_resize': True,
            'size': 20,
            'do_center_crop': True,
            'crop_size': 18,
            'do_normalize': True,
            'image_mean': [0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73],
            'image_std': [0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11],
        }
        lowercase_					:       List[str]     				=					os.path.join(self.tmpdirname       ,				_a				)
        with open(self.image_processor_file       ,				'w'       ,				encoding='utf-8'				) as fp:
             json.dump(_a       ,				_a				)
   def 		_lowerCamelCase (self       ,				**_a				)    ->     Tuple:
        return BertTokenizer.from_pretrained(self.tmpdirname       ,				**_a				)
   def 		_lowerCamelCase (self       ,				**_a				)    ->     Tuple:
        return BertTokenizerFast.from_pretrained(self.tmpdirname       ,				**_a				)
   def 		_lowerCamelCase (self       ,				**_a				)    ->     int:
        return EfficientNetImageProcessor.from_pretrained(self.tmpdirname       ,				**_a				)
   def 		_lowerCamelCase (self				)    ->     Dict:
        shutil.rmtree(self.tmpdirname				)
   def 		_lowerCamelCase (self				)    ->     str:
        lowercase_					:       List[str]     				=					[np.random.randint(255       ,				size=(3, 30, 400)       ,				dtype=np.uinta				)]
        lowercase_					:       Optional[Any]     				=					[Image.fromarray(np.moveaxis(_a       ,				0       ,				-1				)				) for x in image_inputs]
        return image_inputs
   def 		_lowerCamelCase (self				)    ->     Optional[Any]:
        lowercase_					:       str     				=					self.get_tokenizer()
        lowercase_					:       Tuple     				=					self.get_rust_tokenizer()
        lowercase_					:       int     				=					self.get_image_processor()

        lowercase_					:       int     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)
        processor_slow.save_pretrained(self.tmpdirname				)
        lowercase_					:       List[str]     				=					AlignProcessor.from_pretrained(self.tmpdirname       ,				use_fast=_a				)

        lowercase_					:       Optional[Any]     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)
        processor_fast.save_pretrained(self.tmpdirname				)
        lowercase_					:       Any     				=					AlignProcessor.from_pretrained(self.tmpdirname				)

        self.assertEqual(processor_slow.tokenizer.get_vocab()       ,				tokenizer_slow.get_vocab()				)
        self.assertEqual(processor_fast.tokenizer.get_vocab()       ,				tokenizer_fast.get_vocab()				)
        self.assertEqual(tokenizer_slow.get_vocab()       ,				tokenizer_fast.get_vocab()				)
        self.assertIsInstance(processor_slow.tokenizer       ,				_a				)
        self.assertIsInstance(processor_fast.tokenizer       ,				_a				)

        self.assertEqual(processor_slow.image_processor.to_json_string()       ,				image_processor.to_json_string()				)
        self.assertEqual(processor_fast.image_processor.to_json_string()       ,				image_processor.to_json_string()				)
        self.assertIsInstance(processor_slow.image_processor       ,				_a				)
        self.assertIsInstance(processor_fast.image_processor       ,				_a				)
   def 		_lowerCamelCase (self				)    ->     Tuple:
        lowercase_					:       str     				=					AlignProcessor(tokenizer=self.get_tokenizer()       ,				image_processor=self.get_image_processor()				)
        processor.save_pretrained(self.tmpdirname				)

        lowercase_					:       Tuple     				=					self.get_tokenizer(bos_token='(BOS)'       ,				eos_token='(EOS)'				)
        lowercase_					:       str     				=					self.get_image_processor(do_normalize=_a       ,				padding_value=1.0				)

        lowercase_					:       Optional[Any]     				=					AlignProcessor.from_pretrained(
            self.tmpdirname       ,				bos_token='(BOS)'       ,				eos_token='(EOS)'       ,				do_normalize=_a       ,				padding_value=1.0				)

        self.assertEqual(processor.tokenizer.get_vocab()       ,				tokenizer_add_kwargs.get_vocab()				)
        self.assertIsInstance(processor.tokenizer       ,				_a				)

        self.assertEqual(processor.image_processor.to_json_string()       ,				image_processor_add_kwargs.to_json_string()				)
        self.assertIsInstance(processor.image_processor       ,				_a				)
   def 		_lowerCamelCase (self				)    ->     List[str]:
        lowercase_					:       Union[str, Any]     				=					self.get_image_processor()
        lowercase_					:       Union[str, Any]     				=					self.get_tokenizer()

        lowercase_					:       Optional[int]     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)

        lowercase_					:       Union[str, Any]     				=					self.prepare_image_inputs()

        lowercase_					:       Union[str, Any]     				=					image_processor(_a       ,				return_tensors='np'				)
        lowercase_					:       Union[str, Any]     				=					processor(images=_a       ,				return_tensors='np'				)

        for key in input_image_proc.keys():
             self.assertAlmostEqual(input_image_proc[key].sum()       ,				input_processor[key].sum()       ,				delta=1e-2				)
   def 		_lowerCamelCase (self				)    ->     Dict:
        lowercase_					:       List[Any]     				=					self.get_image_processor()
        lowercase_					:       int     				=					self.get_tokenizer()

        lowercase_					:       Any     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)

        lowercase_					:       Tuple     				=					'lower newer'

        lowercase_					:       Optional[Any]     				=					processor(text=_a				)

        lowercase_					:       List[Any]     				=					tokenizer(_a       ,				padding='max_length'       ,				max_length=64				)

        for key in encoded_tok.keys():
             self.assertListEqual(encoded_tok[key]       ,				encoded_processor[key]				)
   def 		_lowerCamelCase (self				)    ->     List[Any]:
        lowercase_					:       Any     				=					self.get_image_processor()
        lowercase_					:       List[Any]     				=					self.get_tokenizer()

        lowercase_					:       List[str]     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)

        lowercase_					:       Optional[Any]     				=					'lower newer'
        lowercase_					:       List[Any]     				=					self.prepare_image_inputs()

        lowercase_					:       List[str]     				=					processor(text=_a       ,				images=_a				)

        self.assertListEqual(list(inputs.keys()				)       ,				['input_ids', 'token_type_ids', 'attention_mask', 'pixel_values']				)

        # test if it raises when no input is passed
        with pytest.raises(_a				):
             processor()
   def 		_lowerCamelCase (self				)    ->     Tuple:
        lowercase_					:       Dict     				=					self.get_image_processor()
        lowercase_					:       Optional[Any]     				=					self.get_tokenizer()

        lowercase_					:       Optional[Any]     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)

        lowercase_					:       Union[str, Any]     				=					[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

        lowercase_					:       str     				=					processor.batch_decode(_a				)
        lowercase_					:       Dict     				=					tokenizer.batch_decode(_a				)

        self.assertListEqual(_a       ,				_a				)





   def 		_lowerCamelCase (self				)    ->     List[str]:
        lowercase_					:       str     				=					self.get_image_processor()
        lowercase_					:       str     				=					self.get_tokenizer()

        lowercase_					:       str     				=					AlignProcessor(tokenizer=_a       ,				image_processor=_a				)

        lowercase_					:       Optional[int]     				=					'lower newer'
        lowercase_					:       Union[str, Any]     				=					self.prepare_image_inputs()

        lowercase_					:       Union[str, Any]     				=					processor(text=_a       ,				images=_a				)

        self.assertListEqual(list(inputs.keys()				)       ,				processor.model_input_names				)




