





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 					A__   (       __snake_case			):
      _UpperCAmelCase	:jnp.ndarray
      _UpperCAmelCase	:jnp.ndarray



class 					A__   (       nn.Module			):
      _UpperCAmelCase	:int
      _UpperCAmelCase	:Tuple[int]						    =   (1_6, 3_2, 9_6, 2_5_6)
      _UpperCAmelCase	:jnp.dtype						    =   jnp.floataa





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Tuple       	=   nn.Conv(
           self.block_out_channels[0]       ,							kernel_size=(3, 3)       ,							padding=((1, 1), (1, 1))       ,							dtype=self.dtype       ,							)

       UpperCamelCase	:					List[str]       	=   []
       for i in range(len(self.block_out_channels   ) - 1   ):
        UpperCamelCase	:					List[Any]       	=   self.block_out_channels[i]
        UpperCamelCase	:					Any       	=   self.block_out_channels[i + 1]
        UpperCamelCase	:					Optional[Any]       	=   nn.Conv(
            A_       ,							kernel_size=(3, 3)       ,							padding=((1, 1), (1, 1))       ,							dtype=self.dtype       ,							)
        blocks.append(A_   )
        UpperCamelCase	:					List[Any]       	=   nn.Conv(
            A_       ,							kernel_size=(3, 3)       ,							strides=(2, 2)       ,							padding=((1, 1), (1, 1))       ,							dtype=self.dtype       ,							)
        blocks.append(A_   )
       UpperCamelCase	:					List[str]       	=   blocks

       UpperCamelCase	:					Union[str, Any]       	=   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       ,							A_   ):
       '''simple docstring'''




       UpperCamelCase	:					List[Any]       	=   self.conv_in(A_   )
       UpperCamelCase	:					Optional[int]       	=   nn.silu(A_   )

       for block in self.blocks:
        UpperCamelCase	:					List[str]       	=   block(A_   )
        UpperCamelCase	:					Union[str, Any]       	=   nn.silu(A_   )

       UpperCamelCase	:					Optional[Any]       	=   self.conv_out(A_   )

       return embedding



@flax_register_to_config
class 					A__   (       nn.Module       ,   __snake_case       ,   __snake_case			):
      _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        __UpperCamelCase( self       ,							A_   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   (1, self.in_channels, self.sample_size, self.sample_size)
       UpperCamelCase	:					Optional[Any]       	=   jnp.zeros(A_       ,							dtype=jnp.floataa   )
       UpperCamelCase	:					Any       	=   jnp.ones((1,)       ,							dtype=jnp.intaa   )
       UpperCamelCase	:					Optional[int]       	=   jnp.zeros((1, 1, self.cross_attention_dim)       ,							dtype=jnp.floataa   )
       UpperCamelCase	:					Optional[int]       	=   (1, 3, self.sample_size * 8, self.sample_size * 8)
       UpperCamelCase	:					Optional[Any]       	=   jnp.zeros(A_       ,							dtype=jnp.floataa   )

       UpperCamelCase	,    UpperCamelCase	:					Tuple       	=   jax.random.split(A_   )
       UpperCamelCase	:					Optional[int]       	=   {"params": params_rng, "dropout": dropout_rng}

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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   self.block_out_channels
       UpperCamelCase	:					Union[str, Any]       	=   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.
       UpperCamelCase	:					Tuple       	=   self.num_attention_heads or self.attention_head_dim

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

       # time
       UpperCamelCase	:					Any       	=   FlaxTimesteps(
           block_out_channels[0]       ,							flip_sin_to_cos=self.flip_sin_to_cos       ,							freq_shift=self.config.freq_shift   )
       UpperCamelCase	:					Dict       	=   FlaxTimestepEmbedding(A_       ,							dtype=self.dtype   )

       UpperCamelCase	:					List[Any]       	=   FlaxControlNetConditioningEmbedding(
           conditioning_embedding_channels=block_out_channels[0]       ,							block_out_channels=self.conditioning_embedding_out_channels       ,							)

       UpperCamelCase	:					Optional[int]       	=   self.only_cross_attention
       if isinstance(A_       ,							A_   ):
        UpperCamelCase	:					Optional[int]       	=   (only_cross_attention,) * len(self.down_block_types   )

       if isinstance(A_       ,							A_   ):
        UpperCamelCase	:					Optional[int]       	=   (num_attention_heads,) * len(self.down_block_types   )

       # down
       UpperCamelCase	:					List[Any]       	=   []
       UpperCamelCase	:					Any       	=   []

       UpperCamelCase	:					str       	=   block_out_channels[0]

       UpperCamelCase	:					Dict       	=   nn.Conv(
           A_       ,							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(A_   )

       for i, down_block_type in enumerate(self.down_block_types   ):
        UpperCamelCase	:					Union[str, Any]       	=   output_channel
        UpperCamelCase	:					List[str]       	=   block_out_channels[i]
        UpperCamelCase	:					Any       	=   i == len(A_   ) - 1

        if down_block_type == "CrossAttnDownBlock2D":
         UpperCamelCase	:					List[str]       	=   FlaxCrossAttnDownBlockaD(
             in_channels=A_       ,							out_channels=A_       ,							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:
         UpperCamelCase	:					str       	=   FlaxDownBlockaD(
             in_channels=A_       ,							out_channels=A_       ,							dropout=self.dropout       ,							num_layers=self.layers_per_block       ,							add_downsample=not is_final_block       ,							dtype=self.dtype       ,							)

        down_blocks.append(A_   )

        for _ in range(self.layers_per_block   ):
         UpperCamelCase	:					Optional[int]       	=   nn.Conv(
             A_       ,							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(A_   )

        if not is_final_block:
         UpperCamelCase	:					Optional[Any]       	=   nn.Conv(
             A_       ,							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(A_   )

       UpperCamelCase	:					Optional[Any]       	=   down_blocks
       UpperCamelCase	:					Tuple       	=   controlnet_down_blocks

       # mid
       UpperCamelCase	:					Dict       	=   block_out_channels[-1]
       UpperCamelCase	:					Any       	=   FlaxUNetMidBlockaDCrossAttn(
           in_channels=A_       ,							dropout=self.dropout       ,							num_attention_heads=num_attention_heads[-1]       ,							use_linear_projection=self.use_linear_projection       ,							dtype=self.dtype       ,							)

       UpperCamelCase	:					Optional[Any]       	=   nn.Conv(
           A_       ,							kernel_size=(1, 1)       ,							padding="VALID"       ,							kernel_init=nn.initializers.zeros_init()       ,							bias_init=nn.initializers.zeros_init()       ,							dtype=self.dtype       ,							)





      def __call__( self       ,							A_       ,							A_       ,							A_       ,							A_       ,							A_ = 1.0       ,							A_ = True       ,							A_ = False       ,							):
       '''simple docstring'''




       UpperCamelCase	:					int       	=   self.controlnet_conditioning_channel_order
       if channel_order == "bgr":
        UpperCamelCase	:					str       	=   jnp.flip(A_       ,							axis=1   )

       # 1. time
       if not isinstance(A_       ,							jnp.ndarray   ):
        UpperCamelCase	:					Optional[int]       	=   jnp.array([timesteps]       ,							dtype=jnp.intaa   )
       elif isinstance(A_       ,							jnp.ndarray   ) and len(timesteps.shape   ) == 0:
        UpperCamelCase	:					List[str]       	=   timesteps.astype(dtype=jnp.floataa   )
        UpperCamelCase	:					List[Any]       	=   jnp.expand_dims(A_       ,							0   )

       UpperCamelCase	:					Dict       	=   self.time_proj(A_   )
       UpperCamelCase	:					List[Any]       	=   self.time_embedding(A_   )

       # 2. pre-process
       UpperCamelCase	:					Union[str, Any]       	=   jnp.transpose(A_       ,							(0, 2, 3, 1)   )
       UpperCamelCase	:					Union[str, Any]       	=   self.conv_in(A_   )

       UpperCamelCase	:					Optional[Any]       	=   jnp.transpose(A_       ,							(0, 2, 3, 1)   )
       UpperCamelCase	:					int       	=   self.controlnet_cond_embedding(A_   )
       sample += controlnet_cond

       # 3. down
       UpperCamelCase	:					List[str]       	=   (sample,)
       for down_block in self.down_blocks:
        if isinstance(A_       ,							A_   ):
         UpperCamelCase	,    UpperCamelCase	:					Any       	=   down_block(A_       ,							A_       ,							A_       ,							deterministic=not train   )
        else:
         UpperCamelCase	,    UpperCamelCase	:					Tuple       	=   down_block(A_       ,							A_       ,							deterministic=not train   )
        down_block_res_samples += res_samples

       # 4. mid
       UpperCamelCase	:					Union[str, Any]       	=   self.mid_block(A_       ,							A_       ,							A_       ,							deterministic=not train   )

       # 5. contronet blocks
       UpperCamelCase	:					Union[str, Any]       	=   ()
       for down_block_res_sample, controlnet_block in zip(A_       ,							self.controlnet_down_blocks   ):
        UpperCamelCase	:					Dict       	=   controlnet_block(A_   )
        controlnet_down_block_res_samples += (down_block_res_sample,)

       UpperCamelCase	:					Union[str, Any]       	=   controlnet_down_block_res_samples

       UpperCamelCase	:					Dict       	=   self.controlnet_mid_block(A_   )

       # 6. scaling
       UpperCamelCase	:					Dict       	=   [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=A_       ,							mid_block_res_sample=A_   )







from typing import List, Optional, Tuple, Union

import torch

from ...utils import logging, randn_tensor
from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline


__lowerCamelCase       :     str   =     logging.get_logger(__name__)  # pylint: disable=invalid-name



class 					A__   (       __snake_case			):





      def __init__( self       ,							A_       ,							A_   ):
       '''simple docstring'''




       super().__init__()
       self.register_modules(unet=A_       ,							scheduler=A_   )





      @torch.no_grad()
      def __call__( self       ,							A_ = 1       ,							A_ = 100       ,							A_ = None       ,							A_ = None       ,							A_ = True       ,							):
       '''simple docstring'''




       if audio_length_in_s is None:
        UpperCamelCase	:					str       	=   self.unet.config.sample_size / self.unet.config.sample_rate

       UpperCamelCase	:					Optional[Any]       	=   audio_length_in_s * self.unet.config.sample_rate

       UpperCamelCase	:					Any       	=   2 ** len(self.unet.up_blocks   )
       if sample_size < 3 * down_scale_factor:
        raise ValueError(
            F"""{audio_length_in_s} is too small. Make sure it's bigger or equal to"""
            F""" {3 * down_scale_factor / self.unet.config.sample_rate}."""   )

       UpperCamelCase	:					Union[str, Any]       	=   int(A_   )
       if sample_size % down_scale_factor != 0:
        UpperCamelCase	:					List[str]       	=   (
            (audio_length_in_s * self.unet.config.sample_rate) // down_scale_factor + 1
        ) * down_scale_factor
        logger.info(
            F"""{audio_length_in_s} is increased to {sample_size / self.unet.config.sample_rate} so that it can be handled"""
            F""" by the model. It will be cut to {original_sample_size / self.unet.config.sample_rate} after the denoising"""
            " process."   )
       UpperCamelCase	:					Any       	=   int(A_   )

       UpperCamelCase	:					Union[str, Any]       	=   next(iter(self.unet.parameters()   )   ).dtype
       UpperCamelCase	:					Optional[int]       	=   (batch_size, self.unet.config.in_channels, sample_size)
       if isinstance(A_       ,							A_   ) and len(A_   ) != batch_size:
        raise ValueError(
            F"""You have passed a list of generators of length {len(A_   )}, but requested an effective batch"""
            F""" size of {batch_size}. Make sure the batch size matches the length of the generators."""   )

       UpperCamelCase	:					Optional[Any]       	=   randn_tensor(A_       ,							generator=A_       ,							device=self.device       ,							dtype=A_   )

       # set step values
       self.scheduler.set_timesteps(A_       ,							device=audio.device   )
       UpperCamelCase	:					Optional[int]       	=   self.scheduler.timesteps.to(A_   )

       for t in self.progress_bar(self.scheduler.timesteps   ):
        # 1. predict noise model_output
        UpperCamelCase	:					Dict       	=   self.unet(A_       ,							A_   ).sample

        # 2. compute previous image: x_t -> t_t-1
        UpperCamelCase	:					int       	=   self.scheduler.step(A_       ,							A_       ,							A_   ).prev_sample

       UpperCamelCase	:					Optional[Any]       	=   audio.clamp(-1       ,							1   ).float().cpu().numpy()

       UpperCamelCase	:					Dict       	=   audio[:, :, :original_sample_size]

       if not return_dict:
        return (audio,)

       return AudioPipelineOutput(audios=A_   )
