



import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np

class  __snake_case      :







							'''simple docstring'''






							def _a      (							self		):
											torch.manual_seed(0		)
											a__				      =		TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

											torch.manual_seed(0		)
											a__				      =		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

											torch.manual_seed(0		)
											a__				      =		UNetaDConditionModel(
											    sample_size=32  ,  layers_per_block=1  ,  block_out_channels=[32, 64]  ,  down_block_types=[
											        """ResnetDownsampleBlock2D""",
											        """SimpleCrossAttnDownBlock2D""",
											    ]  ,  mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""  ,  up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]  ,  in_channels=3  ,  out_channels=6  ,  cross_attention_dim=32  ,  encoder_hid_dim=32  ,  attention_head_dim=8  ,  addition_embed_type="""text"""  ,  addition_embed_type_num_heads=2  ,  cross_attention_norm="""group_norm"""  ,  resnet_time_scale_shift="""scale_shift"""  ,  act_fn="""gelu"""  ,  )
											unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

											torch.manual_seed(0		)
											a__				      =		DDPMScheduler(
											    num_train_timesteps=1_000  ,  beta_schedule="""squaredcos_cap_v2"""  ,  beta_start=0.0_001  ,  beta_end=0.02  ,  thresholding=a_  ,  dynamic_thresholding_ratio=0.95  ,  sample_max_value=1.0  ,  prediction_type="""epsilon"""  ,  variance_type="""learned_range"""  ,  )

											torch.manual_seed(0		)
											a__				      =		IFWatermarker()

											return {
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											    "unet": unet,
											    "scheduler": scheduler,
											    "watermarker": watermarker,
											    "safety_checker": None,
											    "feature_extractor": None,
											}


							def _a      (							self		):
											torch.manual_seed(0		)
											a__				      =		TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

											torch.manual_seed(0		)
											a__				      =		AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""		)

											torch.manual_seed(0		)
											a__				      =		UNetaDConditionModel(
											    sample_size=32  ,  layers_per_block=[1, 2]  ,  block_out_channels=[32, 64]  ,  down_block_types=[
											        """ResnetDownsampleBlock2D""",
											        """SimpleCrossAttnDownBlock2D""",
											    ]  ,  mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""  ,  up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]  ,  in_channels=6  ,  out_channels=6  ,  cross_attention_dim=32  ,  encoder_hid_dim=32  ,  attention_head_dim=8  ,  addition_embed_type="""text"""  ,  addition_embed_type_num_heads=2  ,  cross_attention_norm="""group_norm"""  ,  resnet_time_scale_shift="""scale_shift"""  ,  act_fn="""gelu"""  ,  class_embed_type="""timestep"""  ,  mid_block_scale_factor=1.414  ,  time_embedding_act_fn="""gelu"""  ,  time_embedding_dim=32  ,  )
											unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

											torch.manual_seed(0		)
											a__				      =		DDPMScheduler(
											    num_train_timesteps=1_000  ,  beta_schedule="""squaredcos_cap_v2"""  ,  beta_start=0.0_001  ,  beta_end=0.02  ,  thresholding=a_  ,  dynamic_thresholding_ratio=0.95  ,  sample_max_value=1.0  ,  prediction_type="""epsilon"""  ,  variance_type="""learned_range"""  ,  )

											torch.manual_seed(0		)
											a__				      =		DDPMScheduler(
											    num_train_timesteps=1_000  ,  beta_schedule="""squaredcos_cap_v2"""  ,  beta_start=0.0_001  ,  beta_end=0.02  ,  )

											torch.manual_seed(0		)
											a__				      =		IFWatermarker()

											return {
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											    "unet": unet,
											    "scheduler": scheduler,
											    "image_noising_scheduler": image_noising_scheduler,
											    "watermarker": watermarker,
											    "safety_checker": None,
											    "feature_extractor": None,
											}


							def _a      (							self		):
											a__				      =		self.get_dummy_components()
											a__				      =		self.pipeline_class(**a_		)
											pipe.to(a_		)
											pipe.set_progress_bar_config(disable=a_		)

											a__				      =		self.get_dummy_inputs(a_		)

											a__				      =		inputs["""prompt"""]
											a__				      =		inputs["""generator"""]
											a__				      =		inputs["""num_inference_steps"""]
											a__				      =		inputs["""output_type"""]

											if "image" in inputs:
															a__				      =		inputs["""image"""]
											else:
															a__				      =		None

											if "mask_image" in inputs:
															a__				      =		inputs["""mask_image"""]
											else:
															a__				      =		None

											if "original_image" in inputs:
															a__				      =		inputs["""original_image"""]
											else:
															a__				      =		None

											a__      ,			a__				      =		pipe.encode_prompt(a_		)

											# inputs with prompt converted to embeddings
											a__				      =		{
											    """prompt_embeds""": prompt_embeds,
											    """negative_prompt_embeds""": negative_prompt_embeds,
											    """generator""": generator,
											    """num_inference_steps""": num_inference_steps,
											    """output_type""": output_type,
											}

											if image is not None:
															a__				      =		image

											if mask_image is not None:
															a__				      =		mask_image

											if original_image is not None:
															a__				      =		original_image

											# set all optional components to None
											for optional_component in pipe._optional_components:
															setattr(a_  ,  a_  ,  a_		)

											a__				      =		pipe(**a_		)[0]

											with tempfile.TemporaryDirectory() as tmpdir:
															pipe.save_pretrained(a_		)
															a__				      =		self.pipeline_class.from_pretrained(a_		)
															pipe_loaded.to(a_		)
															pipe_loaded.set_progress_bar_config(disable=a_		)

											pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

											for optional_component in pipe._optional_components:
															self.assertTrue(
															    getattr(a_  ,  a_		) is None  ,  F'''`{optional_component}` did not stay set to None after loading.'''  ,  )

											a__				      =		self.get_dummy_inputs(a_		)

											a__				      =		inputs["""generator"""]
											a__				      =		inputs["""num_inference_steps"""]
											a__				      =		inputs["""output_type"""]

											# inputs with prompt converted to embeddings
											a__				      =		{
											    """prompt_embeds""": prompt_embeds,
											    """negative_prompt_embeds""": negative_prompt_embeds,
											    """generator""": generator,
											    """num_inference_steps""": num_inference_steps,
											    """output_type""": output_type,
											}

											if image is not None:
															a__				      =		image

											if mask_image is not None:
															a__				      =		mask_image

											if original_image is not None:
															a__				      =		original_image

											a__				      =		pipe_loaded(**a_		)[0]

											a__				      =		np.abs(to_np(a_		) - to_np(a_		)		).max()
											self.assertLess(a_  ,  1E-4		)







							def _a      (							self		):
											a__				      =		self.get_dummy_components()
											a__				      =		self.pipeline_class(**a_		)
											pipe.to(a_		)
											pipe.set_progress_bar_config(disable=a_		)

											a__				      =		self.get_dummy_inputs(a_		)
											a__				      =		pipe(**a_		)[0]

											with tempfile.TemporaryDirectory() as tmpdir:
															pipe.save_pretrained(a_		)
															a__				      =		self.pipeline_class.from_pretrained(a_		)
															pipe_loaded.to(a_		)
															pipe_loaded.set_progress_bar_config(disable=a_		)

											pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

											a__				      =		self.get_dummy_inputs(a_		)
											a__				      =		pipe_loaded(**a_		)[0]

											a__				      =		np.abs(to_np(a_		) - to_np(a_		)		).max()
											self.assertLess(a_  ,  1E-4		)






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


UpperCAmelCase ,       UpperCAmelCase ,       UpperCAmelCase							 =					False, False, False

@dataclass
class  __snake_case      :







							'''simple docstring'''






							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=SCREAMING_SNAKE_CASE			,repr=SCREAMING_SNAKE_CASE)


							def __call__(							self		):
											return self.pa_type


							def _a      (							self  ,  a_		):
											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
															a__				      =		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!)
																							a__				      =		np.frombuffer(value["""bytes"""]  ,  dtype=np.intaa		).astype(np.floataa		) / 32_767
																			else:
																							a__				      =		np.memmap(value["""path"""]  ,  dtype="""h"""  ,  mode="""r"""		).astype(np.floataa		) / 32_767

																			a__				      =		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 _a      (							self  ,  a_  ,  a_ = None		):
											if not self.decode:
															raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""		)

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

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

															with xopen(a_  ,  """rb"""  ,  use_auth_token=a_		) as f:
																			a__      ,			a__				      =		sf.read(a_		)

											else:
															a__      ,			a__				      =		sf.read(a_		)

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

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


							def _a      (							self		):
											from .features import Value

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


							def _a      (							self  ,  a_		):
											if pa.types.is_string(storage.type		):
															a__				      =		pa.array([None] * len(a_		)  ,  type=pa.binary()		)
															a__				      =		pa.StructArray.from_arrays([bytes_array, storage]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()		)
											elif pa.types.is_binary(storage.type		):
															a__				      =		pa.array([None] * len(a_		)  ,  type=pa.string()		)
															a__				      =		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"""		):
															a__				      =		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:
																			a__				      =		storage.field("""bytes"""		)
															else:
																			a__				      =		pa.array([None] * len(a_		)  ,  type=pa.binary()		)
															if storage.type.get_field_index("""path"""		) >= 0:
																			a__				      =		storage.field("""path"""		)
															else:
																			a__				      =		pa.array([None] * len(a_		)  ,  type=pa.string()		)
															a__				      =		pa.StructArray.from_arrays([bytes_array, path_array]  ,  ["""bytes""", """path"""]  ,  mask=storage.is_null()		)
											return array_cast(a_  ,  self.pa_type		)







							def _a      (							self  ,  a_		):

											@no_op_if_value_is_null
											def path_to_bytes(a_		):
															with xopen(a_  ,  """rb"""		) as f:
																			a__				      =		f.read()
															return bytes_

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

