



'''simple docstring'''




import tempfile

import torch

from diffusers import PNDMScheduler

from .test_schedulers import SchedulerCommonTest

class 			UpperCamelCase_ (							__magic_name__			):
	lowercase          = (PNDMScheduler,)
	lowercase          = (('num_inference_steps', 50),)





	def  _lowercase(  self		,		**A    )			->			Dict:
								UpperCAmelCase     :		Union[str, Any]							     =		{
								    """num_train_timesteps""": 1000,
								    """beta_start""": 0.0_0_0_1,
								    """beta_end""": 0.0_2,
								    """beta_schedule""": """linear""",
								}

								config.update(**A    )
								return config





	def  _lowercase(  self		,		A=0		,		**A    )			->			int:
								UpperCAmelCase     :		List[str]							     =		dict(self.forward_default_kwargs    )
								UpperCAmelCase     :		str							     =		kwargs.pop("""num_inference_steps"""		,		A    )
								UpperCAmelCase     :		Tuple							     =		self.dummy_sample
								UpperCAmelCase     :		Optional[int]							     =		0.1 * sample
								UpperCAmelCase     :		str							     =		[residual + 0.2, residual + 0.1_5, residual + 0.1, residual + 0.0_5]

								for scheduler_class in self.scheduler_classes:
															UpperCAmelCase     :		Optional[int]							     =		self.get_scheduler_config(**A    )
															UpperCAmelCase     :		Tuple							     =		scheduler_class(**A    )
															scheduler.set_timesteps(A    )
															# copy over dummy past residuals
															UpperCAmelCase     :		Optional[int]							     =		dummy_past_residuals[:]

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

															UpperCAmelCase     :		List[Any]							     =		scheduler.step_prk(A		,		A		,		A		,		**A    ).prev_sample
															UpperCAmelCase     :		str							     =		new_scheduler.step_prk(A		,		A		,		A		,		**A    ).prev_sample

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

															UpperCAmelCase     :		Optional[Any]							     =		scheduler.step_plms(A		,		A		,		A		,		**A    ).prev_sample
															UpperCAmelCase     :		Optional[int]							     =		new_scheduler.step_plms(A		,		A		,		A		,		**A    ).prev_sample

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





	def  _lowercase(  self    )			->			Dict:
								pass





	def  _lowercase(  self		,		A=0		,		**A    )			->			Union[str, Any]:
								UpperCAmelCase     :		Tuple							     =		dict(self.forward_default_kwargs    )
								UpperCAmelCase     :		Optional[int]							     =		kwargs.pop("""num_inference_steps"""		,		A    )
								UpperCAmelCase     :		List[Any]							     =		self.dummy_sample
								UpperCAmelCase     :		Dict							     =		0.1 * sample
								UpperCAmelCase     :		Optional[Any]							     =		[residual + 0.2, residual + 0.1_5, residual + 0.1, residual + 0.0_5]

								for scheduler_class in self.scheduler_classes:
															UpperCAmelCase     :		Any							     =		self.get_scheduler_config()
															UpperCAmelCase     :		str							     =		scheduler_class(**A    )
															scheduler.set_timesteps(A    )

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

															with tempfile.TemporaryDirectory() as tmpdirname:
																						scheduler.save_config(A    )
																						UpperCAmelCase     :		Tuple							     =		scheduler_class.from_pretrained(A    )
																						# copy over dummy past residuals
																						new_scheduler.set_timesteps(A    )

																						# copy over dummy past residual (must be after setting timesteps)
																						UpperCAmelCase     :		Optional[int]							     =		dummy_past_residuals[:]

															UpperCAmelCase     :		Union[str, Any]							     =		scheduler.step_prk(A		,		A		,		A		,		**A    ).prev_sample
															UpperCAmelCase     :		Union[str, Any]							     =		new_scheduler.step_prk(A		,		A		,		A		,		**A    ).prev_sample

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

															UpperCAmelCase     :		Union[str, Any]							     =		scheduler.step_plms(A		,		A		,		A		,		**A    ).prev_sample
															UpperCAmelCase     :		Dict							     =		new_scheduler.step_plms(A		,		A		,		A		,		**A    ).prev_sample

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





	def  _lowercase(  self		,		**A    )			->			Dict:
								UpperCAmelCase     :		int							     =		self.scheduler_classes[0]
								UpperCAmelCase     :		str							     =		self.get_scheduler_config(**A    )
								UpperCAmelCase     :		int							     =		scheduler_class(**A    )

								UpperCAmelCase     :		List[Any]							     =		10
								UpperCAmelCase     :		List[str]							     =		self.dummy_model()
								UpperCAmelCase     :		str							     =		self.dummy_sample_deter
								scheduler.set_timesteps(A    )

								for i, t in enumerate(scheduler.prk_timesteps    ):
															UpperCAmelCase     :		Optional[Any]							     =		model(A		,		A    )
															UpperCAmelCase     :		Tuple							     =		scheduler.step_prk(A		,		A		,		A    ).prev_sample

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

								return sample





	def  _lowercase(  self    )			->			Tuple:
								UpperCAmelCase     :		Optional[int]							     =		dict(self.forward_default_kwargs    )

								UpperCAmelCase     :		Optional[int]							     =		kwargs.pop("""num_inference_steps"""		,		A    )

								for scheduler_class in self.scheduler_classes:
															UpperCAmelCase     :		List[str]							     =		self.get_scheduler_config()
															UpperCAmelCase     :		List[Any]							     =		scheduler_class(**A    )

															UpperCAmelCase     :		Union[str, Any]							     =		self.dummy_sample
															UpperCAmelCase     :		Tuple							     =		0.1 * sample

															if num_inference_steps is not None and hasattr(A		,		"""set_timesteps"""    ):
																						scheduler.set_timesteps(A    )
															elif num_inference_steps is not None and not hasattr(A		,		"""set_timesteps"""    ):
																						UpperCAmelCase     :		Union[str, Any]							     =		num_inference_steps

															# copy over dummy past residuals (must be done after set_timesteps)
															UpperCAmelCase     :		Union[str, Any]							     =		[residual + 0.2, residual + 0.1_5, residual + 0.1, residual + 0.0_5]
															UpperCAmelCase     :		Union[str, Any]							     =		dummy_past_residuals[:]

															UpperCAmelCase     :		Optional[Any]							     =		scheduler.step_prk(A		,		0		,		A		,		**A    ).prev_sample
															UpperCAmelCase     :		Dict							     =		scheduler.step_prk(A		,		1		,		A		,		**A    ).prev_sample

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

															UpperCAmelCase     :		Dict							     =		scheduler.step_plms(A		,		0		,		A		,		**A    ).prev_sample
															UpperCAmelCase     :		int							     =		scheduler.step_plms(A		,		1		,		A		,		**A    ).prev_sample

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





	def  _lowercase(  self    )			->			Optional[Any]:
								for timesteps in [100, 1000]:
															self.check_over_configs(num_train_timesteps=A    )





	def  _lowercase(  self    )			->			Tuple:
								for steps_offset in [0, 1]:
															self.check_over_configs(steps_offset=A    )

								UpperCAmelCase     :		Optional[Any]							     =		self.scheduler_classes[0]
								UpperCAmelCase     :		int							     =		self.get_scheduler_config(steps_offset=1    )
								UpperCAmelCase     :		Optional[int]							     =		scheduler_class(**A    )
								scheduler.set_timesteps(10    )
								assert torch.equal(
								    scheduler.timesteps		,		torch.LongTensor(
								        [901, 851, 851, 801, 801, 751, 751, 701, 701, 651, 651, 601, 601, 501, 401, 301, 201, 101, 1]    )		,		)





	def  _lowercase(  self    )			->			Union[str, Any]:
								for beta_start, beta_end in zip([0.0_0_0_1, 0.0_0_1]		,		[0.0_0_2, 0.0_2]    ):
															self.check_over_configs(beta_start=A		,		beta_end=A    )





	def  _lowercase(  self    )			->			str:
								for schedule in ["linear", "squaredcos_cap_v2"]:
															self.check_over_configs(beta_schedule=A    )





	def  _lowercase(  self    )			->			Any:
								for prediction_type in ["epsilon", "v_prediction"]:
															self.check_over_configs(prediction_type=A    )





	def  _lowercase(  self    )			->			Union[str, Any]:
								for t in [1, 5, 10]:
															self.check_over_forward(time_step=A    )





	def  _lowercase(  self    )			->			Optional[int]:
								for t, num_inference_steps in zip([1, 5, 10]		,		[10, 50, 100]    ):
															self.check_over_forward(num_inference_steps=A    )





	def  _lowercase(  self    )			->			List[str]:
								# earlier version of set_timesteps() caused an error indexing alpha's with inference steps as power of 3
								UpperCAmelCase     :		Dict							     =		27

								for scheduler_class in self.scheduler_classes:
															UpperCAmelCase     :		Optional[int]							     =		self.dummy_sample
															UpperCAmelCase     :		Tuple							     =		0.1 * sample

															UpperCAmelCase     :		str							     =		self.get_scheduler_config()
															UpperCAmelCase     :		List[str]							     =		scheduler_class(**A    )

															scheduler.set_timesteps(A    )

															# 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]    ):
																						UpperCAmelCase     :		List[Any]							     =		scheduler.step_prk(A		,		A		,		A    ).prev_sample





	def  _lowercase(  self    )			->			List[str]:
								with self.assertRaises(A    ):
															UpperCAmelCase     :		List[Any]							     =		self.scheduler_classes[0]
															UpperCAmelCase     :		Optional[Any]							     =		self.get_scheduler_config()
															UpperCAmelCase     :		List[Any]							     =		scheduler_class(**A    )

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





	def  _lowercase(  self    )			->			str:
								UpperCAmelCase     :		Dict							     =		self.full_loop()
								UpperCAmelCase     :		Optional[int]							     =		torch.sum(torch.abs(A    )    )
								UpperCAmelCase     :		List[str]							     =		torch.mean(torch.abs(A    )    )

								assert abs(result_sum.item() - 1_9_8.1_3_1_8    ) < 1e-2
								assert abs(result_mean.item() - 0.2_5_8_0    ) < 1e-3





	def  _lowercase(  self    )			->			Tuple:
								UpperCAmelCase     :		Optional[int]							     =		self.full_loop(prediction_type="""v_prediction"""    )
								UpperCAmelCase     :		List[str]							     =		torch.sum(torch.abs(A    )    )
								UpperCAmelCase     :		str							     =		torch.mean(torch.abs(A    )    )

								assert abs(result_sum.item() - 6_7.3_9_8_6    ) < 1e-2
								assert abs(result_mean.item() - 0.0_8_7_8    ) < 1e-3





	def  _lowercase(  self    )			->			int:
								# We specify different beta, so that the first alpha is 0.99
								UpperCAmelCase     :		str							     =		self.full_loop(set_alpha_to_one=A		,		beta_start=0.0_1    )
								UpperCAmelCase     :		Any							     =		torch.sum(torch.abs(A    )    )
								UpperCAmelCase     :		Dict							     =		torch.mean(torch.abs(A    )    )

								assert abs(result_sum.item() - 2_3_0.0_3_9_9    ) < 1e-2
								assert abs(result_mean.item() - 0.2_9_9_5    ) < 1e-3






	def  _lowercase(  self    )			->			str:
								# We specify different beta, so that the first alpha is 0.99
								UpperCAmelCase     :		Union[str, Any]							     =		self.full_loop(set_alpha_to_one=A		,		beta_start=0.0_1    )
								UpperCAmelCase     :		int							     =		torch.sum(torch.abs(A    )    )
								UpperCAmelCase     :		int							     =		torch.mean(torch.abs(A    )    )

								assert abs(result_sum.item() - 1_8_6.9_4_8_2    ) < 1e-2
								assert abs(result_mean.item() - 0.2_4_3_4    ) < 1e-3






'''simple docstring'''




from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
							from PIL import Image

							from ..image_utils import load_image

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
							from ..tf_utils import stable_softmax

a   :			Union[str, Any]					  =    logging.get_logger(__name__)

@add_end_docstrings(__magic_name__			)
class 			UpperCamelCase_ (							__magic_name__			):





	def __init__(  self		,		**A    )			->			List[str]:
								super().__init__(**A    )

								requires_backends(self		,		"""vision"""    )
								self.check_model_type(
								    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
								    if self.framework == """tf"""
								    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING    )





	def __call__(  self		,		A		,		**A    )			->			Optional[Any]:
								return super().__call__(A		,		**A    )





	def  _lowercase(  self		,		**A    )			->			Optional[Any]:
								UpperCAmelCase     :		List[Any]							     =		{}
								if "candidate_labels" in kwargs:
															UpperCAmelCase     :		Dict							     =		kwargs["""candidate_labels"""]
								if "hypothesis_template" in kwargs:
															UpperCAmelCase     :		Optional[Any]							     =		kwargs["""hypothesis_template"""]

								return preprocess_params, {}, {}





	def  _lowercase(  self		,		A		,		A=None		,		A="This is a photo of {}."    )			->			Optional[Any]:
								UpperCAmelCase     :		int							     =		load_image(A    )
								UpperCAmelCase     :		List[str]							     =		self.image_processor(images=[image]		,		return_tensors=self.framework    )
								UpperCAmelCase     :		List[str]							     =		candidate_labels
								UpperCAmelCase     :		Tuple							     =		[hypothesis_template.format(A    ) for x in candidate_labels]
								UpperCAmelCase     :		Union[str, Any]							     =		self.tokenizer(A		,		return_tensors=self.framework		,		padding=A    )
								UpperCAmelCase     :		Union[str, Any]							     =		[text_inputs]
								return inputs





	def  _lowercase(  self		,		A    )			->			Optional[int]:
								UpperCAmelCase     :		List[Any]							     =		model_inputs.pop("""candidate_labels"""    )
								UpperCAmelCase     :		Optional[Any]							     =		model_inputs.pop("""text_inputs"""    )
								if isinstance(text_inputs[0]		,		A    ):
															UpperCAmelCase     :		Optional[Any]							     =		text_inputs[0]
								else:
															# Batching case.
															UpperCAmelCase     :		Any							     =		text_inputs[0][0]

								UpperCAmelCase     :		Dict							     =		self.model(**A		,		**A    )

								UpperCAmelCase     :		List[Any]							     =		{
								    """candidate_labels""": candidate_labels,
								    """logits""": outputs.logits_per_image,
								}
								return model_outputs






	def  _lowercase(  self		,		A    )			->			Union[str, Any]:
								UpperCAmelCase     :		int							     =		model_outputs.pop("""candidate_labels"""    )
								UpperCAmelCase     :		int							     =		model_outputs["""logits"""][0]
								if self.framework == "pt":
															UpperCAmelCase     :		Optional[int]							     =		logits.softmax(dim=-1    ).squeeze(-1    )
															UpperCAmelCase     :		Any							     =		probs.tolist()
															if not isinstance(A		,		A    ):
																						UpperCAmelCase     :		Any							     =		[scores]
								elif self.framework == "tf":
															UpperCAmelCase     :		List[str]							     =		stable_softmax(A		,		axis=-1    )
															UpperCAmelCase     :		Union[str, Any]							     =		probs.numpy().tolist()
								else:
															raise ValueError(f'''Unsupported framework: {self.framework}'''    )

								UpperCAmelCase     :		Any							     =		[
								    {"""score""": score, """label""": candidate_label}
								    for score, candidate_label in sorted(zip(A		,		A    )		,		key=lambda A    : -x[0]    )
								]
								return result

