

import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
      from .features import FeatureType


lowerCAmelCase__, lowerCAmelCase__, lowerCAmelCase__       = False, False, False






@dataclass
class 						__magic_name__    :
 UpperCAmelCase							    =						None
 UpperCAmelCase							    =						True
 UpperCAmelCase							    =						True
 UpperCAmelCase							    =						None
 # Automatically constructed
 UpperCAmelCase							    =						"dict"
 UpperCAmelCase							    =						pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}    )
 UpperCAmelCase							    =						field(default="""Audio"""	,						init=_snake_case	,						repr=_snake_case    )


 def __call__(						self		: List[str]    )    ->      List[Any]:
   return self.pa_type


 def _UpperCamelCase	(						self		: str			,						lowerCAmelCase__		: Union[str, bytes, dict]    )    ->      dict:
   try:
     import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
   except ImportError as err:
     raise ImportError("To support encoding audio data, please install 'soundfile'."    ) from err
   if isinstance(lowerCAmelCase__			,						lowerCAmelCase__    ):
     return {"bytes": None, "path": value}
   elif isinstance(lowerCAmelCase__			,						lowerCAmelCase__    ):
     return {"bytes": value, "path": None}
   elif "array" in value:
     # convert the audio array to wav bytes
     UpperCAmelCase					      =  BytesIO()
     sf.write(lowerCAmelCase__			,						value["array"]			,						value["sampling_rate"]			,						format="wav"    )
     return {"bytes": buffer.getvalue(), "path": None}
   elif value.get("path"    ) is not None and os.path.isfile(value["path"]    ):
     # we set "bytes": None to not duplicate the data if they're already available locally
     if value["path"].endswith("pcm"    ):
       # "PCM" only has raw audio bytes
       if value.get("sampling_rate"    ) is None:
         # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
         raise KeyError("To use PCM files, please specify a 'sampling_rate' in Audio object"    )
       if value.get("bytes"    ):
         # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
         UpperCAmelCase					      =  np.frombuffer(value["bytes"]			,						dtype=np.intaa    ).astype(np.floataa    ) / 3_2_7_6_7
       else:
         UpperCAmelCase					      =  np.memmap(value["path"]			,						dtype="h"			,						mode="r"    ).astype(np.floataa    ) / 3_2_7_6_7

       UpperCAmelCase					      =  BytesIO(bytes()    )
       sf.write(lowerCAmelCase__			,						lowerCAmelCase__			,						value["sampling_rate"]			,						format="wav"    )
       return {"bytes": buffer.getvalue(), "path": None}
     else:
       return {"bytes": None, "path": value.get("path"    )}
   elif value.get("bytes"    ) is not None or value.get("path"    ) is not None:
     # store the audio bytes, and path is used to infer the audio format using the file extension
     return {"bytes": value.get("bytes"    ), "path": value.get("path"    )}
   else:
     raise ValueError(
         f"An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."    )


 def _UpperCamelCase	(						self		: Any			,						lowerCAmelCase__		: dict			,						lowerCAmelCase__		: Optional[Dict[str, Union[str, bool, None]]] = None    )    ->      dict:
   if not self.decode:
     raise RuntimeError("Decoding is disabled for this feature. Please use Audio(decode=True) instead."    )

   UpperCAmelCase      ,	UpperCAmelCase					      =  (value["path"], BytesIO(value["bytes"]    )) if value["bytes"] is not None else (value["path"], None)
   if path is None and file is None:
     raise ValueError(f"An audio sample should have one of 'path' or 'bytes' but both are None in {value}."    )

   try:
     import librosa
     import soundfile as sf
   except ImportError as err:
     raise ImportError("To support decoding audio files, please install 'librosa' and 'soundfile'."    ) from err

   UpperCAmelCase					      =  xsplitext(lowerCAmelCase__    )[1][1:].lower() if path is not None else None
   if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
     raise RuntimeError(
         "Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, "
         "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "    )
   elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
     raise RuntimeError(
         "Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, "
         "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "    )

   if file is None:
     UpperCAmelCase					      =  token_per_repo_id or {}
     UpperCAmelCase					      =  path.split("::"    )[-1]
     try:
       UpperCAmelCase					      =  string_to_dict(lowerCAmelCase__			,						config.HUB_DATASETS_URL    )["repo_id"]
       UpperCAmelCase					      =  token_per_repo_id[repo_id]
     except (ValueError, KeyError):
       UpperCAmelCase					      =  None

     with xopen(lowerCAmelCase__			,						"rb"			,						use_auth_token=lowerCAmelCase__    ) as f:
       UpperCAmelCase      ,	UpperCAmelCase					      =  sf.read(lowerCAmelCase__    )

   else:
     UpperCAmelCase      ,	UpperCAmelCase					      =  sf.read(lowerCAmelCase__    )

   UpperCAmelCase					      =  array.T
   if self.mono:
     UpperCAmelCase					      =  librosa.to_mono(lowerCAmelCase__    )
   if self.sampling_rate and self.sampling_rate != sampling_rate:
     UpperCAmelCase					      =  librosa.resample(lowerCAmelCase__			,						orig_sr=lowerCAmelCase__			,						target_sr=self.sampling_rate    )
     UpperCAmelCase					      =  self.sampling_rate

   return {"path": path, "array": array, "sampling_rate": sampling_rate}


 def _UpperCamelCase	(						self		: Any    )    ->      Union["FeatureType", Dict[str, "FeatureType"]]:
   from .features import Value

   if self.decode:
     raise ValueError("Cannot flatten a decoded Audio feature."    )
   return {
       "bytes": Value("binary"    ),
       "path": Value("string"    ),
   }


 def _UpperCamelCase	(						self		: Dict			,						lowerCAmelCase__		: Union[pa.StringArray, pa.StructArray]    )    ->      pa.StructArray:
   if pa.types.is_string(storage.type    ):
     UpperCAmelCase					      =  pa.array([None] * len(lowerCAmelCase__    )			,						type=pa.binary()    )
     UpperCAmelCase					      =  pa.StructArray.from_arrays([bytes_array, storage]			,						["bytes", "path"]			,						mask=storage.is_null()    )
   elif pa.types.is_binary(storage.type    ):
     UpperCAmelCase					      =  pa.array([None] * len(lowerCAmelCase__    )			,						type=pa.string()    )
     UpperCAmelCase					      =  pa.StructArray.from_arrays([storage, path_array]			,						["bytes", "path"]			,						mask=storage.is_null()    )
   elif pa.types.is_struct(storage.type    ) and storage.type.get_all_field_indices("array"    ):
     UpperCAmelCase					      =  pa.array([Audio().encode_example(lowerCAmelCase__    ) if x is not None else None for x in storage.to_pylist()]    )
   elif pa.types.is_struct(storage.type    ):
     if storage.type.get_field_index("bytes"    ) >= 0:
       UpperCAmelCase					      =  storage.field("bytes"    )
     else:
       UpperCAmelCase					      =  pa.array([None] * len(lowerCAmelCase__    )			,						type=pa.binary()    )
     if storage.type.get_field_index("path"    ) >= 0:
       UpperCAmelCase					      =  storage.field("path"    )
     else:
       UpperCAmelCase					      =  pa.array([None] * len(lowerCAmelCase__    )			,						type=pa.string()    )
     UpperCAmelCase					      =  pa.StructArray.from_arrays([bytes_array, path_array]			,						["bytes", "path"]			,						mask=storage.is_null()    )
   return array_cast(lowerCAmelCase__			,						self.pa_type    )


 def _UpperCamelCase	(						self		: Dict			,						lowerCAmelCase__		: pa.StructArray    )    ->      pa.StructArray:

   @no_op_if_value_is_null
   def path_to_bytes(lowerCAmelCase__		: List[Any]    ):
     with xopen(lowerCAmelCase__			,						"rb"    ) as f:
       UpperCAmelCase					      =  f.read()
     return bytes_

   UpperCAmelCase					      =  pa.array(
       [
           (path_to_bytes(x["path"]    ) if x["bytes"] is None else x["bytes"]) if x is not None else None
           for x in storage.to_pylist()
       ]			,						type=pa.binary()			,						)
   UpperCAmelCase					      =  pa.array(
       [os.path.basename(lowerCAmelCase__    ) if path is not None else None for path in storage.field("path"    ).to_pylist()]			,						type=pa.string()			,						)
   UpperCAmelCase					      =  pa.StructArray.from_arrays([bytes_array, path_array]			,						["bytes", "path"]			,						mask=bytes_array.is_null()    )
   return array_cast(lowerCAmelCase__			,						self.pa_type    )






import gc
import random
import unittest

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

from diffusers import DDIMScheduler, KandinskyInpaintPipeline, 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__    (       _snake_case	,						unittest.TestCase    ):
 UpperCAmelCase							    =						KandinskyInpaintPipeline
 UpperCAmelCase							    =						["""prompt""", """image_embeds""", """negative_image_embeds""", """image""", """mask_image"""]
 UpperCAmelCase							    =						[
     """prompt""",
     """negative_prompt""",
     """image_embeds""",
     """negative_image_embeds""",
     """image""",
     """mask_image""",
 ]
 UpperCAmelCase							    =						[
     """generator""",
     """height""",
     """width""",
     """latents""",
     """guidance_scale""",
     """negative_prompt""",
     """num_inference_steps""",
     """return_dict""",
     """guidance_scale""",
     """num_images_per_prompt""",
     """output_type""",
     """return_dict""",
 ]
 UpperCAmelCase							    =						False


 @property
 def _UpperCamelCase	(						self		: Union[str, Any]    )    ->      Tuple:
   return 3_2


 @property
 def _UpperCamelCase	(						self		: int    )    ->      List[Any]:
   return 3_2


 @property
 def _UpperCamelCase	(						self		: List[Any]    )    ->      List[Any]:
   return self.time_input_dim


 @property
 def _UpperCamelCase	(						self		: Tuple    )    ->      Tuple:
   return self.time_input_dim * 4


 @property
 def _UpperCamelCase	(						self		: Any    )    ->      Optional[int]:
   return 1_0_0


 @property
 def _UpperCamelCase	(						self		: Optional[int]    )    ->      Optional[int]:
   UpperCAmelCase					      =  XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"    )
   return tokenizer


 @property
 def _UpperCamelCase	(						self		: int    )    ->      Dict:
   torch.manual_seed(0    )
   UpperCAmelCase					      =  MCLIPConfig(
       numDims=self.cross_attention_dim			,						transformerDimensions=self.text_embedder_hidden_size			,						hidden_size=self.text_embedder_hidden_size			,						intermediate_size=3_7			,						num_attention_heads=4			,						num_hidden_layers=5			,						vocab_size=1_0_0_5			,						)

   UpperCAmelCase					      =  MultilingualCLIP(lowerCAmelCase__    )
   UpperCAmelCase					      =  text_encoder.eval()

   return text_encoder


 @property
 def _UpperCamelCase	(						self		: Dict    )    ->      Optional[int]:
   torch.manual_seed(0    )

   UpperCAmelCase					      =  {
       "in_channels": 9,
       # 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,
   }

   UpperCAmelCase					      =  UNetaDConditionModel(**lowerCAmelCase__    )
   return model


 @property
 def _UpperCamelCase	(						self		: str    )    ->      Optional[Any]:
   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		: Dict    )    ->      List[Any]:
   torch.manual_seed(0    )
   UpperCAmelCase					      =  VQModel(**self.dummy_movq_kwargs    )
   return model


 def _UpperCamelCase	(						self		: Tuple    )    ->      Any:
   UpperCAmelCase					      =  self.dummy_text_encoder
   UpperCAmelCase					      =  self.dummy_tokenizer
   UpperCAmelCase					      =  self.dummy_unet
   UpperCAmelCase					      =  self.dummy_movq

   UpperCAmelCase					      =  DDIMScheduler(
       num_train_timesteps=1_0_0_0			,						beta_schedule="linear"			,						beta_start=0.00_085			,						beta_end=0.012			,						clip_sample=lowerCAmelCase__			,						set_alpha_to_one=lowerCAmelCase__			,						steps_offset=1			,						prediction_type="epsilon"			,						thresholding=lowerCAmelCase__			,						)

   UpperCAmelCase					      =  {
       "text_encoder": text_encoder,
       "tokenizer": tokenizer,
       "unet": unet,
       "scheduler": scheduler,
       "movq": movq,
   }

   return components


 def _UpperCamelCase	(						self		: Union[str, Any]			,						lowerCAmelCase__		: Any			,						lowerCAmelCase__		: Tuple=0    )    ->      str:
   UpperCAmelCase					      =  floats_tensor((1, self.cross_attention_dim)			,						rng=random.Random(lowerCAmelCase__    )    ).to(lowerCAmelCase__    )
   UpperCAmelCase					      =  floats_tensor((1, self.cross_attention_dim)			,						rng=random.Random(seed + 1    )    ).to(lowerCAmelCase__    )
   # create init_image
   UpperCAmelCase					      =  floats_tensor((1, 3, 6_4, 6_4)			,						rng=random.Random(lowerCAmelCase__    )    ).to(lowerCAmelCase__    )
   UpperCAmelCase					      =  image.cpu().permute(0			,						2			,						3			,						1    )[0]
   UpperCAmelCase					      =  Image.fromarray(np.uinta(lowerCAmelCase__    )    ).convert("RGB"    ).resize((2_5_6, 2_5_6)    )
   # create mask
   UpperCAmelCase					      =  np.ones((6_4, 6_4)			,						dtype=np.floataa    )
   UpperCAmelCase					      =  0

   if str(lowerCAmelCase__    ).startswith("mps"    ):
     UpperCAmelCase					      =  torch.manual_seed(lowerCAmelCase__    )
   else:
     UpperCAmelCase					      =  torch.Generator(device=lowerCAmelCase__    ).manual_seed(lowerCAmelCase__    )
   UpperCAmelCase					      =  {
       "prompt": "horse",
       "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		: Dict    )    ->      List[str]:
   UpperCAmelCase					      =  "cpu"

   UpperCAmelCase					      =  self.get_dummy_components()

   UpperCAmelCase					      =  self.pipeline_class(**lowerCAmelCase__    )
   UpperCAmelCase					      =  pipe.to(lowerCAmelCase__    )

   pipe.set_progress_bar_config(disable=lowerCAmelCase__    )

   UpperCAmelCase					      =  pipe(**self.get_dummy_inputs(lowerCAmelCase__    )    )
   UpperCAmelCase					      =  output.images

   UpperCAmelCase					      =  pipe(
       **self.get_dummy_inputs(lowerCAmelCase__    )			,						return_dict=lowerCAmelCase__			,						)[0]

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

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

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

   UpperCAmelCase					      =  np.array(
       [0.8_326_919, 0.73_790_467, 0.20_918_581, 0.9_309_612, 0.5_511_791, 0.43_713_328, 0.5_513_321, 0.49_922_934, 0.59_497_786]    )

   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		: str    )    ->      Tuple:
   super().test_inference_batch_single_identical(expected_max_diff=3e-3    )






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


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


 def _UpperCamelCase	(						self		: Tuple    )    ->      int:
   UpperCAmelCase					      =  load_numpy(
       "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
       "/kandinsky/kandinsky_inpaint_cat_with_hat_fp16.npy"    )

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

   UpperCAmelCase					      =  "a hat"

   UpperCAmelCase					      =  KandinskyPriorPipeline.from_pretrained(
       "kandinsky-community/kandinsky-2-1-prior"			,						torch_dtype=torch.floataa    )
   pipe_prior.to(lowerCAmelCase__    )

   UpperCAmelCase					      =  KandinskyInpaintPipeline.from_pretrained(
       "kandinsky-community/kandinsky-2-1-inpaint"			,						torch_dtype=torch.floataa    )
   UpperCAmelCase					      =  pipeline.to(lowerCAmelCase__    )
   pipeline.set_progress_bar_config(disable=lowerCAmelCase__    )

   UpperCAmelCase					      =  torch.Generator(device="cpu"    ).manual_seed(0    )
   UpperCAmelCase      ,	UpperCAmelCase					      =  pipe_prior(
       lowerCAmelCase__			,						generator=lowerCAmelCase__			,						num_inference_steps=5			,						negative_prompt=""			,						).to_tuple()

   UpperCAmelCase					      =  pipeline(
       lowerCAmelCase__			,						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"			,						)

   UpperCAmelCase					      =  output.images[0]

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

   assert_mean_pixel_difference(lowerCAmelCase__			,						lowerCAmelCase__    )



