



'''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        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       )      ->      np.array:
  __lowerCamelCase       :       Any           = F'{sampling_rate}'
  __lowerCamelCase       :       List[str]           = '1'
  __lowerCamelCase       :       int           = 'f32le'
  __lowerCamelCase       :       Dict           = [
      '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:
      __lowerCamelCase       :       Tuple           = 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
  __lowerCamelCase       :       Any           = output_stream[0]
  __lowerCamelCase       :       Union[str, Any]           = np.frombuffer(_lowerCAmelCase       ,np.floataa       )
  if audio.shape[0] == 0:
    raise ValueError('Malformed soundfile'       )
  return audio

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase = "f32le"       ,)      ->      Dict:
  __lowerCamelCase       :       Optional[Any]           = F'{sampling_rate}'
  __lowerCamelCase       :       Optional[int]           = '1'
  if format_for_conversion == "s16le":
    __lowerCamelCase       :       List[Any]           = 2
  elif format_for_conversion == "f32le":
    __lowerCamelCase       :       Tuple           = 4
  else:
    raise ValueError(F'Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'       )

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

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

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase = None       ,_lowerCAmelCase = None       ,_lowerCAmelCase = "f32le"       ,)      ->      List[str]:
  if stream_chunk_s is not None:
    __lowerCamelCase       :       int           = stream_chunk_s
  else:
    __lowerCamelCase       :       List[Any]           = chunk_length_s

  __lowerCamelCase       :       Dict           = ffmpeg_microphone(_lowerCAmelCase       ,_lowerCAmelCase       ,format_for_conversion=_lowerCAmelCase       )
  if format_for_conversion == "s16le":
    __lowerCamelCase       :       List[str]           = np.intaa
    __lowerCamelCase       :       Union[str, Any]           = 2
  elif format_for_conversion == "f32le":
    __lowerCamelCase       :       Union[str, Any]           = np.floataa
    __lowerCamelCase       :       Optional[Any]           = 4
  else:
    raise ValueError(F'Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'       )

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

  __lowerCamelCase       :       Union[str, Any]           = int(round(sampling_rate * stride_length_s[0]       )       ) * size_of_sample
  __lowerCamelCase       :       Optional[Any]           = int(round(sampling_rate * stride_length_s[1]       )       ) * size_of_sample
  __lowerCamelCase       :       Dict           = datetime.datetime.now()
  __lowerCamelCase       :       Any           = 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
    __lowerCamelCase       :       Optional[int]           = np.frombuffer(item['raw']       ,dtype=_lowerCAmelCase       )
    __lowerCamelCase       :       Tuple           = (
        item['stride'][0] // size_of_sample,
        item['stride'][1] // size_of_sample,
    )
    __lowerCamelCase       :       Optional[int]           = sampling_rate
    audio_time += delta
    if datetime.datetime.now() > audio_time + 10 * delta:
      # We're late !! SKIP
      continue
    yield item

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase = False       )      ->      str:
  __lowerCamelCase       :       Optional[int]           = b''
  __lowerCamelCase       ,__lowerCamelCase       :       Any           = 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}'       )
  __lowerCamelCase       :       str           = 0
  for raw in iterator:
    acc += raw
    if stream and len(_lowerCAmelCase       ) < chunk_len:
      __lowerCamelCase       :       Any           = (_stride_left, 0)
      yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
    else:
      while len(_lowerCAmelCase       ) >= chunk_len:
        # We are flushing the accumulator
        __lowerCamelCase       :       Any           = (_stride_left, stride_right)
        __lowerCamelCase       :       Optional[int]           = {'raw': acc[:chunk_len], 'stride': stride}
        if stream:
          __lowerCamelCase       :       List[str]           = False
        yield item
        __lowerCamelCase       :       Tuple           = stride_left
        __lowerCamelCase       :       Union[str, Any]           = acc[chunk_len - stride_left - stride_right :]
    # Last chunk
  if len(_lowerCAmelCase       ) > stride_left:
    __lowerCamelCase       :       Tuple           = {'raw': acc, 'stride': (_stride_left, 0)}
    if stream:
      __lowerCamelCase       :       List[str]           = False
    yield item


def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       )      ->      Tuple:
  __lowerCamelCase       :       int           = 2**24  # 16Mo
  try:
    with subprocess.Popen(_lowerCAmelCase       ,stdout=subprocess.PIPE       ,bufsize=_lowerCAmelCase       ) as ffmpeg_process:
      while True:
        __lowerCamelCase       :       Union[str, 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






