






"""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 					_A			(__a    ,				__a=0.9_99    ,				__a="cosine"    ,				) ->       Optional[Any]:
    """simple docstring"""






    if alpha_transform_type == "cosine":

        def alpha_bar_fn(__a							):
            return math.cos((t + 0.0_08) / 1.0_08 * math.pi / 2							) ** 2

    elif alpha_transform_type == "exp":

        def alpha_bar_fn(__a							):
            return math.exp(t * -12.0							)

    else:
        raise ValueError(f'Unsupported alpha_tranform_type: {alpha_transform_type}'							)

    SCREAMING_SNAKE_CASE_      :  str    						=		[]
    for i in range(_lowerCamelCase							):
        SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		i / num_diffusion_timesteps
        SCREAMING_SNAKE_CASE_      :  str    						=		(i + 1) / num_diffusion_timesteps
        betas.append(min(1 - alpha_bar_fn(_lowerCamelCase							) / alpha_bar_fn(_lowerCamelCase							)    ,				_lowerCamelCase							)							)
    return torch.tensor(_lowerCamelCase    ,				dtype=torch.floataa							)





class 							lowerCAmelCase__				(			SCREAMING_SNAKE_CASE_				,		SCREAMING_SNAKE_CASE_ ):






      '''simple docstring'''

      __UpperCamelCase	 =  [e.name for e in KarrasDiffusionSchedulers]
      __UpperCamelCase	 =  2
      @register_to_config
      def __init__(						self     :  Any						,	lowercase_     :  Any = 1000						,	lowercase_     :  Union[str, Any] = 0.0_00_85						,	lowercase_     :  List[str] = 0.0_12						,	lowercase_     :  Union[str, Any] = "linear"						,	lowercase_     :  str = None						,	lowercase_     :  Any = "epsilon"						,	lowercase_     :  str = "linspace"						,	lowercase_     :  int = 0						,	):




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_      :  List[str]    						=		1.0 - self.betas
          SCREAMING_SNAKE_CASE_      :  Optional[int]    						=		torch.cumprod(self.alphas						,	dim=0)

          #  set all values
          self.set_timesteps(a__						,	a__						,	a__)
      def _SCREAMING_SNAKE_CASE (						self     :  Dict						,	lowercase_     :  Dict						,	lowercase_     :  Optional[Any]=None):




          '''simple docstring'''


          if schedule_timesteps is None:
              SCREAMING_SNAKE_CASE_      :  Dict    						=		self.timesteps

          SCREAMING_SNAKE_CASE_      :  int    						=		(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:
              SCREAMING_SNAKE_CASE_      :  List[str]    						=		1 if len(a__) > 1 else 0
          else:
              SCREAMING_SNAKE_CASE_      :  Optional[int]    						=		timestep.cpu().item() if torch.is_tensor(a__) else timestep
              SCREAMING_SNAKE_CASE_      :  Optional[int]    						=		self._index_counter[timestep_int]

          return indices[pos].item()
      @property
      def _SCREAMING_SNAKE_CASE (						self     :  List[Any]):




          '''simple docstring'''


          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     :  Dict						,	lowercase_     :  List[str]						,	lowercase_     :  Union[str, Any]						,	):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		self.index_for_timestep(a__)

          if self.state_in_first_order:
              SCREAMING_SNAKE_CASE_      :  Optional[int]    						=		self.sigmas[step_index]
          else:
              SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		self.sigmas_interpol[step_index]

          SCREAMING_SNAKE_CASE_      :  List[str]    						=		sample / ((sigma**2 + 1) ** 0.5)
          return sample
      def _SCREAMING_SNAKE_CASE (						self     :  Union[str, Any]						,	lowercase_     :  Tuple						,	lowercase_     :  str = None						,	lowercase_     :  int = None						,	):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[Any]    						=		num_inference_steps

          SCREAMING_SNAKE_CASE_      :  Tuple    						=		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":
              SCREAMING_SNAKE_CASE_      :  List[str]    						=		np.linspace(0						,	num_train_timesteps - 1						,	a__						,	dtype=a__)[::-1].copy()
          elif self.config.timestep_spacing == "leading":
              SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		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
              SCREAMING_SNAKE_CASE_      :  List[str]    						=		(np.arange(0						,	a__) * step_ratio).round()[::-1].copy().astype(a__)
              timesteps += self.config.steps_offset
          elif self.config.timestep_spacing == "trailing":
              SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		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
              SCREAMING_SNAKE_CASE_      :  str    						=		(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\'.')

          SCREAMING_SNAKE_CASE_      :  Dict    						=		np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5)
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		torch.from_numpy(np.log(a__)).to(a__)

          SCREAMING_SNAKE_CASE_      :  List[str]    						=		np.interp(a__						,	np.arange(0						,	len(a__))						,	a__)
          SCREAMING_SNAKE_CASE_      :  Dict    						=		np.concatenate([sigmas, [0.0]]).astype(np.floataa)
          SCREAMING_SNAKE_CASE_      :  Any    						=		torch.from_numpy(a__).to(device=a__)

          # interpolate sigmas
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		sigmas.log().lerp(sigmas.roll(1).log()						,	0.5).exp()

          SCREAMING_SNAKE_CASE_      :  Dict    						=		torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2), sigmas[-1:]])
          SCREAMING_SNAKE_CASE_      :  Union[str, 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
              SCREAMING_SNAKE_CASE_      :  List[Any]    						=		torch.from_numpy(a__).to(a__						,	dtype=torch.floataa)
          else:
              SCREAMING_SNAKE_CASE_      :  List[str]    						=		torch.from_numpy(a__).to(a__)

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

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

          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		None

          # for exp beta schedules, such as the one for `pipeline_shap_e.py`
          # we need an index counter
          SCREAMING_SNAKE_CASE_      :  Tuple    						=		defaultdict(a__)
      def _SCREAMING_SNAKE_CASE (						self     :  Optional[int]						,	lowercase_     :  Optional[int]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  str    						=		sigma.log()

          # get distribution
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		log_sigma - self.log_sigmas[:, None]

          # get sigmas range
          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		dists.ge(0).cumsum(dim=0).argmax(dim=0).clamp(max=self.log_sigmas.shape[0] - 2)
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		low_idx + 1

          SCREAMING_SNAKE_CASE_      :  str    						=		self.log_sigmas[low_idx]
          SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		self.log_sigmas[high_idx]

          # interpolate sigmas
          SCREAMING_SNAKE_CASE_      :  List[Any]    						=		(low - log_sigma) / (low - high)
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		w.clamp(0						,	1)

          # transform interpolation to time range
          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		(1 - w) * low_idx + w * high_idx
          SCREAMING_SNAKE_CASE_      :  Optional[int]    						=		t.view(sigma.shape)
          return t
      @property
      def _SCREAMING_SNAKE_CASE (						self     :  int):




          '''simple docstring'''


          return self.sample is None
      def _SCREAMING_SNAKE_CASE (						self     :  List[str]						,	lowercase_     :  str						,	lowercase_     :  Optional[Any]						,	lowercase_     :  Optional[Any]						,	lowercase_     :  int = True						,	):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[str]    						=		self.index_for_timestep(a__)

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

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

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

              # 3. delta timestep
              SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		sigma_next - sigma_hat

              SCREAMING_SNAKE_CASE_      :  int    						=		self.sample
              SCREAMING_SNAKE_CASE_      :  List[Any]    						=		None

          SCREAMING_SNAKE_CASE_      :  Tuple    						=		sample + derivative * dt

          if not return_dict:
              return (prev_sample,)

          return SchedulerOutput(prev_sample=a__)
      def _SCREAMING_SNAKE_CASE (						self     :  Dict						,	lowercase_     :  Optional[Any]						,	lowercase_     :  Tuple						,	lowercase_     :  Tuple						,	):




          '''simple docstring'''


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

          SCREAMING_SNAKE_CASE_      :  str    						=		[self.index_for_timestep(a__						,	a__) for t in timesteps]

          SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		sigmas[step_indices].flatten()
          while len(sigma.shape) < len(original_samples.shape):
              SCREAMING_SNAKE_CASE_      :  str    						=		sigma.unsqueeze(-1)

          SCREAMING_SNAKE_CASE_      :  Tuple    						=		original_samples + noise * sigma
          return noisy_samples



      def __len__(						self     :  str):




          '''simple docstring'''


          return self.config.num_train_timesteps



"""simple docstring"""




import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_electra import ElectraTokenizer


_a      :		List[Any]           =   {'vocab_file': 'vocab.txt', 'tokenizer_file': 'tokenizer.json'}

_a      :		Union[str, Any]           =   {
    'vocab_file': {
        'google/electra-small-generator': (
            'https://huggingface.co/google/electra-small-generator/resolve/main/vocab.txt'
        ),
        'google/electra-base-generator': 'https://huggingface.co/google/electra-base-generator/resolve/main/vocab.txt',
        'google/electra-large-generator': (
            'https://huggingface.co/google/electra-large-generator/resolve/main/vocab.txt'
        ),
        'google/electra-small-discriminator': (
            'https://huggingface.co/google/electra-small-discriminator/resolve/main/vocab.txt'
        ),
        'google/electra-base-discriminator': (
            'https://huggingface.co/google/electra-base-discriminator/resolve/main/vocab.txt'
        ),
        'google/electra-large-discriminator': (
            'https://huggingface.co/google/electra-large-discriminator/resolve/main/vocab.txt'
        ),
    },
    'tokenizer_file': {
        'google/electra-small-generator': (
            'https://huggingface.co/google/electra-small-generator/resolve/main/tokenizer.json'
        ),
        'google/electra-base-generator': (
            'https://huggingface.co/google/electra-base-generator/resolve/main/tokenizer.json'
        ),
        'google/electra-large-generator': (
            'https://huggingface.co/google/electra-large-generator/resolve/main/tokenizer.json'
        ),
        'google/electra-small-discriminator': (
            'https://huggingface.co/google/electra-small-discriminator/resolve/main/tokenizer.json'
        ),
        'google/electra-base-discriminator': (
            'https://huggingface.co/google/electra-base-discriminator/resolve/main/tokenizer.json'
        ),
        'google/electra-large-discriminator': (
            'https://huggingface.co/google/electra-large-discriminator/resolve/main/tokenizer.json'
        ),
    },
}

_a      :		Optional[Any]           =   {
    'google/electra-small-generator': 512,
    'google/electra-base-generator': 512,
    'google/electra-large-generator': 512,
    'google/electra-small-discriminator': 512,
    'google/electra-base-discriminator': 512,
    'google/electra-large-discriminator': 512,
}

_a      :		Any           =   {
    'google/electra-small-generator': {'do_lower_case': True},
    'google/electra-base-generator': {'do_lower_case': True},
    'google/electra-large-generator': {'do_lower_case': True},
    'google/electra-small-discriminator': {'do_lower_case': True},
    'google/electra-base-discriminator': {'do_lower_case': True},
    'google/electra-large-discriminator': {'do_lower_case': True},
}
class 						__A							( SCREAMING_SNAKE_CASE_		):
     _UpperCamelCase							:			Tuple							=		VOCAB_FILES_NAMES
     _UpperCamelCase							:			List[Any]							=		PRETRAINED_VOCAB_FILES_MAP
     _UpperCamelCase							:			List[Any]							=		PRETRAINED_INIT_CONFIGURATION
     _UpperCamelCase							:			Union[str, Any]							=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _UpperCamelCase							:			Optional[Any]							=		ElectraTokenizer






     def __init__(		self				,     a__=None				,     a__=None				,     a__=True				,     a__="[UNK]"				,     a__="[SEP]"				,     a__="[PAD]"				,     a__="[CLS]"				,     a__="[MASK]"				,     a__=True				,     a__=None				,     **a__				,     ):
       super().__init__(
           a__				,     tokenizer_file=a__				,     do_lower_case=a__				,     unk_token=a__				,     sep_token=a__				,     pad_token=a__				,     cls_token=a__				,     mask_token=a__				,     tokenize_chinese_chars=a__				,     strip_accents=a__				,     **a__				,     )

       _lowerCAmelCase						:       int  			=     json.loads(self.backend_tokenizer.normalizer.__getstate__()  )
       if (
           normalizer_state.get("""lowercase"""				,     a__  ) != do_lower_case
           or normalizer_state.get("""strip_accents"""				,     a__  ) != strip_accents
           or normalizer_state.get("""handle_chinese_chars"""				,     a__  ) != tokenize_chinese_chars
       ):
         _lowerCAmelCase						:       Dict  			=     getattr(a__				,     normalizer_state.pop("""type"""  )  )
         _lowerCAmelCase						:       int  			=     do_lower_case
         _lowerCAmelCase						:       str  			=     strip_accents
         _lowerCAmelCase						:       Dict  			=     tokenize_chinese_chars
         _lowerCAmelCase						:       str  			=     normalizer_class(**a__  )

       _lowerCAmelCase						:       List[str]  			=     do_lower_case






     def     __A  (		self				,     a__				,     a__=None  ):
       _lowerCAmelCase						:       int  			=     [self.cls_token_id] + token_ids_a + [self.sep_token_id]

       if token_ids_a:
         output += token_ids_a + [self.sep_token_id]

       return output






     def     __A  (		self				,     a__				,     a__ = None  ):
       _lowerCAmelCase						:       List[str]  			=     [self.sep_token_id]
       _lowerCAmelCase						:       Any  			=     [self.cls_token_id]
       if token_ids_a is None:
         return len(cls + token_ids_a + sep  ) * [0]
       return len(cls + token_ids_a + sep  ) * [0] + len(token_ids_a + sep  ) * [1]






     def     __A  (		self				,     a__				,     a__ = None  ):
       _lowerCAmelCase						:       Optional[Any]  			=     self._tokenizer.model.save(a__				,     name=a__  )
       return tuple(a__  )




