






# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion
import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.schedulers.scheduling_utils import SchedulerMixin
from diffusers.utils import BaseOutput, deprecate




@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->DDIM
class     _lowerCAmelCase(	UpperCAmelCase_ ):





       """simple docstring"""







       a    :		torch.FloatTensor
       a    :		Optional[torch.FloatTensor]									=None







def   snake_case (UpperCAmelCase__     ,   UpperCAmelCase__=0.999     ,   UpperCAmelCase__="cosine"     ,   )	->    str:
  if alpha_transform_type == "cosine":

    def alpha_bar_fn(UpperCAmelCase__  ):
      return math.cos((t + 0.008) / 1.008 * math.pi / 2  ) ** 2

  elif alpha_transform_type == "exp":

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

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

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




class     _lowerCAmelCase(	UpperCAmelCase_							,     UpperCAmelCase_ ):





       """simple docstring"""







       a    :		str									=1





       @register_to_config
       def __init__( self  ,      _lowerCamelCase = 1_0_0_0  ,      _lowerCamelCase = 0.0_0_0_1  ,      _lowerCamelCase = 0.0_2  ,      _lowerCamelCase = "linear"  ,      _lowerCamelCase = None  ,      _lowerCamelCase = True  ,      _lowerCamelCase = True  ,      _lowerCamelCase = 0  ,      _lowerCamelCase = "epsilon"  ,      _lowerCamelCase = 1.0  ,      **_lowerCamelCase  ,      ):
         if kwargs.get('set_alpha_to_one'  ,      _lowerCamelCase							) is not None:
           UpperCamelCase_: Dict   				=    (
               'The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead.'
           )
           deprecate('set_alpha_to_one'  ,      '1.0.0'  ,      _lowerCamelCase  ,      standard_warn=_lowerCamelCase							)
           UpperCamelCase_: List[str]   				=    kwargs['set_alpha_to_one']
         if trained_betas is not None:
           UpperCamelCase_: Union[str, Any]   				=    torch.tensor(_lowerCamelCase  ,      dtype=torch.floataa							)
         elif beta_schedule == "linear":
           UpperCamelCase_: Optional[Any]   				=    torch.linspace(_lowerCamelCase  ,      _lowerCamelCase  ,      _lowerCamelCase  ,      dtype=torch.floataa							)
         elif beta_schedule == "scaled_linear":
           # this schedule is very specific to the latent diffusion model.
           UpperCamelCase_: List[str]   				=    (
               torch.linspace(beta_start**0.5  ,      beta_end**0.5  ,      _lowerCamelCase  ,      dtype=torch.floataa							) ** 2
           )
         elif beta_schedule == "squaredcos_cap_v2":
           # Glide cosine schedule
           UpperCamelCase_: Dict   				=    betas_for_alpha_bar(_lowerCamelCase							)
         else:
           raise NotImplementedError(f'''{beta_schedule} does is not implemented for {self.__class__}'''							)

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

         # At every step in inverted ddim, we are looking into the next alphas_cumprod
         # For the final step, there is no next alphas_cumprod, and the index is out of bounds
         # `set_alpha_to_zero` decides whether we set this parameter simply to zero
         # in this case, self.step() just output the predicted noise
         # or whether we use the final alpha of the "non-previous" one.
         UpperCamelCase_: Optional[int]   				=    torch.tensor(0.0							) if set_alpha_to_zero else self.alphas_cumprod[-1]

         # standard deviation of the initial noise distribution
         UpperCamelCase_: Any   				=    1.0

         # setable values
         UpperCamelCase_: List[str]   				=    None
         UpperCamelCase_: Dict   				=    torch.from_numpy(np.arange(0  ,      _lowerCamelCase							).copy().astype(np.intaa							)							)





       def       _a   ( self  ,      _lowerCamelCase  ,      _lowerCamelCase = None							):
         return sample





       def       _a   ( self  ,      _lowerCamelCase  ,      _lowerCamelCase = None							):

         if num_inference_steps > self.config.num_train_timesteps:
           raise ValueError(
               f'''`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:'''
               f''' {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle'''
               f''' maximal {self.config.num_train_timesteps} timesteps.'''							)

         UpperCamelCase_: List[str]   				=    num_inference_steps
         UpperCamelCase_: List[Any]   				=    self.config.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
         UpperCamelCase_: List[Any]   				=    (np.arange(0  ,      _lowerCamelCase							) * step_ratio).round().copy().astype(np.intaa							)
         UpperCamelCase_: Optional[int]   				=    torch.from_numpy(_lowerCamelCase							).to(_lowerCamelCase							)
         self.timesteps += self.config.steps_offset





       def       _a   ( self  ,      _lowerCamelCase  ,      _lowerCamelCase  ,      _lowerCamelCase  ,      _lowerCamelCase = 0.0  ,      _lowerCamelCase = False  ,      _lowerCamelCase = None  ,      _lowerCamelCase = True  ,      ):
         # 1. get previous step value (=t+1)
         UpperCamelCase_: Dict   				=    timestep + self.config.num_train_timesteps // self.num_inference_steps

         # 2. compute alphas, betas
         # change original implementation to exactly match noise levels for analogous forward process
         UpperCamelCase_: Optional[int]   				=    self.alphas_cumprod[timestep]
         UpperCamelCase_: Union[str, Any]   				=    (
             self.alphas_cumprod[prev_timestep]
             if prev_timestep < self.config.num_train_timesteps
             else self.final_alpha_cumprod
         )

         UpperCamelCase_: int   				=    1 - alpha_prod_t

         # 3. compute predicted original sample from predicted noise also called
         # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
         if self.config.prediction_type == "epsilon":
           UpperCamelCase_: str   				=    (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
           UpperCamelCase_: Dict   				=    model_output
         elif self.config.prediction_type == "sample":
           UpperCamelCase_: int   				=    model_output
           UpperCamelCase_: List[Any]   				=    (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
         elif self.config.prediction_type == "v_prediction":
           UpperCamelCase_: Dict   				=    (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
           UpperCamelCase_: str   				=    (alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
         else:
           raise ValueError(
               f'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or'''
               ' `v_prediction`'							)

         # 4. Clip or threshold "predicted x_0"
         if self.config.clip_sample:
           UpperCamelCase_: List[Any]   				=    pred_original_sample.clamp(
               -self.config.clip_sample_range  ,      self.config.clip_sample_range							)

         # 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
         UpperCamelCase_: List[str]   				=    (1 - alpha_prod_t_prev) ** 0.5 * pred_epsilon

         # 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
         UpperCamelCase_: str   				=    alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

         if not return_dict:
           return (prev_sample, pred_original_sample)
         return DDIMSchedulerOutput(prev_sample=_lowerCamelCase  ,      pred_original_sample=_lowerCamelCase							)





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







import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
   from test_module.custom_tokenization_fast import CustomTokenizerFast




class     _lowerCAmelCase(	unittest.TestCase ):





       """simple docstring"""







       def       _a   ( self							):
         # A mock response for an HTTP head request to emulate server down
         UpperCamelCase_: Any   				=    mock.Mock()
         UpperCamelCase_: Dict   				=    5_0_0
         UpperCamelCase_: Any   				=    {}
         UpperCamelCase_: Tuple   				=    HTTPError
         UpperCamelCase_: List[str]   				=    {}

         # Download this model to make sure it's in the cache.
         UpperCamelCase_: int   				=    BertTokenizer.from_pretrained('hf-internal-testing/tiny-random-bert'							)

         # Under the mock environment we get a 500 error when trying to reach the tokenizer.
         with mock.patch('requests.Session.request'  ,      return_value=_lowerCamelCase							) as mock_head:
           UpperCamelCase_: Optional[int]   				=    BertTokenizer.from_pretrained('hf-internal-testing/tiny-random-bert'							)
           # This check we did call the fake head request
           mock_head.assert_called()





       @require_tokenizers
       def       _a   ( self							):
         # A mock response for an HTTP head request to emulate server down
         UpperCamelCase_: Union[str, Any]   				=    mock.Mock()
         UpperCamelCase_: Union[str, Any]   				=    5_0_0
         UpperCamelCase_: str   				=    {}
         UpperCamelCase_: List[str]   				=    HTTPError
         UpperCamelCase_: Optional[int]   				=    {}

         # Download this model to make sure it's in the cache.
         UpperCamelCase_: List[str]   				=    GPTaTokenizerFast.from_pretrained('gpt2'							)

         # Under the mock environment we get a 500 error when trying to reach the tokenizer.
         with mock.patch('requests.Session.request'  ,      return_value=_lowerCamelCase							) as mock_head:
           UpperCamelCase_: str   				=    GPTaTokenizerFast.from_pretrained('gpt2'							)
           # This check we did call the fake head request
           mock_head.assert_called()





       def       _a   ( self							):
         # This test is for deprecated behavior and can be removed in v5
         try:
           UpperCamelCase_: Optional[int]   				=    tempfile.mktemp()
           with open(_lowerCamelCase  ,      'wb'							) as f:
             http_get('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'  ,      _lowerCamelCase							)

           UpperCamelCase_: Tuple   				=    AlbertTokenizer.from_pretrained(_lowerCamelCase							)
         finally:
           os.remove(_lowerCamelCase							)

         # Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
         # the current folder and have the right name.
         if os.path.isfile('tokenizer.json'							):
           # We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
           return
         try:
           with open('tokenizer.json'  ,      'wb'							) as f:
             http_get('https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json'  ,      _lowerCamelCase							)
           UpperCamelCase_: List[str]   				=    AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'							)
           # The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
           self.assertEqual(tokenizer.vocab_size  ,      1_0_0_0							)
           # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

         finally:
           os.remove('tokenizer.json'							)





       def       _a   ( self							):
         # This test is for deprecated behavior and can be removed in v5
         UpperCamelCase_: Any   				=    AlbertTokenizer.from_pretrained('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'							)







@is_staging_test
class     _lowerCAmelCase(	unittest.TestCase ):





       """simple docstring"""







       a    :		Dict									=['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']





       @classmethod
       def       _a   ( cls							):
         UpperCamelCase_: Optional[int]   				=    TOKEN
         HfFolder.save_token(_lowerCamelCase							)





       @classmethod
       def       _a   ( cls							):
         try:
           delete_repo(token=cls._token  ,      repo_id='test-tokenizer'							)
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token  ,      repo_id='valid_org/test-tokenizer-org'							)
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token  ,      repo_id='test-dynamic-tokenizer'							)
         except HTTPError:
           pass





       def       _a   ( self							):
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: List[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)
           UpperCamelCase_: int   				=    BertTokenizer(_lowerCamelCase							)

         tokenizer.push_to_hub('test-tokenizer'  ,      use_auth_token=self._token							)
         UpperCamelCase_: Union[str, Any]   				=    BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)

         # Reset repo
         delete_repo(token=self._token  ,      repo_id='test-tokenizer'							)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           tokenizer.save_pretrained(_lowerCamelCase  ,      repo_id='test-tokenizer'  ,      push_to_hub=_lowerCamelCase  ,      use_auth_token=self._token							)

         UpperCamelCase_: List[str]   				=    BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)





       def       _a   ( self							):
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: Optional[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)
           UpperCamelCase_: Union[str, Any]   				=    BertTokenizer(_lowerCamelCase							)

         tokenizer.push_to_hub('valid_org/test-tokenizer-org'  ,      use_auth_token=self._token							)
         UpperCamelCase_: Dict   				=    BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)

         # Reset repo
         delete_repo(token=self._token  ,      repo_id='valid_org/test-tokenizer-org'							)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           tokenizer.save_pretrained(
               _lowerCamelCase  ,      repo_id='valid_org/test-tokenizer-org'  ,      push_to_hub=_lowerCamelCase  ,      use_auth_token=self._token							)

         UpperCamelCase_: Optional[Any]   				=    BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)





       @require_tokenizers
       def       _a   ( self							):
         CustomTokenizer.register_for_auto_class()
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: Optional[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)
           UpperCamelCase_: Optional[int]   				=    CustomTokenizer(_lowerCamelCase							)

         # No fast custom tokenizer
         tokenizer.push_to_hub('test-dynamic-tokenizer'  ,      use_auth_token=self._token							)

         UpperCamelCase_: str   				=    AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''  ,      trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__  ,      'CustomTokenizer'							)

         # Fast and slow custom tokenizer
         CustomTokenizerFast.register_for_auto_class()
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: Optional[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)

           UpperCamelCase_: Dict   				=    BertTokenizerFast.from_pretrained(_lowerCamelCase							)
           bert_tokenizer.save_pretrained(_lowerCamelCase							)
           UpperCamelCase_: List[str]   				=    CustomTokenizerFast.from_pretrained(_lowerCamelCase							)

         tokenizer.push_to_hub('test-dynamic-tokenizer'  ,      use_auth_token=self._token							)

         UpperCamelCase_: Tuple   				=    AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''  ,      trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__  ,      'CustomTokenizerFast'							)
         UpperCamelCase_: int   				=    AutoTokenizer.from_pretrained(
             f'''{USER}/test-dynamic-tokenizer'''  ,      use_fast=_lowerCamelCase  ,      trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__  ,      'CustomTokenizer'							)







class     _lowerCAmelCase(	unittest.TestCase ):





       """simple docstring"""







       def       _a   ( self							):
         UpperCamelCase_: Dict   				=    Trie()
         trie.add('Hello 友達'							)
         self.assertEqual(trie.data  ,      {'H': {'e': {'l': {'l': {'o': {' ': {'友': {'達': {'': 1}}}}}}}}}							)
         trie.add('Hello'							)
         trie.data
         self.assertEqual(trie.data  ,      {'H': {'e': {'l': {'l': {'o': {'': 1, ' ': {'友': {'達': {'': 1}}}}}}}}}							)





       def       _a   ( self							):
         UpperCamelCase_: Optional[int]   				=    Trie()
         self.assertEqual(trie.split('[CLS] This is a extra_id_100'							)  ,      ['[CLS] This is a extra_id_100']							)
         trie.add('[CLS]'							)
         trie.add('extra_id_1'							)
         trie.add('extra_id_100'							)
         self.assertEqual(trie.split('[CLS] This is a extra_id_100'							)  ,      ['[CLS]', ' This is a ', 'extra_id_100']							)





       def       _a   ( self							):
         UpperCamelCase_: int   				=    Trie()
         trie.add('A'							)
         self.assertEqual(trie.split('ABC'							)  ,      ['A', 'BC']							)
         self.assertEqual(trie.split('BCA'							)  ,      ['BC', 'A']							)





       def       _a   ( self							):
         UpperCamelCase_: str   				=    Trie()
         trie.add('TOKEN]'							)
         trie.add('[SPECIAL_TOKEN]'							)
         self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'							)  ,      ['This is something ', '[SPECIAL_TOKEN]']							)





       def       _a   ( self							):
         UpperCamelCase_: Union[str, Any]   				=    Trie()
         trie.add('A'							)
         trie.add('P'							)
         trie.add('[SPECIAL_TOKEN]'							)
         self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'							)  ,      ['This is something ', '[SPECIAL_TOKEN]']							)





       def       _a   ( self							):
         UpperCamelCase_: List[str]   				=    Trie()
         trie.add('AB'							)
         trie.add('B'							)
         trie.add('C'							)
         self.assertEqual(trie.split('ABC'							)  ,      ['AB', 'C']							)





       def       _a   ( self							):
         UpperCamelCase_: List[str]   				=    Trie()
         trie.add('ABC'							)
         trie.add('B'							)
         trie.add('CD'							)
         self.assertEqual(trie.split('ABCD'							)  ,      ['ABC', 'D']							)





       def       _a   ( self							):
         # Even if the offsets are wrong, we necessarily output correct string
         # parts.
         UpperCamelCase_: Union[str, Any]   				=    Trie()
         UpperCamelCase_: Any   				=    trie.cut_text('ABC'  ,      [0, 0, 2, 1, 2, 3]							)
         self.assertEqual(_lowerCamelCase  ,      ['AB', 'C']							)