




'''simple docstring'''


import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available

class      A       (						lowerCAmelCase__       ):
							lowercase_							     =							["""image_processor""", """tokenizer"""]
							lowercase_							     =							"""OwlViTImageProcessor"""
							lowercase_							     =							("""CLIPTokenizer""", """CLIPTokenizerFast""")







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


														"""simple docstring"""


														_a      	=							None
														if "feature_extractor" in kwargs:
																					warnings.warn(
																					    '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
																					    ''' instead.'''				,     __a				,     )
																					_a      	=							kwargs.pop('''feature_extractor'''   )

														_a      	=							image_processor if image_processor is not None else feature_extractor
														if image_processor is None:
																					raise ValueError('''You need to specify an `image_processor`.'''   )
														if tokenizer is None:
																					raise ValueError('''You need to specify a `tokenizer`.'''   )

														super().__init__(__a				,     __a   )







							def __call__(    self			:    Union[str, Any]				,     lowerCAmelCase_			:    Optional[int]=None				,     lowerCAmelCase_			:    Dict=None				,     lowerCAmelCase_			:    List[str]=None				,     lowerCAmelCase_			:    Dict="max_length"				,     lowerCAmelCase_			:    int="np"				,     **lowerCAmelCase_			:    Optional[Any]   )     ->				List[str]:


														"""simple docstring"""


														if text is None and query_images is None and images is None:
																					raise ValueError(
																					    '''You have to specify at least one text or query image or image. All three cannot be none.'''   )

														if text is not None:
																					if isinstance(__a				,     __a   ) or (isinstance(__a				,     __a   ) and not isinstance(text[0]				,     __a   )):
																												_a      	=							[self.tokenizer(__a				,     padding=__a				,     return_tensors=__a				,     **__a   )]

																					elif isinstance(__a				,     __a   ) and isinstance(text[0]				,     __a   ):
																												_a      	=							[]

																												# Maximum number of queries across batch
																												_a      	=							max([len(__a   ) for t in text]   )

																												# Pad all batch samples to max number of text queries
																												for t in text:
																																			if len(__a   ) != max_num_queries:
																																										_a      	=							t + [''' '''] * (max_num_queries - len(__a   ))

																																			_a      	=							self.tokenizer(__a				,     padding=__a				,     return_tensors=__a				,     **__a   )
																																			encodings.append(__a   )
																					else:
																												raise TypeError('''Input text should be a string, a list of strings or a nested list of strings'''   )

																					if return_tensors == "np":
																												_a      	=							np.concatenate([encoding['''input_ids'''] for encoding in encodings]				,     axis=0   )
																												_a      	=							np.concatenate([encoding['''attention_mask'''] for encoding in encodings]				,     axis=0   )

																					elif return_tensors == "jax" and is_flax_available():
																												import jax.numpy as jnp

																												_a      	=							jnp.concatenate([encoding['''input_ids'''] for encoding in encodings]				,     axis=0   )
																												_a      	=							jnp.concatenate([encoding['''attention_mask'''] for encoding in encodings]				,     axis=0   )

																					elif return_tensors == "pt" and is_torch_available():
																												import torch

																												_a      	=							torch.cat([encoding['''input_ids'''] for encoding in encodings]				,     dim=0   )
																												_a      	=							torch.cat([encoding['''attention_mask'''] for encoding in encodings]				,     dim=0   )

																					elif return_tensors == "tf" and is_tf_available():
																												import tensorflow as tf

																												_a      	=							tf.stack([encoding['''input_ids'''] for encoding in encodings]				,     axis=0   )
																												_a      	=							tf.stack([encoding['''attention_mask'''] for encoding in encodings]				,     axis=0   )

																					else:
																												raise ValueError('''Target return tensor type could not be returned'''   )

																					_a      	=							BatchEncoding()
																					_a      	=							input_ids
																					_a      	=							attention_mask

														if query_images is not None:
																					_a      	=							BatchEncoding()
																					_a      	=							self.image_processor(
																					    __a				,     return_tensors=__a				,     **__a   ).pixel_values
																					_a      	=							query_pixel_values

														if images is not None:
																					_a      	=							self.image_processor(__a				,     return_tensors=__a				,     **__a   )

														if text is not None and images is not None:
																					_a      	=							image_features.pixel_values
																					return encoding
														elif query_images is not None and images is not None:
																					_a      	=							image_features.pixel_values
																					return encoding
														elif text is not None or query_images is not None:
																					return encoding
														else:
																					return BatchEncoding(data=dict(**__a   )				,     tensor_type=__a   )







							def   __lowerCAmelCase     (    self			:    List[Any]				,     *lowerCAmelCase_			:    int				,     **lowerCAmelCase_			:    Optional[Any]   )     ->				Dict:


														"""simple docstring"""


														return self.image_processor.post_process(*__a				,     **__a   )







							def   __lowerCAmelCase     (    self			:    Optional[int]				,     *lowerCAmelCase_			:    str				,     **lowerCAmelCase_			:    List[Any]   )     ->				Any:


														"""simple docstring"""


														return self.image_processor.post_process_object_detection(*__a				,     **__a   )







							def   __lowerCAmelCase     (    self			:    Optional[int]				,     *lowerCAmelCase_			:    Optional[Any]				,     **lowerCAmelCase_			:    Dict   )     ->				Dict:


														"""simple docstring"""


														return self.image_processor.post_process_image_guided_detection(*__a				,     **__a   )







							def   __lowerCAmelCase     (    self			:    Optional[Any]				,     *lowerCAmelCase_			:    Dict				,     **lowerCAmelCase_			:    List[str]   )     ->				Any:


														"""simple docstring"""


														return self.tokenizer.batch_decode(*__a				,     **__a   )







							def   __lowerCAmelCase     (    self			:    Dict				,     *lowerCAmelCase_			:    Optional[int]				,     **lowerCAmelCase_			:    Tuple   )     ->				int:


														"""simple docstring"""


														return self.tokenizer.decode(*__a				,     **__a   )







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


														"""simple docstring"""


														warnings.warn(
														    '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''				,     __a				,     )
														return self.image_processor_class







							@property
							def   __lowerCAmelCase     (    self			:    List[str]   )     ->				Optional[int]:


														"""simple docstring"""


														warnings.warn(
														    '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''				,     __a				,     )
														return self.image_processor





"""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	:  List[Any]            =     logging.get_logger(__name__)
class     _UpperCamelCase    (  lowerCAmelCase__						):
							'''simple docstring'''







							def __init__(	self       ,  __a      ):
										super().__init__()
										__lowerCAmelCase							       =     nn.ModuleList(__a      )

							def      snake_case							(	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      )      ):
													__lowerCAmelCase       ,      __lowerCAmelCase							       =     controlnet(
													    __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  __a       ,  )

													# merge samples
													if i == 0:
																__lowerCAmelCase       ,      __lowerCAmelCase							       =     down_samples, mid_sample
													else:
																__lowerCAmelCase							       =     [
																    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      snake_case							(	self       ,  __a       ,  __a = True       ,  __a = None       ,  __a = False       ,  __a = None       ,  ):
										__lowerCAmelCase							       =     0
										__lowerCAmelCase							       =     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
													__lowerCAmelCase							       =     model_path_to_save + f"_{idx}"



							@classmethod
							def      snake_case							(	cls       ,  __a       ,  **__a      ):
										__lowerCAmelCase							       =     0
										__lowerCAmelCase							       =     []

										# 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							       =     pretrained_model_path
										while os.path.isdir(__a      ):
													__lowerCAmelCase							       =     ControlNetModel.from_pretrained(__a       ,  **__a      )
													controlnets.append(__a      )

													idx += 1
													__lowerCAmelCase							       =     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      )




