

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,
    FlaxCrossAttnUpBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
    FlaxUpBlockaD,
)





@flax.struct.dataclass
class 					__lowerCAmelCase	(   SCREAMING_SNAKE_CASE				):
  _a		     =    42





@flax_register_to_config
class 					__lowerCAmelCase	(   nn.Module     ,     SCREAMING_SNAKE_CASE     ,     SCREAMING_SNAKE_CASE				):
  _a		     =    32
  _a		     =    4
  _a		     =    4
  _a		     =    (
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "DownBlock2D",
  )
  _a		     =    ("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
  _a		     =    False
  _a		     =    (320, 640, 1_280, 1_280)
  _a		     =    2
  _a		     =    8
  _a		     =    None
  _a		     =    1_280
  _a		     =    0.0
  _a		     =    False
  _a		     =    jnp.floataa
  _a		     =    True
  _a		     =    0
  _a		     =    False




  def     A__	(							self				,  lowerCAmelCase   )    ->					FrozenDict:
        '''simple docstring'''






        _lowercase      			=(1, self.in_channels, self.sample_size, self.sample_size)
        _lowercase      			=jnp.zeros(lowerCAmelCase				,  dtype=jnp.floataa   )
        _lowercase      			=jnp.ones((1,)				,  dtype=jnp.intaa   )
        _lowercase      			=jnp.zeros((1, 1, self.cross_attention_dim)				,  dtype=jnp.floataa   )

        _lowercase			,			_lowercase      			=jax.random.split(lowerCAmelCase   )
        _lowercase      			={'params': params_rng, 'dropout': dropout_rng}

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




  def     A__	(							self   )    ->					Optional[Any]:
        '''simple docstring'''






        _lowercase      			=self.block_out_channels
        _lowercase      			=block_out_channels[0] * 4

        if self.num_attention_heads is not None:
              raise ValueError(
                  'At the moment it is not possible to define the number of attention heads via `num_attention_heads` because of a naming issue as described in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131. Passing `num_attention_heads` will only be supported in diffusers v0.19.'   )

        # 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(lowerCAmelCase				,  dtype=self.dtype   )

        _lowercase      			=self.only_cross_attention
        if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
              _lowercase      			=(only_cross_attention,) * len(self.down_block_types   )

        if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
              _lowercase      			=(num_attention_heads,) * len(self.down_block_types   )

        # down
        _lowercase      			=[]
        _lowercase      			=block_out_channels[0]
        for i, down_block_type in enumerate(self.down_block_types   ):
              _lowercase      			=output_channel
              _lowercase      			=block_out_channels[i]
              _lowercase      			=i == len(lowerCAmelCase   ) - 1

              if down_block_type == "CrossAttnDownBlock2D":
                    _lowercase      			=FlaxCrossAttnDownBlockaD(
                        in_channels=lowerCAmelCase				,  out_channels=lowerCAmelCase				,  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]				,  use_memory_efficient_attention=self.use_memory_efficient_attention				,  dtype=self.dtype				,  )
              else:
                    _lowercase      			=FlaxDownBlockaD(
                        in_channels=lowerCAmelCase				,  out_channels=lowerCAmelCase				,  dropout=self.dropout				,  num_layers=self.layers_per_block				,  add_downsample=not is_final_block				,  dtype=self.dtype				,  )

              down_blocks.append(lowerCAmelCase   )
        _lowercase      			=down_blocks

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

        # up
        _lowercase      			=[]
        _lowercase      			=list(reversed(lowerCAmelCase   )   )
        _lowercase      			=list(reversed(lowerCAmelCase   )   )
        _lowercase      			=list(reversed(lowerCAmelCase   )   )
        _lowercase      			=reversed_block_out_channels[0]
        for i, up_block_type in enumerate(self.up_block_types   ):
              _lowercase      			=output_channel
              _lowercase      			=reversed_block_out_channels[i]
              _lowercase      			=reversed_block_out_channels[min(i + 1				,  len(lowerCAmelCase   ) - 1   )]

              _lowercase      			=i == len(lowerCAmelCase   ) - 1

              if up_block_type == "CrossAttnUpBlock2D":
                    _lowercase      			=FlaxCrossAttnUpBlockaD(
                        in_channels=lowerCAmelCase				,  out_channels=lowerCAmelCase				,  prev_output_channel=lowerCAmelCase				,  num_layers=self.layers_per_block + 1				,  num_attention_heads=reversed_num_attention_heads[i]				,  add_upsample=not is_final_block				,  dropout=self.dropout				,  use_linear_projection=self.use_linear_projection				,  only_cross_attention=only_cross_attention[i]				,  use_memory_efficient_attention=self.use_memory_efficient_attention				,  dtype=self.dtype				,  )
              else:
                    _lowercase      			=FlaxUpBlockaD(
                        in_channels=lowerCAmelCase				,  out_channels=lowerCAmelCase				,  prev_output_channel=lowerCAmelCase				,  num_layers=self.layers_per_block + 1				,  add_upsample=not is_final_block				,  dropout=self.dropout				,  dtype=self.dtype				,  )

              up_blocks.append(lowerCAmelCase   )
              _lowercase      			=output_channel
        _lowercase      			=up_blocks

        # out
        _lowercase      			=nn.GroupNorm(num_groups=32				,  epsilon=1e-5   )
        _lowercase      			=nn.Conv(
            self.out_channels				,  kernel_size=(3, 3)				,  strides=(1, 1)				,  padding=((1, 1), (1, 1))				,  dtype=self.dtype				,  )




  def __call__(							self				,  lowerCAmelCase				,  lowerCAmelCase				,  lowerCAmelCase				,  lowerCAmelCase=None				,  lowerCAmelCase=None				,  lowerCAmelCase = True				,  lowerCAmelCase = False				,  )    ->					Union[FlaxUNetaDConditionOutput, Tuple]:
        '''simple docstring'''






        if not isinstance(lowerCAmelCase				,  jnp.ndarray   ):
              _lowercase      			=jnp.array([timesteps]				,  dtype=jnp.intaa   )
        elif isinstance(lowerCAmelCase				,  jnp.ndarray   ) and len(timesteps.shape   ) == 0:
              _lowercase      			=timesteps.astype(dtype=jnp.floataa   )
              _lowercase      			=jnp.expand_dims(lowerCAmelCase				,  0   )

        _lowercase      			=self.time_proj(lowerCAmelCase   )
        _lowercase      			=self.time_embedding(lowerCAmelCase   )

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

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

        if down_block_additional_residuals is not None:
              _lowercase      			=()

              for down_block_res_sample, down_block_additional_residual in zip(
                  lowerCAmelCase				,  lowerCAmelCase   ):
                    down_block_res_sample += down_block_additional_residual
                    new_down_block_res_samples += (down_block_res_sample,)

              _lowercase      			=new_down_block_res_samples

        # 4. mid
        _lowercase      			=self.mid_block(lowerCAmelCase				,  lowerCAmelCase				,  lowerCAmelCase				,  deterministic=not train   )

        if mid_block_additional_residual is not None:
              sample += mid_block_additional_residual

        # 5. up
        for up_block in self.up_blocks:
              _lowercase      			=down_block_res_samples[-(self.layers_per_block + 1) :]
              _lowercase      			=down_block_res_samples[: -(self.layers_per_block + 1)]
              if isinstance(lowerCAmelCase				,  lowerCAmelCase   ):
                    _lowercase      			=up_block(
                        lowerCAmelCase				,  temb=lowerCAmelCase				,  encoder_hidden_states=lowerCAmelCase				,  res_hidden_states_tuple=lowerCAmelCase				,  deterministic=not train				,  )
              else:
                    _lowercase      			=up_block(lowerCAmelCase				,  temb=lowerCAmelCase				,  res_hidden_states_tuple=lowerCAmelCase				,  deterministic=not train   )

        # 6. post-process
        _lowercase      			=self.conv_norm_out(lowerCAmelCase   )
        _lowercase      			=nn.silu(lowerCAmelCase   )
        _lowercase      			=self.conv_out(lowerCAmelCase   )
        _lowercase      			=jnp.transpose(lowerCAmelCase				,  (0, 3, 1, 2)   )

        if not return_dict:
              return (sample,)

        return FlaxUNetaDConditionOutput(sample=lowerCAmelCase   )





# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings('ignore', category=UserWarning, module='torch.optim.lr_scheduler')





class 					__lowerCAmelCase	:




  def __init__(							self				,  lowerCAmelCase				,  lowerCAmelCase				,  lowerCAmelCase = True				,  lowerCAmelCase = False   )    ->					str:
        '''simple docstring'''






        _lowercase      			=scheduler
        _lowercase      			=optimizers if isinstance(lowerCAmelCase				,  (list, tuple)   ) else [optimizers]
        _lowercase      			=split_batches
        _lowercase      			=step_with_optimizer
        _lowercase      			=GradientState()




  def     A__	(							self				,  *lowerCAmelCase				,  **lowerCAmelCase   )    ->					List[Any]:
        '''simple docstring'''






        if not self.step_with_optimizer:
              # No link between scheduler and optimizer -> just step
              self.scheduler.step(*lowerCAmelCase				,  **lowerCAmelCase   )
              return

        # Otherwise, first make sure the optimizer was stepped.
        if not self.gradient_state.sync_gradients:
              if self.gradient_state.adjust_scheduler:
                    self.scheduler._step_count += 1
              return

        for opt in self.optimizers:
              if opt.step_was_skipped:
                    return
        if self.split_batches:
              # Split batches -> the training dataloader batch size is not changed so one step per training step
              self.scheduler.step(*lowerCAmelCase				,  **lowerCAmelCase   )
        else:
              # Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
              # num_processes steps per training step
              _lowercase      			=AcceleratorState().num_processes
              for _ in range(lowerCAmelCase   ):
                    # Special case when using OneCycle and `drop_last` was not used
                    if hasattr(self.scheduler				,  'total_steps'   ):
                          if self.scheduler._step_count <= self.scheduler.total_steps:
                                self.scheduler.step(*lowerCAmelCase				,  **lowerCAmelCase   )
                    else:
                          self.scheduler.step(*lowerCAmelCase				,  **lowerCAmelCase   )




  def     A__	(							self   )    ->					str:
        '''simple docstring'''






        return self.scheduler.get_last_lr()




  def     A__	(							self   )    ->					Tuple:
        '''simple docstring'''






        return self.scheduler.state_dict()




  def     A__	(							self				,  lowerCAmelCase   )    ->					str:
        '''simple docstring'''






        self.scheduler.load_state_dict(lowerCAmelCase   )




  def     A__	(							self   )    ->					int:
        '''simple docstring'''






        return self.scheduler.get_lr()




  def     A__	(							self				,  *lowerCAmelCase				,  **lowerCAmelCase   )    ->					Optional[int]:
        '''simple docstring'''






        return self.scheduler.print_lr(*lowerCAmelCase				,  **lowerCAmelCase   )


