


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     __lowerCAmelCase						(							snake_case	:    List[Any]						,      snake_case	:    str=0.999						,      snake_case	:    Dict="cosine"						,      )       ->							Dict:

   '''simple docstring'''





   if alpha_transform_type == "cosine":

      def alpha_bar_fn(snake_case	:    Optional[Any] ):
         return math.cos((t + 0.008) / 1.008 * 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}' )

   __lowerCamelCase:				List[Any]     =	[]
   for i in range(snake_case__ ):
      __lowerCamelCase:				Tuple     =	i / num_diffusion_timesteps
      __lowerCamelCase:				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        a					(			__a     ,__a   ):
     UpperCAmelCase__ :  Any						      =	[e.name for e in KarrasDiffusionSchedulers]
     UpperCAmelCase__ :  Dict						      =	2





     @register_to_config
     def __init__(    self					:				Tuple    ,	SCREAMING_SNAKE_CASE_					:				int = 1000    ,	SCREAMING_SNAKE_CASE_					:				float = 0.0_0085    ,	SCREAMING_SNAKE_CASE_					:				float = 0.012    ,	SCREAMING_SNAKE_CASE_					:				str = "linear"    ,	SCREAMING_SNAKE_CASE_					:				Optional[Union[np.ndarray, List[float]]] = None    ,	SCREAMING_SNAKE_CASE_					:				str = "epsilon"    ,	SCREAMING_SNAKE_CASE_					:				str = "linspace"    ,	SCREAMING_SNAKE_CASE_					:				int = 0    ,	):



        if trained_betas is not None:
           __lowerCamelCase:				Tuple     =	torch.tensor(a_    ,	dtype=torch.floataa							)
        elif beta_schedule == "linear":
           __lowerCamelCase:				Dict     =	torch.linspace(a_    ,	a_    ,	a_    ,	dtype=torch.floataa							)
        elif beta_schedule == "scaled_linear":
           # this schedule is very specific to the latent diffusion model.
           __lowerCamelCase:				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
           __lowerCamelCase:				str     =	betas_for_alpha_bar(a_							)
        else:
           raise NotImplementedError(F'{beta_schedule} does is not implemented for {self.__class__}'							)

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

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





     def  SCREAMING_SNAKE_CASE__       (    self					:				Tuple    ,	SCREAMING_SNAKE_CASE_					:				List[str]    ,	SCREAMING_SNAKE_CASE_					:				Dict=None							):



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

        __lowerCamelCase:				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:
           __lowerCamelCase:				List[Any]     =	1 if len(a_							) > 1 else 0
        else:
           __lowerCamelCase:				int     =	timestep.cpu().item() if torch.is_tensor(a_							) else timestep
           __lowerCamelCase:				Tuple     =	self._index_counter[timestep_int]

        return indices[pos].item()





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



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

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





     def  SCREAMING_SNAKE_CASE__       (    self					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				torch.FloatTensor    ,	SCREAMING_SNAKE_CASE_					:				Union[float, torch.FloatTensor]    ,	):
        __lowerCamelCase:				Union[str, Any]     =	self.index_for_timestep(a_							)

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

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





     def  SCREAMING_SNAKE_CASE__       (    self					:				List[str]    ,	SCREAMING_SNAKE_CASE_					:				int    ,	SCREAMING_SNAKE_CASE_					:				Union[str, torch.device] = None    ,	SCREAMING_SNAKE_CASE_					:				Optional[int] = None    ,	):
        __lowerCamelCase:				Union[str, Any]     =	num_inference_steps

        __lowerCamelCase:				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":
           __lowerCamelCase:				Optional[int]     =	np.linspace(0    ,	num_train_timesteps - 1    ,	a_    ,	dtype=a_							)[::-1].copy()
        elif self.config.timestep_spacing == "leading":
           __lowerCamelCase:				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
           __lowerCamelCase:				int     =	(np.arange(0    ,	a_							) * step_ratio).round()[::-1].copy().astype(a_							)
           timesteps += self.config.steps_offset
        elif self.config.timestep_spacing == "trailing":
           __lowerCamelCase:				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
           __lowerCamelCase:				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\'.'							)

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

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

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

        __lowerCamelCase:				Optional[int]     =	torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2							), sigmas[-1:]]							)
        __lowerCamelCase:				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
           __lowerCamelCase:				Dict     =	torch.from_numpy(a_							).to(a_    ,	dtype=torch.floataa							)
        else:
           __lowerCamelCase:				Optional[Any]     =	torch.from_numpy(a_							).to(a_							)

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

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

        __lowerCamelCase:				Optional[Any]     =	None

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





     def  SCREAMING_SNAKE_CASE__       (    self					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				Tuple							):
        __lowerCamelCase:				Tuple     =	sigma.log()

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

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

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

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

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





     @property
     def  SCREAMING_SNAKE_CASE__       (    self					:				Any							):
        return self.sample is None





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

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

        if self.state_in_first_order:
           __lowerCamelCase:				Union[str, Any]     =	self.sigmas[step_index]
           __lowerCamelCase:				Optional[Any]     =	self.sigmas_interpol[step_index + 1]
           __lowerCamelCase:				Union[str, Any]     =	self.sigmas[step_index + 1]
        else:
           # 2nd order / KDPM2's method
           __lowerCamelCase:				int     =	self.sigmas[step_index - 1]
           __lowerCamelCase:				Any     =	self.sigmas_interpol[step_index]
           __lowerCamelCase:				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
        __lowerCamelCase:				Optional[int]     =	0
        __lowerCamelCase:				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":
           __lowerCamelCase:				str     =	sigma_hat if self.state_in_first_order else sigma_interpol
           __lowerCamelCase:				Optional[Any]     =	sample - sigma_input * model_output
        elif self.config.prediction_type == "v_prediction":
           __lowerCamelCase:				Dict     =	sigma_hat if self.state_in_first_order else sigma_interpol
           __lowerCamelCase:				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
           __lowerCamelCase:				int     =	(sample - pred_original_sample) / sigma_hat
           # 3. delta timestep
           __lowerCamelCase:				Tuple     =	sigma_interpol - sigma_hat

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

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

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

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

        if not return_dict:
           return (prev_sample,)

        return SchedulerOutput(prev_sample=a_							)





     def  SCREAMING_SNAKE_CASE__       (    self					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				torch.FloatTensor    ,	SCREAMING_SNAKE_CASE_					:				torch.FloatTensor    ,	SCREAMING_SNAKE_CASE_					:				torch.FloatTensor    ,	):
        __lowerCamelCase:				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
           __lowerCamelCase:				Dict     =	self.timesteps.to(original_samples.device    ,	dtype=torch.floataa							)
           __lowerCamelCase:				List[Any]     =	timesteps.to(original_samples.device    ,	dtype=torch.floataa							)
        else:
           __lowerCamelCase:				Optional[Any]     =	self.timesteps.to(original_samples.device							)
           __lowerCamelCase:				Any     =	timesteps.to(original_samples.device							)

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

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

        __lowerCamelCase:				Any     =	original_samples + noise * sigma
        return noisy_samples







     def __len__(    self					:				Union[str, Any]							):
        return self.config.num_train_timesteps






def     __lowerCAmelCase						(							)       ->							Optional[Any]:
   __lowerCamelCase:				List[str]     =	[31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]

   __lowerCamelCase:				List[str]     =	6
   __lowerCamelCase:				Optional[int]     =	1
   __lowerCamelCase:				int     =	1901

   __lowerCamelCase:				Tuple     =	0

   while year < 2001:
      day += 7

      if (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0):
         if day > days_per_month[month - 1] and month != 2:
            month += 1
            __lowerCamelCase:				Union[str, Any]     =	day - days_per_month[month - 2]
         elif day > 29 and month == 2:
            month += 1
            __lowerCamelCase:				List[Any]     =	day - 29
      else:
         if day > days_per_month[month - 1]:
            month += 1
            __lowerCamelCase:				Any     =	day - days_per_month[month - 2]

      if month > 12:
         year += 1
         __lowerCamelCase:				Dict     =	1

      if year < 2001 and day == 1:
         sundays += 1
   return sundays


if __name__ == "__main__":
   print(solution())


