"""simple docstring"""
import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    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    __snake_case       (     __lowerCAmelCase       ,      unittest.TestCase				):
      a__        =		KandinskyVaaControlnetPipeline
      a__        =		["""image_embeds""", """negative_image_embeds""", """hint"""]
      a__        =		["""image_embeds""", """negative_image_embeds""", """hint"""]
      a__        =		[
          """generator""",
          """height""",
          """width""",
          """latents""",
          """guidance_scale""",
          """num_inference_steps""",
          """return_dict""",
          """guidance_scale""",
          """num_images_per_prompt""",
          """output_type""",
          """return_dict""",
      ]
      a__        =		False



      @property
      def   lowerCamelCase_   (						self)	->			int:





        '''simple docstring'''





        return 32



      @property
      def   lowerCamelCase_   (						self)	->			int:





        '''simple docstring'''





        return 32



      @property
      def   lowerCamelCase_   (						self)	->			List[str]:





        '''simple docstring'''





        return self.time_input_dim



      @property
      def   lowerCamelCase_   (						self)	->			List[Any]:





        '''simple docstring'''





        return self.time_input_dim * 4



      @property
      def   lowerCamelCase_   (						self)	->			Optional[Any]:





        '''simple docstring'''





        return 1_00



      @property
      def   lowerCamelCase_   (						self)	->			str:





        '''simple docstring'''





        torch.manual_seed(0)

        a__:     Union[str, Any]										=	{
            'in_channels': 8,
            # Out channels is double in channels because predicts mean and variance
            'out_channels': 8,
            'addition_embed_type': 'image_hint',
            '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__:     Optional[int]										=	UNetaDConditionModel(**lowercase)
        return model



      @property
      def   lowerCamelCase_   (						self)	->			Optional[Any]:





        '''simple docstring'''





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



      @property
      def   lowerCamelCase_   (						self)	->			Tuple:





        '''simple docstring'''





        torch.manual_seed(0)
        a__:     str										=	VQModel(**self.dummy_movq_kwargs)
        return model



      def   lowerCamelCase_   (						self)	->			Union[str, Any]:





        '''simple docstring'''





        a__:     str										=	self.dummy_unet
        a__:     List[Any]										=	self.dummy_movq

        a__:     str										=	DDIMScheduler(
            num_train_timesteps=10_00    ,			beta_schedule='linear'    ,			beta_start=0.00085    ,			beta_end=0.012    ,			clip_sample=lowercase    ,			set_alpha_to_one=lowercase    ,			steps_offset=1    ,			prediction_type='epsilon'    ,			thresholding=lowercase    ,			)

        a__:     Union[str, Any]										=	{
            'unet': unet,
            'scheduler': scheduler,
            'movq': movq,
        }
        return components



      def   lowerCamelCase_   (						self    ,			lowercase    ,			lowercase=0)	->			List[str]:





        '''simple docstring'''





        a__:     Optional[Any]										=	floats_tensor((1, self.text_embedder_hidden_size)    ,			rng=random.Random(lowercase)).to(lowercase)
        a__:     Union[str, Any]										=	floats_tensor((1, self.text_embedder_hidden_size)    ,			rng=random.Random(seed + 1)).to(
            lowercase)

        # create hint
        a__:     Union[str, Any]										=	floats_tensor((1, 3, 64, 64)    ,			rng=random.Random(lowercase)).to(lowercase)

        if str(lowercase).startswith('mps'):
          a__:     List[str]										=	torch.manual_seed(lowercase)
        else:
          a__:     Union[str, Any]										=	torch.Generator(device=lowercase).manual_seed(lowercase)
        a__:     Union[str, Any]										=	{
            'image_embeds': image_embeds,
            'negative_image_embeds': negative_image_embeds,
            'hint': hint,
            'generator': generator,
            'height': 64,
            'width': 64,
            'guidance_scale': 4.0,
            'num_inference_steps': 2,
            'output_type': 'np',
        }
        return inputs



      def   lowerCamelCase_   (						self)	->			Optional[Any]:





        '''simple docstring'''





        a__:     str										=	'cpu'

        a__:     int										=	self.get_dummy_components()

        a__:     Optional[int]										=	self.pipeline_class(**lowercase)
        a__:     Any										=	pipe.to(lowercase)

        pipe.set_progress_bar_config(disable=lowercase)

        a__:     Optional[Any]										=	pipe(**self.get_dummy_inputs(lowercase))
        a__:     Any										=	output.images

        a__:     Any										=	pipe(
            **self.get_dummy_inputs(lowercase)    ,			return_dict=lowercase    ,			)[0]

        a__:     Optional[Any]										=	image[0, -3:, -3:, -1]
        a__:     Optional[int]										=	image_from_tuple[0, -3:, -3:, -1]

        assert image.shape == (1, 64, 64, 3)

        a__:     List[Any]										=	np.array(
            [0.6959826, 0.868279, 0.7558092, 0.68769467, 0.85805804, 0.65977496, 0.44885302, 0.5959111, 0.4251595])

        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()}'







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



      def   lowerCamelCase_   (						self)	->			Optional[Any]:





        '''simple docstring'''





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



      def   lowerCamelCase_   (						self)	->			List[Any]:





        '''simple docstring'''





        a__:     List[str]										=	load_numpy(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy')

        a__:     Any										=	load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/kandinskyv22/hint_image_cat.png')
        a__:     int										=	torch.from_numpy(np.array(lowercase)).float() / 255.0
        a__:     Tuple										=	hint.permute(2    ,			0    ,			1).unsqueeze(0)

        a__:     List[str]										=	KandinskyVaaPriorPipeline.from_pretrained(
            'kandinsky-community/kandinsky-2-2-prior'    ,			torch_dtype=torch.floataa)
        pipe_prior.to(lowercase)

        a__:     Tuple										=	KandinskyVaaControlnetPipeline.from_pretrained(
            'kandinsky-community/kandinsky-2-2-controlnet-depth'    ,			torch_dtype=torch.floataa)
        a__:     Union[str, Any]										=	pipeline.to(lowercase)
        pipeline.set_progress_bar_config(disable=lowercase)

        a__:     Union[str, Any]										=	'A robot, 4k photo'

        a__:     List[str]										=	torch.Generator(device='cuda').manual_seed(0)
        a__							,    a__:     Union[str, Any]										=	pipe_prior(
            lowercase    ,			generator=lowercase    ,			num_inference_steps=5    ,			negative_prompt=''    ,			).to_tuple()

        a__:     Optional[Any]										=	torch.Generator(device='cuda').manual_seed(0)
        a__:     Optional[int]										=	pipeline(
            image_embeds=lowercase    ,			negative_image_embeds=lowercase    ,			hint=lowercase    ,			generator=lowercase    ,			num_inference_steps=1_00    ,			output_type='np'    ,			)

        a__:     Dict										=	output.images[0]

        assert image.shape == (5_12, 5_12, 3)

        assert_mean_pixel_difference(lowercase    ,			lowercase)



"""simple docstring"""
import itertools
import random
import unittest

import numpy as np

from transformers import WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST, WavaVecaConfig, WavaVecaFeatureExtractor
from transformers.testing_utils import require_torch, slow

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


lowercase__					       =	random.Random()







def 						__a   (	_SCREAMING_SNAKE_CASE		,     _SCREAMING_SNAKE_CASE=1.0		,     _SCREAMING_SNAKE_CASE=None		,     _SCREAMING_SNAKE_CASE=None   )	->Optional[int]:
  if rng is None:
    a__:     Any										=	global_rng

  a__:     int										=	[]
  for batch_idx in range(shape[0]   ):
    values.append([]   )
    for _ in range(shape[1]   ):
      values[-1].append(rng.random() * scale   )

  return values




class    __snake_case       (     unittest.TestCase				):



      def __init__(						self    ,			lowercase    ,			lowercase=7    ,			lowercase=4_00    ,			lowercase=20_00    ,			lowercase=1    ,			lowercase=0.0    ,			lowercase=1_60_00    ,			lowercase=True    ,			lowercase=True    ,			)	->			Union[str, Any]:





        '''simple docstring'''





        a__:     Tuple										=	parent
        a__:     Optional[int]										=	batch_size
        a__:     Optional[Any]										=	min_seq_length
        a__:     Optional[int]										=	max_seq_length
        a__:     Tuple										=	(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        a__:     Dict										=	feature_size
        a__:     Any										=	padding_value
        a__:     Optional[Any]										=	sampling_rate
        a__:     Optional[Any]										=	return_attention_mask
        a__:     str										=	do_normalize



      def   lowerCamelCase_   (						self)	->			Dict:





        '''simple docstring'''





        return {
            "feature_size": self.feature_size,
            "padding_value": self.padding_value,
            "sampling_rate": self.sampling_rate,
            "return_attention_mask": self.return_attention_mask,
            "do_normalize": self.do_normalize,
        }



      def   lowerCamelCase_   (						self    ,			lowercase=False    ,			lowercase=False)	->			Tuple:





        '''simple docstring'''





        def _flatten(lowercase):
          return list(itertools.chain(*lowercase))

        if equal_length:
          a__:     Dict										=	floats_list((self.batch_size, self.max_seq_length))
        else:
          # make sure that inputs increase in size
          a__:     List[Any]										=	[
              _flatten(floats_list((x, self.feature_size)))
              for x in range(self.min_seq_length    ,			self.max_seq_length    ,			self.seq_length_diff)
          ]

        if numpify:
          a__:     str										=	[np.asarray(lowercase) for x in speech_inputs]

        return speech_inputs







class    __snake_case       (     __lowerCAmelCase       ,      unittest.TestCase				):
      a__        =		WavaVecaFeatureExtractor



      def   lowerCamelCase_   (						self)	->			Tuple:





        '''simple docstring'''





        a__:     Optional[int]										=	WavaVecaFeatureExtractionTester(self)



      def   lowerCamelCase_   (						self    ,			lowercase)	->			List[Any]:





        '''simple docstring'''





        self.assertTrue(np.all(np.mean(lowercase    ,			axis=0) < 1e-3))
        self.assertTrue(np.all(np.abs(np.var(lowercase    ,			axis=0) - 1) < 1e-3))



      def   lowerCamelCase_   (						self)	->			List[str]:





        '''simple docstring'''





        a__:     List[Any]										=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict())
        # create three inputs of length 800, 1000, and 1200
        a__:     Optional[Any]										=	[floats_list((1, x))[0] for x in range(8_00    ,			14_00    ,			2_00)]
        a__:     List[str]										=	[np.asarray(lowercase) for speech_input in speech_inputs]

        # Test not batched input
        a__:     Optional[Any]										=	feat_extract(speech_inputs[0]    ,			return_tensors='np').input_values
        a__:     Dict										=	feat_extract(np_speech_inputs[0]    ,			return_tensors='np').input_values
        self.assertTrue(np.allclose(lowercase    ,			lowercase    ,			atol=1e-3))

        # Test batched
        a__:     Dict										=	feat_extract(lowercase    ,			return_tensors='np').input_values
        a__:     int										=	feat_extract(lowercase    ,			return_tensors='np').input_values
        for enc_seq_a, enc_seq_a in zip(lowercase    ,			lowercase):
          self.assertTrue(np.allclose(lowercase    ,			lowercase    ,			atol=1e-3))

        # Test 2-D numpy arrays are batched.
        a__:     int										=	[floats_list((1, x))[0] for x in (8_00, 8_00, 8_00)]
        a__:     Union[str, Any]										=	np.asarray(lowercase)
        a__:     int										=	feat_extract(lowercase    ,			return_tensors='np').input_values
        a__:     Any										=	feat_extract(lowercase    ,			return_tensors='np').input_values
        for enc_seq_a, enc_seq_a in zip(lowercase    ,			lowercase):
          self.assertTrue(np.allclose(lowercase    ,			lowercase    ,			atol=1e-3))



      def   lowerCamelCase_   (						self)	->			List[Any]:





        '''simple docstring'''





        a__:     Tuple										=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict())
        a__:     List[Any]										=	[floats_list((1, x))[0] for x in range(8_00    ,			14_00    ,			2_00)]

        a__:     Optional[int]										=	['longest', 'max_length', 'do_not_pad']
        a__:     List[Any]										=	[None, 16_00, None]
        for max_length, padding in zip(lowercase    ,			lowercase):
          a__:     Dict										=	feat_extract(lowercase    ,			padding=lowercase    ,			max_length=lowercase    ,			return_tensors='np')
          a__:     Union[str, Any]										=	processed.input_values

          self._check_zero_mean_unit_variance(input_values[0][:8_00])
          self.assertTrue(input_values[0][8_00:].sum() < 1e-6)
          self._check_zero_mean_unit_variance(input_values[1][:10_00])
          self.assertTrue(input_values[0][10_00:].sum() < 1e-6)
          self._check_zero_mean_unit_variance(input_values[2][:12_00])



      def   lowerCamelCase_   (						self)	->			Optional[int]:





        '''simple docstring'''





        a__:     str										=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict())
        a__:     Optional[int]										=	range(8_00    ,			14_00    ,			2_00)
        a__:     List[str]										=	[floats_list((1, x))[0] for x in lengths]

        a__:     Tuple										=	['longest', 'max_length', 'do_not_pad']
        a__:     Dict										=	[None, 16_00, None]

        for max_length, padding in zip(lowercase    ,			lowercase):
          a__:     int										=	feat_extract(lowercase    ,			max_length=lowercase    ,			padding=lowercase)
          a__:     Any										=	processed.input_values

          self._check_zero_mean_unit_variance(input_values[0][:8_00])
          self._check_zero_mean_unit_variance(input_values[1][:10_00])
          self._check_zero_mean_unit_variance(input_values[2][:12_00])



      def   lowerCamelCase_   (						self)	->			List[str]:





        '''simple docstring'''





        a__:     Any										=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict())
        a__:     Any										=	[floats_list((1, x))[0] for x in range(8_00    ,			14_00    ,			2_00)]
        a__:     Dict										=	feat_extract(
            lowercase    ,			truncation=lowercase    ,			max_length=10_00    ,			padding='max_length'    ,			return_tensors='np')
        a__:     int										=	processed.input_values

        self._check_zero_mean_unit_variance(input_values[0, :8_00])
        self._check_zero_mean_unit_variance(input_values[1])
        self._check_zero_mean_unit_variance(input_values[2])



      def   lowerCamelCase_   (						self)	->			Dict:





        '''simple docstring'''





        a__:     Tuple										=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict())
        a__:     int										=	[floats_list((1, x))[0] for x in range(8_00    ,			14_00    ,			2_00)]
        a__:     str										=	feat_extract(
            lowercase    ,			truncation=lowercase    ,			max_length=10_00    ,			padding='longest'    ,			return_tensors='np')
        a__:     Any										=	processed.input_values

        self._check_zero_mean_unit_variance(input_values[0, :8_00])
        self._check_zero_mean_unit_variance(input_values[1, :10_00])
        self._check_zero_mean_unit_variance(input_values[2])

        # make sure that if max_length < longest -> then pad to max_length
        self.assertTrue(input_values.shape == (3, 10_00))

        a__:     Dict										=	[floats_list((1, x))[0] for x in range(8_00    ,			14_00    ,			2_00)]
        a__:     Tuple										=	feat_extract(
            lowercase    ,			truncation=lowercase    ,			max_length=20_00    ,			padding='longest'    ,			return_tensors='np')
        a__:     str										=	processed.input_values

        self._check_zero_mean_unit_variance(input_values[0, :8_00])
        self._check_zero_mean_unit_variance(input_values[1, :10_00])
        self._check_zero_mean_unit_variance(input_values[2])

        # make sure that if max_length > longest -> then pad to longest
        self.assertTrue(input_values.shape == (3, 12_00))



      @require_torch
      def   lowerCamelCase_   (						self)	->			Tuple:





        '''simple docstring'''





        import torch

        a__:     Optional[int]										=	self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict())
        a__:     Tuple										=	np.random.rand(1_00).astype(np.floataa)
        a__:     Tuple										=	np_speech_inputs.tolist()

        for inputs in [py_speech_inputs, np_speech_inputs]:
          a__:     Any										=	feature_extractor.pad([{'input_values': inputs}]    ,			return_tensors='np')
          self.assertTrue(np_processed.input_values.dtype == np.floataa)
          a__:     Optional[Any]										=	feature_extractor.pad([{'input_values': inputs}]    ,			return_tensors='pt')
          self.assertTrue(pt_processed.input_values.dtype == torch.floataa)



      @slow
      @require_torch
      def   lowerCamelCase_   (						self)	->			Optional[Any]:





        '''simple docstring'''





        for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
          a__:     str										=	WavaVecaConfig.from_pretrained(lowercase)
          a__:     str										=	WavaVecaFeatureExtractor.from_pretrained(lowercase)

          # only "layer" feature extraction norm should make use of
          # attention_mask
          self.assertEqual(feat_extract.return_attention_mask    ,			config.feat_extract_norm == 'layer')


