





import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaInpaintPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()





class UpperCAmelCase_				( _UpperCamelCase     ,		unittest.TestCase ):
  __lowerCAmelCase   :						List[Any]							=   KandinskyVaaInpaintPipeline
  __lowerCAmelCase   :						Union[str, Any]							=   ["""image_embeds""", """negative_image_embeds""", """image""", """mask_image"""]
  __lowerCAmelCase   :						Optional[int]							=   [
      """image_embeds""",
      """negative_image_embeds""",
      """image""",
      """mask_image""",
  ]
  __lowerCAmelCase   :						str							=   [
      """generator""",
      """height""",
      """width""",
      """latents""",
      """guidance_scale""",
      """num_inference_steps""",
      """return_dict""",
      """guidance_scale""",
      """num_images_per_prompt""",
      """output_type""",
      """return_dict""",
  ]
  __lowerCAmelCase   :						int							=   False


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return 3_2


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return 3_2


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self.time_input_dim


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self.time_input_dim * 4


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return 1_0_0


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       torch.manual_seed(0 )

       A_          =       {
           'in_channels': 9,
           # Out channels is double in channels because predicts mean and variance
           'out_channels': 8,
           'addition_embed_type': 'image',
           'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
           'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
           'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
           'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
           'layers_per_block': 1,
           'encoder_hid_dim': self.text_embedder_hidden_size,
           'encoder_hid_dim_type': 'image_proj',
           'cross_attention_dim': self.cross_attention_dim,
           'attention_head_dim': 4,
           'resnet_time_scale_shift': 'scale_shift',
           'class_embed_type': None,
       }

       A_          =       UNetaDConditionModel(**lowerCamelCase_ )
       return model


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return {
           "block_out_channels": [3_2, 6_4],
           "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
           "in_channels": 3,
           "latent_channels": 4,
           "layers_per_block": 1,
           "norm_num_groups": 8,
           "norm_type": "spatial",
           "num_vq_embeddings": 1_2,
           "out_channels": 3,
           "up_block_types": [
               "AttnUpDecoderBlock2D",
               "UpDecoderBlock2D",
           ],
           "vq_embed_dim": 4,
       }


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       torch.manual_seed(0 )
       A_          =       VQModel(**self.dummy_movq_kwargs )
       return model


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self.dummy_unet
       A_          =       self.dummy_movq

       A_          =       DDIMScheduler(
           num_train_timesteps=1_0_0_0 ,beta_schedule='''linear''' ,beta_start=0.0_0085 ,beta_end=0.012 ,clip_sample=lowerCamelCase_ ,set_alpha_to_one=lowerCamelCase_ ,steps_offset=1 ,prediction_type='''epsilon''' ,thresholding=lowerCamelCase_ ,)

       A_          =       {
           'unet': unet,
           'scheduler': scheduler,
           'movq': movq,
       }

       return components


  def 						__UpperCAmelCase     (							self ,__snake_case ,__snake_case=0 ):



       """simple docstring"""





       A_          =       floats_tensor((1, self.text_embedder_hidden_size) ,rng=random.Random(lowerCamelCase_ ) ).to(lowerCamelCase_ )
       A_          =       floats_tensor((1, self.text_embedder_hidden_size) ,rng=random.Random(seed + 1 ) ).to(
           lowerCamelCase_ )
       # create init_image
       A_          =       floats_tensor((1, 3, 6_4, 6_4) ,rng=random.Random(lowerCamelCase_ ) ).to(lowerCamelCase_ )
       A_          =       image.cpu().permute(0 ,2 ,3 ,1 )[0]
       A_          =       Image.fromarray(np.uinta(lowerCamelCase_ ) ).convert('''RGB''' ).resize((2_5_6, 2_5_6) )
       # create mask
       A_          =       np.ones((6_4, 6_4) ,dtype=np.floataa )
       A_          =       0

       if str(lowerCamelCase_ ).startswith('''mps''' ):
            A_          =       torch.manual_seed(lowerCamelCase_ )
       else:
            A_          =       torch.Generator(device=lowerCamelCase_ ).manual_seed(lowerCamelCase_ )
       A_          =       {
           'image': init_image,
           'mask_image': mask,
           'image_embeds': image_embeds,
           'negative_image_embeds': negative_image_embeds,
           'generator': generator,
           'height': 6_4,
           'width': 6_4,
           'num_inference_steps': 2,
           'guidance_scale': 4.0,
           'output_type': 'np',
       }
       return inputs


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       'cpu'

       A_          =       self.get_dummy_components()

       A_          =       self.pipeline_class(**lowerCamelCase_ )
       A_          =       pipe.to(lowerCamelCase_ )

       pipe.set_progress_bar_config(disable=lowerCamelCase_ )

       A_          =       pipe(**self.get_dummy_inputs(lowerCamelCase_ ) )
       A_          =       output.images

       A_          =       pipe(
           **self.get_dummy_inputs(lowerCamelCase_ ) ,return_dict=lowerCamelCase_ ,)[0]

       A_          =       image[0, -3:, -3:, -1]
       A_          =       image_from_tuple[0, -3:, -3:, -1]

       print(f'image.shape {image.shape}' )

       assert image.shape == (1, 6_4, 6_4, 3)

       A_          =       np.array(
           [0.5077_5903, 0.4952_7195, 0.4882_4543, 0.5019_2237, 0.4864_4906, 0.4937_3814, 0.478_0598, 0.4723_4827, 0.4832_7848] )

       assert (
           np.abs(image_slice.flatten() - expected_slice ).max() < 1E-2
       ), f' expected_slice {expected_slice}, but got {image_slice.flatten()}'
       assert (
           np.abs(image_from_tuple_slice.flatten() - expected_slice ).max() < 1E-2
       ), f' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       super().test_inference_batch_single_identical(expected_max_diff=3E-3 )





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


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       load_numpy(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
           '''/kandinskyv22/kandinskyv22_inpaint_cat_with_hat_fp16.npy''' )

       A_          =       load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png''' )
       A_          =       np.ones((7_6_8, 7_6_8) ,dtype=np.floataa )
       A_          =       0

       A_          =       'a hat'

       A_          =       KandinskyVaaPriorPipeline.from_pretrained(
           '''kandinsky-community/kandinsky-2-2-prior''' ,torch_dtype=torch.floataa )
       pipe_prior.to(lowerCamelCase_ )

       A_          =       KandinskyVaaInpaintPipeline.from_pretrained(
           '''kandinsky-community/kandinsky-2-2-decoder-inpaint''' ,torch_dtype=torch.floataa )
       A_          =       pipeline.to(lowerCamelCase_ )
       pipeline.set_progress_bar_config(disable=lowerCamelCase_ )

       A_          =       torch.Generator(device='''cpu''' ).manual_seed(0 )
       A_          =       pipe_prior(
           lowerCamelCase_ ,generator=lowerCamelCase_ ,num_inference_steps=5 ,negative_prompt='''''' ,).to_tuple()

       A_          =       pipeline(
           image=lowerCamelCase_ ,mask_image=lowerCamelCase_ ,image_embeds=lowerCamelCase_ ,negative_image_embeds=lowerCamelCase_ ,generator=lowerCamelCase_ ,num_inference_steps=1_0_0 ,height=7_6_8 ,width=7_6_8 ,output_type='''np''' ,)

       A_          =       output.images[0]

       assert image.shape == (7_6_8, 7_6_8, 3)

       assert_mean_pixel_difference(lowerCamelCase_ ,lowerCamelCase_ )










import unittest

import numpy as np

from transformers import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING, TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
from transformers.pipelines import AudioClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torchaudio,
    slow,
)

from .test_pipelines_common import ANY


@is_pipeline_test
class       __lowerCAmelCase					(       unittest.TestCase      ):


	'''simple docstring'''





	_A													=				MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
	_A													=				TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

	def    snake_case__(						self:						Optional[int],					lowerCamelCase_:						Tuple,					lowerCamelCase_:						List[Any],					lowerCamelCase_:						Optional[Any]			):
			lowercase__					:      int      =		AudioClassificationPipeline(model=lowerCamelCase_,					feature_extractor=lowerCamelCase_			)

			# test with a raw waveform
			lowercase__					:      Dict      =		np.zeros((34000,)			)
			lowercase__					:      Optional[int]      =		np.zeros((14000,)			)
			return audio_classifier, [audioa, audio]

	def    snake_case__(						self:						Optional[int],					lowerCamelCase_:						Tuple,					lowerCamelCase_:						str			):
			lowercase__   ,				lowercase__					:      List[str]      =		examples
			lowercase__					:      int      =		audio_classifier(lowerCamelCase_			)
			# by default a model is initialized with num_labels=2
			self.assertEqual(
			    lowerCamelCase_,					[
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			    ],					)
			lowercase__					:      List[Any]      =		audio_classifier(lowerCamelCase_,					top_k=1			)
			self.assertEqual(
			    lowerCamelCase_,					[
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			    ],					)

			self.run_torchaudio(lowerCamelCase_			)

	@require_torchaudio
	def    snake_case__(						self:						Dict,					lowerCamelCase_:						str			):
			import datasets

			# test with a local file
			lowercase__					:      Optional[Any]      =		datasets.load_dataset('hf-internal-testing/librispeech_asr_dummy',					'clean',					split='validation'			)
			lowercase__					:      int      =		dataset[0]['audio']['array']
			lowercase__					:      Tuple      =		audio_classifier(lowerCamelCase_			)
			self.assertEqual(
			    lowerCamelCase_,					[
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			    ],					)

	@require_torch
	def    snake_case__(						self:						Optional[int]			):
			lowercase__					:      List[str]      =		'anton-l/wav2vec2-random-tiny-classifier'

			lowercase__					:      Tuple      =		pipeline('audio-classification',					model=lowerCamelCase_			)

			lowercase__					:      List[str]      =		np.ones((8000,)			)
			lowercase__					:      List[Any]      =		audio_classifier(lowerCamelCase_,					top_k=4			)

			lowercase__					:      List[str]      =		[
			    {'score': 0.0_8_4_2, 'label': 'no'},
			    {'score': 0.0_8_3_8, 'label': 'up'},
			    {'score': 0.0_8_3_7, 'label': 'go'},
			    {'score': 0.0_8_3_4, 'label': 'right'},
			]
			lowercase__					:      Optional[int]      =		[
			    {'score': 0.0_8_4_5, 'label': 'stop'},
			    {'score': 0.0_8_4_4, 'label': 'on'},
			    {'score': 0.0_8_4_1, 'label': 'right'},
			    {'score': 0.0_8_3_4, 'label': 'left'},
			]
			self.assertIn(nested_simplify(lowerCamelCase_,					decimals=4			),					[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]			)

			lowercase__					:      Tuple      =		{'array': np.ones((8000,)			), 'sampling_rate': audio_classifier.feature_extractor.sampling_rate}
			lowercase__					:      Any      =		audio_classifier(lowerCamelCase_,					top_k=4			)
			self.assertIn(nested_simplify(lowerCamelCase_,					decimals=4			),					[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]			)

	@require_torch
	@slow
	def    snake_case__(						self:						Optional[int]			):
			import datasets

			lowercase__					:      str      =		'superb/wav2vec2-base-superb-ks'

			lowercase__					:      Optional[int]      =		pipeline('audio-classification',					model=lowerCamelCase_			)
			lowercase__					:      List[Any]      =		datasets.load_dataset('anton-l/superb_dummy',					'ks',					split='test'			)

			lowercase__					:      Optional[Any]      =		np.array(dataset[3]['speech'],					dtype=np.floataa			)
			lowercase__					:      Optional[int]      =		audio_classifier(lowerCamelCase_,					top_k=4			)
			self.assertEqual(
			    nested_simplify(lowerCamelCase_,					decimals=3			),					[
			        {'score': 0.9_8_1, 'label': 'go'},
			        {'score': 0.0_0_7, 'label': 'up'},
			        {'score': 0.0_0_6, 'label': '_unknown_'},
			        {'score': 0.0_0_1, 'label': 'down'},
			    ],					)





	@require_tf
	@unittest.skip('Audio classification is not implemented for TF'			)
	def    snake_case__(						self:						int			):
			pass


