

"""simple docstring"""





import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


_a							=							logging.get_logger(__name__)




class  _lowerCAmelCase  (     lowercase       ):







 """simple docstring"""







 __UpperCAmelCase     : List[Any]       		=     ["input_features"]


 def __init__(      self							:      Any, UpperCAmelCase__							:      Union[str, Any]=8_0, UpperCAmelCase__							:      Union[str, Any]=1_6_0_0_0, UpperCAmelCase__							:      Optional[int]=1_6_0, UpperCAmelCase__							:      List[str]=3_0, UpperCAmelCase__							:      Optional[int]=4_0_0, UpperCAmelCase__							:      Optional[Any]=0.0, UpperCAmelCase__							:      List[Any]=False, **UpperCAmelCase__							:      List[Any], ):
    super().__init__(
        feature_size=UpperCAmelCase__, sampling_rate=UpperCAmelCase__, padding_value=UpperCAmelCase__, return_attention_mask=UpperCAmelCase__, **UpperCAmelCase__, )
    __lowercase								=						n_fft
    __lowercase								=						hop_length
    __lowercase								=						chunk_length
    __lowercase								=						chunk_length * sampling_rate
    __lowercase								=						self.n_samples // hop_length
    __lowercase								=						sampling_rate
    __lowercase								=						mel_filter_bank(
        num_frequency_bins=1 + n_fft // 2, num_mel_filters=UpperCAmelCase__, min_frequency=0.0, max_frequency=8_000.0, sampling_rate=UpperCAmelCase__, norm="slaney", mel_scale="slaney", )


 def  _lowercase			(      self							:      Optional[int], UpperCAmelCase__							:      np.array     ):
    __lowercase								=						spectrogram(
        UpperCAmelCase__, window_function(self.n_fft, "hann"     ), frame_length=self.n_fft, hop_length=self.hop_length, power=2.0, mel_filters=self.mel_filters, log_mel="log10", )
    __lowercase								=						log_spec[:, :-1]
    __lowercase								=						np.maximum(UpperCAmelCase__, log_spec.max() - 8.0     )
    __lowercase								=						(log_spec + 4.0) / 4.0
    return log_spec


 @staticmethod
 # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
 def  _lowercase			(      UpperCAmelCase__							:      List[np.ndarray], UpperCAmelCase__							:      List[np.ndarray], UpperCAmelCase__							:      float = 0.0     ):
    if attention_mask is not None:
       __lowercase								=						np.array(UpperCAmelCase__, np.intaa     )
       __lowercase								=						[]

       for vector, length in zip(UpperCAmelCase__, attention_mask.sum(-1     )     ):
          __lowercase								=						(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7     )
          if length < normed_slice.shape[0]:
             __lowercase								=						padding_value

          normed_input_values.append(UpperCAmelCase__     )
    else:
       __lowercase								=						[(x - x.mean()) / np.sqrt(x.var() + 1E-7     ) for x in input_values]

    return normed_input_values


 def __call__(      self							:      Optional[int], UpperCAmelCase__							:      Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]], UpperCAmelCase__							:      bool = True, UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[Union[str, TensorType]] = None, UpperCAmelCase__							:      Optional[bool] = None, UpperCAmelCase__							:      Optional[str] = "max_length", UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[bool] = None, **UpperCAmelCase__							:      Optional[int], ):

    if sampling_rate is not None:
       if sampling_rate != self.sampling_rate:
          raise ValueError(
              F"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
              F""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
              F""" was sampled with {self.sampling_rate} and not {sampling_rate}."""     )
    else:
       logger.warning(
           "It is strongly recommended to pass the `sampling_rate` argument to this function. "
           "Failing to do so can result in silent errors that might be hard to debug."     )

    __lowercase								=						isinstance(UpperCAmelCase__, np.ndarray     ) and len(raw_speech.shape     ) > 1
    if is_batched_numpy and len(raw_speech.shape     ) > 2:
       raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""     )
    __lowercase								=						is_batched_numpy or (
        isinstance(UpperCAmelCase__, (list, tuple)     ) and (isinstance(raw_speech[0], (np.ndarray, tuple, list)     ))
    )

    if is_batched:
       __lowercase								=						[np.asarray([speech], dtype=np.floataa     ).T for speech in raw_speech]
    elif not is_batched and not isinstance(UpperCAmelCase__, np.ndarray     ):
       __lowercase								=						np.asarray(UpperCAmelCase__, dtype=np.floataa     )
    elif isinstance(UpperCAmelCase__, np.ndarray     ) and raw_speech.dtype is np.dtype(np.floataa     ):
       __lowercase								=						raw_speech.astype(np.floataa     )

    # always return batch
    if not is_batched:
       __lowercase								=						[np.asarray([raw_speech]     ).T]

    __lowercase								=						BatchFeature({"input_features": raw_speech}     )

    # convert into correct format for padding

    __lowercase								=						self.pad(
        UpperCAmelCase__, padding=UpperCAmelCase__, max_length=max_length if max_length else self.n_samples, truncation=UpperCAmelCase__, pad_to_multiple_of=UpperCAmelCase__, return_attention_mask=return_attention_mask or do_normalize, )

    # zero-mean and unit-variance normalization
    if do_normalize:
       __lowercase								=						self.zero_mean_unit_var_norm(
           padded_inputs["input_features"], attention_mask=padded_inputs["attention_mask"], padding_value=self.padding_value, )
       __lowercase								=						np.stack(padded_inputs["input_features"], axis=0     )

    # make sure list is in array format
    __lowercase								=						padded_inputs.get("input_features"     ).transpose(2, 0, 1     )

    __lowercase								=						[self._np_extract_fbank_features(UpperCAmelCase__     ) for waveform in input_features[0]]

    if isinstance(input_features[0], UpperCAmelCase__     ):
       __lowercase								=						[np.asarray(UpperCAmelCase__, dtype=np.floataa     ) for feature in input_features]
    else:
       __lowercase								=						input_features

    if return_attention_mask:
       # rescale from sample (48000) to feature (3000)
       __lowercase								=						padded_inputs["attention_mask"][:, :: self.hop_length]

    if return_tensors is not None:
       __lowercase								=						padded_inputs.convert_to_tensors(UpperCAmelCase__     )

    return padded_inputs



 def  _lowercase			(      self							:      Optional[Any]     ):
    __lowercase								=						copy.deepcopy(self.__dict__     )
    __lowercase								=						self.__class__.__name__
    if "mel_filters" in output:
       del output["mel_filters"]
    return output







"""simple docstring"""





from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)




@flax.struct.dataclass
class  _lowerCAmelCase  (     lowercase       ):







 """simple docstring"""







 __UpperCAmelCase     : jnp.ndarray
 __UpperCAmelCase     : jnp.ndarray







class  _lowerCAmelCase  (     nn.Module       ):







 """simple docstring"""







 __UpperCAmelCase     : int
 __UpperCAmelCase     : Tuple[int]       		=     (1_6, 3_2, 9_6, 2_5_6)
 __UpperCAmelCase     : jnp.dtype       		=     jnp.floataa


 def  _lowercase			(      self							:      Union[str, Any]     ):
    __lowercase								=						nn.Conv(
        self.block_out_channels[0], kernel_size=(3, 3), padding=((1, 1), (1, 1)), dtype=self.dtype, )

    __lowercase								=						[]
    for i in range(len(self.block_out_channels     ) - 1     ):
       __lowercase								=						self.block_out_channels[i]
       __lowercase								=						self.block_out_channels[i + 1]
       __lowercase								=						nn.Conv(
           UpperCAmelCase__, kernel_size=(3, 3), padding=((1, 1), (1, 1)), dtype=self.dtype, )
       blocks.append(UpperCAmelCase__     )
       __lowercase								=						nn.Conv(
           UpperCAmelCase__, kernel_size=(3, 3), strides=(2, 2), padding=((1, 1), (1, 1)), dtype=self.dtype, )
       blocks.append(UpperCAmelCase__     )
    __lowercase								=						blocks

    __lowercase								=						nn.Conv(
        self.conditioning_embedding_channels, kernel_size=(3, 3), padding=((1, 1), (1, 1)), kernel_init=nn.initializers.zeros_init(), bias_init=nn.initializers.zeros_init(), dtype=self.dtype, )



 def __call__(      self							:      Any, UpperCAmelCase__							:      Union[str, Any]     ):
    __lowercase								=						self.conv_in(UpperCAmelCase__     )
    __lowercase								=						nn.silu(UpperCAmelCase__     )

    for block in self.blocks:
       __lowercase								=						block(UpperCAmelCase__     )
       __lowercase								=						nn.silu(UpperCAmelCase__     )

    __lowercase								=						self.conv_out(UpperCAmelCase__     )

    return embedding







@flax_register_to_config
class  _lowerCAmelCase  (     nn.Module					,lowercase					,lowercase       ):







 """simple docstring"""







 __UpperCAmelCase     : int       		=     3_2
 __UpperCAmelCase     : int       		=     4
 __UpperCAmelCase     : Tuple[str]       		=     (
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "DownBlock2D",
 )
 __UpperCAmelCase     : Union[bool, Tuple[bool]]       		=     False
 __UpperCAmelCase     : Tuple[int]       		=     (3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
 __UpperCAmelCase     : int       		=     2
 __UpperCAmelCase     : Union[int, Tuple[int]]       		=     8
 __UpperCAmelCase     : Optional[Union[int, Tuple[int]]]       		=     None
 __UpperCAmelCase     : int       		=     1_2_8_0
 __UpperCAmelCase     : float       		=     0.0
 __UpperCAmelCase     : bool       		=     False
 __UpperCAmelCase     : jnp.dtype       		=     jnp.floataa
 __UpperCAmelCase     : bool       		=     True
 __UpperCAmelCase     : int       		=     0
 __UpperCAmelCase     : str       		=     "rgb"
 __UpperCAmelCase     : Tuple[int]       		=     (1_6, 3_2, 9_6, 2_5_6)


 def  _lowercase			(      self							:      List[Any], UpperCAmelCase__							:      jax.random.KeyArray     ):
    # init input tensors
    __lowercase								=						(1, self.in_channels, self.sample_size, self.sample_size)
    __lowercase								=						jnp.zeros(UpperCAmelCase__, dtype=jnp.floataa     )
    __lowercase								=						jnp.ones((1,), dtype=jnp.intaa     )
    __lowercase								=						jnp.zeros((1, 1, self.cross_attention_dim), dtype=jnp.floataa     )
    __lowercase								=						(1, 3, self.sample_size * 8, self.sample_size * 8)
    __lowercase								=						jnp.zeros(UpperCAmelCase__, dtype=jnp.floataa     )

    __lowercase		,__lowercase								=						jax.random.split(UpperCAmelCase__     )
    __lowercase								=						{"params": params_rng, "dropout": dropout_rng}

    return self.init(UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__     )["params"]


 def  _lowercase			(      self							:      Union[str, Any]     ):
    __lowercase								=						self.block_out_channels
    __lowercase								=						block_out_channels[0] * 4

    # If `num_attention_heads` is not defined (which is the case for most models)
    # it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
    # The reason for this behavior is to correct for incorrectly named variables that were introduced
    # when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
    # Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
    # which is why we correct for the naming here.
    __lowercase								=						self.num_attention_heads or self.attention_head_dim

    # input
    __lowercase								=						nn.Conv(
        block_out_channels[0], kernel_size=(3, 3), strides=(1, 1), padding=((1, 1), (1, 1)), dtype=self.dtype, )

    # time
    __lowercase								=						FlaxTimesteps(
        block_out_channels[0], flip_sin_to_cos=self.flip_sin_to_cos, freq_shift=self.config.freq_shift     )
    __lowercase								=						FlaxTimestepEmbedding(UpperCAmelCase__, dtype=self.dtype     )

    __lowercase								=						FlaxControlNetConditioningEmbedding(
        conditioning_embedding_channels=block_out_channels[0], block_out_channels=self.conditioning_embedding_out_channels, )

    __lowercase								=						self.only_cross_attention
    if isinstance(UpperCAmelCase__, UpperCAmelCase__     ):
       __lowercase								=						(only_cross_attention,) * len(self.down_block_types     )

    if isinstance(UpperCAmelCase__, UpperCAmelCase__     ):
       __lowercase								=						(num_attention_heads,) * len(self.down_block_types     )

    # down
    __lowercase								=						[]
    __lowercase								=						[]

    __lowercase								=						block_out_channels[0]

    __lowercase								=						nn.Conv(
        UpperCAmelCase__, kernel_size=(1, 1), padding="VALID", kernel_init=nn.initializers.zeros_init(), bias_init=nn.initializers.zeros_init(), dtype=self.dtype, )
    controlnet_down_blocks.append(UpperCAmelCase__     )

    for i, down_block_type in enumerate(self.down_block_types     ):
       __lowercase								=						output_channel
       __lowercase								=						block_out_channels[i]
       __lowercase								=						i == len(UpperCAmelCase__     ) - 1

       if down_block_type == "CrossAttnDownBlock2D":
          __lowercase								=						FlaxCrossAttnDownBlockaD(
              in_channels=UpperCAmelCase__, out_channels=UpperCAmelCase__, dropout=self.dropout, num_layers=self.layers_per_block, num_attention_heads=num_attention_heads[i], add_downsample=not is_final_block, use_linear_projection=self.use_linear_projection, only_cross_attention=only_cross_attention[i], dtype=self.dtype, )
       else:
          __lowercase								=						FlaxDownBlockaD(
              in_channels=UpperCAmelCase__, out_channels=UpperCAmelCase__, dropout=self.dropout, num_layers=self.layers_per_block, add_downsample=not is_final_block, dtype=self.dtype, )

       down_blocks.append(UpperCAmelCase__     )

       for _ in range(self.layers_per_block     ):
          __lowercase								=						nn.Conv(
              UpperCAmelCase__, kernel_size=(1, 1), padding="VALID", kernel_init=nn.initializers.zeros_init(), bias_init=nn.initializers.zeros_init(), dtype=self.dtype, )
          controlnet_down_blocks.append(UpperCAmelCase__     )

       if not is_final_block:
          __lowercase								=						nn.Conv(
              UpperCAmelCase__, kernel_size=(1, 1), padding="VALID", kernel_init=nn.initializers.zeros_init(), bias_init=nn.initializers.zeros_init(), dtype=self.dtype, )
          controlnet_down_blocks.append(UpperCAmelCase__     )

    __lowercase								=						down_blocks
    __lowercase								=						controlnet_down_blocks

    # mid
    __lowercase								=						block_out_channels[-1]
    __lowercase								=						FlaxUNetMidBlockaDCrossAttn(
        in_channels=UpperCAmelCase__, dropout=self.dropout, num_attention_heads=num_attention_heads[-1], use_linear_projection=self.use_linear_projection, dtype=self.dtype, )

    __lowercase								=						nn.Conv(
        UpperCAmelCase__, kernel_size=(1, 1), padding="VALID", kernel_init=nn.initializers.zeros_init(), bias_init=nn.initializers.zeros_init(), dtype=self.dtype, )



 def __call__(      self							:      str, UpperCAmelCase__							:      Union[str, Any], UpperCAmelCase__							:      List[Any], UpperCAmelCase__							:      Tuple, UpperCAmelCase__							:      List[str], UpperCAmelCase__							:      float = 1.0, UpperCAmelCase__							:      bool = True, UpperCAmelCase__							:      bool = False, ):
    __lowercase								=						self.controlnet_conditioning_channel_order
    if channel_order == "bgr":
       __lowercase								=						jnp.flip(UpperCAmelCase__, axis=1     )

    # 1. time
    if not isinstance(UpperCAmelCase__, jnp.ndarray     ):
       __lowercase								=						jnp.array([timesteps], dtype=jnp.intaa     )
    elif isinstance(UpperCAmelCase__, jnp.ndarray     ) and len(timesteps.shape     ) == 0:
       __lowercase								=						timesteps.astype(dtype=jnp.floataa     )
       __lowercase								=						jnp.expand_dims(UpperCAmelCase__, 0     )

    __lowercase								=						self.time_proj(UpperCAmelCase__     )
    __lowercase								=						self.time_embedding(UpperCAmelCase__     )

    # 2. pre-process
    __lowercase								=						jnp.transpose(UpperCAmelCase__, (0, 2, 3, 1)     )
    __lowercase								=						self.conv_in(UpperCAmelCase__     )

    __lowercase								=						jnp.transpose(UpperCAmelCase__, (0, 2, 3, 1)     )
    __lowercase								=						self.controlnet_cond_embedding(UpperCAmelCase__     )
    sample += controlnet_cond

    # 3. down
    __lowercase								=						(sample,)
    for down_block in self.down_blocks:
       if isinstance(UpperCAmelCase__, UpperCAmelCase__     ):
          __lowercase		,__lowercase								=						down_block(UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__, deterministic=not train     )
       else:
          __lowercase		,__lowercase								=						down_block(UpperCAmelCase__, UpperCAmelCase__, deterministic=not train     )
       down_block_res_samples += res_samples

    # 4. mid
    __lowercase								=						self.mid_block(UpperCAmelCase__, UpperCAmelCase__, UpperCAmelCase__, deterministic=not train     )

    # 5. contronet blocks
    __lowercase								=						()
    for down_block_res_sample, controlnet_block in zip(UpperCAmelCase__, self.controlnet_down_blocks     ):
       __lowercase								=						controlnet_block(UpperCAmelCase__     )
       controlnet_down_block_res_samples += (down_block_res_sample,)

    __lowercase								=						controlnet_down_block_res_samples

    __lowercase								=						self.controlnet_mid_block(UpperCAmelCase__     )

    # 6. scaling
    __lowercase								=						[sample * conditioning_scale for sample in down_block_res_samples]
    mid_block_res_sample *= conditioning_scale

    if not return_dict:
       return (down_block_res_samples, mid_block_res_sample)

    return FlaxControlNetOutput(
        down_block_res_samples=UpperCAmelCase__, mid_block_res_sample=UpperCAmelCase__     )




