import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetImgaImgPipeline,
    KandinskyVaaPriorEmbaEmbPipeline,
    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 					lowercase__						(   __snake_case	,							unittest.TestCase):
						UpperCamelCase_     =		KandinskyVaaControlnetImgaImgPipeline
						UpperCamelCase_     =		["""image_embeds""", """negative_image_embeds""", """image""", """hint"""]
						UpperCamelCase_     =		["""image_embeds""", """negative_image_embeds""", """image""", """hint"""]
						UpperCamelCase_     =		[
						    """generator""",
						    """height""",
						    """width""",
						    """strength""",
						    """guidance_scale""",
						    """num_inference_steps""",
						    """return_dict""",
						    """guidance_scale""",
						    """num_images_per_prompt""",
						    """output_type""",
						    """return_dict""",
						]
						UpperCamelCase_     =		False


						@property
						def        __A       (       self	:				Dict						):





												'''simple docstring'''





												return 32


						@property
						def        __A       (       self	:				List[Any]						):





												'''simple docstring'''





												return 32


						@property
						def        __A       (       self	:				Union[str, Any]						):





												'''simple docstring'''





												return self.time_input_dim


						@property
						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												return self.time_input_dim * 4


						@property
						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												return 100


						@property
						def        __A       (       self	:				int						):





												'''simple docstring'''





												torch.manual_seed(0						)

												SCREAMING_SNAKE_CASE							:    List[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,
												}

												SCREAMING_SNAKE_CASE							:    List[str]							=							UNetaDConditionModel(**UpperCamelCase__						)
												return model


						@property
						def        __A       (       self	:				Dict						):





												'''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        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





												torch.manual_seed(0						)
												SCREAMING_SNAKE_CASE							:    int							=							VQModel(**self.dummy_movq_kwargs						)
												return model


						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[str]							=							self.dummy_unet
												SCREAMING_SNAKE_CASE							:    Tuple							=							self.dummy_movq

												SCREAMING_SNAKE_CASE							:    List[str]							=							{
												    '''num_train_timesteps''': 1000,
												    '''beta_schedule''': '''linear''',
												    '''beta_start''': 0.0_0085,
												    '''beta_end''': 0.012,
												    '''clip_sample''': False,
												    '''set_alpha_to_one''': False,
												    '''steps_offset''': 0,
												    '''prediction_type''': '''epsilon''',
												    '''thresholding''': False,
												}

												SCREAMING_SNAKE_CASE							:    str							=							DDIMScheduler(**UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    Tuple							=							{
												    '''unet''': unet,
												    '''scheduler''': scheduler,
												    '''movq''': movq,
												}

												return components


						def        __A       (       self	:				List[Any]    ,		UpperCamelCase__	:				Tuple    ,		UpperCamelCase__	:				str=0						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    int							=							floats_tensor((1, self.text_embedder_hidden_size)    ,		rng=random.Random(UpperCamelCase__						)						).to(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Any							=							floats_tensor((1, self.text_embedder_hidden_size)    ,		rng=random.Random(seed + 1						)						).to(
												    UpperCamelCase__						)
												# create init_image
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							floats_tensor((1, 3, 64, 64)    ,		rng=random.Random(UpperCamelCase__						)						).to(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    List[str]							=							image.cpu().permute(0    ,		2    ,		3    ,		1						)[0]
												SCREAMING_SNAKE_CASE							:    Tuple							=							Image.fromarray(np.uinta(UpperCamelCase__						)						).convert('''RGB'''						).resize((256, 256)						)
												# create hint
												SCREAMING_SNAKE_CASE							:    Any							=							floats_tensor((1, 3, 64, 64)    ,		rng=random.Random(UpperCamelCase__						)						).to(UpperCamelCase__						)

												if str(UpperCamelCase__						).startswith('''mps'''						):
																		SCREAMING_SNAKE_CASE							:    str							=							torch.manual_seed(UpperCamelCase__						)
												else:
																		SCREAMING_SNAKE_CASE							:    List[str]							=							torch.Generator(device=UpperCamelCase__						).manual_seed(UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							{
												    '''image''': init_image,
												    '''image_embeds''': image_embeds,
												    '''negative_image_embeds''': negative_image_embeds,
												    '''hint''': hint,
												    '''generator''': generator,
												    '''height''': 64,
												    '''width''': 64,
												    '''num_inference_steps''': 10,
												    '''guidance_scale''': 7.0,
												    '''strength''': 0.2,
												    '''output_type''': '''np''',
												}
												return inputs



						def        __A       (       self	:				List[Any]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							'''cpu'''

												SCREAMING_SNAKE_CASE							:    Dict							=							self.get_dummy_components()

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							self.pipeline_class(**UpperCamelCase__						)
												SCREAMING_SNAKE_CASE							:    Dict							=							pipe.to(UpperCamelCase__						)

												pipe.set_progress_bar_config(disable=UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    Optional[int]							=							pipe(**self.get_dummy_inputs(UpperCamelCase__						)						)
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							output.images

												SCREAMING_SNAKE_CASE							:    str							=							pipe(
												    **self.get_dummy_inputs(UpperCamelCase__						)    ,		return_dict=UpperCamelCase__    ,		)[0]

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

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

												SCREAMING_SNAKE_CASE							:    List[Any]							=							np.array(
												    [0.5498_5034, 0.5550_9365, 0.5256_1504, 0.557_0494, 0.559_3818, 0.526_3979, 0.5028_5643, 0.506_9846, 0.5119_6736]						)
												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 					lowercase__						(   unittest.TestCase):


						def        __A       (       self	:				Optional[Any]						):





												'''simple docstring'''





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



						def        __A       (       self	:				Optional[int]						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							load_numpy(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
												    '''/kandinskyv22/kandinskyv22_controlnet_img2img_robotcat_fp16.npy'''						)

												SCREAMING_SNAKE_CASE							:    int							=							load_image(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''						)
												SCREAMING_SNAKE_CASE							:    List[Any]							=							init_image.resize((512, 512)						)

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

												SCREAMING_SNAKE_CASE							:    int							=							'''A robot, 4k photo'''

												SCREAMING_SNAKE_CASE							:    str							=							KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
												    '''kandinsky-community/kandinsky-2-2-prior'''    ,		torch_dtype=torch.floataa						)
												pipe_prior.to(UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    Optional[int]							=							KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
												    '''kandinsky-community/kandinsky-2-2-controlnet-depth'''    ,		torch_dtype=torch.floataa						)
												SCREAMING_SNAKE_CASE							:    List[Any]							=							pipeline.to(UpperCamelCase__						)

												pipeline.set_progress_bar_config(disable=UpperCamelCase__						)

												SCREAMING_SNAKE_CASE							:    Dict							=							torch.Generator(device='''cpu'''						).manual_seed(0						)

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							pipe_prior(
												    UpperCamelCase__    ,		image=UpperCamelCase__    ,		strength=0.85    ,		generator=UpperCamelCase__    ,		negative_prompt=''''''    ,		).to_tuple()

												SCREAMING_SNAKE_CASE							:    str							=							pipeline(
												    image=UpperCamelCase__    ,		image_embeds=UpperCamelCase__    ,		negative_image_embeds=UpperCamelCase__    ,		hint=UpperCamelCase__    ,		generator=UpperCamelCase__    ,		num_inference_steps=100    ,		height=512    ,		width=512    ,		strength=0.5    ,		output_type='''np'''    ,		)

												SCREAMING_SNAKE_CASE							:    str							=							output.images[0]

												assert image.shape == (512, 512, 3)

												assert_mean_pixel_difference(UpperCamelCase__    ,		UpperCamelCase__						)







'''simple docstring'''




from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


lowercase			:				Optional[int]    		=			logging.get_logger(__name__)





class  A				(							__snake_case						):
     __magic_name__						=		['''input_features''', '''attention_mask''']





     def __init__(							self						,			SCREAMING_SNAKE_CASE=80						,			SCREAMING_SNAKE_CASE=16000						,			SCREAMING_SNAKE_CASE=80						,			SCREAMING_SNAKE_CASE=0.0						,			SCREAMING_SNAKE_CASE=True						,			SCREAMING_SNAKE_CASE=True						,			SCREAMING_SNAKE_CASE=True						,			**SCREAMING_SNAKE_CASE						,			) ->				int:






         """simple docstring"""






         super().__init__(feature_size=SCREAMING_SNAKE_CASE						,			sampling_rate=SCREAMING_SNAKE_CASE						,			padding_value=SCREAMING_SNAKE_CASE						,			**SCREAMING_SNAKE_CASE	)
         A       :   Optional[int]					=						num_mel_bins
         A       :   Tuple					=						do_ceptral_normalize
         A       :   Dict					=						normalize_means
         A       :   List[Any]					=						normalize_vars
         A       :   List[str]					=						True





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			) ->				np.ndarray:






         """simple docstring"""






         A       :   List[Any]					=						waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
         A       :   Any					=						torch.from_numpy(SCREAMING_SNAKE_CASE	).unsqueeze(0	)
         A       :   Any					=						ta_kaldi.fbank(SCREAMING_SNAKE_CASE						,			num_mel_bins=self.num_mel_bins						,			sample_frequency=self.sampling_rate	)
         return features.numpy()





     @staticmethod
     def 				__lowerCAmelCase							(							SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = True						,			SCREAMING_SNAKE_CASE = True						,			SCREAMING_SNAKE_CASE = 0.0						,			) ->				np.ndarray:






         """simple docstring"""






         if normalize_means:
             A       :   Dict					=						x[:input_length].mean(axis=0	)
             A       :   Optional[Any]					=						np.subtract(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)
         if normalize_vars:
             A       :   str					=						x[:input_length].std(axis=0	)
             A       :   int					=						np.divide(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)

         if input_length < x.shape[0]:
             A       :   List[str]					=						padding_value

         # make sure array is in float32
         A       :   Tuple					=						x.astype(np.floataa	)

         return x





     def 				__lowerCAmelCase							(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = None	) ->				List[np.ndarray]:






         """simple docstring"""






         A       :   List[Any]					=						attention_mask.sum(-1	) if attention_mask is not None else [x.shape[0] for x in input_features]
         return [
             self.utterance_cmvn(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE						,			self.normalize_means						,			self.normalize_vars						,			self.padding_value	)
             for x, n in zip(SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE	)
         ]





     def __call__(							self						,			SCREAMING_SNAKE_CASE						,			SCREAMING_SNAKE_CASE = False						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = False						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = None						,			SCREAMING_SNAKE_CASE = None						,			**SCREAMING_SNAKE_CASE						,			) ->				BatchFeature:






         """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.'''	)

         A       :   List[Any]					=						isinstance(SCREAMING_SNAKE_CASE						,			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}'	)
         A       :   Tuple					=						is_batched_numpy or (
             isinstance(SCREAMING_SNAKE_CASE						,			(list, tuple)	) and (isinstance(raw_speech[0]						,			(np.ndarray, tuple, list)	))
         )

         if is_batched:
             A       :   Union[str, Any]					=						[np.asarray(SCREAMING_SNAKE_CASE						,			dtype=np.floataa	) for speech in raw_speech]
         elif not is_batched and not isinstance(SCREAMING_SNAKE_CASE						,			np.ndarray	):
             A       :   Union[str, Any]					=						np.asarray(SCREAMING_SNAKE_CASE						,			dtype=np.floataa	)
         elif isinstance(SCREAMING_SNAKE_CASE						,			np.ndarray	) and raw_speech.dtype is np.dtype(np.floataa	):
             A       :   Optional[int]					=						raw_speech.astype(np.floataa	)

         # always return batch
         if not is_batched:
             A       :   Any					=						[raw_speech]

         # extract fbank features
         A       :   List[str]					=						[self._extract_fbank_features(SCREAMING_SNAKE_CASE	) for waveform in raw_speech]

         # convert into correct format for padding
         A       :   str					=						BatchFeature({'''input_features''': features}	)

         A       :   Union[str, Any]					=						self.pad(
             SCREAMING_SNAKE_CASE						,			padding=SCREAMING_SNAKE_CASE						,			max_length=SCREAMING_SNAKE_CASE						,			truncation=SCREAMING_SNAKE_CASE						,			pad_to_multiple_of=SCREAMING_SNAKE_CASE						,			return_attention_mask=SCREAMING_SNAKE_CASE						,			**SCREAMING_SNAKE_CASE						,			)

         # make sure list is in array format
         A       :   List[str]					=						padded_inputs.get('''input_features'''	)
         if isinstance(input_features[0]						,			SCREAMING_SNAKE_CASE	):
             A       :   str					=						[np.asarray(SCREAMING_SNAKE_CASE						,			dtype=np.floataa	) for feature in input_features]

         A       :   Union[str, Any]					=						padded_inputs.get('''attention_mask'''	)
         if attention_mask is not None:
             A       :   Union[str, Any]					=						[np.asarray(SCREAMING_SNAKE_CASE						,			dtype=np.intaa	) for array in attention_mask]

         # Utterance-level cepstral mean and variance normalization
         if self.do_ceptral_normalize:
             A       :   Dict					=						(
                 np.array(SCREAMING_SNAKE_CASE						,			dtype=np.intaa	)
                 if self._get_padding_strategies(SCREAMING_SNAKE_CASE						,			max_length=SCREAMING_SNAKE_CASE	) is not PaddingStrategy.DO_NOT_PAD
                 else None
             )
             A       :   List[Any]					=						self.normalize(
                 padded_inputs['''input_features''']						,			attention_mask=SCREAMING_SNAKE_CASE	)

         if return_tensors is not None:
             A       :   int					=						padded_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE	)

         return padded_inputs




