


from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
						from PIL import Image

						from ..image_utils import load_image

if is_torch_available():
						import torch

						from transformers.modeling_outputs import BaseModelOutput

						from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

_UpperCAmelCase			:  str     =logging.get_logger(__name__)







@add_end_docstrings(lowercase_							)
class 				snake_case__(							lowercase_							):







						'''simple docstring'''




						def __init__(				self   ,						**__lowercase							)					->    Optional[int]:
										super().__init__(**lowerCamelCase_							)

										if self.framework == "tf":
														raise ValueError(f"""The {self.__class__} is only available in PyTorch."""							)

										requires_backends(self   ,						'''vision'''							)
										self.check_model_type(lowerCamelCase_							)




						def __call__(				self   ,						__lowercase   ,						__lowercase = None   ,						**__lowercase   ,						)					->    Optional[int]:



										if "text_queries" in kwargs:
														lowerCAmelCase_ :      List[str]    				=				kwargs.pop('''text_queries'''							)

										if isinstance(lowerCamelCase_   ,						(str, Image.Image)							):
														lowerCAmelCase_ :      int    				=				{"""image""": image, """candidate_labels""": candidate_labels}
										else:
														lowerCAmelCase_ :      Optional[Any]    				=				image
										lowerCAmelCase_ :      List[str]    				=				super().__call__(lowerCamelCase_   ,						**lowerCamelCase_							)
										return results




						def 						lowercase_ (				self   ,						**__lowercase							)					->    Tuple:
										lowerCAmelCase_ :      List[str]    				=				{}
										if "threshold" in kwargs:
														lowerCAmelCase_ :      Any    				=				kwargs["""threshold"""]
										if "top_k" in kwargs:
														lowerCAmelCase_ :      Tuple    				=				kwargs["""top_k"""]
										return {}, {}, postprocess_params




						def 						lowercase_ (				self   ,						__lowercase							)					->    List[Any]:
										lowerCAmelCase_ :      int    				=				load_image(inputs['''image''']							)
										lowerCAmelCase_ :      Optional[Any]    				=				inputs["""candidate_labels"""]
										if isinstance(lowerCamelCase_   ,						lowerCamelCase_							):
														lowerCAmelCase_ :      Dict    				=				candidate_labels.split(''','''							)

										lowerCAmelCase_ :      Union[str, Any]    				=				torch.tensor([[image.height, image.width]]   ,						dtype=torch.intaa							)
										for i, candidate_label in enumerate(lowerCamelCase_							):
														lowerCAmelCase_ :      Any    				=				self.tokenizer(lowerCamelCase_   ,						return_tensors=self.framework							)
														lowerCAmelCase_ :      List[str]    				=				self.image_processor(lowerCamelCase_   ,						return_tensors=self.framework							)
														yield {
														    "is_last": i == len(lowerCamelCase_							) - 1,
														    "target_size": target_size,
														    "candidate_label": candidate_label,
														    **text_inputs,
														    **image_features,
														}




						def 						lowercase_ (				self   ,						__lowercase							)					->    List[Any]:
										lowerCAmelCase_ :      Union[str, Any]    				=				model_inputs.pop('''target_size'''							)
										lowerCAmelCase_ :      Any    				=				model_inputs.pop('''candidate_label'''							)
										lowerCAmelCase_ :      str    				=				model_inputs.pop('''is_last'''							)

										lowerCAmelCase_ :      List[Any]    				=				self.model(**lowerCamelCase_							)

										lowerCAmelCase_ :      List[str]    				=				{"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
										return model_outputs




						def 						lowercase_ (				self   ,						__lowercase   ,						__lowercase=0.1   ,						__lowercase=None							)					->    Optional[int]:
										lowerCAmelCase_ :      Any    				=				[]
										for model_output in model_outputs:
														lowerCAmelCase_ :      List[str]    				=				model_output["""candidate_label"""]
														lowerCAmelCase_ :      Union[str, Any]    				=				BaseModelOutput(lowerCamelCase_							)
														lowerCAmelCase_ :      Any    				=				self.image_processor.post_process_object_detection(
														    outputs=lowerCamelCase_   ,						threshold=lowerCamelCase_   ,						target_sizes=model_output['''target_size''']							)[0]

														for index in outputs["scores"].nonzero():
																		lowerCAmelCase_ :      str    				=				outputs["""scores"""][index].item()
																		lowerCAmelCase_ :      List[Any]    				=				self._get_bounding_box(outputs['''boxes'''][index][0]							)

																		lowerCAmelCase_ :      Any    				=				{"""score""": score, """label""": label, """box""": box}
																		results.append(lowerCamelCase_							)

										lowerCAmelCase_ :      List[str]    				=				sorted(lowerCamelCase_   ,						key=lambda __lowercase							: x["score"]   ,						reverse=lowerCamelCase_							)
										if top_k:
														lowerCAmelCase_ :      str    				=				results[:top_k]

										return results




						def 						lowercase_ (				self   ,						__lowercase							)					->    Optional[Any]:



										if self.framework != "pt":
														raise ValueError('''The ZeroShotObjectDetectionPipeline is only available in PyTorch.'''							)
										lowerCAmelCase_ :      Union[str, Any]    				=				box.int().tolist()
										lowerCAmelCase_ :      Dict    				=				{
										    """xmin""": xmin,
										    """ymin""": ymin,
										    """xmax""": xmax,
										    """ymax""": ymax,
										}
										return bbox



import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


_UpperCAmelCase			:  Tuple     =logging.get_logger(__name__)







class 				snake_case__(							UpperCAmelCase__							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__      :  List[str]					=							"""linear"""
						SCREAMING_SNAKE_CASE__      :  Union[str, Any]					=							"""cosine"""
						SCREAMING_SNAKE_CASE__      :  Dict					=							"""cosine_with_restarts"""
						SCREAMING_SNAKE_CASE__      :  List[str]					=							"""polynomial"""
						SCREAMING_SNAKE_CASE__      :  Dict					=							"""constant"""
						SCREAMING_SNAKE_CASE__      :  List[str]					=							"""constant_with_warmup"""
						SCREAMING_SNAKE_CASE__      :  str					=							"""piecewise_constant"""






def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ = -1 )-> Tuple:
				return LambdaLR(lowerCAmelCase_ ,   lambda lowerCAmelCase_ : 1 ,   last_epoch=lowerCAmelCase_ )



def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ = -1 )-> List[Any]:

				def lr_lambda(lowerCAmelCase_ ):
								if current_step < num_warmup_steps:
												return float(lowerCAmelCase_ ) / float(max(1.0 ,   lowerCAmelCase_ ) )
								return 1.0

				return LambdaLR(lowerCAmelCase_ ,   lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ )



def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ = -1 )-> int:
				lowerCAmelCase_ :      Optional[int]    				=				{}
				lowerCAmelCase_ :      Union[str, Any]    				=				step_rules.split(''',''' )
				for rule_str in rule_list[:-1]:
								lowerCAmelCase_				,  lowerCAmelCase_ :      str    				=				rule_str.split(''':''' )
								lowerCAmelCase_ :      int    				=				int(lowerCAmelCase_ )
								lowerCAmelCase_ :      str    				=				float(lowerCAmelCase_ )
								lowerCAmelCase_ :      List[Any]    				=				value
				lowerCAmelCase_ :      int    				=				float(rule_list[-1] )

				def create_rules_function(lowerCAmelCase_ ,   lowerCAmelCase_ ):
								def rule_func(lowerCAmelCase_ ) -> float:
												lowerCAmelCase_ :      Tuple    				=				sorted(rules_dict.keys() )
												for i, sorted_step in enumerate(lowerCAmelCase_ ):
																if steps < sorted_step:
																				return rules_dict[sorted_steps[i]]
												return last_lr_multiple

								return rule_func

				lowerCAmelCase_ :      Tuple    				=				create_rules_function(lowerCAmelCase_ ,   lowerCAmelCase_ )

				return LambdaLR(lowerCAmelCase_ ,   lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ )



def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_=-1 )-> Optional[int]:

				def lr_lambda(lowerCAmelCase_ ):
								if current_step < num_warmup_steps:
												return float(lowerCAmelCase_ ) / float(max(1 ,   lowerCAmelCase_ ) )
								return max(
								    0.0 ,   float(num_training_steps - current_step ) / float(max(1 ,   num_training_steps - num_warmup_steps ) ) )

				return LambdaLR(lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ )



def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ = 0.5 ,   lowerCAmelCase_ = -1 )-> List[Any]:

				def lr_lambda(lowerCAmelCase_ ):
								if current_step < num_warmup_steps:
												return float(lowerCAmelCase_ ) / float(max(1 ,   lowerCAmelCase_ ) )
								lowerCAmelCase_ :      Tuple    				=				float(current_step - num_warmup_steps ) / float(max(1 ,   num_training_steps - num_warmup_steps ) )
								return max(0.0 ,   0.5 * (1.0 + math.cos(math.pi * float(lowerCAmelCase_ ) * 2.0 * progress )) )

				return LambdaLR(lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ )



def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ = 1 ,   lowerCAmelCase_ = -1 )-> Dict:

				def lr_lambda(lowerCAmelCase_ ):
								if current_step < num_warmup_steps:
												return float(lowerCAmelCase_ ) / float(max(1 ,   lowerCAmelCase_ ) )
								lowerCAmelCase_ :      List[Any]    				=				float(current_step - num_warmup_steps ) / float(max(1 ,   num_training_steps - num_warmup_steps ) )
								if progress >= 1.0:
												return 0.0
								return max(0.0 ,   0.5 * (1.0 + math.cos(math.pi * ((float(lowerCAmelCase_ ) * progress) % 1.0) )) )

				return LambdaLR(lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ )



def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_=1e-7 ,   lowerCAmelCase_=1.0 ,   lowerCAmelCase_=-1 )-> Any:
				lowerCAmelCase_ :      Dict    				=				optimizer.defaults['''lr''']
				if not (lr_init > lr_end):
								raise ValueError(f"""lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})""" )

				def lr_lambda(lowerCAmelCase_ ):
								if current_step < num_warmup_steps:
												return float(lowerCAmelCase_ ) / float(max(1 ,   lowerCAmelCase_ ) )
								elif current_step > num_training_steps:
												return lr_end / lr_init  # as LambdaLR multiplies by lr_init
								else:
												lowerCAmelCase_ :      List[Any]    				=				lr_init - lr_end
												lowerCAmelCase_ :      Optional[Any]    				=				num_training_steps - num_warmup_steps
												lowerCAmelCase_ :      Any    				=				1 - (current_step - num_warmup_steps) / decay_steps
												lowerCAmelCase_ :      List[Any]    				=				lr_range * pct_remaining**power + lr_end
												return decay / lr_init  # as LambdaLR multiplies by lr_init

				return LambdaLR(lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ )


_UpperCAmelCase			:  Union[str, Any]     ={
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}




def 			lowerCAmelCase (		lowerCAmelCase_ ,   lowerCAmelCase_ ,   lowerCAmelCase_ = None ,   lowerCAmelCase_ = None ,   lowerCAmelCase_ = None ,   lowerCAmelCase_ = 1 ,   lowerCAmelCase_ = 1.0 ,   lowerCAmelCase_ = -1 ,   )-> Optional[int]:
				lowerCAmelCase_ :      Union[str, Any]    				=				SchedulerType(lowerCAmelCase_ )
				lowerCAmelCase_ :      Optional[int]    				=				TYPE_TO_SCHEDULER_FUNCTION[name]
				if name == SchedulerType.CONSTANT:
								return schedule_func(lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ )

				if name == SchedulerType.PIECEWISE_CONSTANT:
								return schedule_func(lowerCAmelCase_ ,   step_rules=lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ )

				# All other schedulers require `num_warmup_steps`
				if num_warmup_steps is None:
								raise ValueError(f"""{name} requires `num_warmup_steps`, please provide that argument.""" )

				if name == SchedulerType.CONSTANT_WITH_WARMUP:
								return schedule_func(lowerCAmelCase_ ,   num_warmup_steps=lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ )

				# All other schedulers require `num_training_steps`
				if num_training_steps is None:
								raise ValueError(f"""{name} requires `num_training_steps`, please provide that argument.""" )

				if name == SchedulerType.COSINE_WITH_RESTARTS:
								return schedule_func(
								    lowerCAmelCase_ ,   num_warmup_steps=lowerCAmelCase_ ,   num_training_steps=lowerCAmelCase_ ,   num_cycles=lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ ,   )

				if name == SchedulerType.POLYNOMIAL:
								return schedule_func(
								    lowerCAmelCase_ ,   num_warmup_steps=lowerCAmelCase_ ,   num_training_steps=lowerCAmelCase_ ,   power=lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ ,   )

				return schedule_func(
				    lowerCAmelCase_ ,   num_warmup_steps=lowerCAmelCase_ ,   num_training_steps=lowerCAmelCase_ ,   last_epoch=lowerCAmelCase_ )