
"""simple docstring"""






from importlib import import_module

from .logging import get_logger


lowerCamelCase__ :       int						     =			get_logger(__name__)




class 				lowercase__:
	'''simple docstring'''

	def __init__(						self    :int		, lowerCamelCase_    :List[str]		, lowerCamelCase_    :Optional[Any]=None					)							->			int:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     str				= attrs or []
			if module is not None:
					for key in module.__dict__:
							if key in attrs or not key.startswith('''__'''					):
									setattr(self		, lowerCamelCase_		, getattr(lowerCamelCase_		, lowerCamelCase_					)					)
			SCREAMING_SNAKE_CASE       :     Tuple				= module._original_module if isinstance(lowerCamelCase_		, _PatchedModuleObj					) else module





class 				lowercase__:
	'''simple docstring'''

	UpperCamelCase				=  []





	def __init__(						self    :str		, lowerCamelCase_    :List[Any]		, lowerCamelCase_    :str		, lowerCamelCase_    :Union[str, Any]		, lowerCamelCase_    :str=None					)							->			int:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[int]				= obj
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= target
			SCREAMING_SNAKE_CASE       :     List[str]				= new
			SCREAMING_SNAKE_CASE       :     str				= target.split('''.'''					)[0]
			SCREAMING_SNAKE_CASE       :     Optional[int]				= {}
			SCREAMING_SNAKE_CASE       :     Any				= attrs or []





	def __enter__(						self    :str					)							->			Tuple:







			'''simple docstring'''




			*SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     Any				= self.target.split('''.'''					)

			# Patch modules:
			# it's used to patch attributes of submodules like "os.path.join";
			# in this case we need to patch "os" and "os.path"

			for i in range(len(lowerCamelCase_					)					):
					try:
							SCREAMING_SNAKE_CASE       :     Any				= import_module('''.'''.join(submodules[: i + 1]					)					)
					except ModuleNotFoundError:
							continue
					# We iterate over all the globals in self.obj in case we find "os" or "os.path"
					for attr in self.obj.__dir__():
							SCREAMING_SNAKE_CASE       :     List[str]				= getattr(self.obj		, lowerCamelCase_					)
							# We don't check for the name of the global, but rather if its value *is* "os" or "os.path".
							# This allows to patch renamed modules like "from os import path as ospath".
							if obj_attr is submodule or (
							    (isinstance(lowerCamelCase_		, _PatchedModuleObj					) and obj_attr._original_module is submodule)
							):
									SCREAMING_SNAKE_CASE       :     List[Any]				= obj_attr
									# patch at top level
									setattr(self.obj		, lowerCamelCase_		, _PatchedModuleObj(lowerCamelCase_		, attrs=self.attrs					)					)
									SCREAMING_SNAKE_CASE       :     Union[str, Any]				= getattr(self.obj		, lowerCamelCase_					)
									# construct lower levels patches
									for key in submodules[i + 1 :]:
											setattr(lowerCamelCase_		, lowerCamelCase_		, _PatchedModuleObj(getattr(lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_					)		, attrs=self.attrs					)					)
											SCREAMING_SNAKE_CASE       :     Union[str, Any]				= getattr(lowerCamelCase_		, lowerCamelCase_					)
									# finally set the target attribute
									setattr(lowerCamelCase_		, lowerCamelCase_		, self.new					)

        # Patch attribute itself:
        # it's used for builtins like "open",
        # and also to patch "os.path.join" we may also need to patch "join"
        # itself if it was imported as "from os.path import join".

			if submodules:  # if it's an attribute of a submodule like "os.path.join"
					try:
							SCREAMING_SNAKE_CASE       :     Any				= getattr(import_module('''.'''.join(lowerCamelCase_					)					)		, lowerCamelCase_					)
					except (AttributeError, ModuleNotFoundError):
							return
					# We iterate over all the globals in self.obj in case we find "os.path.join"
					for attr in self.obj.__dir__():
							# We don't check for the name of the global, but rather if its value *is* "os.path.join".
							# This allows to patch renamed attributes like "from os.path import join as pjoin".
							if getattr(self.obj		, lowerCamelCase_					) is attr_value:
									SCREAMING_SNAKE_CASE       :     Any				= getattr(self.obj		, lowerCamelCase_					)
									setattr(self.obj		, lowerCamelCase_		, self.new					)
			elif target_attr in globals()["__builtins__"]:  # if it'a s builtin like "open"
					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= globals()['''__builtins__'''][target_attr]
					setattr(self.obj		, lowerCamelCase_		, self.new					)
			else:
					raise RuntimeError(f"Tried to patch attribute {target_attr} instead of a submodule."					)





	def __exit__(						self    :Union[str, Any]		, *lowerCamelCase_    :Optional[Any]					)							->			List[Any]:







			'''simple docstring'''




			for attr in list(self.original					):
					setattr(self.obj		, lowerCamelCase_		, self.original.pop(lowerCamelCase_					)					)





	def   __lowerCAmelCase (						self    :List[Any]					)							->			Dict:







			'''simple docstring'''




			self.__enter__()
			self._active_patches.append(self					)







	def   __lowerCAmelCase (						self    :Optional[Any]					)							->			Optional[int]:







			'''simple docstring'''




			try:
					self._active_patches.remove(self					)
			except ValueError:
					# If the patch hasn't been started this will fail
					return None

			return self.__exit__()







"""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


lowerCamelCase__ :       List[Any]						     =			logging.get_logger(__name__)




class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	def __init__(						self    :List[str]		, lowerCamelCase_    :Union[List[ControlNetModel], Tuple[ControlNetModel]]					)							->			Union[str, Any]:







			'''simple docstring'''




			super().__init__()
			SCREAMING_SNAKE_CASE       :     Optional[Any]				= nn.ModuleList(lowerCamelCase_					)





	def   __lowerCAmelCase (						self    :Any		, lowerCamelCase_    :torch.FloatTensor		, lowerCamelCase_    :Union[torch.Tensor, float, int]		, lowerCamelCase_    :torch.Tensor		, lowerCamelCase_    :List[torch.tensor]		, lowerCamelCase_    :List[float]		, lowerCamelCase_    :Optional[torch.Tensor] = None		, lowerCamelCase_    :Optional[torch.Tensor] = None		, lowerCamelCase_    :Optional[torch.Tensor] = None		, lowerCamelCase_    :Optional[Dict[str, Any]] = None		, lowerCamelCase_    :bool = False		, lowerCamelCase_    :bool = True		, )							->			Union[ControlNetOutput, Tuple]:







			'''simple docstring'''




			for i, (image, scale, controlnet) in enumerate(zip(lowerCamelCase_		, lowerCamelCase_		, self.nets					)					):
					SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     List[Any]				= controlnet(
					    lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, lowerCamelCase_		, )

					# merge samples
					if i == 0:
							SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     int				= down_samples, mid_sample
					else:
							SCREAMING_SNAKE_CASE       :     Optional[int]				= [
							    samples_prev + samples_curr
							    for samples_prev, samples_curr in zip(lowerCamelCase_		, lowerCamelCase_					)
							]
							mid_block_res_sample += mid_sample

			return down_block_res_samples, mid_block_res_sample





	def   __lowerCAmelCase (						self    :Any		, lowerCamelCase_    :Union[str, os.PathLike]		, lowerCamelCase_    :bool = True		, lowerCamelCase_    :Callable = None		, lowerCamelCase_    :bool = False		, lowerCamelCase_    :Optional[str] = None		, )							->			Optional[int]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= 0
			SCREAMING_SNAKE_CASE       :     Any				= save_directory
			for controlnet in self.nets:
					controlnet.save_pretrained(
					    lowerCamelCase_		, is_main_process=lowerCamelCase_		, save_function=lowerCamelCase_		, safe_serialization=lowerCamelCase_		, variant=lowerCamelCase_		, )

					idx += 1
					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= model_path_to_save + f"_{idx}"







	@classmethod
	def   __lowerCAmelCase (						cls    :Dict		, lowerCamelCase_    :Optional[Union[str, os.PathLike]]		, **lowerCamelCase_    :Tuple					)							->			List[Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     str				= 0
			SCREAMING_SNAKE_CASE       :     Optional[int]				= []

			# 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`, ...
			SCREAMING_SNAKE_CASE       :     Dict				= pretrained_model_path
			while os.path.isdir(lowerCamelCase_					):
					SCREAMING_SNAKE_CASE       :     Tuple				= ControlNetModel.from_pretrained(lowerCamelCase_		, **lowerCamelCase_					)
					controlnets.append(lowerCamelCase_					)

					idx += 1
					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= pretrained_model_path + f"_{idx}"

			logger.info(f"{len(lowerCamelCase_					)} controlnets loaded from {pretrained_model_path}."					)

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

			return cls(lowerCamelCase_					)





