



'''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__    :   Optional[Any]           =					logging.get_logger(__name__)



class  snake_case__						(    SCREAMING_SNAKE_CASE_				):
 def __init__(   self						:							Tuple      ,     __a						:							Union[List[ControlNetModel], Tuple[ControlNetModel]]  )      ->   Union[str, Any]:







       '''simple docstring'''
       super().__init__()
       __snake_case :   Dict													=    nn.ModuleList(__a  )
 def A_   (   self						:							Dict      ,     __a						:							torch.FloatTensor      ,     __a						:							Union[torch.Tensor, float, int]      ,     __a						:							torch.Tensor      ,     __a						:							List[torch.tensor]      ,     __a						:							List[float]      ,     __a						:							Optional[torch.Tensor] = None      ,     __a						:							Optional[torch.Tensor] = None      ,     __a						:							Optional[torch.Tensor] = None      ,     __a						:							Optional[Dict[str, Any]] = None      ,     __a						:							bool = False      ,     __a						:							bool = True      ,     )      ->   Union[ControlNetOutput, Tuple]:







       '''simple docstring'''
       for i, (image, scale, controlnet) in enumerate(zip(__a      ,     __a      ,     self.nets  )  ):
             __snake_case							,     __snake_case :   Optional[int]													=    controlnet(
                 __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     __a      ,     )

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

       return down_block_res_samples, mid_block_res_sample
 def A_   (   self						:							List[str]      ,     __a						:							Union[str, os.PathLike]      ,     __a						:							bool = True      ,     __a						:							Callable = None      ,     __a						:							bool = False      ,     __a						:							Optional[str] = None      ,     )      ->   str:







       '''simple docstring'''
       __snake_case :   str													=    0
       __snake_case :   int													=    save_directory
       for controlnet in self.nets:
             controlnet.save_pretrained(
                 __a      ,     is_main_process=__a      ,     save_function=__a      ,     safe_serialization=__a      ,     variant=__a      ,     )

             idx += 1
             __snake_case :   str													=    model_path_to_save + f'''_{idx}'''





 @classmethod
 def A_   (   cls						:							Optional[int]      ,     __a						:							Optional[Union[str, os.PathLike]]      ,     **__a						:							Union[str, Any]  )      ->   Any:







       '''simple docstring'''
       __snake_case :   List[str]													=    0
       __snake_case :   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`, ...
       __snake_case :   Optional[Any]													=    pretrained_model_path
       while os.path.isdir(__a  ):
             __snake_case :   Union[str, Any]													=    ControlNetModel.from_pretrained(__a      ,     **__a  )
             controlnets.append(__a  )

             idx += 1
             __snake_case :   str													=    pretrained_model_path + f'''_{idx}'''

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

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

       return cls(__a  )








'''simple docstring'''





from abc import ABC, abstractmethod
from argparse import ArgumentParser



class  snake_case__						(    SCREAMING_SNAKE_CASE_				):
 @staticmethod
 @abstractmethod
 def A_   (   __a						:							ArgumentParser  )      ->   List[str]:







       '''simple docstring'''
       raise NotImplementedError()





 @abstractmethod
 def A_   (   self						:							Optional[Any]  )      ->   Optional[Any]:







       '''simple docstring'''
       raise NotImplementedError()



