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






@flax.struct.dataclass
class    UpperCAmelCase_				(	_UpperCAmelCase					):
 UpperCAmelCase_		    =						42






@flax_register_to_config
class    UpperCAmelCase_				(	nn.Module    ,						_UpperCAmelCase    ,						_UpperCAmelCase					):
 UpperCAmelCase_		    =						3_2
 UpperCAmelCase_		    =						4
 UpperCAmelCase_		    =						4
 UpperCAmelCase_		    =						(
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "DownBlock2D",
 )
 UpperCAmelCase_		    =						("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
 UpperCAmelCase_		    =						False
 UpperCAmelCase_		    =						(3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
 UpperCAmelCase_		    =						2
 UpperCAmelCase_		    =						8
 UpperCAmelCase_		    =						None
 UpperCAmelCase_		    =						1_2_8_0
 UpperCAmelCase_		    =						0.0
 UpperCAmelCase_		    =						False
 UpperCAmelCase_		    =						jnp.floataa
 UpperCAmelCase_		    =						True
 UpperCAmelCase_		    =						0
 UpperCAmelCase_		    =						False





 def        snake_case__		(    self ,       lowercase_):
     # init input tensors
     snake_case_ :    str							=						(1, self.in_channels, self.sample_size, self.sample_size)
     snake_case_ :    Any							=						jnp.zeros(lowercase__ ,       dtype=jnp.floataa)
     snake_case_ :    Optional[Any]							=						jnp.ones((1,) ,       dtype=jnp.intaa)
     snake_case_ :    Optional[int]							=						jnp.zeros((1, 1, self.cross_attention_dim) ,       dtype=jnp.floataa)

     snake_case_ :    str							=						jax.random.split(lowercase__)
     snake_case_ :    Optional[int]							=						{"""params""": params_rng, """dropout""": dropout_rng}

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





 def        snake_case__		(    self):
     snake_case_ :    Any							=						self.block_out_channels
     snake_case_ :    Union[str, Any]							=						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.
     snake_case_ :    Dict							=						self.num_attention_heads or self.attention_head_dim

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

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

     snake_case_ :    Optional[Any]							=						self.only_cross_attention
     if isinstance(lowercase__ ,       lowercase__):
         snake_case_ :    int							=						(only_cross_attention,) * len(self.down_block_types)

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

     # down
     snake_case_ :    str							=						[]
     snake_case_ :    Tuple							=						block_out_channels[0]
     for i, down_block_type in enumerate(self.down_block_types):
         snake_case_ :    Tuple							=						output_channel
         snake_case_ :    Any							=						block_out_channels[i]
         snake_case_ :    int							=						i == len(lowercase__) - 1

         if down_block_type == "CrossAttnDownBlock2D":
             snake_case_ :    Optional[Any]							=						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] ,       use_memory_efficient_attention=self.use_memory_efficient_attention ,       dtype=self.dtype ,       )
         else:
             snake_case_ :    Optional[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__)
     snake_case_ :    Optional[int]							=						down_blocks

     # mid
     snake_case_ :    str							=						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
     snake_case_ :    int							=						[]
     snake_case_ :    Any							=						list(reversed(lowercase__))
     snake_case_ :    Tuple							=						list(reversed(lowercase__))
     snake_case_ :    str							=						list(reversed(lowercase__))
     snake_case_ :    Optional[Any]							=						reversed_block_out_channels[0]
     for i, up_block_type in enumerate(self.up_block_types):
         snake_case_ :    Optional[Any]							=						output_channel
         snake_case_ :    Union[str, Any]							=						reversed_block_out_channels[i]
         snake_case_ :    str							=						reversed_block_out_channels[min(i + 1 ,       len(lowercase__) - 1)]

         snake_case_ :    Tuple							=						i == len(lowercase__) - 1

         if up_block_type == "CrossAttnUpBlock2D":
             snake_case_ :    Dict							=						FlaxCrossAttnUpBlockaD(
                 in_channels=lowercase__ ,       out_channels=lowercase__ ,       prev_output_channel=lowercase__ ,       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:
             snake_case_ :    List[str]							=						FlaxUpBlockaD(
                 in_channels=lowercase__ ,       out_channels=lowercase__ ,       prev_output_channel=lowercase__ ,       num_layers=self.layers_per_block + 1 ,       add_upsample=not is_final_block ,       dropout=self.dropout ,       dtype=self.dtype ,       )

         up_blocks.append(lowercase__)
         snake_case_ :    Optional[Any]							=						output_channel
     snake_case_ :    Optional[int]							=						up_blocks

     # out
     snake_case_ :    Dict							=						nn.GroupNorm(num_groups=32 ,       epsilon=1E-5)
     snake_case_ :    Dict							=						nn.Conv(
         self.out_channels ,       kernel_size=(3, 3) ,       strides=(1, 1) ,       padding=((1, 1), (1, 1)) ,       dtype=self.dtype ,       )





 def __call__(    self ,       lowercase_ ,       lowercase_ ,       lowercase_ ,       lowercase_=None ,       lowercase_=None ,       lowercase_ = True ,       lowercase_ = False ,       ):
     # 1. time
     if not isinstance(lowercase__ ,       jnp.ndarray):
         snake_case_ :    Dict							=						jnp.array([timesteps] ,       dtype=jnp.intaa)
     elif isinstance(lowercase__ ,       jnp.ndarray) and len(timesteps.shape) == 0:
         snake_case_ :    Any							=						timesteps.astype(dtype=jnp.floataa)
         snake_case_ :    List[Any]							=						jnp.expand_dims(lowercase__ ,       0)

     snake_case_ :    List[str]							=						self.time_proj(lowercase__)
     snake_case_ :    Any							=						self.time_embedding(lowercase__)

     # 2. pre-process
     snake_case_ :    int							=						jnp.transpose(lowercase__ ,       (0, 2, 3, 1))
     snake_case_ :    Any							=						self.conv_in(lowercase__)

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

     if down_block_additional_residuals is not None:
         snake_case_ :    Optional[Any]							=						()

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

         snake_case_ :    str							=						new_down_block_res_samples

     # 4. mid
     snake_case_ :    Tuple							=						self.mid_block(lowercase__ ,       lowercase__ ,       lowercase__ ,       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:
         snake_case_ :    Optional[Any]							=						down_block_res_samples[-(self.layers_per_block + 1) :]
         snake_case_ :    List[Any]							=						down_block_res_samples[: -(self.layers_per_block + 1)]
         if isinstance(lowercase__ ,       lowercase__):
             snake_case_ :    Union[str, Any]							=						up_block(
                 lowercase__ ,       temb=lowercase__ ,       encoder_hidden_states=lowercase__ ,       res_hidden_states_tuple=lowercase__ ,       deterministic=not train ,       )
         else:
             snake_case_ :    List[Any]							=						up_block(lowercase__ ,       temb=lowercase__ ,       res_hidden_states_tuple=lowercase__ ,       deterministic=not train)

        # 6. post-process
     snake_case_ :    List[Any]							=						self.conv_norm_out(lowercase__)
     snake_case_ :    Tuple							=						nn.silu(lowercase__)
     snake_case_ :    List[Any]							=						self.conv_out(lowercase__)
     snake_case_ :    Optional[Any]							=						jnp.transpose(lowercase__ ,       (0, 3, 1, 2))

     if not return_dict:
         return (sample,)

     return FlaxUNetaDConditionOutput(sample=lowercase__)






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

from collections.abc import Callable
from typing import Generic, TypeVar

a_											=						TypeVar("T")
a_											=						TypeVar("U")






class    UpperCAmelCase_				(	Generic[T, U]					):





 def __init__(    self ,       lowercase_ ,       lowercase_):
     snake_case_ :    Any     			=						key
     snake_case_ :    List[Any]     			=						val
     snake_case_ :    DoubleLinkedListNode[T, U] | None     			=						None
     snake_case_ :    DoubleLinkedListNode[T, U] | None     			=						None





 def __repr__(    self):
     return (
         F'Node: key: {self.key}, val: {self.val}, '
         F'has next: {bool(self.next)}, has prev: {bool(self.prev)}'
     )






class    UpperCAmelCase_				(	Generic[T, U]					):





 def __init__(    self):
     snake_case_ :    DoubleLinkedListNode[T, U]     			=						DoubleLinkedListNode(lowercase_ ,       lowercase_)
     snake_case_ :    DoubleLinkedListNode[T, U]     			=						DoubleLinkedListNode(lowercase_ ,       lowercase_)
     snake_case_			, snake_case_ :    Union[str, Any]     			=						self.rear, self.head





 def __repr__(    self):
     snake_case_ :    Dict     			=						["DoubleLinkedList"]
     snake_case_ :    Dict     			=						self.head
     while node.next is not None:
         rep.append(str(lowercase_))
         snake_case_ :    List[str]     			=						node.next
     rep.append(str(self.rear))
     return ",\n    ".join(lowercase_)





 def        snake_case__		(    self ,       lowercase_):
     snake_case_ :    List[str]     			=						self.rear.prev

     # All nodes other than self.head are guaranteed to have non-None previous
     assert previous is not None

     snake_case_ :    Tuple     			=						node
     snake_case_ :    str     			=						previous
     snake_case_ :    Optional[Any]     			=						node
     snake_case_ :    Any     			=						self.rear





 def        snake_case__		(    self ,       lowercase_):

     if node.prev is None or node.next is None:
         return None

     snake_case_ :    Union[str, Any]     			=						node.next
     snake_case_ :    Optional[int]     			=						node.prev
     snake_case_ :    str     			=						None
     snake_case_ :    int     			=						None
     return node






class    UpperCAmelCase_				(	Generic[T, U]					):
 UpperCAmelCase_		    =						{}





 def __init__(    self ,       lowercase_):
     snake_case_ :    DoubleLinkedList[T, U]     			=						DoubleLinkedList()
     snake_case_ :    List[str]     			=						capacity
     snake_case_ :    Any     			=						0
     snake_case_ :    Dict     			=						0
     snake_case_ :    Union[str, Any]     			=						0
     snake_case_ :    dict[T, DoubleLinkedListNode[T, U]]     			=						{}





 def __repr__(    self):
     return (
         F'CacheInfo(hits={self.hits}, misses={self.miss}, '
         F'capacity={self.capacity}, current size={self.num_keys})'
     )





 def __contains__(    self ,       lowercase_):
     return key in self.cache





 def        snake_case__		(    self ,       lowercase_):
     # Note: pythonic interface would throw KeyError rather than return None

     if key in self.cache:
         self.hits += 1
         snake_case_ :    DoubleLinkedListNode[T, U]     			=						self.cache[key]
         snake_case_ :    Optional[Any]     			=						self.list.remove(self.cache[key])
         assert node == value_node

         # node is guaranteed not None because it is in self.cache
         assert node is not None
         self.list.add(lowercase_)
         return node.val
     self.miss += 1
     return None





 def        snake_case__		(    self ,       lowercase_ ,       lowercase_):

     if key not in self.cache:
         if self.num_keys >= self.capacity:
             # delete first node (oldest) when over capacity
             snake_case_ :    Optional[Any]     			=						self.list.head.next

             # guaranteed to have a non-None first node when num_keys > 0
             # explain to type checker via assertions
             assert first_node is not None
             assert first_node.key is not None
             assert (
                 self.list.remove(lowercase_) is not None
             )  # node guaranteed to be in list assert node.key is not None

             del self.cache[first_node.key]
             self.num_keys -= 1
         snake_case_ :    Dict     			=						DoubleLinkedListNode(lowercase_ ,       lowercase_)
         self.list.add(self.cache[key])
         self.num_keys += 1

     else:
         # bump node to the end of the list, update value
         snake_case_ :    List[Any]     			=						self.list.remove(self.cache[key])
         assert node is not None  # node guaranteed to be in list
         snake_case_ :    Optional[Any]     			=						value
         self.list.add(lowercase_)





 @classmethod
 def        snake_case__		(    cls ,       lowercase_ = 1_28):

     def cache_decorator_inner(lowercase_) -> Callable[..., U]:
         def cache_decorator_wrapper(*lowercase_) -> U:
             if func not in cls.decorator_function_to_instance_map:
                 snake_case_ :    List[str]     			=						LRUCache(lowercase_)

             snake_case_ :    Optional[Any]     			=						cls.decorator_function_to_instance_map[func].get(args[0])
             if result is None:
                 snake_case_ :    Any     			=						func(*lowercase_)
                 cls.decorator_function_to_instance_map[func].put(args[0] ,       lowercase_)
             return result

         def cache_info() -> LRUCache[T, U]:
             return cls.decorator_function_to_instance_map[func]

         setattr(lowercase_ ,       "cache_info" ,       lowercase_)  # noqa: B010

         return cache_decorator_wrapper

     return cache_decorator_inner


if __name__ == "__main__":
  import doctest

  doctest.testmod()




