
'''simple docstring'''





import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class 						a   (				__magic_name__				,__magic_name__				,unittest.TestCase    ):
				_snake_case		    = StableDiffusionSAGPipeline
				_snake_case		    = TEXT_TO_IMAGE_PARAMS
				_snake_case		    = TEXT_TO_IMAGE_BATCH_PARAMS
				_snake_case		    = TEXT_TO_IMAGE_IMAGE_PARAMS
				_snake_case		    = TEXT_TO_IMAGE_IMAGE_PARAMS
				_snake_case		    = False




				def 	__snake_case							(							self	:				Optional[Any]      ):
										torch.manual_seed(0      )
										snake_case       :     Dict =    UNetaDConditionModel(
										    block_out_channels=(32, 64),    layers_per_block=2,    sample_size=32,    in_channels=4,    out_channels=4,    down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D'''),    up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D'''),    cross_attention_dim=32,    )
										snake_case       :     List[str] =    DDIMScheduler(
										    beta_start=0.00085,    beta_end=0.012,    beta_schedule='''scaled_linear''',    clip_sample=SCREAMING_SNAKE_CASE_,    set_alpha_to_one=SCREAMING_SNAKE_CASE_,    )
										torch.manual_seed(0      )
										snake_case       :     Optional[Any] =    AutoencoderKL(
										    block_out_channels=[32, 64],    in_channels=3,    out_channels=3,    down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''],    up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''],    latent_channels=4,    )
										torch.manual_seed(0      )
										snake_case       :     int =    CLIPTextConfig(
										    bos_token_id=0,    eos_token_id=2,    hidden_size=32,    intermediate_size=37,    layer_norm_eps=1e-05,    num_attention_heads=4,    num_hidden_layers=5,    pad_token_id=1,    vocab_size=10_00,    )
										snake_case       :     Dict =    CLIPTextModel(SCREAMING_SNAKE_CASE_      )
										snake_case       :     str =    CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

										snake_case       :     str =    {
										    '''unet''': unet,
										    '''scheduler''': scheduler,
										    '''vae''': vae,
										    '''text_encoder''': text_encoder,
										    '''tokenizer''': tokenizer,
										    '''safety_checker''': None,
										    '''feature_extractor''': None,
										}
										return components




				def 	__snake_case							(							self	:				Any,    SCREAMING_SNAKE_CASE_	:				str,    SCREAMING_SNAKE_CASE_	:				Optional[int]=0      ):
										if str(SCREAMING_SNAKE_CASE_      ).startswith('''mps'''      ):
																snake_case       :     Optional[Any] =    torch.manual_seed(SCREAMING_SNAKE_CASE_      )
										else:
																snake_case       :     int =    torch.Generator(device=SCREAMING_SNAKE_CASE_      ).manual_seed(SCREAMING_SNAKE_CASE_      )
										snake_case       :     int =    {
										    '''prompt''': '''.''',
										    '''generator''': generator,
										    '''num_inference_steps''': 2,
										    '''guidance_scale''': 1.0,
										    '''sag_scale''': 1.0,
										    '''output_type''': '''numpy''',
										}
										return inputs




				def 	__snake_case							(							self	:				List[Any]      ):
										super().test_inference_batch_single_identical(expected_max_diff=3e-3      )




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




				def 	__snake_case							(							self	:				Union[str, Any]      ):
										# clean up the VRAM after each test
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()




				def 	__snake_case							(							self	:				Dict      ):
										snake_case       :     List[Any] =    StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''      )
										snake_case       :     Tuple =    sag_pipe.to(SCREAMING_SNAKE_CASE_      )
										sag_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

										snake_case       :     List[Any] =    '''.'''
										snake_case       :     Tuple =    torch.manual_seed(0      )
										snake_case       :     Optional[int] =    sag_pipe(
										    [prompt],    generator=SCREAMING_SNAKE_CASE_,    guidance_scale=7.5,    sag_scale=1.0,    num_inference_steps=20,    output_type='''np'''      )

										snake_case       :     Union[str, Any] =    output.images

										snake_case       :     int =    image[0, -3:, -3:, -1]

										assert image.shape == (1, 5_12, 5_12, 3)
										snake_case       :     int =    np.array([0.1568, 0.1738, 0.1695, 0.1693, 0.1507, 0.1705, 0.1547, 0.1751, 0.1949]      )

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




				def 	__snake_case							(							self	:				str      ):
										snake_case       :     Optional[Any] =    StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''      )
										snake_case       :     Dict =    sag_pipe.to(SCREAMING_SNAKE_CASE_      )
										sag_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

										snake_case       :     Tuple =    '''.'''
										snake_case       :     Union[str, Any] =    torch.manual_seed(0      )
										snake_case       :     str =    sag_pipe(
										    [prompt],    generator=SCREAMING_SNAKE_CASE_,    guidance_scale=7.5,    sag_scale=1.0,    num_inference_steps=20,    output_type='''np'''      )

										snake_case       :     Union[str, Any] =    output.images

										snake_case       :     Optional[int] =    image[0, -3:, -3:, -1]

										assert image.shape == (1, 5_12, 5_12, 3)
										snake_case       :     str =    np.array([0.3459, 0.2876, 0.2537, 0.3002, 0.2671, 0.2160, 0.3026, 0.2262, 0.2371]      )

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




				def 	__snake_case							(							self	:				Any      ):
										snake_case       :     Tuple =    StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''      )
										snake_case       :     int =    sag_pipe.to(SCREAMING_SNAKE_CASE_      )
										sag_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

										snake_case       :     str =    '''.'''
										snake_case       :     List[Any] =    torch.manual_seed(0      )
										snake_case       :     Tuple =    sag_pipe(
										    [prompt],    width=7_68,    height=5_12,    generator=SCREAMING_SNAKE_CASE_,    guidance_scale=7.5,    sag_scale=1.0,    num_inference_steps=20,    output_type='''np''',    )

										snake_case       :     Optional[int] =    output.images

										assert image.shape == (1, 5_12, 7_68, 3)



'''simple docstring'''





import json
import re
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
						if is_torch_available():
												import torch
						if is_tf_available():
												import tensorflow as tf

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_codegen import CodeGenTokenizer


UpperCAmelCase    					=   logging.get_logger(__name__)

UpperCAmelCase    					=   {"vocab_file": "vocab.json", "merges_file": "merges.txt", "tokenizer_file": "tokenizer.json"}

UpperCAmelCase    					=   {
    "vocab_file": {
        "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json",
    },
    "merges_file": {
        "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt",
    },
    "tokenizer_file": {
        "Salesforce/codegen-350M-mono": (
            "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/tokenizer.json"
        ),
    },
}

UpperCAmelCase    					=   {
    "Salesforce/codegen-350M-mono": 2_048,
}

class 						a   (				__magic_name__    ):
				_snake_case		    = VOCAB_FILES_NAMES
				_snake_case		    = PRETRAINED_VOCAB_FILES_MAP
				_snake_case		    = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				_snake_case		    = ['''input_ids''', '''attention_mask''']
				_snake_case		    = CodeGenTokenizer




				def __init__(							self	:				Any,    SCREAMING_SNAKE_CASE_	:				Dict=None,    SCREAMING_SNAKE_CASE_	:				Optional[Any]=None,    SCREAMING_SNAKE_CASE_	:				Optional[Any]=None,    SCREAMING_SNAKE_CASE_	:				Optional[int]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				List[Any]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				List[str]="<|endoftext|>",    SCREAMING_SNAKE_CASE_	:				int=False,    **SCREAMING_SNAKE_CASE_	:				Optional[int],    ):
										super().__init__(
										    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    tokenizer_file=SCREAMING_SNAKE_CASE_,    unk_token=SCREAMING_SNAKE_CASE_,    bos_token=SCREAMING_SNAKE_CASE_,    eos_token=SCREAMING_SNAKE_CASE_,    add_prefix_space=SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_,    )

										if kwargs.pop('''add_bos_token''',    SCREAMING_SNAKE_CASE_      ):
																snake_case       :     Dict =    kwargs.pop('''name_or_path''',    ''''''      )
																raise ValueError(
																    '''Currenty GPT2\'s fast tokenizer does NOT support adding a BOS token.'''
																    '''Instead you should use GPT2\'s slow tokenizer class `CodeGenTokenizer` as follows: \n'''
																    F"""`CodeGenTokenizer.from_pretrained('{model_id}')`\nor\n"""
																    F"""`AutoTokenizer.from_pretrained('{model_id}', use_fast=False)`\n"""
																    '''This issue will be fixed soon, see: https://github.com/huggingface/tokenizers/pull/1005.'''
																    ''' so that the fast tokenizer works correctly.'''      )

										snake_case       :     int =    json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()      )
										if pre_tok_state.get('''add_prefix_space''',    SCREAMING_SNAKE_CASE_      ) != add_prefix_space:
																snake_case       :     Optional[Any] =    getattr(SCREAMING_SNAKE_CASE_,    pre_tok_state.pop('''type'''      )      )
																snake_case       :     Union[str, Any] =    add_prefix_space
																snake_case       :     Optional[Any] =    pre_tok_class(**SCREAMING_SNAKE_CASE_      )

										snake_case       :     int =    add_prefix_space




				def 	__snake_case							(							self	:				List[Any],    *SCREAMING_SNAKE_CASE_	:				Tuple,    **SCREAMING_SNAKE_CASE_	:				int      ):
										snake_case       :     Dict =    kwargs.get('''is_split_into_words''',    SCREAMING_SNAKE_CASE_      )
										assert self.add_prefix_space or not is_split_into_words, (
										    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._batch_encode_plus(*SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				List[str],    *SCREAMING_SNAKE_CASE_	:				str,    **SCREAMING_SNAKE_CASE_	:				Optional[int]      ):
										snake_case       :     Dict =    kwargs.get('''is_split_into_words''',    SCREAMING_SNAKE_CASE_      )

										assert self.add_prefix_space or not is_split_into_words, (
										    F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
										    "to use it with pretokenized inputs."
										)

										return super()._encode_plus(*SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Tuple,    SCREAMING_SNAKE_CASE_	:				str,    SCREAMING_SNAKE_CASE_	:				Optional[str] = None      ):
										snake_case       :     Optional[int] =    self._tokenizer.model.save(SCREAMING_SNAKE_CASE_,    name=SCREAMING_SNAKE_CASE_      )
										return tuple(SCREAMING_SNAKE_CASE_      )




				def 	__snake_case							(							self	:				Dict,    SCREAMING_SNAKE_CASE_	:				Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"],    SCREAMING_SNAKE_CASE_	:				bool = False,    SCREAMING_SNAKE_CASE_	:				bool = None,    SCREAMING_SNAKE_CASE_	:				Optional[List[str]] = None,    **SCREAMING_SNAKE_CASE_	:				Union[str, Any],    ):
										snake_case       :     Dict =    super().decode(
										    token_ids=SCREAMING_SNAKE_CASE_,    skip_special_tokens=SCREAMING_SNAKE_CASE_,    clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_,    )

										if truncate_before_pattern is not None and len(SCREAMING_SNAKE_CASE_      ) > 0:
																snake_case       :     Optional[int] =    self.truncate(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      )

										return decoded_text




				def 	__snake_case							(							self	:				int,    SCREAMING_SNAKE_CASE_	:				Optional[Any],    SCREAMING_SNAKE_CASE_	:				Tuple      ):
										def find_re(SCREAMING_SNAKE_CASE_	:				Optional[int],    SCREAMING_SNAKE_CASE_	:				Dict,    SCREAMING_SNAKE_CASE_	:				Optional[Any]      ):
																snake_case       :     Optional[Any] =    pattern.search(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      )
																return m.start() if m else -1

										snake_case       :     Union[str, Any] =    [re.compile(SCREAMING_SNAKE_CASE_,    re.MULTILINE      ) for pattern in truncate_before_pattern]

										snake_case       :     Union[str, Any] =    list(re.finditer('''^print''',    SCREAMING_SNAKE_CASE_,    re.MULTILINE      )      )

										if len(SCREAMING_SNAKE_CASE_      ) > 1:
																snake_case       :     Tuple =    completion[: prints[1].start()]

										snake_case       :     List[str] =    list(re.finditer('''^def''',    SCREAMING_SNAKE_CASE_,    re.MULTILINE      )      )

										if len(SCREAMING_SNAKE_CASE_      ) > 1:
																snake_case       :     Optional[Any] =    completion[: defs[1].start()]

										snake_case       :     Tuple =    0

										snake_case       :     List[Any] =    [
										    pos for pos in [find_re(SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_,    SCREAMING_SNAKE_CASE_      ) for terminal in terminals] if pos != -1
										]

										if len(SCREAMING_SNAKE_CASE_      ) > 0:
																return completion[: min(SCREAMING_SNAKE_CASE_      )]
										else:
																return completion

