






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


__UpperCAmelCase							:   Union[str, Any]							=  logging.get_logger(__name__)


class 					_snake_case						(			_A ):






	def __init__(		self					,UpperCamelCase       )						->    List[Any]:
					super().__init__()
					snake_case__      :Union[str, Any]         =   nn.ModuleList(UpperCamelCase       )






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

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

					return down_block_res_samples, mid_block_res_sample






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = True					,UpperCamelCase = None					,UpperCamelCase = False					,UpperCamelCase = None					,)						->    Union[str, Any]:
					snake_case__      :List[str]         =   0
					snake_case__      :Optional[int]         =   save_directory
					for controlnet in self.nets:
									controlnet.save_pretrained(
									    UpperCamelCase					,is_main_process=UpperCamelCase					,save_function=UpperCamelCase					,safe_serialization=UpperCamelCase					,variant=UpperCamelCase					,)

									idx += 1
									snake_case__      :List[Any]         =   model_path_to_save + f'_{idx}'






	@classmethod
	def    lowerCAmelCase_							(		cls					,UpperCamelCase					,**UpperCamelCase       )						->    Dict:
					snake_case__      :List[Any]         =   0
					snake_case__      :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`, ...
					snake_case__      :Any         =   pretrained_model_path
					while os.path.isdir(UpperCamelCase       ):
									snake_case__      :List[str]         =   ControlNetModel.from_pretrained(UpperCamelCase					,**UpperCamelCase       )
									controlnets.append(UpperCamelCase       )

									idx += 1
									snake_case__      :Optional[Any]         =   pretrained_model_path + f'_{idx}'

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

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

					return cls(UpperCamelCase       )







from ...utils import (
    OptionalDependencyNotAvailable,
    is_flax_available,
    is_torch_available,
    is_transformers_available,
)


try:
	if not (is_transformers_available() and is_torch_available()):
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	from ...utils.dummy_torch_and_transformers_objects import *  # noqa F403
else:
	from .multicontrolnet import MultiControlNetModel
	from .pipeline_controlnet import StableDiffusionControlNetPipeline
	from .pipeline_controlnet_imgaimg import StableDiffusionControlNetImgaImgPipeline
	from .pipeline_controlnet_inpaint import StableDiffusionControlNetInpaintPipeline


if is_transformers_available() and is_flax_available():
	from .pipeline_flax_controlnet import FlaxStableDiffusionControlNetPipeline