


'''simple docstring'''



import os
from typing import Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn

from ...models.controlnet import ControlNetModel, ControlNetOutput
from ...models.modeling_utils import ModelMixin
from ...utils import logging


A__						:   Dict				=    logging.get_logger(__name__)





class   UpperCAmelCase_    (_UpperCAmelCase    ):





    """simple docstring"""






    def __init__(				self     ,	SCREAMING_SNAKE_CASE_      )							->	List[str]:
         super().__init__()
         __lowerCamelCase  :      str														=						nn.ModuleList(SCREAMING_SNAKE_CASE_      )







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = False     ,	SCREAMING_SNAKE_CASE_ = True     ,	)							->	Union[ControlNetOutput, Tuple]:
         for i, (image, scale, controlnet) in enumerate(zip(SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	self.nets      )      ):
              __lowerCamelCase							,			__lowerCamelCase  :      str														=						controlnet(
                  SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_     ,	)

              # merge samples
              if i == 0:
                   __lowerCamelCase							,			__lowerCamelCase  :      Tuple														=						down_samples, mid_sample
              else:
                   __lowerCamelCase  :      str														=						[
                       samples_prev + samples_curr
                       for samples_prev, samples_curr in zip(SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_      )
                   ]
                   mid_block_res_sample += mid_sample

         return down_block_res_samples, mid_block_res_sample







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = True     ,	SCREAMING_SNAKE_CASE_ = None     ,	SCREAMING_SNAKE_CASE_ = False     ,	SCREAMING_SNAKE_CASE_ = None     ,	)							->	List[Any]:
         __lowerCamelCase  :      Any														=						0
         __lowerCamelCase  :      Union[str, Any]														=						save_directory
         for controlnet in self.nets:
              controlnet.save_pretrained(
                  SCREAMING_SNAKE_CASE_     ,	is_main_process=SCREAMING_SNAKE_CASE_     ,	save_function=SCREAMING_SNAKE_CASE_     ,	safe_serialization=SCREAMING_SNAKE_CASE_     ,	variant=SCREAMING_SNAKE_CASE_     ,	)

              idx += 1
              __lowerCamelCase  :      List[str]														=						model_path_to_save + f'_{idx}'







    @classmethod
    def  lowercase_							(				cls     ,	SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )							->	Optional[Any]:
         __lowerCamelCase  :      str														=						0
         __lowerCamelCase  :      Any														=						[]

         # load controlnet and append to list until no controlnet directory exists anymore
         # first controlnet has to be saved under `./mydirectory/controlnet` to be compliant with `DiffusionPipeline.from_prertained`
         # second, third, ... controlnets have to be saved under `./mydirectory/controlnet_1`, `./mydirectory/controlnet_2`, ...
         __lowerCamelCase  :      str														=						pretrained_model_path
         while os.path.isdir(SCREAMING_SNAKE_CASE_      ):
              __lowerCamelCase  :      Optional[int]														=						ControlNetModel.from_pretrained(SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
              controlnets.append(SCREAMING_SNAKE_CASE_      )

              idx += 1
              __lowerCamelCase  :      Dict														=						pretrained_model_path + f'_{idx}'

         logger.info(f'{len(SCREAMING_SNAKE_CASE_      )} controlnets loaded from {pretrained_model_path}.'      )

         if len(SCREAMING_SNAKE_CASE_      ) == 0:
              raise ValueError(
                  f'No ControlNets found under {os.path.dirname(SCREAMING_SNAKE_CASE_      )}. Expected at least {pretrained_model_path + "_0"}.'      )

         return cls(SCREAMING_SNAKE_CASE_      )












'''simple docstring'''





import re

from flax.core.frozen_dict import freeze
from flax.traverse_util import flatten_dict, unflatten_dict
from jax.experimental import PartitionSpec as P


# Sentinels
UpperCamelCase_		:     Optional[int]          =		object()

# For specifying empty leaf dict `{}`
UpperCamelCase_		:     Optional[int]          =		object()





def       __a      (  _UpperCamelCase:       Union[str, Any]   ,		_UpperCamelCase:       Optional[Any]       )		->					Optional[Any]:

     """simple docstring"""




     _snake_case				= tuple((re.compile(x + "$"       ) for x in qs)       )
     for i in range(len(_UpperCamelCase       ) - len(_UpperCamelCase       ) + 1       ):
          _snake_case				= [x.match(_UpperCamelCase       ) for x, y in zip(_UpperCamelCase   ,		ks[i:]       )]
          if matches and all(_UpperCamelCase       ):
               return True
     return False





def       __a      (  _UpperCamelCase:       Optional[Any]       )		->					Union[str, Any]:

     """simple docstring"""




     def replace(_UpperCamelCase:       Tuple   ,		_UpperCamelCase:       List[str]       ):
          for rule, replacement in rules:
               if _match(_UpperCamelCase   ,		_UpperCamelCase       ):
                    return replacement
          return val

     return replace





def       __a      (  )		->					Any:

     """simple docstring"""




     return [
         # embeddings
         (("transformer", "wpe", "embedding"), P("mp"   ,		_UpperCamelCase       )),
         (("transformer", "wte", "embedding"), P("mp"   ,		_UpperCamelCase       )),
         # atention
         (("attention", "(q_proj|k_proj|v_proj)", "kernel"), P(_UpperCamelCase   ,		"mp"       )),
         (("attention", "out_proj", "kernel"), P("mp"   ,		_UpperCamelCase       )),
         (("attention", "out_proj", "bias"), None),
         # mlp
         (("mlp", "c_fc", "kernel"), P(_UpperCamelCase   ,		"mp"       )),
         (("mlp", "c_fc", "bias"), P("mp"       )),
         (("mlp", "c_proj", "kernel"), P("mp"   ,		_UpperCamelCase       )),
         (("mlp", "c_proj", "bias"), None),
         # layer norms
         ((r"ln_\d+", "bias"), None),
         ((r"\d+", r"ln_\d+", "scale"), None),
         (("ln_f", "bias"), None),
         (("ln_f", "scale"), None),
     ]





def       __a      (  _UpperCamelCase:       Union[str, Any]       )		->					Any:

     """simple docstring"""




     _snake_case				= _get_partition_rules()
     _snake_case				= _replacement_rules(_UpperCamelCase       )
     _snake_case				= {k: _unmatched for k in flatten_dict(_UpperCamelCase       )}
     _snake_case				= {k: replace(_UpperCamelCase   ,		_UpperCamelCase       ) for k, v in initd.items()}
     assert _unmatched not in result.values(), "Incomplete partition spec."
     return freeze(unflatten_dict(_UpperCamelCase       )       )


