


'''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					(    snake_case_				):




 '''simple docstring'''
 __lowercase				:		jnp.ndarray
 __lowercase				:		jnp.ndarray


class 						_lowerCamelCase					(    nn.Module				):




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


 def  snake_case__   (      self						):

  """simple docstring"""







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

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

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

  """simple docstring"""







  __A  :		Optional[Any]       =			self.conv_in(__lowercase						)
  __A  :		Any       =			nn.silu(__lowercase						)

  for block in self.blocks:
   __A  :		str       =			block(__lowercase						)
   __A  :		List[str]       =			nn.silu(__lowercase						)

  __A  :		List[Any]       =			self.conv_out(__lowercase						)

  return embedding


@flax_register_to_config
class 						_lowerCamelCase					(    nn.Module    ,		snake_case_    ,		snake_case_				):




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


 def  snake_case__   (      self      ,      __lowercase						):

  """simple docstring"""







  __A  :		Any       =			(1, self.in_channels, self.sample_size, self.sample_size)
  __A  :		Tuple       =			jnp.zeros(__lowercase      ,      dtype=jnp.floataa						)
  __A  :		Optional[Any]       =			jnp.ones((1,)      ,      dtype=jnp.intaa						)
  __A  :		Tuple       =			jnp.zeros((1, 1, self.cross_attention_dim)      ,      dtype=jnp.floataa						)
  __A  :		Any       =			(1, 3, self.sample_size * 8, self.sample_size * 8)
  __A  :		Optional[Any]       =			jnp.zeros(__lowercase      ,      dtype=jnp.floataa						)

  __A     ,__A  :		Optional[int]       =			jax.random.split(__lowercase						)
  __A  :		List[Any]       =			{'params': params_rng, 'dropout': dropout_rng}

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


 def  snake_case__   (      self						):

  """simple docstring"""







  __A  :		List[str]       =			self.block_out_channels
  __A  :		int       =			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.
  __A  :		Dict       =			self.num_attention_heads or self.attention_head_dim

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

  # time
  __A  :		Optional[int]       =			FlaxTimesteps(
      block_out_channels[0]      ,      flip_sin_to_cos=self.flip_sin_to_cos      ,      freq_shift=self.config.freq_shift						)
  __A  :		str       =			FlaxTimestepEmbedding(__lowercase      ,      dtype=self.dtype						)

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

  __A  :		int       =			self.only_cross_attention
  if isinstance(__lowercase      ,      __lowercase						):
   __A  :		Any       =			(only_cross_attention,) * len(self.down_block_types						)

  if isinstance(__lowercase      ,      __lowercase						):
   __A  :		List[str]       =			(num_attention_heads,) * len(self.down_block_types						)

  # down
  __A  :		str       =			[]
  __A  :		Any       =			[]

  __A  :		List[Any]       =			block_out_channels[0]

  __A  :		List[Any]       =			nn.Conv(
      __lowercase      ,      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(__lowercase						)

  for i, down_block_type in enumerate(self.down_block_types						):
   __A  :		Tuple       =			output_channel
   __A  :		Any       =			block_out_channels[i]
   __A  :		Union[str, Any]       =			i == len(__lowercase						) - 1

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

   down_blocks.append(__lowercase						)

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

   if not is_final_block:
    __A  :		int       =			nn.Conv(
        __lowercase      ,      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(__lowercase						)

  __A  :		str       =			down_blocks
  __A  :		Optional[Any]       =			controlnet_down_blocks

  # mid
  __A  :		List[str]       =			block_out_channels[-1]
  __A  :		int       =			FlaxUNetMidBlockaDCrossAttn(
      in_channels=__lowercase      ,      dropout=self.dropout      ,      num_attention_heads=num_attention_heads[-1]      ,      use_linear_projection=self.use_linear_projection      ,      dtype=self.dtype      ,      )

  __A  :		List[str]       =			nn.Conv(
      __lowercase      ,      kernel_size=(1, 1)      ,      padding='VALID'      ,      kernel_init=nn.initializers.zeros_init()      ,      bias_init=nn.initializers.zeros_init()      ,      dtype=self.dtype      ,      )







 def __call__(      self      ,      __lowercase      ,      __lowercase      ,      __lowercase      ,      __lowercase      ,      __lowercase = 1.0      ,      __lowercase = True      ,      __lowercase = False      ,      ):

  """simple docstring"""







  __A  :		Any       =			self.controlnet_conditioning_channel_order
  if channel_order == "bgr":
   __A  :		int       =			jnp.flip(__lowercase      ,      axis=1						)

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

  __A  :		Union[str, Any]       =			self.time_proj(__lowercase						)
  __A  :		List[Any]       =			self.time_embedding(__lowercase						)

  # 2. pre-process
  __A  :		Tuple       =			jnp.transpose(__lowercase      ,      (0, 2, 3, 1)						)
  __A  :		int       =			self.conv_in(__lowercase						)

  __A  :		str       =			jnp.transpose(__lowercase      ,      (0, 2, 3, 1)						)
  __A  :		Optional[Any]       =			self.controlnet_cond_embedding(__lowercase						)
  sample += controlnet_cond

  # 3. down
  __A  :		Optional[int]       =			(sample,)
  for down_block in self.down_blocks:
   if isinstance(__lowercase      ,      __lowercase						):
    __A     ,__A  :		Dict       =			down_block(__lowercase      ,      __lowercase      ,      __lowercase      ,      deterministic=not train						)
   else:
    __A     ,__A  :		List[str]       =			down_block(__lowercase      ,      __lowercase      ,      deterministic=not train						)
   down_block_res_samples += res_samples

  # 4. mid
  __A  :		Dict       =			self.mid_block(__lowercase      ,      __lowercase      ,      __lowercase      ,      deterministic=not train						)

  # 5. contronet blocks
  __A  :		List[str]       =			()
  for down_block_res_sample, controlnet_block in zip(__lowercase      ,      self.controlnet_down_blocks						):
   __A  :		int       =			controlnet_block(__lowercase						)
   controlnet_down_block_res_samples += (down_block_res_sample,)

  __A  :		List[str]       =			controlnet_down_block_res_samples

  __A  :		Dict       =			self.controlnet_mid_block(__lowercase						)

  # 6. scaling
  __A  :		Optional[Any]       =			[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=__lowercase      ,      mid_block_res_sample=__lowercase						)










'''simple docstring'''
from __future__ import annotations

class 						_lowerCamelCase					:




 '''simple docstring'''


 def __init__(      self      ,      __lowercase						):

  """simple docstring"""







  __A  :		Dict       =			order

  # a_{0} ... a_{k}
  __A  :		str       =			[1.0] + [0.0] * order
  # b_{0} ... b_{k}
  __A  :		Any       =			[1.0] + [0.0] * order

  # x[n-1] ... x[n-k]
  __A  :		Optional[int]       =			[0.0] * self.order
  # y[n-1] ... y[n-k]
  __A  :		Dict       =			[0.0] * self.order


 def  snake_case__   (      self      ,      __lowercase      ,      __lowercase						):

  """simple docstring"""







  if len(__lowercase						) < self.order:
   __A  :		List[Any]       =			[1.0, *a_coeffs]

  if len(__lowercase						) != self.order + 1:
   __A  :		Tuple       =			(
       F"""Expected a_coeffs to have {self.order + 1} elements """
       F"""for {self.order}-order filter, got {len(__lowercase						)}"""
   )
   raise ValueError(__lowercase						)

  if len(__lowercase						) != self.order + 1:
   __A  :		List[Any]       =			(
       F"""Expected b_coeffs to have {self.order + 1} elements """
       F"""for {self.order}-order filter, got {len(__lowercase						)}"""
   )
   raise ValueError(__lowercase						)

  __A  :		Optional[int]       =			a_coeffs
  __A  :		Optional[int]       =			b_coeffs







 def  snake_case__   (      self      ,      __lowercase						):

  """simple docstring"""







  __A  :		Union[str, Any]       =			0.0

  # Start at index 1 and do index 0 at the end.
  for i in range(1      ,      self.order + 1						):
   result += (
       self.b_coeffs[i] * self.input_history[i - 1]
       - self.a_coeffs[i] * self.output_history[i - 1]
   )

  __A  :		Optional[int]       =			(result + self.b_coeffs[0] * sample) / self.a_coeffs[0]

  __A  :		List[str]       =			self.input_history[:-1]
  __A  :		Optional[int]       =			self.output_history[:-1]

  __A  :		Optional[int]       =			sample
  __A  :		Dict       =			result

  return result






