

"""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   __lowercase      (     __lowerCamelCase       ):
  snake_case_												=							42
  snake_case_												=							42







class   __lowercase      (     nn.Module       ):
  snake_case_												=							42
  snake_case_												=							(1_6, 3_2, 9_6, 2_5_6)
  snake_case_												=							jnp.floataa

  def  __lowercase							(	self		:      int     ):






   '''simple docstring'''
   UpperCAmelCase__       :	Optional[int]        =       nn.Conv(
       self.block_out_channels[0]       ,kernel_size=(3, 3)       ,padding=((1, 1), (1, 1))       ,dtype=self.dtype       ,)

   UpperCAmelCase__       :	Tuple        =       []
   for i in range(len(self.block_out_channels     ) - 1     ):
    UpperCAmelCase__       :	Union[str, Any]        =       self.block_out_channels[i]
    UpperCAmelCase__       :	Dict        =       self.block_out_channels[i + 1]
    UpperCAmelCase__       :	Optional[int]        =       nn.Conv(
        A       ,kernel_size=(3, 3)       ,padding=((1, 1), (1, 1))       ,dtype=self.dtype       ,)
    blocks.append(A     )
    UpperCAmelCase__       :	Optional[int]        =       nn.Conv(
        A       ,kernel_size=(3, 3)       ,strides=(2, 2)       ,padding=((1, 1), (1, 1))       ,dtype=self.dtype       ,)
    blocks.append(A     )
   UpperCAmelCase__       :	List[Any]        =       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		:      Dict       ,A		:      int     ):






   '''simple docstring'''
   UpperCAmelCase__       :	Tuple        =       self.conv_in(A     )
   UpperCAmelCase__       :	Tuple        =       nn.silu(A     )

   for block in self.blocks:
    UpperCAmelCase__       :	Tuple        =       block(A     )
    UpperCAmelCase__       :	str        =       nn.silu(A     )

   UpperCAmelCase__       :	Dict        =       self.conv_out(A     )

   return embedding







@flax_register_to_config
class   __lowercase      (     nn.Module						,      __lowerCamelCase						,      __lowerCamelCase       ):
  snake_case_												=							3_2
  snake_case_												=							4
  snake_case_												=							(
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "DownBlock2D",
  )
  snake_case_												=							False
  snake_case_												=							(3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
  snake_case_												=							2
  snake_case_												=							8
  snake_case_												=							None
  snake_case_												=							1_2_8_0
  snake_case_												=							0.0
  snake_case_												=							False
  snake_case_												=							jnp.floataa
  snake_case_												=							True
  snake_case_												=							0
  snake_case_												=							"rgb"
  snake_case_												=							(1_6, 3_2, 9_6, 2_5_6)

  def  __lowercase							(	self		:      Tuple       ,A		:      jax.random.KeyArray     ):






   '''simple docstring'''
   # init input tensors
   UpperCAmelCase__       :	Optional[int]        =       (1, self.in_channels, self.sample_size, self.sample_size)
   UpperCAmelCase__       :	Optional[Any]        =       jnp.zeros(A       ,dtype=jnp.floataa     )
   UpperCAmelCase__       :	Union[str, Any]        =       jnp.ones((1,)       ,dtype=jnp.intaa     )
   UpperCAmelCase__       :	int        =       jnp.zeros((1, 1, self.cross_attention_dim)       ,dtype=jnp.floataa     )
   UpperCAmelCase__       :	Dict        =       (1, 3, self.sample_size * 8, self.sample_size * 8)
   UpperCAmelCase__       :	str        =       jnp.zeros(A       ,dtype=jnp.floataa     )

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

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

  def  __lowercase							(	self		:      Any     ):






   '''simple docstring'''
   UpperCAmelCase__       :	Optional[int]        =       self.block_out_channels
   UpperCAmelCase__       :	Dict        =       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__       :	Dict        =       self.num_attention_heads or self.attention_head_dim

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

   # time
   UpperCAmelCase__       :	List[str]        =       FlaxTimesteps(
       block_out_channels[0]       ,flip_sin_to_cos=self.flip_sin_to_cos       ,freq_shift=self.config.freq_shift     )
   UpperCAmelCase__       :	List[Any]        =       FlaxTimestepEmbedding(A       ,dtype=self.dtype     )

   UpperCAmelCase__       :	Dict        =       FlaxControlNetConditioningEmbedding(
       conditioning_embedding_channels=block_out_channels[0]       ,block_out_channels=self.conditioning_embedding_out_channels       ,)

   UpperCAmelCase__       :	Any        =       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[Any]        =       (num_attention_heads,) * len(self.down_block_types     )

   # down
   UpperCAmelCase__       :	Dict        =       []
   UpperCAmelCase__       :	Optional[int]        =       []

   UpperCAmelCase__       :	Dict        =       block_out_channels[0]

   UpperCAmelCase__       :	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     )

   for i, down_block_type in enumerate(self.down_block_types     ):
    UpperCAmelCase__       :	Optional[Any]        =       output_channel
    UpperCAmelCase__       :	Any        =       block_out_channels[i]
    UpperCAmelCase__       :	List[str]        =       i == len(A     ) - 1

    if down_block_type == "CrossAttnDownBlock2D":
     UpperCAmelCase__       :	Tuple        =       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__       :	List[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__       :	Tuple        =       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[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     )

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

   # mid
   UpperCAmelCase__       :	Dict        =       block_out_channels[-1]
   UpperCAmelCase__       :	Optional[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__       :	Union[str, 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		:      List[str]       ,A		:      Any       ,A		:      Tuple       ,A		:      str       ,A		:      Optional[int]       ,A		:      float = 1.0       ,A		:      bool = True       ,A		:      bool = False       ,):






   '''simple docstring'''
   UpperCAmelCase__       :	Union[str, Any]        =       self.controlnet_conditioning_channel_order
   if channel_order == "bgr":
    UpperCAmelCase__       :	Optional[Any]        =       jnp.flip(A       ,axis=1     )

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

   UpperCAmelCase__       :	str        =       self.time_proj(A     )
   UpperCAmelCase__       :	List[Any]        =       self.time_embedding(A     )

   # 2. pre-process
   UpperCAmelCase__       :	Dict        =       jnp.transpose(A       ,(0, 2, 3, 1)     )
   UpperCAmelCase__       :	Tuple        =       self.conv_in(A     )

   UpperCAmelCase__       :	List[Any]        =       jnp.transpose(A       ,(0, 2, 3, 1)     )
   UpperCAmelCase__       :	Dict        =       self.controlnet_cond_embedding(A     )
   sample += controlnet_cond

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

   # 4. mid
   UpperCAmelCase__       :	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__       :	Tuple        =       controlnet_block(A     )
    controlnet_down_block_res_samples += (down_block_res_sample,)

   UpperCAmelCase__       :	int        =       controlnet_down_block_res_samples

   UpperCAmelCase__       :	Tuple        =       self.controlnet_mid_block(A     )

   # 6. scaling
   UpperCAmelCase__       :	List[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=A       ,mid_block_res_sample=A     )








import gc
import threading
import time

import psutil
import torch





class        lowerCAmelCase__  :



       '''simple docstring'''







       def __init__(							self     ):







             '''simple docstring'''



             __A    =psutil.Process()
             __A    =False

       def     __UpperCamelCase     (							self     ):







             '''simple docstring'''



             __A    =-1

             while True:
                   __A    =max(self.process.memory_info().rss ,   self.cpu_memory_peak     )

                   # can't sleep or will not catch the peak right (this comment is here on purpose)
                   if not self.peak_monitoring:
                         break

       def     __UpperCamelCase     (							self     ):







             '''simple docstring'''



             __A    =True
             __A    =threading.Thread(target=self.peak_monitor     )
             __A    =True
             self.thread.start()


       def     __UpperCamelCase     (							self     ):







             '''simple docstring'''



             __A    =False
             self.thread.join()
             return self.cpu_memory_peak

_lowerCamelCase      :					Union[str, Any]					    =      PeakCPUMemory()





def      A__					(			)					->Dict:
      # Time
      __A    ={'''time''': time.time()}

      gc.collect()
      torch.cuda.empty_cache()

      # CPU mem
      __A    =psutil.Process().memory_info().rss
      cpu_peak_tracker.start()

      # GPU mem
      for i in range(torch.cuda.device_count()			):
            __A    =torch.cuda.memory_allocated(__A			)
      torch.cuda.reset_peak_memory_stats()

      return measures





def      A__					(			__A  :						Any			)					->Any:
      # Time
      __A    ={'''time''': time.time() - start_measures['''time''']}

      gc.collect()
      torch.cuda.empty_cache()

      # CPU mem
      __A    =(psutil.Process().memory_info().rss - start_measures['''cpu''']) / 2**20
      __A    =(cpu_peak_tracker.stop() - start_measures['''cpu''']) / 2**20

      # GPU mem
      for i in range(torch.cuda.device_count()			):
            __A    =(torch.cuda.memory_allocated(__A			) - start_measures[str(__A			)]) / 2**20
            __A    =(torch.cuda.max_memory_allocated(__A			) - start_measures[str(__A			)]) / 2**20

      return measures





def      A__					(			__A  :						Union[str, Any]     ,				__A  :						Any			)					->Union[str, Any]:
      print(F'''{description}:'''			)
      print(F'''- Time: {measures['time']:.2f}s'''			)
      for i in range(torch.cuda.device_count()			):
            print(F'''- GPU {i} allocated: {measures[str(__A			)]:.2f}MiB'''			)
            __A    =measures[F'''{i}-peak''']
            print(F'''- GPU {i} peak: {peak:.2f}MiB'''			)
      print(F'''- CPU RAM allocated: {measures['cpu']:.2f}MiB'''			)
      print(F'''- CPU RAM peak: {measures['cpu-peak']:.2f}MiB'''			)


