





"""simple docstring"""





from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings

@maybe_allow_in_graph
class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase      ,						lowercase      ,						lowercase=0.0      ,						lowercase = None      ,						lowercase = "geglu"      ,						lowercase = None      ,						lowercase = False      ,						lowercase = False      ,						lowercase = False      ,						lowercase = False      ,						lowercase = True      ,						lowercase = "layer_norm"      ,						lowercase = False      ,						)		->     Dict:
            super().__init__()
            lowerCAmelCase  						=       only_cross_attention

            lowerCAmelCase  						=       (num_embeds_ada_norm is not None) and norm_type == """ada_norm_zero"""
            lowerCAmelCase  						=       (num_embeds_ada_norm is not None) and norm_type == """ada_norm"""

            if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
                  raise ValueError(
                      f'`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to'
                      f' define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}.' )

            # Define 3 blocks. Each block has its own normalization layer.
            # 1. Self-Attn
            if self.use_ada_layer_norm:
                  lowerCAmelCase  						=       AdaLayerNorm(lowercase      ,						lowercase )
            elif self.use_ada_layer_norm_zero:
                  lowerCAmelCase  						=       AdaLayerNormZero(lowercase      ,						lowercase )
            else:
                  lowerCAmelCase  						=       nn.LayerNorm(lowercase      ,						elementwise_affine=lowercase )
            lowerCAmelCase  						=       Attention(
                query_dim=lowercase      ,						heads=lowercase      ,						dim_head=lowercase      ,						dropout=lowercase      ,						bias=lowercase      ,						cross_attention_dim=cross_attention_dim if only_cross_attention else None      ,						upcast_attention=lowercase      ,						)

            # 2. Cross-Attn
            if cross_attention_dim is not None or double_self_attention:
                  # We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
                  # I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
                  # the second cross attention block.
                  lowerCAmelCase  						=       (
                      AdaLayerNorm(lowercase      ,						lowercase )
                      if self.use_ada_layer_norm
                      else nn.LayerNorm(lowercase      ,						elementwise_affine=lowercase )
                  )
                  lowerCAmelCase  						=       Attention(
                      query_dim=lowercase      ,						cross_attention_dim=cross_attention_dim if not double_self_attention else None      ,						heads=lowercase      ,						dim_head=lowercase      ,						dropout=lowercase      ,						bias=lowercase      ,						upcast_attention=lowercase      ,						)  # is self-attn if encoder_hidden_states is none
            else:
                  lowerCAmelCase  						=       None
                  lowerCAmelCase  						=       None

            # 3. Feed-forward
            lowerCAmelCase  						=       nn.LayerNorm(lowercase      ,						elementwise_affine=lowercase )
            lowerCAmelCase  						=       FeedForward(lowercase      ,						dropout=lowercase      ,						activation_fn=lowercase      ,						final_dropout=lowercase )

            # let chunk size default to None
            lowerCAmelCase  						=       None
            lowerCAmelCase  						=       0





      def     _snake_case			(       self      ,						lowercase      ,						lowercase )		->     List[Any]:
            # Sets chunk feed-forward
            lowerCAmelCase  						=       chunk_size
            lowerCAmelCase  						=       dim





      def     _snake_case			(       self      ,						lowercase      ,						lowercase = None      ,						lowercase = None      ,						lowercase = None      ,						lowercase = None      ,						lowercase = None      ,						lowercase = None      ,						)		->     List[str]:
            # Notice that normalization is always applied before the real computation in the following blocks.
            # 1. Self-Attention
            if self.use_ada_layer_norm:
                  lowerCAmelCase  						=       self.norma(lowercase      ,						lowercase )
            elif self.use_ada_layer_norm_zero:
                  lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase  						=       self.norma(
                      lowercase      ,						lowercase      ,						lowercase      ,						hidden_dtype=hidden_states.dtype )
            else:
                  lowerCAmelCase  						=       self.norma(lowercase )

            lowerCAmelCase  						=       cross_attention_kwargs if cross_attention_kwargs is not None else {}

            lowerCAmelCase  						=       self.attna(
                lowercase      ,						encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None      ,						attention_mask=lowercase      ,						**lowercase      ,						)
            if self.use_ada_layer_norm_zero:
                  lowerCAmelCase  						=       gate_msa.unsqueeze(1 ) * attn_output
            lowerCAmelCase  						=       attn_output + hidden_states

            # 2. Cross-Attention
            if self.attna is not None:
                  lowerCAmelCase  						=       (
                      self.norma(lowercase      ,						lowercase ) if self.use_ada_layer_norm else self.norma(lowercase )
                  )

                  lowerCAmelCase  						=       self.attna(
                      lowercase      ,						encoder_hidden_states=lowercase      ,						attention_mask=lowercase      ,						**lowercase      ,						)
                  lowerCAmelCase  						=       attn_output + hidden_states

            # 3. Feed-forward
            lowerCAmelCase  						=       self.norma(lowercase )

            if self.use_ada_layer_norm_zero:
                  lowerCAmelCase  						=       norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

            if self._chunk_size is not None:
                  # "feed_forward_chunk_size" can be used to save memory
                  if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
                        raise ValueError(
                            f'`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`.' )

                  lowerCAmelCase  						=       norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
                  lowerCAmelCase  						=       torch.cat(
                      [self.ff(lowercase ) for hid_slice in norm_hidden_states.chunk(lowercase      ,						dim=self._chunk_dim )]      ,						dim=self._chunk_dim      ,						)
            else:
                  lowerCAmelCase  						=       self.ff(lowercase )

            if self.use_ada_layer_norm_zero:
                  lowerCAmelCase  						=       gate_mlp.unsqueeze(1 ) * ff_output

            lowerCAmelCase  						=       ff_output + hidden_states

            return hidden_states







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase = None      ,						lowercase = 4      ,						lowercase = 0.0      ,						lowercase = "geglu"      ,						lowercase = False      ,						)		->     Any:
            super().__init__()
            lowerCAmelCase  						=       int(dim * mult )
            lowerCAmelCase  						=       dim_out if dim_out is not None else dim

            if activation_fn == "gelu":
                  lowerCAmelCase  						=       GELU(lowercase      ,						lowercase )
            if activation_fn == "gelu-approximate":
                  lowerCAmelCase  						=       GELU(lowercase      ,						lowercase      ,						approximate="""tanh""" )
            elif activation_fn == "geglu":
                  lowerCAmelCase  						=       GEGLU(lowercase      ,						lowercase )
            elif activation_fn == "geglu-approximate":
                  lowerCAmelCase  						=       ApproximateGELU(lowercase      ,						lowercase )

            lowerCAmelCase  						=       nn.ModuleList([] )
            # project in
            self.net.append(lowercase )
            # project dropout
            self.net.append(nn.Dropout(lowercase ) )
            # project out
            self.net.append(nn.Linear(lowercase      ,						lowercase ) )
            # FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
            if final_dropout:
                  self.net.append(nn.Dropout(lowercase ) )





      def     _snake_case			(       self      ,						lowercase )		->     Tuple:
            for module in self.net:
                  lowerCAmelCase  						=       module(lowercase )
            return hidden_states







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase      ,						lowercase = "none" )		->     Optional[int]:
            super().__init__()
            lowerCAmelCase  						=       nn.Linear(lowercase      ,						lowercase )
            lowerCAmelCase  						=       approximate





      def     _snake_case			(       self      ,						lowercase )		->     Optional[Any]:
            if gate.device.type != "mps":
                  return F.gelu(lowercase      ,						approximate=self.approximate )
            # mps: gelu is not implemented for float16
            return F.gelu(gate.to(dtype=torch.floataa )      ,						approximate=self.approximate ).to(dtype=gate.dtype )





      def     _snake_case			(       self      ,						lowercase )		->     List[str]:
            lowerCAmelCase  						=       self.proj(lowercase )
            lowerCAmelCase  						=       self.gelu(lowercase )
            return hidden_states







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase )		->     int:
            super().__init__()
            lowerCAmelCase  						=       nn.Linear(lowercase      ,						dim_out * 2 )





      def     _snake_case			(       self      ,						lowercase )		->     Dict:
            if gate.device.type != "mps":
                  return F.gelu(lowercase )
            # mps: gelu is not implemented for float16
            return F.gelu(gate.to(dtype=torch.floataa ) ).to(dtype=gate.dtype )





      def     _snake_case			(       self      ,						lowercase )		->     Union[str, Any]:
            lowerCAmelCase						,		lowerCAmelCase  						=       self.proj(lowercase ).chunk(2      ,						dim=-1 )
            return hidden_states * self.gelu(lowercase )







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase )		->     Union[str, Any]:
            super().__init__()
            lowerCAmelCase  						=       nn.Linear(lowercase      ,						lowercase )





      def     _snake_case			(       self      ,						lowercase )		->     int:
            lowerCAmelCase  						=       self.proj(lowercase )
            return x * torch.sigmoid(1.702 * x )







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase )		->     Tuple:
            super().__init__()
            lowerCAmelCase  						=       nn.Embedding(lowercase      ,						lowercase )
            lowerCAmelCase  						=       nn.SiLU()
            lowerCAmelCase  						=       nn.Linear(lowercase      ,						embedding_dim * 2 )
            lowerCAmelCase  						=       nn.LayerNorm(lowercase      ,						elementwise_affine=lowercase )





      def     _snake_case			(       self      ,						lowercase      ,						lowercase )		->     Any:
            lowerCAmelCase  						=       self.linear(self.silu(self.emb(lowercase ) ) )
            lowerCAmelCase						,		lowerCAmelCase  						=       torch.chunk(lowercase      ,						2 )
            lowerCAmelCase  						=       self.norm(lowercase ) * (1 + scale) + shift
            return x







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase )		->     Any:
            super().__init__()

            lowerCAmelCase  						=       CombinedTimestepLabelEmbeddings(lowercase      ,						lowercase )

            lowerCAmelCase  						=       nn.SiLU()
            lowerCAmelCase  						=       nn.Linear(lowercase      ,						6 * embedding_dim      ,						bias=lowercase )
            lowerCAmelCase  						=       nn.LayerNorm(lowercase      ,						elementwise_affine=lowercase      ,						eps=1e-6 )





      def     _snake_case			(       self      ,						lowercase      ,						lowercase      ,						lowercase      ,						lowercase=None )		->     int:
            lowerCAmelCase  						=       self.linear(self.silu(self.emb(lowercase      ,						lowercase      ,						hidden_dtype=lowercase ) ) )
            lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase						,		lowerCAmelCase  						=       emb.chunk(6      ,						dim=1 )
            lowerCAmelCase  						=       self.norm(lowercase ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
            return x, gate_msa, shift_mlp, scale_mlp, gate_mlp







class 		lowercase					(					nn.Module					):





      def __init__(       self      ,						lowercase      ,						lowercase      ,						lowercase      ,						lowercase = None      ,						lowercase = 1e-5 )		->     int:
            super().__init__()
            lowerCAmelCase  						=       num_groups
            lowerCAmelCase  						=       eps

            if act_fn is None:
                  lowerCAmelCase  						=       None
            else:
                  lowerCAmelCase  						=       get_activation(lowercase )

            lowerCAmelCase  						=       nn.Linear(lowercase      ,						out_dim * 2 )





      def     _snake_case			(       self      ,						lowercase      ,						lowercase )		->     Union[str, Any]:
            if self.act:
                  lowerCAmelCase  						=       self.act(lowercase )
            lowerCAmelCase  						=       self.linear(lowercase )
            lowerCAmelCase  						=       emb[:, :, None, None]
            lowerCAmelCase						,		lowerCAmelCase  						=       emb.chunk(2      ,						dim=1 )

            lowerCAmelCase  						=       F.group_norm(lowercase      ,						self.num_groups      ,						eps=self.eps )
            lowerCAmelCase  						=       x * (1 + scale) + shift
            return x













"""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


SCREAMING_SNAKE_CASE__,      SCREAMING_SNAKE_CASE__,      SCREAMING_SNAKE_CASE__					=   False, False, False

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





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





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

                        lowerCAmelCase  						=       BytesIO(bytes() )
                        sf.write(lowercase      ,						lowercase      ,						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      ,						lowercase      ,						lowercase = None )		->     dict:
            if not self.decode:
                  raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead.""" )

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

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

                  with xopen(lowercase      ,						"""rb"""      ,						use_auth_token=lowercase ) as f:
                        lowerCAmelCase						,		lowerCAmelCase  						=       sf.read(lowercase )

            else:
                  lowerCAmelCase						,		lowerCAmelCase  						=       sf.read(lowercase )

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

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





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





      def     _snake_case			(       self      ,						lowercase )		->     pa.StructArray:

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

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






