



"""simple docstring"""






import tempfile

import torch

from diffusers import PNDMScheduler

from .test_schedulers import SchedulerCommonTest






class 	__snake_case	(     _lowercase):
  snake_case__    :       List[Any]        =						(PNDMScheduler,)
  snake_case__    :       Optional[Any]        =						(("num_inference_steps", 5_0),)




  def 						SCREAMING_SNAKE_CASE    (			self					:						str						, **__lowerCAmelCase					:						int		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			{
             '''num_train_timesteps''': 1_0_0_0,
             '''beta_start''': 0.00_01,
             '''beta_end''': 0.02,
             '''beta_schedule''': '''linear''',
         }

         config.update(**__lowerCAmelCase		)
         return config




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]						, __lowerCAmelCase					:						List[Any]=0						, **__lowerCAmelCase					:						Dict		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			dict(self.forward_default_kwargs		)
         _lowerCamelCase						:    Union[str, Any]         =			kwargs.pop('''num_inference_steps'''						, __lowerCAmelCase		)
         _lowerCamelCase						:    Dict         =			self.dummy_sample
         _lowerCamelCase						:    Optional[int]         =			0.1 * sample
         _lowerCamelCase						:    Any         =			[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

         for scheduler_class in self.scheduler_classes:
                _lowerCamelCase						:    Union[str, Any]         =			self.get_scheduler_config(**__lowerCAmelCase		)
                _lowerCamelCase						:    List[Any]         =			scheduler_class(**__lowerCAmelCase		)
                scheduler.set_timesteps(__lowerCAmelCase		)
                # copy over dummy past residuals
                _lowerCamelCase						:    Optional[Any]         =			dummy_past_residuals[:]

                with tempfile.TemporaryDirectory() as tmpdirname:
                       scheduler.save_config(__lowerCAmelCase		)
                       _lowerCamelCase						:    Any         =			scheduler_class.from_pretrained(__lowerCAmelCase		)
                       new_scheduler.set_timesteps(__lowerCAmelCase		)
                       # copy over dummy past residuals
                       _lowerCamelCase						:    List[Any]         =			dummy_past_residuals[:]

                _lowerCamelCase						:    Tuple         =			scheduler.step_prk(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample
                _lowerCamelCase						:    str         =			new_scheduler.step_prk(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample

                assert torch.sum(torch.abs(output - new_output		)		) < 1E-5, "Scheduler outputs are not identical"

                _lowerCamelCase						:    int         =			scheduler.step_plms(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample
                _lowerCamelCase						:    List[str]         =			new_scheduler.step_plms(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample

                assert torch.sum(torch.abs(output - new_output		)		) < 1E-5, "Scheduler outputs are not identical"




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[int]		):






         """simple docstring"""






         pass




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]						, __lowerCAmelCase					:						List[Any]=0						, **__lowerCAmelCase					:						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			dict(self.forward_default_kwargs		)
         _lowerCamelCase						:    List[str]         =			kwargs.pop('''num_inference_steps'''						, __lowerCAmelCase		)
         _lowerCamelCase						:    Any         =			self.dummy_sample
         _lowerCamelCase						:    Any         =			0.1 * sample
         _lowerCamelCase						:    Tuple         =			[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

         for scheduler_class in self.scheduler_classes:
                _lowerCamelCase						:    int         =			self.get_scheduler_config()
                _lowerCamelCase						:    List[Any]         =			scheduler_class(**__lowerCAmelCase		)
                scheduler.set_timesteps(__lowerCAmelCase		)

                # copy over dummy past residuals (must be after setting timesteps)
                _lowerCamelCase						:    str         =			dummy_past_residuals[:]

                with tempfile.TemporaryDirectory() as tmpdirname:
                       scheduler.save_config(__lowerCAmelCase		)
                       _lowerCamelCase						:    Optional[int]         =			scheduler_class.from_pretrained(__lowerCAmelCase		)
                       # copy over dummy past residuals
                       new_scheduler.set_timesteps(__lowerCAmelCase		)

                       # copy over dummy past residual (must be after setting timesteps)
                       _lowerCamelCase						:    List[Any]         =			dummy_past_residuals[:]

                _lowerCamelCase						:    Any         =			scheduler.step_prk(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample
                _lowerCamelCase						:    Optional[int]         =			new_scheduler.step_prk(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample

                assert torch.sum(torch.abs(output - new_output		)		) < 1E-5, "Scheduler outputs are not identical"

                _lowerCamelCase						:    Optional[int]         =			scheduler.step_plms(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample
                _lowerCamelCase						:    List[Any]         =			new_scheduler.step_plms(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample

                assert torch.sum(torch.abs(output - new_output		)		) < 1E-5, "Scheduler outputs are not identical"




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]						, **__lowerCAmelCase					:						Optional[int]		):






         """simple docstring"""






         _lowerCamelCase						:    Tuple         =			self.scheduler_classes[0]
         _lowerCamelCase						:    Any         =			self.get_scheduler_config(**__lowerCAmelCase		)
         _lowerCamelCase						:    Dict         =			scheduler_class(**__lowerCAmelCase		)

         _lowerCamelCase						:    Optional[Any]         =			1_0
         _lowerCamelCase						:    Optional[Any]         =			self.dummy_model()
         _lowerCamelCase						:    Optional[Any]         =			self.dummy_sample_deter
         scheduler.set_timesteps(__lowerCAmelCase		)

         for i, t in enumerate(scheduler.prk_timesteps		):
                _lowerCamelCase						:    Dict         =			model(__lowerCAmelCase						, __lowerCAmelCase		)
                _lowerCamelCase						:    List[str]         =			scheduler.step_prk(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase		).prev_sample

         for i, t in enumerate(scheduler.plms_timesteps		):
                _lowerCamelCase						:    Optional[int]         =			model(__lowerCAmelCase						, __lowerCAmelCase		)
                _lowerCamelCase						:    Dict         =			scheduler.step_plms(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase		).prev_sample

         return sample




  def 						SCREAMING_SNAKE_CASE    (			self					:						int		):






         """simple docstring"""






         _lowerCamelCase						:    str         =			dict(self.forward_default_kwargs		)

         _lowerCamelCase						:    List[str]         =			kwargs.pop('''num_inference_steps'''						, __lowerCAmelCase		)

         for scheduler_class in self.scheduler_classes:
                _lowerCamelCase						:    Union[str, Any]         =			self.get_scheduler_config()
                _lowerCamelCase						:    Union[str, Any]         =			scheduler_class(**__lowerCAmelCase		)

                _lowerCamelCase						:    Any         =			self.dummy_sample
                _lowerCamelCase						:    Tuple         =			0.1 * sample

                if num_inference_steps is not None and hasattr(__lowerCAmelCase						, '''set_timesteps'''		):
                       scheduler.set_timesteps(__lowerCAmelCase		)
                elif num_inference_steps is not None and not hasattr(__lowerCAmelCase						, '''set_timesteps'''		):
                       _lowerCamelCase						:    List[Any]         =			num_inference_steps

                # copy over dummy past residuals (must be done after set_timesteps)
                _lowerCamelCase						:    Optional[int]         =			[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
                _lowerCamelCase						:    Dict         =			dummy_past_residuals[:]

                _lowerCamelCase						:    Any         =			scheduler.step_prk(__lowerCAmelCase						, 0						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample
                _lowerCamelCase						:    str         =			scheduler.step_prk(__lowerCAmelCase						, 1						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample

                self.assertEqual(output_a.shape						, sample.shape		)
                self.assertEqual(output_a.shape						, output_a.shape		)

                _lowerCamelCase						:    Union[str, Any]         =			scheduler.step_plms(__lowerCAmelCase						, 0						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample
                _lowerCamelCase						:    Optional[int]         =			scheduler.step_plms(__lowerCAmelCase						, 1						, __lowerCAmelCase						, **__lowerCAmelCase		).prev_sample

                self.assertEqual(output_a.shape						, sample.shape		)
                self.assertEqual(output_a.shape						, output_a.shape		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         for timesteps in [1_0_0, 1_0_0_0]:
                self.check_over_configs(num_train_timesteps=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]		):






         """simple docstring"""






         for steps_offset in [0, 1]:
                self.check_over_configs(steps_offset=__lowerCAmelCase		)

         _lowerCamelCase						:    Dict         =			self.scheduler_classes[0]
         _lowerCamelCase						:    List[str]         =			self.get_scheduler_config(steps_offset=1		)
         _lowerCamelCase						:    int         =			scheduler_class(**__lowerCAmelCase		)
         scheduler.set_timesteps(1_0		)
         assert torch.equal(
             scheduler.timesteps						, torch.LongTensor(
                 [9_0_1, 8_5_1, 8_5_1, 8_0_1, 8_0_1, 7_5_1, 7_5_1, 7_0_1, 7_0_1, 6_5_1, 6_5_1, 6_0_1, 6_0_1, 5_0_1, 4_0_1, 3_0_1, 2_0_1, 1_0_1, 1]		)						, )




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]		):






         """simple docstring"""






         for beta_start, beta_end in zip([0.00_01, 0.0_01]						, [0.0_02, 0.02]		):
                self.check_over_configs(beta_start=__lowerCAmelCase						, beta_end=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[int]		):






         """simple docstring"""






         for schedule in ["linear", "squaredcos_cap_v2"]:
                self.check_over_configs(beta_schedule=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]		):






         """simple docstring"""






         for prediction_type in ["epsilon", "v_prediction"]:
                self.check_over_configs(prediction_type=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]		):






         """simple docstring"""






         for t in [1, 5, 1_0]:
                self.check_over_forward(time_step=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]		):






         """simple docstring"""






         for t, num_inference_steps in zip([1, 5, 1_0]						, [1_0, 5_0, 1_0_0]		):
                self.check_over_forward(num_inference_steps=__lowerCAmelCase		)




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]		):






         """simple docstring"""






         _lowerCamelCase						:    List[str]         =			2_7

         for scheduler_class in self.scheduler_classes:
                _lowerCamelCase						:    Dict         =			self.dummy_sample
                _lowerCamelCase						:    Dict         =			0.1 * sample

                _lowerCamelCase						:    Any         =			self.get_scheduler_config()
                _lowerCamelCase						:    int         =			scheduler_class(**__lowerCAmelCase		)

                scheduler.set_timesteps(__lowerCAmelCase		)

                # before power of 3 fix, would error on first step, so we only need to do two
                for i, t in enumerate(scheduler.prk_timesteps[:2]		):
                       _lowerCamelCase						:    Dict         =			scheduler.step_prk(__lowerCAmelCase						, __lowerCAmelCase						, __lowerCAmelCase		).prev_sample




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]		):






         """simple docstring"""






         with self.assertRaises(__lowerCAmelCase		):
                _lowerCamelCase						:    Optional[Any]         =			self.scheduler_classes[0]
                _lowerCamelCase						:    Tuple         =			self.get_scheduler_config()
                _lowerCamelCase						:    Optional[int]         =			scheduler_class(**__lowerCAmelCase		)

                scheduler.step_plms(self.dummy_sample						, 1						, self.dummy_sample		).prev_sample




  def 						SCREAMING_SNAKE_CASE    (			self					:						Dict		):






         """simple docstring"""






         _lowerCamelCase						:    List[Any]         =			self.full_loop()
         _lowerCamelCase						:    List[str]         =			torch.sum(torch.abs(__lowerCAmelCase		)		)
         _lowerCamelCase						:    Optional[Any]         =			torch.mean(torch.abs(__lowerCAmelCase		)		)

         assert abs(result_sum.item() - 1_98.13_18		) < 1E-2
         assert abs(result_mean.item() - 0.25_80		) < 1E-3




  def 						SCREAMING_SNAKE_CASE    (			self					:						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:    List[Any]         =			self.full_loop(prediction_type='''v_prediction'''		)
         _lowerCamelCase						:    Optional[Any]         =			torch.sum(torch.abs(__lowerCAmelCase		)		)
         _lowerCamelCase						:    int         =			torch.mean(torch.abs(__lowerCAmelCase		)		)

         assert abs(result_sum.item() - 67.39_86		) < 1E-2
         assert abs(result_mean.item() - 0.08_78		) < 1E-3




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any		):






         """simple docstring"""






         _lowerCamelCase						:    Tuple         =			self.full_loop(set_alpha_to_one=__lowerCAmelCase						, beta_start=0.01		)
         _lowerCamelCase						:    List[str]         =			torch.sum(torch.abs(__lowerCAmelCase		)		)
         _lowerCamelCase						:    Optional[int]         =			torch.mean(torch.abs(__lowerCAmelCase		)		)

         assert abs(result_sum.item() - 2_30.03_99		) < 1E-2
         assert abs(result_mean.item() - 0.29_95		) < 1E-3




  def 						SCREAMING_SNAKE_CASE    (			self					:						Dict		):






         """simple docstring"""






         _lowerCamelCase						:    Any         =			self.full_loop(set_alpha_to_one=__lowerCAmelCase						, beta_start=0.01		)
         _lowerCamelCase						:    Dict         =			torch.sum(torch.abs(__lowerCAmelCase		)		)
         _lowerCamelCase						:    Dict         =			torch.mean(torch.abs(__lowerCAmelCase		)		)

         assert abs(result_sum.item() - 1_86.94_82		) < 1E-2
         assert abs(result_mean.item() - 0.24_34		) < 1E-3







"""simple docstring"""






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

lowerCAmelCase__      =    logging.get_logger(__name__)






@add_end_docstrings(_lowercase)
class 	__snake_case	(     _lowercase):




  def __init__(			self					:						Any						, **__lowerCAmelCase					:						Union[str, Any]		):






         """simple docstring"""






         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					:						Dict						, __lowerCAmelCase					:						Union[str, "Image.Image", List[Dict[str, Any]]]						, __lowerCAmelCase					:						Union[str, List[str]] = None						, **__lowerCAmelCase					:						int						, ):






         """simple docstring"""






         if "text_queries" in kwargs:
                _lowerCamelCase						:    List[Any]         =			kwargs.pop('''text_queries'''		)

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




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]						, **__lowerCAmelCase					:						int		):






         """simple docstring"""






         _lowerCamelCase						:    int         =			{}
         if "threshold" in kwargs:
                _lowerCamelCase						:    Optional[Any]         =			kwargs['''threshold''']
         if "top_k" in kwargs:
                _lowerCamelCase						:    int         =			kwargs['''top_k''']
         return {}, {}, postprocess_params




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]						, __lowerCAmelCase					:						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:    int         =			load_image(inputs['''image''']		)
         _lowerCamelCase						:    Optional[Any]         =			inputs['''candidate_labels''']
         if isinstance(__lowerCAmelCase						, __lowerCAmelCase		):
                _lowerCamelCase						:    int         =			candidate_labels.split(''','''		)

         _lowerCamelCase						:    Tuple         =			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						:    Optional[Any]         =			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 						SCREAMING_SNAKE_CASE    (			self					:						Any						, __lowerCAmelCase					:						List[Any]		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			model_inputs.pop('''target_size'''		)
         _lowerCamelCase						:    List[Any]         =			model_inputs.pop('''candidate_label'''		)
         _lowerCamelCase						:    Dict         =			model_inputs.pop('''is_last'''		)

         _lowerCamelCase						:    str         =			self.model(**__lowerCAmelCase		)

         _lowerCamelCase						:    Optional[Any]         =			{'''target_size''': target_size, '''candidate_label''': candidate_label, '''is_last''': is_last, **outputs}
         return model_outputs




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]						, __lowerCAmelCase					:						Any						, __lowerCAmelCase					:						Union[str, Any]=0.1						, __lowerCAmelCase					:						Optional[Any]=None		):






         """simple docstring"""






         _lowerCamelCase						:    str         =			[]
         for model_output in model_outputs:
                _lowerCamelCase						:    Any         =			model_output['''candidate_label''']
                _lowerCamelCase						:    Union[str, Any]         =			BaseModelOutput(__lowerCAmelCase		)
                _lowerCamelCase						:    Tuple         =			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						:    Tuple         =			outputs['''scores'''][index].item()
                       _lowerCamelCase						:    Optional[Any]         =			self._get_bounding_box(outputs['''boxes'''][index][0]		)

                       _lowerCamelCase						:    Optional[Any]         =			{'''score''': score, '''label''': label, '''box''': box}
                       results.append(__lowerCAmelCase		)

         _lowerCamelCase						:    int         =			sorted(__lowerCAmelCase						, key=lambda __lowerCAmelCase		: x["score"]						, reverse=__lowerCAmelCase		)
         if top_k:
                _lowerCamelCase						:    Dict         =			results[:top_k]

         return results




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any						, __lowerCAmelCase					:						"torch.Tensor"		):






         """simple docstring"""






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


