



'''simple docstring'''

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
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 	__magic_name__						(    _UpperCamelCase			,						unittest.TestCase   ):
							lowerCAmelCase       :    str   				=					KandinskyImgaImgPipeline
							lowerCAmelCase       :    List[Any]   				=					['prompt', 'image_embeds', 'negative_image_embeds', 'image']
							lowerCAmelCase       :    List[str]   				=					[
							    'prompt',
							    'negative_prompt',
							    'image_embeds',
							    'negative_image_embeds',
							    'image',
							]
							lowerCAmelCase       :    Union[str, Any]   				=					[
							    'generator',
							    'height',
							    'width',
							    'strength',
							    'guidance_scale',
							    'negative_prompt',
							    'num_inference_steps',
							    'return_dict',
							    'guidance_scale',
							    'num_images_per_prompt',
							    'output_type',
							    'return_dict',
							]
							lowerCAmelCase       :    str   				=					False







							@property
							def 							__lowercase			(							self					:       str		):
									return 32







							@property
							def 							__lowercase			(							self					:       Union[str, Any]		):
									return 32







							@property
							def 							__lowercase			(							self					:       Optional[int]		):
									return self.time_input_dim







							@property
							def 							__lowercase			(							self					:       List[Any]		):
									return self.time_input_dim * 4







							@property
							def 							__lowercase			(							self					:       int		):
									return 100







							@property
							def 							__lowercase			(							self					:       Any		):
									_a     :      Optional[Any]        =							XLMRobertaTokenizerFast.from_pretrained('YiYiXu/tiny-random-mclip-base'		)
									return tokenizer







							@property
							def 							__lowercase			(							self					:       List[Any]		):
									torch.manual_seed(0		)
									_a     :      Union[str, Any]        =							MCLIPConfig(
									    numDims=self.cross_attention_dim	,transformerDimensions=self.text_embedder_hidden_size	,hidden_size=self.text_embedder_hidden_size	,intermediate_size=37	,num_attention_heads=4	,num_hidden_layers=5	,vocab_size=1005	,)

									_a     :      int        =							MultilingualCLIP(_UpperCAmelCase		)
									_a     :      List[Any]        =							text_encoder.eval()

									return text_encoder







							@property
							def 							__lowercase			(							self					:       Optional[Any]		):
									torch.manual_seed(0		)

									_a     :      List[Any]        =							{
									    'in_channels': 4,
									    # Out channels is double in channels because predicts mean and variance
									    'out_channels': 8,
									    'addition_embed_type': 'text_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': 'text_image_proj',
									    'cross_attention_dim': self.cross_attention_dim,
									    'attention_head_dim': 4,
									    'resnet_time_scale_shift': 'scale_shift',
									    'class_embed_type': None,
									}

									_a     :      Union[str, Any]        =							UNetaDConditionModel(**_UpperCAmelCase		)
									return model







							@property
							def 							__lowercase			(							self					:       Union[str, Any]		):
									return {
									    "block_out_channels": [32, 64],
									    "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": 12,
									    "out_channels": 3,
									    "up_block_types": [
									        "AttnUpDecoderBlock2D",
									        "UpDecoderBlock2D",
									    ],
									    "vq_embed_dim": 4,
									}







							@property
							def 							__lowercase			(							self					:       Optional[int]		):
									torch.manual_seed(0		)
									_a     :      Any        =							VQModel(**self.dummy_movq_kwargs		)
									return model







							def 							__lowercase			(							self					:       int		):
									_a     :      List[Any]        =							self.dummy_text_encoder
									_a     :      Optional[int]        =							self.dummy_tokenizer
									_a     :      List[Any]        =							self.dummy_unet
									_a     :      Any        =							self.dummy_movq

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

									_a     :      List[str]        =							DDIMScheduler(**_UpperCAmelCase		)

									_a     :      Any        =							{
									    'text_encoder': text_encoder,
									    'tokenizer': tokenizer,
									    'unet': unet,
									    'scheduler': scheduler,
									    'movq': movq,
									}

									return components







							def 							__lowercase			(							self					:       Union[str, Any]	,_UpperCAmelCase					:       Dict	,_UpperCAmelCase					:       Optional[int]=0		):
									_a     :      Any        =							floats_tensor((1, self.cross_attention_dim)	,rng=random.Random(_UpperCAmelCase		)		).to(_UpperCAmelCase		)
									_a     :      Union[str, Any]        =							floats_tensor((1, self.cross_attention_dim)	,rng=random.Random(seed + 1		)		).to(_UpperCAmelCase		)
									# create init_image
									_a     :      Optional[int]        =							floats_tensor((1, 3, 64, 64)	,rng=random.Random(_UpperCAmelCase		)		).to(_UpperCAmelCase		)
									_a     :      Optional[int]        =							image.cpu().permute(0	,2	,3	,1		)[0]
									_a     :      Optional[int]        =							Image.fromarray(np.uinta(_UpperCAmelCase		)		).convert('RGB'		).resize((256, 256)		)

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







							def 							__lowercase			(							self					:       Union[str, Any]		):
									_a     :      Any        =							'cpu'

									_a     :      List[Any]        =							self.get_dummy_components()

									_a     :      Tuple        =							self.pipeline_class(**_UpperCAmelCase		)
									_a     :      Optional[Any]        =							pipe.to(_UpperCAmelCase		)

									pipe.set_progress_bar_config(disable=_UpperCAmelCase		)

									_a     :      Tuple        =							pipe(**self.get_dummy_inputs(_UpperCAmelCase		)		)
									_a     :      Union[str, Any]        =							output.images

									_a     :      Any        =							pipe(
									    **self.get_dummy_inputs(_UpperCAmelCase		)	,return_dict=_UpperCAmelCase	,)[0]

									_a     :      List[Any]        =							image[0, -3:, -3:, -1]
									_a     :      int        =							image_from_tuple[0, -3:, -3:, -1]

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

									_a     :      Tuple        =							np.array(
									    [0.61_47_49_43, 0.6_07_35_39, 0.43_30_85_44, 0.5_92_82_69, 0.47_49_35_95, 0.46_75_59_73, 0.4_61_38_38, 0.45_36_87_97, 0.50_11_92_33]		)
									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 	__magic_name__						(    unittest.TestCase   ):







							def 							__lowercase			(							self					:       str		):
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()







							def 							__lowercase			(							self					:       List[str]		):
									_a     :      Optional[Any]        =							load_numpy(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
									    '/kandinsky/kandinsky_img2img_frog.npy'		)

									_a     :      Any        =							load_image(
									    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main' '/kandinsky/cat.png'		)
									_a     :      Optional[int]        =							'A red cartoon frog, 4k'

									_a     :      List[Any]        =							KandinskyPriorPipeline.from_pretrained(
									    'kandinsky-community/kandinsky-2-1-prior'	,torch_dtype=torch.floataa		)
									pipe_prior.to(_UpperCAmelCase		)

									_a     :      List[Any]        =							KandinskyImgaImgPipeline.from_pretrained(
									    'kandinsky-community/kandinsky-2-1'	,torch_dtype=torch.floataa		)
									_a     :      List[Any]        =							pipeline.to(_UpperCAmelCase		)

									pipeline.set_progress_bar_config(disable=_UpperCAmelCase		)

									_a     :      Any        =							torch.Generator(device='cpu'		).manual_seed(0		)
									_a       ,					_a     :      Dict        =							pipe_prior(
									    _UpperCAmelCase	,generator=_UpperCAmelCase	,num_inference_steps=5	,negative_prompt=''	,).to_tuple()

									_a     :      int        =							pipeline(
									    _UpperCAmelCase	,image=_UpperCAmelCase	,image_embeds=_UpperCAmelCase	,negative_image_embeds=_UpperCAmelCase	,generator=_UpperCAmelCase	,num_inference_steps=100	,height=768	,width=768	,strength=0.2	,output_type='np'	,)

									_a     :      str        =							output.images[0]

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

									assert_mean_pixel_difference(_UpperCAmelCase	,_UpperCAmelCase		)









'''simple docstring'''

import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np



def      __lowerCamelCase     (      lowerCAmelCase_						,			lowerCAmelCase_  )			->      np.array:
		_a     :      Optional[int]        =							f"""{sampling_rate}"""
		_a     :      Any        =							'1'
		_a     :      Optional[int]        =							'f32le'
		_a     :      Any        =							[
		    'ffmpeg',
		    '-i',
		    'pipe:0',
		    '-ac',
		    ac,
		    '-ar',
		    ar,
		    '-f',
		    format_for_conversion,
		    '-hide_banner',
		    '-loglevel',
		    'quiet',
		    'pipe:1',
		]

		try:
				with subprocess.Popen(lowerCAmelCase_						,			stdin=subprocess.PIPE						,			stdout=subprocess.PIPE  ) as ffmpeg_process:
						_a     :      int        =							ffmpeg_process.communicate(lowerCAmelCase_  )
		except FileNotFoundError as error:
				raise ValueError('ffmpeg was not found but is required to load audio files from filename'  ) from error
		_a     :      int        =							output_stream[0]
		_a     :      List[str]        =							np.frombuffer(lowerCAmelCase_						,			np.floataa  )
		if audio.shape[0] == 0:
				raise ValueError('Malformed soundfile'  )
		return audio



def      __lowerCamelCase     (      lowerCAmelCase_						,			lowerCAmelCase_						,			lowerCAmelCase_ = "f32le"						,			)			->      Union[str, Any]:
		_a     :      List[str]        =							f"""{sampling_rate}"""
		_a     :      List[str]        =							'1'
		if format_for_conversion == "s16le":
				_a     :      List[Any]        =							2
		elif format_for_conversion == "f32le":
				_a     :      Dict        =							4
		else:
				raise ValueError(f"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""  )

		_a     :      Any        =							platform.system()
		if system == "Linux":
				_a     :      Union[str, Any]        =							'alsa'
				_a     :      Union[str, Any]        =							'default'
		elif system == "Darwin":
				_a     :      Any        =							'avfoundation'
				_a     :      Optional[int]        =							':0'
		elif system == "Windows":
				_a     :      str        =							'dshow'
				_a     :      Tuple        =							'default'

		_a     :      str        =							[
		    'ffmpeg',
		    '-f',
		    format_,
		    '-i',
		    input_,
		    '-ac',
		    ac,
		    '-ar',
		    ar,
		    '-f',
		    format_for_conversion,
		    '-fflags',
		    'nobuffer',
		    '-hide_banner',
		    '-loglevel',
		    'quiet',
		    'pipe:1',
		]
		_a     :      Any        =							int(round(sampling_rate * chunk_length_s  )  ) * size_of_sample
		_a     :      Union[str, Any]        =							_ffmpeg_stream(lowerCAmelCase_						,			lowerCAmelCase_  )
		for item in iterator:
				yield item



def      __lowerCamelCase     (      lowerCAmelCase_						,			lowerCAmelCase_						,			lowerCAmelCase_ = None						,			lowerCAmelCase_ = None						,			lowerCAmelCase_ = "f32le"						,			)			->      str:
		if stream_chunk_s is not None:
				_a     :      str        =							stream_chunk_s
		else:
				_a     :      List[str]        =							chunk_length_s

		_a     :      int        =							ffmpeg_microphone(lowerCAmelCase_						,			lowerCAmelCase_						,			format_for_conversion=lowerCAmelCase_  )
		if format_for_conversion == "s16le":
				_a     :      Optional[Any]        =							np.intaa
				_a     :      List[Any]        =							2
		elif format_for_conversion == "f32le":
				_a     :      Tuple        =							np.floataa
				_a     :      Any        =							4
		else:
				raise ValueError(f"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""  )

		if stride_length_s is None:
				_a     :      str        =							chunk_length_s / 6
		_a     :      Any        =							int(round(sampling_rate * chunk_length_s  )  ) * size_of_sample
		if isinstance(lowerCAmelCase_						,			(int, float)  ):
				_a     :      List[str]        =							[stride_length_s, stride_length_s]

		_a     :      str        =							int(round(sampling_rate * stride_length_s[0]  )  ) * size_of_sample
		_a     :      List[str]        =							int(round(sampling_rate * stride_length_s[1]  )  ) * size_of_sample
		_a     :      Any        =							datetime.datetime.now()
		_a     :      Dict        =							datetime.timedelta(seconds=lowerCAmelCase_  )
		for item in chunk_bytes_iter(lowerCAmelCase_						,			lowerCAmelCase_						,			stride=(stride_left, stride_right)						,			stream=lowerCAmelCase_  ):
				# Put everything back in numpy scale
				_a     :      List[Any]        =							np.frombuffer(item['raw']						,			dtype=lowerCAmelCase_  )
				_a     :      List[str]        =							(
				    item['stride'][0] // size_of_sample,
				    item['stride'][1] // size_of_sample,
				)
				_a     :      Union[str, Any]        =							sampling_rate
				audio_time += delta
				if datetime.datetime.now() > audio_time + 10 * delta:
						# We're late !! SKIP
						continue
				yield item



def      __lowerCamelCase     (      lowerCAmelCase_						,			lowerCAmelCase_						,			lowerCAmelCase_						,			lowerCAmelCase_ = False  )			->      List[Any]:
		_a     :      Tuple        =							B''
		_a       ,					_a     :      str        =							stride
		if stride_left + stride_right >= chunk_len:
				raise ValueError(
				    f"""Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}"""  )
		_a     :      Optional[int]        =							0
		for raw in iterator:
				acc += raw
				if stream and len(lowerCAmelCase_  ) < chunk_len:
						_a     :      str        =							(_stride_left, 0)
						yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
				else:
						while len(lowerCAmelCase_  ) >= chunk_len:
								# We are flushing the accumulator
								_a     :      Union[str, Any]        =							(_stride_left, stride_right)
								_a     :      Dict        =							{'raw': acc[:chunk_len], 'stride': stride}
								if stream:
										_a     :      List[str]        =							False
								yield item
								_a     :      int        =							stride_left
								_a     :      List[Any]        =							acc[chunk_len - stride_left - stride_right :]
    # Last chunk
		if len(lowerCAmelCase_  ) > stride_left:
				_a     :      str        =							{'raw': acc, 'stride': (_stride_left, 0)}
				if stream:
						_a     :      str        =							False
				yield item



def      __lowerCamelCase     (      lowerCAmelCase_						,			lowerCAmelCase_  )			->      Tuple:
		_a     :      Optional[Any]        =							2**24  # 16Mo
		try:
				with subprocess.Popen(lowerCAmelCase_						,			stdout=subprocess.PIPE						,			bufsize=lowerCAmelCase_  ) as ffmpeg_process:
						while True:
								_a     :      Any        =							ffmpeg_process.stdout.read(lowerCAmelCase_  )
								if raw == b"":
										break
								yield raw
		except FileNotFoundError as error:
				raise ValueError('ffmpeg was not found but is required to stream audio files from filename'  ) from error




