



'''simple docstring'''

# DISCLAIMER: This file is strongly influenced by https://github.com/yang-song/score_sde_pytorch

import math
from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin, SchedulerOutput

@dataclass
class 	__magic_name__						(    __UpperCamelCase   ):
							lowerCAmelCase       :    torch.FloatTensor
							lowerCAmelCase       :    torch.FloatTensor

class 	__magic_name__						(    __UpperCamelCase			,						__UpperCamelCase   ):
							lowerCAmelCase       :    str   				=					1







							@register_to_config
							def __init__(							self					:       Tuple	,_UpperCAmelCase					:       int = 2000	,_UpperCAmelCase					:       List[str] = 0.15	,_UpperCAmelCase					:       Dict = 0.01	,_UpperCAmelCase					:       Dict = 1348.0	,_UpperCAmelCase					:       Dict = 1E-5	,_UpperCAmelCase					:       Union[str, Any] = 1	,):
									_a     :      Dict        =							sigma_max

									# setable values
									_a     :      List[Any]        =							None

									self.set_sigmas(lowerCAmelCase__	,lowerCAmelCase__	,lowerCAmelCase__	,lowerCAmelCase__		)







							def 							__lowercase			(							self					:       Dict	,_UpperCAmelCase					:       Dict	,_UpperCAmelCase					:       Optional[int] = None		):
									return sample







							def 							__lowercase			(							self					:       Union[str, Any]	,_UpperCAmelCase					:       List[Any]	,_UpperCAmelCase					:       List[str] = None	,_UpperCAmelCase					:       str = None		):
									_a     :      str        =							sampling_eps if sampling_eps is not None else self.config.sampling_eps

									_a     :      str        =							torch.linspace(1	,lowerCAmelCase__	,lowerCAmelCase__	,device=lowerCAmelCase__		)







							def 							__lowercase			(							self					:       int	,_UpperCAmelCase					:       Any	,_UpperCAmelCase					:       Union[str, Any] = None	,_UpperCAmelCase					:       Tuple = None	,_UpperCAmelCase					:       List[str] = None		):
									_a     :      Optional[Any]        =							sigma_min if sigma_min is not None else self.config.sigma_min
									_a     :      Optional[int]        =							sigma_max if sigma_max is not None else self.config.sigma_max
									_a     :      Optional[int]        =							sampling_eps if sampling_eps is not None else self.config.sampling_eps
									if self.timesteps is None:
											self.set_timesteps(lowerCAmelCase__	,lowerCAmelCase__		)

									_a     :      Any        =							sigma_min * (sigma_max / sigma_min) ** (self.timesteps / sampling_eps)
									_a     :      Any        =							torch.exp(torch.linspace(math.log(lowerCAmelCase__		)	,math.log(lowerCAmelCase__		)	,lowerCAmelCase__		)		)
									_a     :      Union[str, Any]        =							torch.tensor([sigma_min * (sigma_max / sigma_min) ** t for t in self.timesteps]		)







							def 							__lowercase			(							self					:       Optional[int]	,_UpperCAmelCase					:       int	,_UpperCAmelCase					:       List[str]		):
									return torch.where(
									    timesteps == 0	,torch.zeros_like(t.to(timesteps.device		)		)	,self.discrete_sigmas[timesteps - 1].to(timesteps.device		)	,)







							def 							__lowercase			(							self					:       Dict	,_UpperCAmelCase					:       Any	,_UpperCAmelCase					:       List[str]	,_UpperCAmelCase					:       Optional[Any]	,_UpperCAmelCase					:       str = None	,_UpperCAmelCase					:       Union[str, Any] = True	,):
									if self.timesteps is None:
											raise ValueError(
											    '`self.timesteps` is not set, you need to run \'set_timesteps\' after creating the scheduler'		)

									_a     :      Tuple        =							timestep * torch.ones(
									    sample.shape[0]	,device=sample.device		)  # torch.repeat_interleave(timestep, sample.shape[0])
									_a     :      List[str]        =							(timestep * (len(self.timesteps		) - 1)).long()

									# mps requires indices to be in the same device, so we use cpu as is the default with cuda
									_a     :      Any        =							timesteps.to(self.discrete_sigmas.device		)

									_a     :      int        =							self.discrete_sigmas[timesteps].to(sample.device		)
									_a     :      Optional[int]        =							self.get_adjacent_sigma(lowerCAmelCase__	,lowerCAmelCase__		).to(sample.device		)
									_a     :      Optional[Any]        =							torch.zeros_like(lowerCAmelCase__		)
									_a     :      Optional[Any]        =							(sigma**2 - adjacent_sigma**2) ** 0.5

									# equation 6 in the paper: the model_output modeled by the network is grad_x log pt(x)
									# also equation 47 shows the analog from SDE models to ancestral sampling methods
									_a     :      Union[str, Any]        =							diffusion.flatten()
									while len(diffusion.shape		) < len(sample.shape		):
											_a     :      List[str]        =							diffusion.unsqueeze(-1		)
									_a     :      List[Any]        =							drift - diffusion**2 * model_output

									#  equation 6: sample noise for the diffusion term of
									_a     :      Union[str, Any]        =							randn_tensor(
									    sample.shape	,layout=sample.layout	,generator=lowerCAmelCase__	,device=sample.device	,dtype=sample.dtype		)
									_a     :      int        =							sample - drift  # subtract because `dt` is a small negative timestep
									# TODO is the variable diffusion the correct scaling term for the noise?
									_a     :      Any        =							prev_sample_mean + diffusion * noise  # add impact of diffusion field g

									if not return_dict:
											return (prev_sample, prev_sample_mean)

									return SdeVeOutput(prev_sample=lowerCAmelCase__	,prev_sample_mean=lowerCAmelCase__		)







							def 							__lowercase			(							self					:       int	,_UpperCAmelCase					:       Dict	,_UpperCAmelCase					:       int	,_UpperCAmelCase					:       Tuple = None	,_UpperCAmelCase					:       Optional[int] = True	,):
									if self.timesteps is None:
											raise ValueError(
											    '`self.timesteps` is not set, you need to run \'set_timesteps\' after creating the scheduler'		)

									# For small batch sizes, the paper "suggest replacing norm(z) with sqrt(d), where d is the dim. of z"
									# sample noise for correction
									_a     :      List[str]        =							randn_tensor(sample.shape	,layout=sample.layout	,generator=lowerCAmelCase__		).to(sample.device		)

									# compute step size from the model_output, the noise, and the snr
									_a     :      Tuple        =							torch.norm(model_output.reshape(model_output.shape[0]	,-1		)	,dim=-1		).mean()
									_a     :      Dict        =							torch.norm(noise.reshape(noise.shape[0]	,-1		)	,dim=-1		).mean()
									_a     :      List[Any]        =							(self.config.snr * noise_norm / grad_norm) ** 2 * 2
									_a     :      int        =							step_size * torch.ones(sample.shape[0]		).to(sample.device		)
									# self.repeat_scalar(step_size, sample.shape[0])

									# compute corrected sample: model_output term and noise term
									_a     :      int        =							step_size.flatten()
									while len(step_size.shape		) < len(sample.shape		):
											_a     :      Tuple        =							step_size.unsqueeze(-1		)
									_a     :      List[str]        =							sample + step_size * model_output
									_a     :      Dict        =							prev_sample_mean + ((step_size * 2) ** 0.5) * noise

									if not return_dict:
											return (prev_sample,)

									return SchedulerOutput(prev_sample=lowerCAmelCase__		)







							def 							__lowercase			(							self					:       Optional[int]	,_UpperCAmelCase					:       str	,_UpperCAmelCase					:       Tuple	,_UpperCAmelCase					:       Tuple	,):
									_a     :      List[str]        =							timesteps.to(original_samples.device		)
									_a     :      List[Any]        =							self.discrete_sigmas.to(original_samples.device		)[timesteps]
									_a     :      int        =							(
									    noise * sigmas[:, None, None, None]
									    if noise is not None
									    else torch.randn_like(lowerCAmelCase__		) * sigmas[:, None, None, None]
									)
									_a     :      str        =							noise + original_samples
									return noisy_samples







							def __len__(							self					:       Any		):
									return self.config.num_train_timesteps








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

__lowerCAmelCase		       =							logging.get_logger(__name__)





@add_end_docstrings(__UpperCamelCase      )
class   __a			(    __UpperCamelCase      ):
 def __init__(   self				, **lowerCAmelCase__ )					->				List[str]:




        '''simple docstring'''
        super().__init__(**lowerCAmelCase__ )

        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				, lowerCAmelCase__				, **lowerCAmelCase__ )					->				Optional[Any]:




        '''simple docstring'''
        return super().__call__(lowerCAmelCase__				, **lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self				, **lowerCAmelCase__ )					->				Dict:




        '''simple docstring'''
        lowercase__:   int  			=   {}
        if "candidate_labels" in kwargs:
               lowercase__:   Dict  			=   kwargs['candidate_labels']
        if "hypothesis_template" in kwargs:
               lowercase__:   List[Any]  			=   kwargs['hypothesis_template']

        return preprocess_params, {}, {}
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__				, lowerCAmelCase__=None				, lowerCAmelCase__="This is a photo of {}." )					->				int:




        '''simple docstring'''
        lowercase__:   Optional[int]  			=   load_image(lowerCAmelCase__ )
        lowercase__:   Dict  			=   self.image_processor(images=[image]				, return_tensors=self.framework )
        lowercase__:   Tuple  			=   candidate_labels
        lowercase__:   List[str]  			=   [hypothesis_template.format(lowerCAmelCase__ ) for x in candidate_labels]
        lowercase__:   Optional[Any]  			=   self.tokenizer(lowerCAmelCase__				, return_tensors=self.framework				, padding=lowerCAmelCase__ )
        lowercase__:   str  			=   [text_inputs]
        return inputs
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__ )					->				int:




        '''simple docstring'''
        lowercase__:   str  			=   model_inputs.pop('candidate_labels' )
        lowercase__:   List[str]  			=   model_inputs.pop('text_inputs' )
        if isinstance(text_inputs[0]				, lowerCAmelCase__ ):
               lowercase__:   Any  			=   text_inputs[0]
        else:
               # Batching case.
               lowercase__:   Optional[int]  			=   text_inputs[0][0]

        lowercase__:   Tuple  			=   self.model(**lowerCAmelCase__				, **lowerCAmelCase__ )

        lowercase__:   Any  			=   {
            'candidate_labels': candidate_labels,
            'logits': outputs.logits_per_image,
        }
        return model_outputs
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__ )					->				Union[str, Any]:




        '''simple docstring'''
        lowercase__:   Tuple  			=   model_outputs.pop('candidate_labels' )
        lowercase__:   Dict  			=   model_outputs['logits'][0]
        if self.framework == "pt":
               lowercase__:   Optional[int]  			=   logits.softmax(dim=-1 ).squeeze(-1 )
               lowercase__:   Dict  			=   probs.tolist()
               if not isinstance(lowerCAmelCase__				, lowerCAmelCase__ ):
                      lowercase__:   Dict  			=   [scores]
        elif self.framework == "tf":
               lowercase__:   Optional[int]  			=   stable_softmax(lowerCAmelCase__				, axis=-1 )
               lowercase__:   Union[str, Any]  			=   probs.numpy().tolist()
        else:
               raise ValueError(F'Unsupported framework: {self.framework}' )

        lowercase__:   List[Any]  			=   [
            {'score': score, 'label': candidate_label}
            for score, candidate_label in sorted(zip(lowerCAmelCase__				, lowerCAmelCase__ )				, key=lambda lowerCAmelCase__ : -x[0] )
        ]
        return result


