
'''simple docstring'''





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					:			List[str]			      =		False, False, False


@dataclass
class 						__lowerCAmelCase	:



      """simple docstring"""







      A__	:			Optional[int]        =    None
      A__	:			bool        =    True
      A__	:			bool        =    True
      A__	:			Optional[str]        =    None
      # Automatically constructed
      A__	:			ClassVar[str]        =    "dict"
      A__	:			ClassVar[Any]        =    pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}      )
      A__	:			str        =    field(default='''Audio''' , init=lowerCAmelCase__ , repr=lowerCAmelCase__      )




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




      def 		snake_case_			(      self		:       List[Any]				,       _snake_case		:       Any				):





          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
              __lowercase		:			List[str]	    =       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!)
                      __lowercase		:			Optional[int]	    =       np.frombuffer(value['''bytes''']				,       dtype=np.intaa				).astype(np.floataa				) / 3_2767
                  else:
                      __lowercase		:			Dict	    =       np.memmap(value['''path''']				,       dtype='''h'''				,       mode='''r'''				).astype(np.floataa				) / 3_2767

                  __lowercase		:			Union[str, Any]	    =       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 		snake_case_			(      self		:       Any				,       _snake_case		:       List[str]				,       _snake_case		:       Union[str, Any] = None				):





          if not self.decode:
              raise RuntimeError('''Decoding is disabled for this feature. Please use Audio(decode=True) instead.'''				)

          __lowercase    ,      __lowercase		:			List[str]	    =       (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

          __lowercase		:			List[Any]	    =       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:
              __lowercase		:			Any	    =       token_per_repo_id or {}
              __lowercase		:			Any	    =       path.split('''::'''				)[-1]
              try:
                  __lowercase		:			str	    =       string_to_dict(_lowerCamelCase				,       config.HUB_DATASETS_URL				)['''repo_id''']
                  __lowercase		:			Dict	    =       token_per_repo_id[repo_id]
              except (ValueError, KeyError):
                  __lowercase		:			str	    =       None

              with xopen(_lowerCamelCase				,       '''rb'''				,       use_auth_token=_lowerCamelCase				) as f:
                  __lowercase    ,      __lowercase		:			List[Any]	    =       sf.read(_lowerCamelCase				)

          else:
              __lowercase    ,      __lowercase		:			Dict	    =       sf.read(_lowerCamelCase				)

          __lowercase		:			Tuple	    =       array.T
          if self.mono:
              __lowercase		:			Dict	    =       librosa.to_mono(_lowerCamelCase				)
          if self.sampling_rate and self.sampling_rate != sampling_rate:
              __lowercase		:			str	    =       librosa.resample(_lowerCamelCase				,       orig_sr=_lowerCamelCase				,       target_sr=self.sampling_rate				)
              __lowercase		:			Optional[Any]	    =       self.sampling_rate

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




      def 		snake_case_			(      self		:       Dict				):
          from .features import Value

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




      def 		snake_case_			(      self		:       Optional[Any]				,       _snake_case		:       Tuple				):





          if pa.types.is_string(storage.type				):
              __lowercase		:			Optional[Any]	    =       pa.array([None] * len(_lowerCamelCase				)				,       type=pa.binary()				)
              __lowercase		:			Optional[Any]	    =       pa.StructArray.from_arrays([bytes_array, storage]				,       ['''bytes''', '''path''']				,       mask=storage.is_null()				)
          elif pa.types.is_binary(storage.type				):
              __lowercase		:			Optional[int]	    =       pa.array([None] * len(_lowerCamelCase				)				,       type=pa.string()				)
              __lowercase		:			int	    =       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'''				):
              __lowercase		:			Optional[int]	    =       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:
                  __lowercase		:			List[str]	    =       storage.field('''bytes'''				)
              else:
                  __lowercase		:			Optional[Any]	    =       pa.array([None] * len(_lowerCamelCase				)				,       type=pa.binary()				)
              if storage.type.get_field_index('''path'''				) >= 0:
                  __lowercase		:			Union[str, Any]	    =       storage.field('''path'''				)
              else:
                  __lowercase		:			List[str]	    =       pa.array([None] * len(_lowerCamelCase				)				,       type=pa.string()				)
              __lowercase		:			str	    =       pa.StructArray.from_arrays([bytes_array, path_array]				,       ['''bytes''', '''path''']				,       mask=storage.is_null()				)
          return array_cast(_lowerCamelCase				,       self.pa_type				)




      def 		snake_case_			(      self		:       Optional[Any]				,       _snake_case		:       List[str]				):





          @no_op_if_value_is_null
          def path_to_bytes(_snake_case		:       Union[str, Any]				):
              with xopen(_lowerCamelCase				,       '''rb'''				) as f:
                  __lowercase		:			List[str]	    =       f.read()
              return bytes_

          __lowercase		:			List[Any]	    =       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()				,       )
          __lowercase		:			Optional[Any]	    =       pa.array(
              [os.path.basename(_lowerCamelCase				) if path is not None else None for path in storage.field('''path'''				).to_pylist()]				,       type=pa.string()				,       )
          __lowercase		:			Tuple	    =       pa.StructArray.from_arrays([bytes_array, path_array]				,       ['''bytes''', '''path''']				,       mask=bytes_array.is_null()				)
          return array_cast(_lowerCamelCase				,       self.pa_type				)




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 						__lowerCAmelCase	:



      """simple docstring"""







      def 		snake_case_			(      self		:       str				):
          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Optional[Any]	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Tuple	    =       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				)
          __lowercase		:			List[str]	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       thresholding=_snake_case				,       dynamic_thresholding_ratio=0.95				,       sample_max_value=1.0				,       prediction_type='''epsilon'''				,       variance_type='''learned_range'''				,       )

          torch.manual_seed(0				)
          __lowercase		:			str	    =       IFWatermarker()

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




      def 		snake_case_			(      self		:       Optional[int]				):
          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       TaEncoderModel.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Tuple	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-t5'''				)

          torch.manual_seed(0				)
          __lowercase		:			Dict	    =       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.4_14				,       time_embedding_act_fn='''gelu'''				,       time_embedding_dim=32				,       )
          unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

          torch.manual_seed(0				)
          __lowercase		:			Dict	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       thresholding=_snake_case				,       dynamic_thresholding_ratio=0.95				,       sample_max_value=1.0				,       prediction_type='''epsilon'''				,       variance_type='''learned_range'''				,       )

          torch.manual_seed(0				)
          __lowercase		:			Optional[int]	    =       DDPMScheduler(
              num_train_timesteps=1000				,       beta_schedule='''squaredcos_cap_v2'''				,       beta_start=0.00_01				,       beta_end=0.02				,       )

          torch.manual_seed(0				)
          __lowercase		:			Any	    =       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 		snake_case_			(      self		:       Any				):
          __lowercase		:			Tuple	    =       self.get_dummy_components()
          __lowercase		:			Any	    =       self.pipeline_class(**_snake_case				)
          pipe.to(_snake_case				)
          pipe.set_progress_bar_config(disable=_snake_case				)

          __lowercase		:			List[str]	    =       self.get_dummy_inputs(_snake_case				)

          __lowercase		:			List[Any]	    =       inputs['''prompt''']
          __lowercase		:			int	    =       inputs['''generator''']
          __lowercase		:			Dict	    =       inputs['''num_inference_steps''']
          __lowercase		:			Optional[int]	    =       inputs['''output_type''']

          if "image" in inputs:
              __lowercase		:			Tuple	    =       inputs['''image''']
          else:
              __lowercase		:			Dict	    =       None

          if "mask_image" in inputs:
              __lowercase		:			Tuple	    =       inputs['''mask_image''']
          else:
              __lowercase		:			List[Any]	    =       None

          if "original_image" in inputs:
              __lowercase		:			List[Any]	    =       inputs['''original_image''']
          else:
              __lowercase		:			Optional[int]	    =       None

          __lowercase    ,      __lowercase		:			Optional[int]	    =       pipe.encode_prompt(_snake_case				)

          # inputs with prompt converted to embeddings
          __lowercase		:			int	    =       {
              '''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:
              __lowercase		:			Tuple	    =       image

          if mask_image is not None:
              __lowercase		:			List[str]	    =       mask_image

          if original_image is not None:
              __lowercase		:			List[Any]	    =       original_image

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

          __lowercase		:			List[str]	    =       pipe(**_snake_case				)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_snake_case				)
              __lowercase		:			List[Any]	    =       self.pipeline_class.from_pretrained(_snake_case				)
              pipe_loaded.to(_snake_case				)
              pipe_loaded.set_progress_bar_config(disable=_snake_case				)

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

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

          __lowercase		:			int	    =       self.get_dummy_inputs(_snake_case				)

          __lowercase		:			Union[str, Any]	    =       inputs['''generator''']
          __lowercase		:			Any	    =       inputs['''num_inference_steps''']
          __lowercase		:			Tuple	    =       inputs['''output_type''']

          # inputs with prompt converted to embeddings
          __lowercase		:			List[str]	    =       {
              '''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:
              __lowercase		:			Dict	    =       image

          if mask_image is not None:
              __lowercase		:			Tuple	    =       mask_image

          if original_image is not None:
              __lowercase		:			Any	    =       original_image

          __lowercase		:			List[str]	    =       pipe_loaded(**_snake_case				)[0]

          __lowercase		:			Optional[int]	    =       np.abs(to_np(_snake_case				) - to_np(_snake_case				)				).max()
          self.assertLess(_snake_case				,       1E-4				)




      def 		snake_case_			(      self		:       Optional[int]				):
          __lowercase		:			Union[str, Any]	    =       self.get_dummy_components()
          __lowercase		:			Dict	    =       self.pipeline_class(**_snake_case				)
          pipe.to(_snake_case				)
          pipe.set_progress_bar_config(disable=_snake_case				)

          __lowercase		:			Dict	    =       self.get_dummy_inputs(_snake_case				)
          __lowercase		:			Any	    =       pipe(**_snake_case				)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_snake_case				)
              __lowercase		:			Dict	    =       self.pipeline_class.from_pretrained(_snake_case				)
              pipe_loaded.to(_snake_case				)
              pipe_loaded.set_progress_bar_config(disable=_snake_case				)

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

          __lowercase		:			Optional[Any]	    =       self.get_dummy_inputs(_snake_case				)
          __lowercase		:			str	    =       pipe_loaded(**_snake_case				)[0]

          __lowercase		:			int	    =       np.abs(to_np(_snake_case				) - to_np(_snake_case				)				).max()
          self.assertLess(_snake_case				,       1E-4				)


