


'''simple docstring'''






import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 		a__    (							__A    ,    unittest.TestCase  ):
							lowercase__      :       Optional[Any]     						=				ConsistencyModelPipeline
							lowercase__      :       Optional[int]     						=				UNCONDITIONAL_IMAGE_GENERATION_PARAMS
							lowercase__      :       int     						=				UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

							# Override required_optional_params to remove num_images_per_prompt
							lowercase__      :       List[Any]     						=				frozenset(
							    [
							        "num_inference_steps",
							        "generator",
							        "latents",
							        "output_type",
							        "return_dict",
							        "callback",
							        "callback_steps",
							    ]  )







							@property
							def   __SCREAMING_SNAKE_CASE							(		self       )	->							Dict:
														lowerCAmelCase__	       =     UNetaDModel.from_pretrained(
														    '''diffusers/consistency-models-test''' ,   subfolder='''test_unet''' ,   )
														return unet







							@property
							def   __SCREAMING_SNAKE_CASE							(		self       )	->							int:
														lowerCAmelCase__	       =     UNetaDModel.from_pretrained(
														    '''diffusers/consistency-models-test''' ,   subfolder='''test_unet_class_cond''' ,   )
														return unet







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_=False       )	->							Union[str, Any]:
														if class_cond:
																					lowerCAmelCase__	       =     self.dummy_cond_unet
														else:
																					lowerCAmelCase__	       =     self.dummy_uncond_unet

														# Default to CM multistep sampler
														lowerCAmelCase__	       =     CMStochasticIterativeScheduler(
														    num_train_timesteps=40 ,   sigma_min=0.002 ,   sigma_max=80.0 ,   )

														lowerCAmelCase__	       =     {
														    '''unet''': unet,
														    '''scheduler''': scheduler,
														}

														return components







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_=0       )	->							str:
														if str(__lowercase       ).startswith('''mps'''       ):
																					lowerCAmelCase__	       =     torch.manual_seed(__lowercase       )
														else:
																					lowerCAmelCase__	       =     torch.Generator(device=__lowercase       ).manual_seed(__lowercase       )

														lowerCAmelCase__	       =     {
														    '''batch_size''': 1,
														    '''num_inference_steps''': None,
														    '''timesteps''': [22, 0],
														    '''generator''': generator,
														    '''output_type''': '''np''',
														}

														return inputs







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							str:
														lowerCAmelCase__	       =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
														lowerCAmelCase__	       =     self.get_dummy_components()
														lowerCAmelCase__	       =     ConsistencyModelPipeline(**__lowercase       )
														lowerCAmelCase__	       =     pipe.to(__lowercase       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_dummy_inputs(__lowercase       )
														lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 32, 32, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]
														lowerCAmelCase__	       =     np.array([0.3_572, 0.6_273, 0.4_031, 0.3_961, 0.4_321, 0.5_730, 0.5_266, 0.4_780, 0.5_004]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							Optional[Any]:
														lowerCAmelCase__	       =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
														lowerCAmelCase__	       =     self.get_dummy_components(class_cond=__lowercase       )
														lowerCAmelCase__	       =     ConsistencyModelPipeline(**__lowercase       )
														lowerCAmelCase__	       =     pipe.to(__lowercase       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_dummy_inputs(__lowercase       )
														lowerCAmelCase__	       =     0
														lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 32, 32, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]
														lowerCAmelCase__	       =     np.array([0.3_572, 0.6_273, 0.4_031, 0.3_961, 0.4_321, 0.5_730, 0.5_266, 0.4_780, 0.5_004]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							Optional[Any]:
														lowerCAmelCase__	       =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
														lowerCAmelCase__	       =     self.get_dummy_components()
														lowerCAmelCase__	       =     ConsistencyModelPipeline(**__lowercase       )
														lowerCAmelCase__	       =     pipe.to(__lowercase       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_dummy_inputs(__lowercase       )
														lowerCAmelCase__	       =     1
														lowerCAmelCase__	       =     None
														lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 32, 32, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]
														lowerCAmelCase__	       =     np.array([0.5_004, 0.5_004, 0.4_994, 0.5_008, 0.4_976, 0.5_018, 0.4_990, 0.4_982, 0.4_987]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							List[str]:
														lowerCAmelCase__	       =     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
														lowerCAmelCase__	       =     self.get_dummy_components(class_cond=__lowercase       )
														lowerCAmelCase__	       =     ConsistencyModelPipeline(**__lowercase       )
														lowerCAmelCase__	       =     pipe.to(__lowercase       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_dummy_inputs(__lowercase       )
														lowerCAmelCase__	       =     1
														lowerCAmelCase__	       =     None
														lowerCAmelCase__	       =     0
														lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 32, 32, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]
														lowerCAmelCase__	       =     np.array([0.5_004, 0.5_004, 0.4_994, 0.5_008, 0.4_976, 0.5_018, 0.4_990, 0.4_982, 0.4_987]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3




@slow
@require_torch_gpu
class 		a__    (							unittest.TestCase  ):







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							Tuple:
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_=0 ,   lowerCamelCase_=False ,   lowerCamelCase_="cpu" ,   lowerCamelCase_=torch.floataa ,   lowerCamelCase_=(1, 3, 64, 64)       )	->							Tuple:
														lowerCAmelCase__	       =     torch.manual_seed(__lowercase       )

														lowerCAmelCase__	       =     {
														    '''num_inference_steps''': None,
														    '''timesteps''': [22, 0],
														    '''class_labels''': 0,
														    '''generator''': generator,
														    '''output_type''': '''np''',
														}

														if get_fixed_latents:
																					lowerCAmelCase__	       =     self.get_fixed_latents(seed=__lowercase ,   device=__lowercase ,   dtype=__lowercase ,   shape=__lowercase       )
																					lowerCAmelCase__	       =     latents

														return inputs







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_=0 ,   lowerCamelCase_="cpu" ,   lowerCamelCase_=torch.floataa ,   lowerCamelCase_=(1, 3, 64, 64)       )	->							List[Any]:
														if type(__lowercase       ) == str:
																					lowerCAmelCase__	       =     torch.device(__lowercase       )
														lowerCAmelCase__	       =     torch.Generator(device=__lowercase       ).manual_seed(__lowercase       )
														lowerCAmelCase__	       =     randn_tensor(__lowercase ,   generator=__lowercase ,   device=__lowercase ,   dtype=__lowercase       )
														return latents







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							List[Any]:
														lowerCAmelCase__	       =     UNetaDModel.from_pretrained('''diffusers/consistency_models''' ,   subfolder='''diffusers_cd_imagenet64_l2'''       )
														lowerCAmelCase__	       =     CMStochasticIterativeScheduler(
														    num_train_timesteps=40 ,   sigma_min=0.002 ,   sigma_max=80.0 ,   )
														lowerCAmelCase__	       =     ConsistencyModelPipeline(unet=__lowercase ,   scheduler=__lowercase       )
														pipe.to(torch_device=__lowercase       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_inputs()
														lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 64, 64, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]

														lowerCAmelCase__	       =     np.array([0.0_888, 0.0_881, 0.0_666, 0.0_479, 0.0_292, 0.0_195, 0.0_201, 0.0_163, 0.0_254]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 2e-2







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							int:
														lowerCAmelCase__	       =     UNetaDModel.from_pretrained('''diffusers/consistency_models''' ,   subfolder='''diffusers_cd_imagenet64_l2'''       )
														lowerCAmelCase__	       =     CMStochasticIterativeScheduler(
														    num_train_timesteps=40 ,   sigma_min=0.002 ,   sigma_max=80.0 ,   )
														lowerCAmelCase__	       =     ConsistencyModelPipeline(unet=__lowercase ,   scheduler=__lowercase       )
														pipe.to(torch_device=__lowercase       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_inputs()
														lowerCAmelCase__	       =     1
														lowerCAmelCase__	       =     None
														lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 64, 64, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]

														lowerCAmelCase__	       =     np.array([0.0_340, 0.0_152, 0.0_063, 0.0_267, 0.0_221, 0.0_107, 0.0_416, 0.0_186, 0.0_217]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 2e-2







							@require_torch_a
							def   __SCREAMING_SNAKE_CASE							(		self       )	->							Optional[Any]:
														lowerCAmelCase__	       =     UNetaDModel.from_pretrained('''diffusers/consistency_models''' ,   subfolder='''diffusers_cd_imagenet64_l2'''       )
														lowerCAmelCase__	       =     CMStochasticIterativeScheduler(
														    num_train_timesteps=40 ,   sigma_min=0.002 ,   sigma_max=80.0 ,   )
														lowerCAmelCase__	       =     ConsistencyModelPipeline(unet=__lowercase ,   scheduler=__lowercase       )
														pipe.to(torch_device=__lowercase ,   torch_dtype=torch.floataa       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_inputs(get_fixed_latents=__lowercase ,   device=__lowercase       )
														# Ensure usage of flash attention in torch 2.0
														with sdp_kernel(enable_flash=__lowercase ,   enable_math=__lowercase ,   enable_mem_efficient=__lowercase       ):
																					lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 64, 64, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]

														lowerCAmelCase__	       =     np.array([0.1_875, 0.1_428, 0.1_289, 0.2_151, 0.2_092, 0.1_477, 0.1_877, 0.1_641, 0.1_353]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3







							@require_torch_a
							def   __SCREAMING_SNAKE_CASE							(		self       )	->							str:
														lowerCAmelCase__	       =     UNetaDModel.from_pretrained('''diffusers/consistency_models''' ,   subfolder='''diffusers_cd_imagenet64_l2'''       )
														lowerCAmelCase__	       =     CMStochasticIterativeScheduler(
														    num_train_timesteps=40 ,   sigma_min=0.002 ,   sigma_max=80.0 ,   )
														lowerCAmelCase__	       =     ConsistencyModelPipeline(unet=__lowercase ,   scheduler=__lowercase       )
														pipe.to(torch_device=__lowercase ,   torch_dtype=torch.floataa       )
														pipe.set_progress_bar_config(disable=__lowercase       )

														lowerCAmelCase__	       =     self.get_inputs(get_fixed_latents=__lowercase ,   device=__lowercase       )
														lowerCAmelCase__	       =     1
														lowerCAmelCase__	       =     None
														# Ensure usage of flash attention in torch 2.0
														with sdp_kernel(enable_flash=__lowercase ,   enable_math=__lowercase ,   enable_mem_efficient=__lowercase       ):
																					lowerCAmelCase__	       =     pipe(**__lowercase       ).images
														assert image.shape == (1, 64, 64, 3)

														lowerCAmelCase__	       =     image[0, -3:, -3:, -1]

														lowerCAmelCase__	       =     np.array([0.1_663, 0.1_948, 0.2_275, 0.1_680, 0.1_204, 0.1_245, 0.1_858, 0.1_338, 0.2_095]       )

														assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1e-3



'''simple docstring'''






import json
from typing import List, Optional, Tuple

from tokenizers import normalizers
from tokenizers.pre_tokenizers import BertPreTokenizer, PreTokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roformer import RoFormerTokenizer
from .tokenization_utils import JiebaPreTokenizer


__UpperCAmelCase       =							logging.get_logger(__name__)

__UpperCAmelCase       =							{'''vocab_file''': '''vocab.txt''', '''tokenizer_file''': '''tokenizer.json'''}

__UpperCAmelCase       =							{
    '''vocab_file''': {
        '''junnyu/roformer_chinese_small''': '''https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/vocab.txt''',
        '''junnyu/roformer_chinese_base''': '''https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/vocab.txt''',
        '''junnyu/roformer_chinese_char_small''': (
            '''https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/vocab.txt'''
        ),
        '''junnyu/roformer_chinese_char_base''': (
            '''https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/vocab.txt'''
        ),
        '''junnyu/roformer_small_discriminator''': (
            '''https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/vocab.txt'''
        ),
        '''junnyu/roformer_small_generator''': (
            '''https://huggingface.co/junnyu/roformer_small_generator/resolve/main/vocab.txt'''
        ),
    }
}

__UpperCAmelCase       =							{
    '''junnyu/roformer_chinese_small''': 1_536,
    '''junnyu/roformer_chinese_base''': 1_536,
    '''junnyu/roformer_chinese_char_small''': 512,
    '''junnyu/roformer_chinese_char_base''': 512,
    '''junnyu/roformer_small_discriminator''': 128,
    '''junnyu/roformer_small_generator''': 128,
}


__UpperCAmelCase       =							{
    '''junnyu/roformer_chinese_small''': {'''do_lower_case''': True},
    '''junnyu/roformer_chinese_base''': {'''do_lower_case''': True},
    '''junnyu/roformer_chinese_char_small''': {'''do_lower_case''': True},
    '''junnyu/roformer_chinese_char_base''': {'''do_lower_case''': True},
    '''junnyu/roformer_small_discriminator''': {'''do_lower_case''': True},
    '''junnyu/roformer_small_generator''': {'''do_lower_case''': True},
}




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       int     						=				VOCAB_FILES_NAMES
							lowercase__      :       Optional[int]     						=				PRETRAINED_VOCAB_FILES_MAP
							lowercase__      :       Any     						=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowercase__      :       Union[str, Any]     						=				PRETRAINED_INIT_CONFIGURATION
							lowercase__      :       Tuple     						=				RoFormerTokenizer







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

														lowerCAmelCase__	       =     json.loads(self.backend_tokenizer.normalizer.__getstate__()       )
														if (
														    pre_tok_state.get('''lowercase''' ,   lowerCamelCase_       ) != do_lower_case
														    or pre_tok_state.get('''strip_accents''' ,   lowerCamelCase_       ) != strip_accents
														):
																					lowerCAmelCase__	       =     getattr(lowerCamelCase_ ,   pre_tok_state.pop('''type'''       )       )
																					lowerCAmelCase__	       =     do_lower_case
																					lowerCAmelCase__	       =     strip_accents
																					lowerCAmelCase__	       =     pre_tok_class(**lowerCamelCase_       )

														lowerCAmelCase__	       =     do_lower_case







							def __getstate__(		self       )	->							Any:
														lowerCAmelCase__	       =     self.__dict__.copy()
														lowerCAmelCase__	       =     BertPreTokenizer()
														return state







							def __setstate__(		self ,   lowerCamelCase_       )	->							List[Any]:
														lowerCAmelCase__	       =     d
														lowerCAmelCase__	       =     self.__dict__['''_tokenizer'''].get_vocab()
														lowerCAmelCase__	       =     PreTokenizer.custom(JiebaPreTokenizer(lowerCamelCase_       )       )







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_=None       )	->							Union[str, Any]:
														lowerCAmelCase__	       =     [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   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = None       )	->							List[int]:
														lowerCAmelCase__	       =     [self.sep_token_id]
														lowerCAmelCase__	       =     [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   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = None       )	->							Tuple[str]:
														lowerCAmelCase__	       =     self._tokenizer.model.save(lowerCamelCase_ ,   name=lowerCamelCase_       )
														return tuple(lowerCamelCase_       )







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_=None ,   lowerCamelCase_=None ,   lowerCamelCase_=False ,   **lowerCamelCase_ ,   )	->							Union[str, Any]:
														lowerCAmelCase__	       =     BertPreTokenizer()
														return super().save_pretrained(lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_ ,   **lowerCamelCase_       )