






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			:	Any								=logging.get_logger(__name__)






class 						UpperCAmelCase__ (							__snake_case							):







		def __init__(					self       ,A__		):
						super().__init__()
						_A	:				Union[str, Any]   =      nn.ModuleList(A__		)







		def  A__      (					self       ,A__       ,A__       ,A__       ,A__       ,A__       ,A__ = None       ,A__ = None       ,A__ = None       ,A__ = None       ,A__ = False       ,A__ = True       ,):
						for i, (image, scale, controlnet) in enumerate(zip(A__       ,A__       ,self.nets		)		):
										_A  ,				_A	:				Optional[int]   =      controlnet(
										    A__       ,A__       ,A__       ,A__       ,A__       ,A__       ,A__       ,A__       ,A__       ,A__       ,A__       ,)

										# merge samples
										if i == 0:
														_A  ,				_A	:				List[Any]   =      down_samples, mid_sample
										else:
														_A	:				str   =      [
														    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       ,A__       ,A__ = True       ,A__ = None       ,A__ = False       ,A__ = None       ,):
						_A	:				int   =      0
						_A	:				Any   =      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
										_A	:				List[str]   =      model_path_to_save + f"""_{idx}"""







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

										idx += 1
										_A	:				Dict   =      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__		)











from ...configuration_utils import PretrainedConfig
from ...utils import logging


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

_UpperCamelCase			:	Optional[Any]								={
    'google/pegasus-large': 'https://huggingface.co/google/pegasus-large/resolve/main/config.json',
    # See all PEGASUS models at https://huggingface.co/models?filter=pegasus
}






class 						UpperCAmelCase__ (							__snake_case							):
		__snake_case      :       Optional[int]     =   "pegasus"
		__snake_case      :       List[str]     =   ["past_key_values"]
		__snake_case      :       Union[str, Any]     =   {"num_attention_heads": "encoder_attention_heads", "hidden_size": "d_model"}







		def __init__(					self       ,A__=50265       ,A__=1024       ,A__=12       ,A__=4096       ,A__=16       ,A__=12       ,A__=4096       ,A__=16       ,A__=0.0       ,A__=0.0       ,A__=True       ,A__=True       ,A__="gelu"       ,A__=1024       ,A__=0.1       ,A__=0.0       ,A__=0.0       ,A__=0.02       ,A__=0       ,A__=False       ,A__=0       ,A__=1       ,A__=1       ,**A__       ,):
						_A	:				List[str]   =      vocab_size
						_A	:				List[Any]   =      max_position_embeddings
						_A	:				Union[str, Any]   =      d_model
						_A	:				Union[str, Any]   =      encoder_ffn_dim
						_A	:				Union[str, Any]   =      encoder_layers
						_A	:				int   =      encoder_attention_heads
						_A	:				List[Any]   =      decoder_ffn_dim
						_A	:				Optional[int]   =      decoder_layers
						_A	:				Tuple   =      decoder_attention_heads
						_A	:				List[str]   =      dropout
						_A	:				Tuple   =      attention_dropout
						_A	:				Union[str, Any]   =      activation_dropout
						_A	:				int   =      activation_function
						_A	:				Union[str, Any]   =      init_std
						_A	:				Any   =      encoder_layerdrop
						_A	:				int   =      decoder_layerdrop
						_A	:				Tuple   =      use_cache
						_A	:				Dict   =      encoder_layers
						_A	:				List[str]   =      scale_embedding  # scale factor will be sqrt(d_model) if True
						super().__init__(
						    pad_token_id=A__       ,eos_token_id=A__       ,is_encoder_decoder=A__       ,decoder_start_token_id=A__       ,forced_eos_token_id=A__       ,**A__       ,)







		@property
		def  A__      (					self		):
						return self.encoder_attention_heads







		@property
		def  A__      (					self		):
						return self.d_model



