
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


_A      ,						_A      ,						_A						      =				False, False, False







@dataclass
class   lowerCamelCase	:
     UpperCAmelCase__       :							Optional[int]     =	None
     UpperCAmelCase__       :							bool     =	True
     UpperCAmelCase__       :							bool     =	True
     UpperCAmelCase__       :							Optional[str]     =	None
     # Automatically constructed
     UpperCAmelCase__       :							ClassVar[str]     =	"dict"
     UpperCAmelCase__       :							ClassVar[Any]     =	pa.struct({"bytes": pa.binary(), "path": pa.string()}						)
     UpperCAmelCase__       :							str     =	field(default="Audio"							,				init=A_							,				repr=A_						)





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





     def 	UpperCAmelCase(self   :					str       ,					_A   :					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(_A       ,					_A     ):
             return {"bytes": None, "path": value}
         elif isinstance(_A       ,					_A     ):
             return {"bytes": value, "path": None}
         elif "array" in value:
             # convert the audio array to wav bytes
             snake_case				  =      BytesIO()
             sf.write(_A       ,					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!)
                     snake_case				  =      np.frombuffer(value["bytes"]       ,					dtype=np.intaa     ).astype(np.floataa     ) / 3_2_7_6_7
                 else:
                     snake_case				  =      np.memmap(value["path"]       ,					dtype="h"       ,					mode="r"     ).astype(np.floataa     ) / 3_2_7_6_7

                 snake_case				  =      BytesIO(bytes()     )
                 sf.write(_A       ,					_A       ,					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       ,					_A   :					dict       ,					_A   :					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."     )

         snake_case  ,							snake_case				  =      (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

         snake_case				  =      xsplitext(_A     )[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:
             snake_case				  =      token_per_repo_id or {}
             snake_case				  =      path.split("::"     )[-1]
             try:
                 snake_case				  =      string_to_dict(_A       ,					config.HUB_DATASETS_URL     )["repo_id"]
                 snake_case				  =      token_per_repo_id[repo_id]
             except (ValueError, KeyError):
                 snake_case				  =      None

             with xopen(_A       ,					"rb"       ,					use_auth_token=_A     ) as f:
                 snake_case  ,							snake_case				  =      sf.read(_A     )

         else:
             snake_case  ,							snake_case				  =      sf.read(_A     )

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

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





     def 	UpperCAmelCase(self   :					Optional[int]     )		->		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   :					str       ,					_A   :					Union[pa.StringArray, pa.StructArray]     )		->		pa.StructArray:
         if pa.types.is_string(storage.type     ):
             snake_case				  =      pa.array([None] * len(_A     )       ,					type=pa.binary()     )
             snake_case				  =      pa.StructArray.from_arrays([bytes_array, storage]       ,					["bytes", "path"]       ,					mask=storage.is_null()     )
         elif pa.types.is_binary(storage.type     ):
             snake_case				  =      pa.array([None] * len(_A     )       ,					type=pa.string()     )
             snake_case				  =      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"     ):
             snake_case				  =      pa.array([Audio().encode_example(_A     ) 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:
                 snake_case				  =      storage.field("bytes"     )
             else:
                 snake_case				  =      pa.array([None] * len(_A     )       ,					type=pa.binary()     )
             if storage.type.get_field_index("path"     ) >= 0:
                 snake_case				  =      storage.field("path"     )
             else:
                 snake_case				  =      pa.array([None] * len(_A     )       ,					type=pa.string()     )
             snake_case				  =      pa.StructArray.from_arrays([bytes_array, path_array]       ,					["bytes", "path"]       ,					mask=storage.is_null()     )
         return array_cast(_A       ,					self.pa_type     )





     def 	UpperCAmelCase(self   :					int       ,					_A   :					pa.StructArray     )		->		pa.StructArray:

         @no_op_if_value_is_null
         def path_to_bytes(_A   :					Optional[Any]     ):
             with xopen(_A       ,					"rb"     ) as f:
                 snake_case				  =      f.read()
             return bytes_

         snake_case				  =      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()       ,					)
         snake_case				  =      pa.array(
             [os.path.basename(_A     ) if path is not None else None for path in storage.field("path"     ).to_pylist()]       ,					type=pa.string()       ,					)
         snake_case				  =      pa.StructArray.from_arrays([bytes_array, path_array]       ,					["bytes", "path"]       ,					mask=bytes_array.is_null()     )
         return array_cast(_A       ,					self.pa_type     )



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   lowerCamelCase	(						A_							,				unittest.TestCase						):
     UpperCAmelCase__       :							Any     =	KandinskyVaaControlnetImgaImgPipeline
     UpperCAmelCase__       :							Optional[Any]     =	["image_embeds", "negative_image_embeds", "image", "hint"]
     UpperCAmelCase__       :							List[Any]     =	["image_embeds", "negative_image_embeds", "image", "hint"]
     UpperCAmelCase__       :							List[str]     =	[
         "generator",
         "height",
         "width",
         "strength",
         "guidance_scale",
         "num_inference_steps",
         "return_dict",
         "guidance_scale",
         "num_images_per_prompt",
         "output_type",
         "return_dict",
     ]
     UpperCAmelCase__       :							List[str]     =	False





     @property
     def 	UpperCAmelCase(self   :					Optional[int]     )		->		Dict:
         return 3_2





     @property
     def 	UpperCAmelCase(self   :					Union[str, Any]     )		->		Optional[int]:
         return 3_2





     @property
     def 	UpperCAmelCase(self   :					Optional[Any]     )		->		Union[str, Any]:
         return self.time_input_dim





     @property
     def 	UpperCAmelCase(self   :					Union[str, Any]     )		->		Union[str, Any]:
         return self.time_input_dim * 4





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





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

         snake_case				  =      {
             "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,
         }

         snake_case				  =      UNetaDConditionModel(**_A     )
         return model





     @property
     def 	UpperCAmelCase(self   :					Any     )		->		List[str]:
         return {
             "block_out_channels": [3_2, 3_2, 6_4, 6_4],
             "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": 1_2,
             "out_channels": 3,
             "up_block_types": ["AttnUpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"],
             "vq_embed_dim": 4,
         }





     @property
     def 	UpperCAmelCase(self   :					Union[str, Any]     )		->		List[str]:
         torch.manual_seed(0     )
         snake_case				  =      VQModel(**self.dummy_movq_kwargs     )
         return model





     def 	UpperCAmelCase(self   :					Any     )		->		Optional[Any]:
         snake_case				  =      self.dummy_unet
         snake_case				  =      self.dummy_movq

         snake_case				  =      {
             "num_train_timesteps": 1_0_0_0,
             "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,
         }

         snake_case				  =      DDIMScheduler(**_A     )

         snake_case				  =      {
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }

         return components





     def 	UpperCAmelCase(self   :					Any       ,					_A   :					Any       ,					_A   :					Optional[Any]=0     )		->		List[Any]:
         snake_case				  =      floats_tensor((1, self.text_embedder_hidden_size)       ,					rng=random.Random(_A     )     ).to(_A     )
         snake_case				  =      floats_tensor((1, self.text_embedder_hidden_size)       ,					rng=random.Random(seed + 1     )     ).to(
             _A     )
         # create init_image
         snake_case				  =      floats_tensor((1, 3, 6_4, 6_4)       ,					rng=random.Random(_A     )     ).to(_A     )
         snake_case				  =      image.cpu().permute(0       ,					2       ,					3       ,					1     )[0]
         snake_case				  =      Image.fromarray(np.uinta(_A     )     ).convert("RGB"     ).resize((2_5_6, 2_5_6)     )
         # create hint
         snake_case				  =      floats_tensor((1, 3, 6_4, 6_4)       ,					rng=random.Random(_A     )     ).to(_A     )

         if str(_A     ).startswith("mps"     ):
             snake_case				  =      torch.manual_seed(_A     )
         else:
             snake_case				  =      torch.Generator(device=_A     ).manual_seed(_A     )
         snake_case				  =      {
             "image": init_image,
             "image_embeds": image_embeds,
             "negative_image_embeds": negative_image_embeds,
             "hint": hint,
             "generator": generator,
             "height": 6_4,
             "width": 6_4,
             "num_inference_steps": 1_0,
             "guidance_scale": 7.0,
             "strength": 0.2,
             "output_type": "np",
         }
         return inputs





     def 	UpperCAmelCase(self   :					str     )		->		Union[str, Any]:
         snake_case				  =      "cpu"

         snake_case				  =      self.get_dummy_components()

         snake_case				  =      self.pipeline_class(**_A     )
         snake_case				  =      pipe.to(_A     )

         pipe.set_progress_bar_config(disable=_A     )

         snake_case				  =      pipe(**self.get_dummy_inputs(_A     )     )
         snake_case				  =      output.images

         snake_case				  =      pipe(
             **self.get_dummy_inputs(_A     )       ,					return_dict=_A       ,					)[0]

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

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

         snake_case				  =      np.array(
             [0.54_98_50_34, 0.55_50_93_65, 0.52_56_15_04, 0.5_57_04_94, 0.5_59_38_18, 0.5_26_39_79, 0.50_28_56_43, 0.5_06_98_46, 0.51_19_67_36]     )
         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   lowerCamelCase	(						unittest.TestCase						):





     def 	UpperCAmelCase(self   :					int     )		->		Optional[int]:
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





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

         snake_case				  =      load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"     )
         snake_case				  =      init_image.resize((5_1_2, 5_1_2)     )

         snake_case				  =      load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinskyv22/hint_image_cat.png"     )
         snake_case				  =      torch.from_numpy(np.array(_A     )     ).float() / 2_55.0
         snake_case				  =      hint.permute(2       ,					0       ,					1     ).unsqueeze(0     )

         snake_case				  =      "A robot, 4k photo"

         snake_case				  =      KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-prior"       ,					torch_dtype=torch.floataa     )
         pipe_prior.to(_A     )

         snake_case				  =      KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-controlnet-depth"       ,					torch_dtype=torch.floataa     )
         snake_case				  =      pipeline.to(_A     )

         pipeline.set_progress_bar_config(disable=_A     )

         snake_case				  =      torch.Generator(device="cpu"     ).manual_seed(0     )

         snake_case  ,							snake_case				  =      pipe_prior(
             _A       ,					image=_A       ,					strength=0.85       ,					generator=_A       ,					negative_prompt=""       ,					).to_tuple()

         snake_case				  =      pipeline(
             image=_A       ,					image_embeds=_A       ,					negative_image_embeds=_A       ,					hint=_A       ,					generator=_A       ,					num_inference_steps=1_0_0       ,					height=5_1_2       ,					width=5_1_2       ,					strength=0.5       ,					output_type="np"       ,					)

         snake_case				  =      output.images[0]

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

         assert_mean_pixel_difference(_A       ,					_A     )

