




"""simple docstring"""


import math
from collections import defaultdict
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput


def       UpperCAmelCase__ (snake_case__							:      List[Any]	,     snake_case__							:      str=0.9_99	,     snake_case__							:      Dict="cosine"	,     ):
  """simple docstring"""

  if alpha_transform_type == "cosine":

    def alpha_bar_fn(snake_case__							:      Optional[Any]					):
      return math.cos((t + 0.0_08) / 1.0_08 * math.pi / 2					) ** 2

  elif alpha_transform_type == "exp":

    def alpha_bar_fn(snake_case__							:      Optional[int]					):
      return math.exp(t * -12.0					)

  else:
    raise ValueError(F"Unsupported alpha_tranform_type: {alpha_transform_type}"					)

  _snake_case							:	List[Any]			=						[]
  for i in range(snake_case__					):
    _snake_case							:	Tuple			=						i / num_diffusion_timesteps
    _snake_case							:	Optional[int]			=						(i + 1) / num_diffusion_timesteps
    betas.append(min(1 - alpha_bar_fn(snake_case__					) / alpha_bar_fn(snake_case__					)	,     snake_case__					)					)
  return torch.tensor(snake_case__	,     dtype=torch.floataa					)






class       lowercase(  __a ,    __a				):







 '''simple docstring'''





 lowercase__				      =      [e.name for e in KarrasDiffusionSchedulers]
 lowercase__				      =      2


 @register_to_config
 def __init__(       self:	Tuple,   a_:	int = 1_000,   a_:	float = 0.00_085,   a_:	float = 0.012,   a_:	str = "linear",   a_:	Optional[Union[np.ndarray, List[float]]] = None,   a_:	str = "epsilon",   a_:	str = "linspace",   a_:	int = 0,   ):




   '''simple docstring'''



   if trained_betas is not None:
     _snake_case							:	Tuple			=						torch.tensor(a_,   dtype=torch.floataa	)
   elif beta_schedule == "linear":
     _snake_case							:	Dict			=						torch.linspace(a_,   a_,   a_,   dtype=torch.floataa	)
   elif beta_schedule == "scaled_linear":
     # this schedule is very specific to the latent diffusion model.
     _snake_case							:	List[str]			=						(
         torch.linspace(beta_start**0.5,   beta_end**0.5,   a_,   dtype=torch.floataa	) ** 2
     )
   elif beta_schedule == "squaredcos_cap_v2":
     # Glide cosine schedule
     _snake_case							:	str			=						betas_for_alpha_bar(a_	)
   else:
     raise NotImplementedError(f"{beta_schedule} does is not implemented for {self.__class__}"	)

   _snake_case							:	int			=						1.0 - self.betas
   _snake_case							:	Optional[Any]			=						torch.cumprod(self.alphas,   dim=0	)

   #  set all values
   self.set_timesteps(a_,   a_,   a_	)


 def 				UpperCamelCase_     (       self:	Tuple,   a_:	List[str],   a_:	Dict=None	):




   '''simple docstring'''



   if schedule_timesteps is None:
     _snake_case							:	List[Any]			=						self.timesteps

   _snake_case							:	Any			=						(schedule_timesteps == timestep).nonzero()

   # The sigma index that is taken for the **very** first `step`
   # is always the second index (or the last index if there is only 1)
   # This way we can ensure we don't accidentally skip a sigma in
   # case we start in the middle of the denoising schedule (e.g. for image-to-image)
   if len(self._index_counter	) == 0:
     _snake_case							:	List[Any]			=						1 if len(a_	) > 1 else 0
   else:
     _snake_case							:	int			=						timestep.cpu().item() if torch.is_tensor(a_	) else timestep
     _snake_case							:	Tuple			=						self._index_counter[timestep_int]

   return indices[pos].item()


 @property
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   if self.config.timestep_spacing in ["linspace", "trailing"]:
     return self.sigmas.max()

   return (self.sigmas.max() ** 2 + 1) ** 0.5


 def 				UpperCamelCase_     (       self:	Optional[Any],   a_:	torch.FloatTensor,   a_:	Union[float, torch.FloatTensor],   ):




   '''simple docstring'''



   _snake_case							:	Union[str, Any]			=						self.index_for_timestep(a_	)

   if self.state_in_first_order:
     _snake_case							:	List[str]			=						self.sigmas[step_index]
   else:
     _snake_case							:	Any			=						self.sigmas_interpol[step_index]

   _snake_case							:	List[str]			=						sample / ((sigma**2 + 1) ** 0.5)
   return sample


 def 				UpperCamelCase_     (       self:	List[str],   a_:	int,   a_:	Union[str, torch.device] = None,   a_:	Optional[int] = None,   ):




   '''simple docstring'''



   _snake_case							:	Union[str, Any]			=						num_inference_steps

   _snake_case							:	Optional[int]			=						num_train_timesteps or self.config.num_train_timesteps

   # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
   if self.config.timestep_spacing == "linspace":
     _snake_case							:	Optional[int]			=						np.linspace(0,   num_train_timesteps - 1,   a_,   dtype=a_	)[::-1].copy()
   elif self.config.timestep_spacing == "leading":
     _snake_case							:	int			=						num_train_timesteps // self.num_inference_steps
     # creates integer timesteps by multiplying by ratio
     # casting to int to avoid issues when num_inference_step is power of 3
     _snake_case							:	int			=						(np.arange(0,   a_	) * step_ratio).round()[::-1].copy().astype(a_	)
     timesteps += self.config.steps_offset
   elif self.config.timestep_spacing == "trailing":
     _snake_case							:	Optional[int]			=						num_train_timesteps / self.num_inference_steps
     # creates integer timesteps by multiplying by ratio
     # casting to int to avoid issues when num_inference_step is power of 3
     _snake_case							:	Tuple			=						(np.arange(a_,   0,   -step_ratio	)).round().copy().astype(a_	)
     timesteps -= 1
   else:
     raise ValueError(
         f"{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."	)

   _snake_case							:	Dict			=						np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5	)
   _snake_case							:	Dict			=						torch.from_numpy(np.log(a_	)	).to(a_	)

   _snake_case							:	int			=						np.interp(a_,   np.arange(0,   len(a_	)	),   a_	)
   _snake_case							:	Tuple			=						np.concatenate([sigmas, [0.0]]	).astype(np.floataa	)
   _snake_case							:	str			=						torch.from_numpy(a_	).to(device=a_	)

   # interpolate sigmas
   _snake_case							:	int			=						sigmas.log().lerp(sigmas.roll(1	).log(),   0.5	).exp()

   _snake_case							:	Optional[int]			=						torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2	), sigmas[-1:]]	)
   _snake_case							:	Optional[Any]			=						torch.cat(
       [sigmas_interpol[:1], sigmas_interpol[1:].repeat_interleave(2	), sigmas_interpol[-1:]]	)

   if str(a_	).startswith("""mps"""	):
     # mps does not support float64
     _snake_case							:	Dict			=						torch.from_numpy(a_	).to(a_,   dtype=torch.floataa	)
   else:
     _snake_case							:	Optional[Any]			=						torch.from_numpy(a_	).to(a_	)

   # interpolate timesteps
   _snake_case							:	Tuple			=						self.sigma_to_t(a_	).to(a_,   dtype=timesteps.dtype	)
   _snake_case							:	List[Any]			=						torch.stack((timesteps_interpol[1:-1, None], timesteps[1:, None]),   dim=-1	).flatten()

   _snake_case							:	List[str]			=						torch.cat([timesteps[:1], interleaved_timesteps]	)

   _snake_case							:	Optional[Any]			=						None

   # for exp beta schedules, such as the one for `pipeline_shap_e.py`
   # we need an index counter
   _snake_case							:	List[str]			=						defaultdict(a_	)


 def 				UpperCamelCase_     (       self:	Optional[Any],   a_:	Tuple	):




   '''simple docstring'''



   _snake_case							:	Tuple			=						sigma.log()

   # get distribution
   _snake_case							:	Any			=						log_sigma - self.log_sigmas[:, None]

   # get sigmas range
   _snake_case							:	int			=						dists.ge(0	).cumsum(dim=0	).argmax(dim=0	).clamp(max=self.log_sigmas.shape[0] - 2	)
   _snake_case							:	Dict			=						low_idx + 1

   _snake_case							:	List[str]			=						self.log_sigmas[low_idx]
   _snake_case							:	int			=						self.log_sigmas[high_idx]

   # interpolate sigmas
   _snake_case							:	Any			=						(low - log_sigma) / (low - high)
   _snake_case							:	Dict			=						w.clamp(0,   1	)

   # transform interpolation to time range
   _snake_case							:	Tuple			=						(1 - w) * low_idx + w * high_idx
   _snake_case							:	List[Any]			=						t.view(sigma.shape	)
   return t


 @property
 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   return self.sample is None


 def 				UpperCamelCase_     (       self:	Any,   a_:	Union[torch.FloatTensor, np.ndarray],   a_:	Union[float, torch.FloatTensor],   a_:	Union[torch.FloatTensor, np.ndarray],   a_:	bool = True,   ):




   '''simple docstring'''



   _snake_case							:	Any			=						self.index_for_timestep(a_	)

   # advance index counter by 1
   _snake_case							:	List[str]			=						timestep.cpu().item() if torch.is_tensor(a_	) else timestep
   self._index_counter[timestep_int] += 1

   if self.state_in_first_order:
     _snake_case							:	Union[str, Any]			=						self.sigmas[step_index]
     _snake_case							:	Optional[Any]			=						self.sigmas_interpol[step_index + 1]
     _snake_case							:	Union[str, Any]			=						self.sigmas[step_index + 1]
   else:
     # 2nd order / KDPM2's method
     _snake_case							:	int			=						self.sigmas[step_index - 1]
     _snake_case							:	Any			=						self.sigmas_interpol[step_index]
     _snake_case							:	Tuple			=						self.sigmas[step_index]

   # currently only gamma=0 is supported. This usually works best anyways.
   # We can support gamma in the future but then need to scale the timestep before
   # passing it to the model which requires a change in API
   _snake_case							:	Optional[int]			=						0
   _snake_case							:	List[str]			=						sigma * (gamma + 1)  # Note: sigma_hat == sigma for now

   # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
   if self.config.prediction_type == "epsilon":
     _snake_case							:	str			=						sigma_hat if self.state_in_first_order else sigma_interpol
     _snake_case							:	Optional[Any]			=						sample - sigma_input * model_output
   elif self.config.prediction_type == "v_prediction":
     _snake_case							:	Dict			=						sigma_hat if self.state_in_first_order else sigma_interpol
     _snake_case							:	Optional[Any]			=						model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
         sample / (sigma_input**2 + 1)
     )
   elif self.config.prediction_type == "sample":
     raise NotImplementedError("""prediction_type not implemented yet: sample"""	)
   else:
     raise ValueError(
         f"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"	)

   if self.state_in_first_order:
     # 2. Convert to an ODE derivative for 1st order
     _snake_case							:	int			=						(sample - pred_original_sample) / sigma_hat
     # 3. delta timestep
     _snake_case							:	Tuple			=						sigma_interpol - sigma_hat

     # store for 2nd order step
     _snake_case							:	str			=						sample
   else:
     # DPM-Solver-2
     # 2. Convert to an ODE derivative for 2nd order
     _snake_case							:	Optional[Any]			=						(sample - pred_original_sample) / sigma_interpol

     # 3. delta timestep
     _snake_case							:	Any			=						sigma_next - sigma_hat

     _snake_case							:	Optional[int]			=						self.sample
     _snake_case							:	Optional[Any]			=						None

   _snake_case							:	Union[str, Any]			=						sample + derivative * dt

   if not return_dict:
     return (prev_sample,)

   return SchedulerOutput(prev_sample=a_	)


 def 				UpperCamelCase_     (       self:	Optional[Any],   a_:	torch.FloatTensor,   a_:	torch.FloatTensor,   a_:	torch.FloatTensor,   ):




   '''simple docstring'''



   _snake_case							:	Union[str, Any]			=						self.sigmas.to(device=original_samples.device,   dtype=original_samples.dtype	)
   if original_samples.device.type == "mps" and torch.is_floating_point(a_	):
     # mps does not support float64
     _snake_case							:	Dict			=						self.timesteps.to(original_samples.device,   dtype=torch.floataa	)
     _snake_case							:	List[Any]			=						timesteps.to(original_samples.device,   dtype=torch.floataa	)
   else:
     _snake_case							:	Optional[Any]			=						self.timesteps.to(original_samples.device	)
     _snake_case							:	Any			=						timesteps.to(original_samples.device	)

   _snake_case							:	List[str]			=						[self.index_for_timestep(a_,   a_	) for t in timesteps]

   _snake_case							:	Tuple			=						sigmas[step_indices].flatten()
   while len(sigma.shape	) < len(original_samples.shape	):
     _snake_case							:	List[Any]			=						sigma.unsqueeze(-1	)

   _snake_case							:	Any			=						original_samples + noise * sigma
   return noisy_samples


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




   '''simple docstring'''



   return self.config.num_train_timesteps








"""simple docstring"""


from ...processing_utils import ProcessorMixin






class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      ["image_processor", "feature_extractor"]
 lowercase__				      =      "TvltImageProcessor"
 lowercase__				      =      "TvltFeatureExtractor"


 def __init__(       self:	Dict,   a_:	Union[str, Any],   a_:	Union[str, Any]	):




   '''simple docstring'''



   super().__init__(image_processor=a_,   feature_extractor=a_	)

   _snake_case							:	Any			=						image_processor
   _snake_case							:	Dict			=						feature_extractor


 def __call__(       self:	int,   a_:	str=None,   a_:	Tuple=None,   a_:	Dict=None,   a_:	str=None,   a_:	Optional[int]=False,   a_:	Tuple=False,   *a_:	List[str],   **a_:	int,   ):




   '''simple docstring'''



   if images is None and audio is None:
     raise ValueError("""You need to specify either an `images` or `audio` input to process."""	)

   _snake_case							:	Optional[int]			=						None
   if images is not None:
     _snake_case							:	Tuple			=						self.image_processor(a_,   mask_pixel=a_,   *a_,   **a_	)
   if images_mixed is not None:
     _snake_case							:	Optional[int]			=						self.image_processor(a_,   is_mixed=a_,   *a_,   **a_	)
   if audio is not None:
     _snake_case							:	Any			=						self.feature_extractor(
         a_,   *a_,   sampling_rate=a_,   mask_audio=a_,   **a_	)

   _snake_case							:	List[str]			=						{}
   if audio is not None:
     output_dict.update(a_	)
   if images is not None:
     output_dict.update(a_	)
   if images_mixed_dict is not None:
     output_dict.update(a_	)
   return output_dict


 @property
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   _snake_case							:	Dict			=						self.image_processor.model_input_names
   _snake_case							:	List[str]			=						self.feature_extractor.model_input_names
   return list(dict.fromkeys(image_processor_input_names + feature_extractor_input_names	)	)


