import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest






class 		SCREAMING_SNAKE_CASE__		(   UpperCamelCase_    ):
    __a  =      (DEISMultistepScheduler,)
    __a  =      (("num_inference_steps", 25),)
    def 						UpperCamelCase_				(  self , **lowerCAmelCase   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   Dict=      {
         '''num_train_timesteps''': 1000,
         '''beta_start''': 0.0001,
         '''beta_end''': 0.02,
         '''beta_schedule''': '''linear''',
         '''solver_order''': 2,
     }

     config.update(**lowerCAmelCase   )
     return config
    def 						UpperCamelCase_				(  self , lowerCAmelCase=0 , **lowerCAmelCase   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   str=      dict(self.forward_default_kwargs   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      kwargs.pop('''num_inference_steps''' , lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   str=      self.dummy_sample
     SCREAMING_SNAKE_CASE__:   str=      0.1 * sample
     SCREAMING_SNAKE_CASE__:   List[str]=      [residual + 0.2, residual + 0.15, residual + 0.10]

     for scheduler_class in self.scheduler_classes:
      SCREAMING_SNAKE_CASE__:   str=      self.get_scheduler_config(**lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      scheduler_class(**lowerCAmelCase   )
      scheduler.set_timesteps(lowerCAmelCase   )
      # copy over dummy past residuals
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      dummy_past_residuals[: scheduler.config.solver_order]

      with tempfile.TemporaryDirectory() as tmpdirname:
       scheduler.save_config(lowerCAmelCase   )
       SCREAMING_SNAKE_CASE__:   Optional[int]=      scheduler_class.from_pretrained(lowerCAmelCase   )
       new_scheduler.set_timesteps(lowerCAmelCase   )
       # copy over dummy past residuals
       SCREAMING_SNAKE_CASE__:   Tuple=      dummy_past_residuals[: new_scheduler.config.solver_order]

      SCREAMING_SNAKE_CASE__:   List[Any]=      sample, sample
      for t in range(lowerCAmelCase , time_step + scheduler.config.solver_order + 1   ):
       SCREAMING_SNAKE_CASE__:   Optional[int]=      scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , **lowerCAmelCase   ).prev_sample
       SCREAMING_SNAKE_CASE__:   str=      new_scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , **lowerCAmelCase   ).prev_sample

       assert torch.sum(torch.abs(output - new_output   )   ) < 1e-5, "Scheduler outputs are not identical"
    def 						UpperCamelCase_				(  self   )       ->					int:
     pass
    def 						UpperCamelCase_				(  self , lowerCAmelCase=0 , **lowerCAmelCase   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   str=      dict(self.forward_default_kwargs   )
     SCREAMING_SNAKE_CASE__:   List[str]=      kwargs.pop('''num_inference_steps''' , lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.dummy_sample
     SCREAMING_SNAKE_CASE__:   int=      0.1 * sample
     SCREAMING_SNAKE_CASE__:   int=      [residual + 0.2, residual + 0.15, residual + 0.10]

     for scheduler_class in self.scheduler_classes:
      SCREAMING_SNAKE_CASE__:   Tuple=      self.get_scheduler_config()
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      scheduler_class(**lowerCAmelCase   )
      scheduler.set_timesteps(lowerCAmelCase   )

      # copy over dummy past residuals (must be after setting timesteps)
      SCREAMING_SNAKE_CASE__:   List[Any]=      dummy_past_residuals[: scheduler.config.solver_order]

      with tempfile.TemporaryDirectory() as tmpdirname:
       scheduler.save_config(lowerCAmelCase   )
       SCREAMING_SNAKE_CASE__:   List[Any]=      scheduler_class.from_pretrained(lowerCAmelCase   )
       # copy over dummy past residuals
       new_scheduler.set_timesteps(lowerCAmelCase   )

       # copy over dummy past residual (must be after setting timesteps)
       SCREAMING_SNAKE_CASE__:   Dict=      dummy_past_residuals[: new_scheduler.config.solver_order]

      SCREAMING_SNAKE_CASE__:   Tuple=      scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , **lowerCAmelCase   ).prev_sample
      SCREAMING_SNAKE_CASE__:   Union[str, Any]=      new_scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , **lowerCAmelCase   ).prev_sample

      assert torch.sum(torch.abs(output - new_output   )   ) < 1e-5, "Scheduler outputs are not identical"
    def 						UpperCamelCase_				(  self , lowerCAmelCase=None , **lowerCAmelCase   )       ->					Any:
     if scheduler is None:
      SCREAMING_SNAKE_CASE__:   Dict=      self.scheduler_classes[0]
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.get_scheduler_config(**lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   int=      scheduler_class(**lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      self.scheduler_classes[0]
     SCREAMING_SNAKE_CASE__:   str=      self.get_scheduler_config(**lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Any=      scheduler_class(**lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Dict=      10
     SCREAMING_SNAKE_CASE__:   Optional[int]=      self.dummy_model()
     SCREAMING_SNAKE_CASE__:   Any=      self.dummy_sample_deter
     scheduler.set_timesteps(lowerCAmelCase   )

     for i, t in enumerate(scheduler.timesteps   ):
      SCREAMING_SNAKE_CASE__:   List[str]=      model(lowerCAmelCase , lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Any=      scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   ).prev_sample

     return sample
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   List[Any]=      dict(self.forward_default_kwargs   )

     SCREAMING_SNAKE_CASE__:   Dict=      kwargs.pop('''num_inference_steps''' , lowerCAmelCase   )

     for scheduler_class in self.scheduler_classes:
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.get_scheduler_config()
      SCREAMING_SNAKE_CASE__:   List[Any]=      scheduler_class(**lowerCAmelCase   )

      SCREAMING_SNAKE_CASE__:   str=      self.dummy_sample
      SCREAMING_SNAKE_CASE__:   List[str]=      0.1 * sample

      if num_inference_steps is not None and hasattr(lowerCAmelCase , '''set_timesteps'''   ):
       scheduler.set_timesteps(lowerCAmelCase   )
      elif num_inference_steps is not None and not hasattr(lowerCAmelCase , '''set_timesteps'''   ):
       SCREAMING_SNAKE_CASE__:   List[str]=      num_inference_steps

      # copy over dummy past residuals (must be done after set_timesteps)
      SCREAMING_SNAKE_CASE__:   Tuple=      [residual + 0.2, residual + 0.15, residual + 0.10]
      SCREAMING_SNAKE_CASE__:   List[str]=      dummy_past_residuals[: scheduler.config.solver_order]

      SCREAMING_SNAKE_CASE__:   Dict=      scheduler.timesteps[5]
      SCREAMING_SNAKE_CASE__:   List[str]=      scheduler.timesteps[6]

      SCREAMING_SNAKE_CASE__:   List[str]=      scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , **lowerCAmelCase   ).prev_sample
      SCREAMING_SNAKE_CASE__:   Optional[int]=      scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , **lowerCAmelCase   ).prev_sample

      self.assertEqual(output_a.shape , sample.shape   )
      self.assertEqual(output_a.shape , output_a.shape   )
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     # make sure that iterating over schedulers with same config names gives same results
     # for defaults
     SCREAMING_SNAKE_CASE__:   str=      DEISMultistepScheduler(**self.get_scheduler_config()   )
     SCREAMING_SNAKE_CASE__:   List[str]=      self.full_loop(scheduler=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Tuple=      torch.mean(torch.abs(lowerCAmelCase   )   )

     assert abs(result_mean.item() - 0.23916   ) < 1e-3

     SCREAMING_SNAKE_CASE__:   int=      DPMSolverSinglestepScheduler.from_config(scheduler.config   )
     SCREAMING_SNAKE_CASE__:   int=      DPMSolverMultistepScheduler.from_config(scheduler.config   )
     SCREAMING_SNAKE_CASE__:   str=      UniPCMultistepScheduler.from_config(scheduler.config   )
     SCREAMING_SNAKE_CASE__:   str=      DEISMultistepScheduler.from_config(scheduler.config   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      self.full_loop(scheduler=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Tuple=      torch.mean(torch.abs(lowerCAmelCase   )   )

     assert abs(result_mean.item() - 0.23916   ) < 1e-3
    def 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     for timesteps in [25, 50, 100, 999, 1000]:
      self.check_over_configs(num_train_timesteps=lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     self.check_over_configs(thresholding=lowerCAmelCase   )
     for order in [1, 2, 3]:
      for solver_type in ["logrho"]:
       for threshold in [0.5, 1.0, 2.0]:
        for prediction_type in ["epsilon", "sample"]:
         self.check_over_configs(
             thresholding=lowerCAmelCase , prediction_type=lowerCAmelCase , sample_max_value=lowerCAmelCase , algorithm_type='''deis''' , solver_order=lowerCAmelCase , solver_type=lowerCAmelCase , )
    def 						UpperCamelCase_				(  self   )       ->					Optional[int]:
     for prediction_type in ["epsilon", "v_prediction"]:
      self.check_over_configs(prediction_type=lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     for algorithm_type in ["deis"]:
      for solver_type in ["logrho"]:
       for order in [1, 2, 3]:
        for prediction_type in ["epsilon", "sample"]:
         self.check_over_configs(
             solver_order=lowerCAmelCase , solver_type=lowerCAmelCase , prediction_type=lowerCAmelCase , algorithm_type=lowerCAmelCase , )
         SCREAMING_SNAKE_CASE__:   List[Any]=      self.full_loop(
             solver_order=lowerCAmelCase , solver_type=lowerCAmelCase , prediction_type=lowerCAmelCase , algorithm_type=lowerCAmelCase , )
         assert not torch.isnan(lowerCAmelCase   ).any(), "Samples have nan numbers"
    def 						UpperCamelCase_				(  self   )       ->					Dict:
     self.check_over_configs(lower_order_final=lowerCAmelCase   )
     self.check_over_configs(lower_order_final=lowerCAmelCase   )
    def 						UpperCamelCase_				(  self   )       ->					int:
     for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1000]:
      self.check_over_forward(num_inference_steps=lowerCAmelCase , time_step=0   )
    def 						UpperCamelCase_				(  self   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   List[Any]=      self.full_loop()
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.mean(torch.abs(lowerCAmelCase   )   )

     assert abs(result_mean.item() - 0.23916   ) < 1e-3
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     SCREAMING_SNAKE_CASE__:   List[Any]=      self.full_loop(prediction_type='''v_prediction'''   )
     SCREAMING_SNAKE_CASE__:   str=      torch.mean(torch.abs(lowerCAmelCase   )   )

     assert abs(result_mean.item() - 0.091   ) < 1e-3



    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      self.scheduler_classes[0]
     SCREAMING_SNAKE_CASE__:   Tuple=      self.get_scheduler_config(thresholding=lowerCAmelCase , dynamic_thresholding_ratio=0   )
     SCREAMING_SNAKE_CASE__:   str=      scheduler_class(**lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      10
     SCREAMING_SNAKE_CASE__:   str=      self.dummy_model()
     SCREAMING_SNAKE_CASE__:   List[str]=      self.dummy_sample_deter.half()
     scheduler.set_timesteps(lowerCAmelCase   )

     for i, t in enumerate(scheduler.timesteps   ):
      SCREAMING_SNAKE_CASE__:   int=      model(lowerCAmelCase , lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Dict=      scheduler.step(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   ).prev_sample

     assert sample.dtype == torch.floataa


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		(   UpperCamelCase_    ):
    __a  =      42
    __a  =      42






class 		_lowerCamelCase		(   nn.Module    ):
    __a  =      42
    __a  =      (16, 32, 96, 256)
    __a  =      jnp.floataa
    def 						UpperCamelCase_				(  self   )       ->					Optional[int]:
     SCREAMING_SNAKE_CASE__:   List[str]=      nn.Conv(
         self.block_out_channels[0] , kernel_size=(3, 3) , padding=((1, 1), (1, 1)) , dtype=self.dtype , )

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

     SCREAMING_SNAKE_CASE__:   Optional[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 , lowerCAmelCase   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Dict=      self.conv_in(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Optional[int]=      nn.silu(lowerCAmelCase   )

     for block in self.blocks:
      SCREAMING_SNAKE_CASE__:   Optional[int]=      block(lowerCAmelCase   )
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      nn.silu(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Any=      self.conv_out(lowerCAmelCase   )

     return embedding






@flax_register_to_config
class 		_lowerCamelCase		(   nn.Module    ,     UpperCamelCase_    ,     UpperCamelCase_    ):
    __a  =      32
    __a  =      4
    __a  =      (
        "CrossAttnDownBlock2D",
        "CrossAttnDownBlock2D",
        "CrossAttnDownBlock2D",
        "DownBlock2D",
    )
    __a  =      False
    __a  =      (320, 640, 1280, 1280)
    __a  =      2
    __a  =      8
    __a  =      None
    __a  =      1280
    __a  =      0.0
    __a  =      False
    __a  =      jnp.floataa
    __a  =      True
    __a  =      0
    __a  =      "rgb"
    __a  =      (16, 32, 96, 256)
    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					FrozenDict:
     # init input tensors
     SCREAMING_SNAKE_CASE__:   Any=      (1, self.in_channels, self.sample_size, self.sample_size)
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      jnp.zeros(lowerCAmelCase , dtype=jnp.floataa   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      jnp.ones((1,) , dtype=jnp.intaa   )
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      jnp.zeros((1, 1, self.cross_attention_dim) , dtype=jnp.floataa   )
     SCREAMING_SNAKE_CASE__:   int=      (1, 3, self.sample_size * 8, self.sample_size * 8)
     SCREAMING_SNAKE_CASE__:   Optional[int]=      jnp.zeros(lowerCAmelCase , dtype=jnp.floataa   )

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

     return self.init(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )["params"]
    def 						UpperCamelCase_				(  self   )       ->					Optional[Any]:
     SCREAMING_SNAKE_CASE__:   Dict=      self.block_out_channels
     SCREAMING_SNAKE_CASE__:   str=      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.
     SCREAMING_SNAKE_CASE__:   int=      self.num_attention_heads or self.attention_head_dim

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

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

     SCREAMING_SNAKE_CASE__:   str=      FlaxControlNetConditioningEmbedding(
         conditioning_embedding_channels=block_out_channels[0] , block_out_channels=self.conditioning_embedding_out_channels , )

     SCREAMING_SNAKE_CASE__:   List[str]=      self.only_cross_attention
     if isinstance(lowerCAmelCase , lowerCAmelCase   ):
      SCREAMING_SNAKE_CASE__:   str=      (only_cross_attention,) * len(self.down_block_types   )

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

     # down
     SCREAMING_SNAKE_CASE__:   Optional[int]=      []
     SCREAMING_SNAKE_CASE__:   List[str]=      []

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      block_out_channels[0]

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

     for i, down_block_type in enumerate(self.down_block_types   ):
      SCREAMING_SNAKE_CASE__:   Tuple=      output_channel
      SCREAMING_SNAKE_CASE__:   Any=      block_out_channels[i]
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      i == len(lowerCAmelCase   ) - 1

      if down_block_type == "CrossAttnDownBlock2D":
       SCREAMING_SNAKE_CASE__:   Any=      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] , dtype=self.dtype , )
      else:
       SCREAMING_SNAKE_CASE__:   Union[str, Any]=      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   )

      for _ in range(self.layers_per_block   ):
       SCREAMING_SNAKE_CASE__:   str=      nn.Conv(
           lowerCAmelCase , 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(lowerCAmelCase   )

      if not is_final_block:
       SCREAMING_SNAKE_CASE__:   str=      nn.Conv(
           lowerCAmelCase , 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(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   int=      down_blocks
     SCREAMING_SNAKE_CASE__:   int=      controlnet_down_blocks

     # mid
     SCREAMING_SNAKE_CASE__:   Optional[int]=      block_out_channels[-1]
     SCREAMING_SNAKE_CASE__:   int=      FlaxUNetMidBlockaDCrossAttn(
         in_channels=lowerCAmelCase , dropout=self.dropout , num_attention_heads=num_attention_heads[-1] , use_linear_projection=self.use_linear_projection , dtype=self.dtype , )

     SCREAMING_SNAKE_CASE__:   str=      nn.Conv(
         lowerCAmelCase , kernel_size=(1, 1) , padding='''VALID''' , kernel_init=nn.initializers.zeros_init() , bias_init=nn.initializers.zeros_init() , dtype=self.dtype , )



    def __call__(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase = 1.0 , lowerCAmelCase = True , lowerCAmelCase = False , )       ->					Union[FlaxControlNetOutput, Tuple]:
     SCREAMING_SNAKE_CASE__:   Dict=      self.controlnet_conditioning_channel_order
     if channel_order == "bgr":
      SCREAMING_SNAKE_CASE__:   Optional[Any]=      jnp.flip(lowerCAmelCase , axis=1   )

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

     SCREAMING_SNAKE_CASE__:   Dict=      self.time_proj(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Dict=      self.time_embedding(lowerCAmelCase   )

     # 2. pre-process
     SCREAMING_SNAKE_CASE__:   int=      jnp.transpose(lowerCAmelCase , (0, 2, 3, 1)   )
     SCREAMING_SNAKE_CASE__:   str=      self.conv_in(lowerCAmelCase   )

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

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

     # 4. mid
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      self.mid_block(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , deterministic=not train   )

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

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      controlnet_down_block_res_samples

     SCREAMING_SNAKE_CASE__:   Optional[int]=      self.controlnet_mid_block(lowerCAmelCase   )

     # 6. scaling
     SCREAMING_SNAKE_CASE__:   Tuple=      [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=lowerCAmelCase , mid_block_res_sample=lowerCAmelCase   )

