


import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_snake_case  :	Optional[Any]			      =       logging.get_logger(__name__)

_snake_case  :	Optional[int]			      =       '''▁'''

_snake_case  :	Dict			      =       {
    '''vocab_file''': '''vocab.json''',
    '''spm_file''': '''sentencepiece.bpe.model''',
}

_snake_case  :	Tuple			      =       {
    '''vocab_file''': {
        '''facebook/s2t-small-librispeech-asr''': (
            '''https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/vocab.json'''
        ),
    },
    '''spm_file''': {
        '''facebook/s2t-small-librispeech-asr''': (
            '''https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/sentencepiece.bpe.model'''
        )
    },
}

_snake_case  :	Dict			      =       {
    '''facebook/s2t-small-librispeech-asr''': 1_024,
}

_snake_case  :	Optional[int]			      =       ['''pt''', '''fr''', '''ru''', '''nl''', '''ro''', '''it''', '''es''', '''de''']

_snake_case  :	List[Any]			      =       {'''mustc''': MUSTC_LANGS}

class        a	(_UpperCAmelCase   ):




							"""simple docstring"""


							__UpperCAmelCase					:							Optional[int]  						=						VOCAB_FILES_NAMES
							__UpperCAmelCase					:							Tuple  						=						PRETRAINED_VOCAB_FILES_MAP
							__UpperCAmelCase					:							Optional[int]  						=						MAX_MODEL_INPUT_SIZES
							__UpperCAmelCase					:							Tuple  						=						['input_ids', 'attention_mask']

							__UpperCAmelCase					:							List[int]  						=						[]

							def __init__(							self :						Any					,			lowerCamelCase :						Union[str, Any]					,			lowerCamelCase :						Dict					,			lowerCamelCase :						List[str]="<s>"					,			lowerCamelCase :						Tuple="</s>"					,			lowerCamelCase :						Any="<pad>"					,			lowerCamelCase :						Optional[int]="<unk>"					,			lowerCamelCase :						Any=False					,			lowerCamelCase :						Dict=False					,			lowerCamelCase :						Optional[Any]=None					,			lowerCamelCase :						Tuple=None					,			lowerCamelCase :						Optional[Dict[str, Any]] = None					,			**lowerCamelCase :						List[str]					,			)			->					None:
														__snake_case	:    int											=     {} if sp_model_kwargs is None else sp_model_kwargs

														super().__init__(
														    bos_token=__UpperCamelCase					,			eos_token=__UpperCamelCase					,			unk_token=__UpperCamelCase					,			pad_token=__UpperCamelCase					,			do_upper_case=__UpperCamelCase					,			do_lower_case=__UpperCamelCase					,			tgt_lang=__UpperCamelCase					,			lang_codes=__UpperCamelCase					,			sp_model_kwargs=self.sp_model_kwargs					,			**__UpperCamelCase					,			)
														__snake_case	:    List[Any]											=     do_upper_case
														__snake_case	:    Optional[int]											=     do_lower_case

														__snake_case	:    Optional[int]											=     load_json(__UpperCamelCase       )
														__snake_case	:    Any											=     {v: k for k, v in self.encoder.items()}
														__snake_case	:    Optional[Any]											=     spm_file
														__snake_case	:    Union[str, Any]											=     load_spm(__UpperCamelCase					,			self.sp_model_kwargs       )

														if lang_codes is not None:
																					__snake_case	:    Optional[int]											=     lang_codes
																					__snake_case	:    int											=     LANGUAGES[lang_codes]
																					__snake_case	:    List[Any]											=     [F'<lang:{lang}>' for lang in self.langs]
																					__snake_case	:    List[Any]											=     {lang: self.sp_model.PieceToId(F'<lang:{lang}>'       ) for lang in self.langs}

																					__snake_case	:    List[Any]											=     self.lang_tokens
																					__snake_case	:    Dict											=     tgt_lang if tgt_lang is not None else self.langs[0]

																					self.set_tgt_lang_special_tokens(self._tgt_lang       )
														else:
																					__snake_case	:    Any											=     {}

							@property
							def 					UpperCamelCase      (							self :						int       )			->					int:
														return len(self.encoder       )

							@property
							def 					UpperCamelCase      (							self :						List[str]       )			->					str:
														return self._tgt_lang

							@tgt_lang.setter
							def 					UpperCamelCase      (							self :						int					,			lowerCamelCase :						Any       )			->					None:
														__snake_case	:    Union[str, Any]											=     new_tgt_lang
														self.set_tgt_lang_special_tokens(__UpperCamelCase       )

							def 					UpperCamelCase      (							self :						Optional[Any]					,			lowerCamelCase :						str       )			->					None:
														__snake_case	:    Optional[Any]											=     self.lang_code_to_id[tgt_lang]
														__snake_case	:    str											=     [lang_code_id]

							def 					UpperCamelCase      (							self :						List[Any]					,			lowerCamelCase :						str       )			->					List[str]:
														return self.sp_model.encode(__UpperCamelCase					,			out_type=__UpperCamelCase       )

							def 					UpperCamelCase      (							self :						str					,			lowerCamelCase :						List[str]       )			->					Optional[int]:
														return self.encoder.get(__UpperCamelCase					,			self.encoder[self.unk_token]       )

							def 					UpperCamelCase      (							self :						Optional[int]					,			lowerCamelCase :						int       )			->					str:
														return self.decoder.get(__UpperCamelCase					,			self.unk_token       )

							def 					UpperCamelCase      (							self :						Union[str, Any]					,			lowerCamelCase :						List[str]       )			->					str:
														__snake_case	:    List[str]											=     []
														__snake_case	:    Any											=     ""
														for token in tokens:
																					# make sure that special tokens are not decoded using sentencepiece model
																					if token in self.all_special_tokens:
																												__snake_case	:    List[str]											=     self.sp_model.decode(__UpperCamelCase       )
																												out_string += (decoded.upper() if self.do_upper_case else decoded) + token + " "
																												__snake_case	:    Any											=     []
																					else:
																												current_sub_tokens.append(__UpperCamelCase       )
														__snake_case	:    Optional[Any]											=     self.sp_model.decode(__UpperCamelCase       )
														out_string += decoded.upper() if self.do_upper_case else decoded
														return out_string.strip()

							def 					UpperCamelCase      (							self :						Optional[int]					,			lowerCamelCase :						Optional[int]					,			lowerCamelCase :						Optional[Any]=None       )			->					List[int]:






														if token_ids_a is None:
																					return self.prefix_tokens + token_ids_a + [self.eos_token_id]
														# We don't expect to process pairs, but leave the pair logic for API consistency
														return self.prefix_tokens + token_ids_a + token_ids_a + [self.eos_token_id]

							def 					UpperCamelCase      (							self :						Dict					,			lowerCamelCase :						List[int]					,			lowerCamelCase :						Optional[List[int]] = None					,			lowerCamelCase :						bool = False       )			->					List[int]:






														if already_has_special_tokens:
																					return super().get_special_tokens_mask(
																					    token_ids_a=__UpperCamelCase					,			token_ids_a=__UpperCamelCase					,			already_has_special_tokens=__UpperCamelCase       )

														__snake_case	:    str											=     [1] * len(self.prefix_tokens       )
														__snake_case	:    Dict											=     [1]
														if token_ids_a is None:
																					return prefix_ones + ([0] * len(__UpperCamelCase       )) + suffix_ones
														return prefix_ones + ([0] * len(__UpperCamelCase       )) + ([0] * len(__UpperCamelCase       )) + suffix_ones

							def 					UpperCamelCase      (							self :						int       )			->					Dict:
														__snake_case	:    int											=     self.encoder.copy()
														vocab.update(self.added_tokens_encoder       )
														return vocab

							def __getstate__(							self :						Any       )			->					Dict:
														__snake_case	:    Optional[Any]											=     self.__dict__.copy()
														__snake_case	:    Any											=     None
														return state

							def __setstate__(							self :						Union[str, Any]					,			lowerCamelCase :						Dict       )			->					None:
														__snake_case	:    str											=     d

														# for backward compatibility
														if not hasattr(self					,			"sp_model_kwargs"       ):
																					__snake_case	:    str											=     {}

														__snake_case	:    Dict											=     load_spm(self.spm_file					,			self.sp_model_kwargs       )






							def 					UpperCamelCase      (							self :						List[str]					,			lowerCamelCase :						str					,			lowerCamelCase :						Optional[str] = None       )			->					Tuple[str]:
														__snake_case	:    int											=     Path(__UpperCamelCase       )
														assert save_dir.is_dir(), F'{save_directory} should be a directory'
														__snake_case	:    Tuple											=     save_dir / (
														    (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["vocab_file"]
														)
														__snake_case	:    Union[str, Any]											=     save_dir / (
														    (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["spm_file"]
														)

														save_json(self.encoder					,			__UpperCamelCase       )

														if os.path.abspath(self.spm_file       ) != os.path.abspath(__UpperCamelCase       ) and os.path.isfile(self.spm_file       ):
																					copyfile(self.spm_file					,			__UpperCamelCase       )
														elif not os.path.isfile(self.spm_file       ):
																					with open(__UpperCamelCase					,			"wb"       ) as fi:
																												__snake_case	:    List[Any]											=     self.sp_model.serialized_model_proto()
																												fi.write(__UpperCamelCase       )

														return (str(__UpperCamelCase       ), str(__UpperCamelCase       ))







def        lowerCAmelCase_  (      __lowerCamelCase							,			__lowerCamelCase			):
							__snake_case	:    Any											=     sentencepiece.SentencePieceProcessor(**_A			)
							spm.Load(str(_A			)			)
							return spm







def        lowerCAmelCase_  (      __lowerCamelCase			):

							with open(_A							,			"r"			) as f:
														return json.load(_A			)







def        lowerCAmelCase_  (      __lowerCamelCase							,			__lowerCamelCase			):

							with open(_A							,			"w"			) as f:
														json.dump(_A							,			_A							,			indent=2			)





import gc
import unittest

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

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

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


enable_full_determinism()

class        a	(_lowerCAmelCase  ,       unittest.TestCase   ):




							"""simple docstring"""


							__UpperCAmelCase					:							Optional[Any]  						=						LDMTextToImagePipeline
							__UpperCAmelCase					:							List[Any]  						=						TEXT_TO_IMAGE_PARAMS - {
							    "negative_prompt",
							    "negative_prompt_embeds",
							    "cross_attention_kwargs",
							    "prompt_embeds",
							}
							__UpperCAmelCase					:							int  						=						PipelineTesterMixin.required_optional_params - {
							    "num_images_per_prompt",
							    "callback",
							    "callback_steps",
							}
							__UpperCAmelCase					:							Any  						=						TEXT_TO_IMAGE_BATCH_PARAMS
							__UpperCAmelCase					:							int  						=						False

							def 					__snake_case      (							self :						Union[str, Any]       )			->					str:
														torch.manual_seed(0       )
														__snake_case	:    List[Any]											=     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	:    Any											=     DDIMScheduler(
														    beta_start=0.0_00_85					,			beta_end=0.0_12					,			beta_schedule="scaled_linear"					,			clip_sample=lowerCamelCase					,			set_alpha_to_one=lowerCamelCase					,			)
														torch.manual_seed(0       )
														__snake_case	:    Dict											=     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	:    Any											=     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=1000					,			)
														__snake_case	:    Any											=     CLIPTextModel(lowerCamelCase       )
														__snake_case	:    Dict											=     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       )

														__snake_case	:    List[str]											=     {
														    "unet": unet,
														    "scheduler": scheduler,
														    "vqvae": vae,
														    "bert": text_encoder,
														    "tokenizer": tokenizer,
														}
														return components

							def 					__snake_case      (							self :						List[Any]					,			lowerCamelCase :						str					,			lowerCamelCase :						Any=0       )			->					str:
														if str(lowerCamelCase       ).startswith("mps"       ):
																					__snake_case	:    Union[str, Any]											=     torch.manual_seed(lowerCamelCase       )
														else:
																					__snake_case	:    str											=     torch.Generator(device=lowerCamelCase       ).manual_seed(lowerCamelCase       )
														__snake_case	:    Dict											=     {
														    "prompt": "A painting of a squirrel eating a burger",
														    "generator": generator,
														    "num_inference_steps": 2,
														    "guidance_scale": 6.0,
														    "output_type": "numpy",
														}
														return inputs






							def 					__snake_case      (							self :						Dict       )			->					Optional[Any]:
														__snake_case	:    List[Any]											=     "cpu"  # ensure determinism for the device-dependent torch.Generator

														__snake_case	:    Union[str, Any]											=     self.get_dummy_components()
														__snake_case	:    int											=     LDMTextToImagePipeline(**lowerCamelCase       )
														pipe.to(lowerCamelCase       )
														pipe.set_progress_bar_config(disable=lowerCamelCase       )

														__snake_case	:    Optional[Any]											=     self.get_dummy_inputs(lowerCamelCase       )
														__snake_case	:    Optional[Any]											=     pipe(**lowerCamelCase       ).images
														__snake_case	:    Optional[int]											=     image[0, -3:, -3:, -1]

														assert image.shape == (1, 16, 16, 3)
														__snake_case	:    int											=     np.array([0.61_01, 0.61_56, 0.56_22, 0.48_95, 0.66_61, 0.38_04, 0.57_48, 0.61_36, 0.50_14]       )

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

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




							"""simple docstring"""


							def 					__snake_case      (							self :						int       )			->					str:
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()

							def 					__snake_case      (							self :						Optional[int]					,			lowerCamelCase :						Dict					,			lowerCamelCase :						Any=torch.floataa					,			lowerCamelCase :						List[Any]=0       )			->					Dict:
														__snake_case	:    List[Any]											=     torch.manual_seed(lowerCamelCase       )
														__snake_case	:    Tuple											=     np.random.RandomState(lowerCamelCase       ).standard_normal((1, 4, 32, 32)       )
														__snake_case	:    int											=     torch.from_numpy(lowerCamelCase       ).to(device=lowerCamelCase					,			dtype=lowerCamelCase       )
														__snake_case	:    List[str]											=     {
														    "prompt": "A painting of a squirrel eating a burger",
														    "latents": latents,
														    "generator": generator,
														    "num_inference_steps": 3,
														    "guidance_scale": 6.0,
														    "output_type": "numpy",
														}
														return inputs






							def 					__snake_case      (							self :						Tuple       )			->					Optional[Any]:
														__snake_case	:    str											=     LDMTextToImagePipeline.from_pretrained("CompVis/ldm-text2im-large-256"       ).to(lowerCamelCase       )
														pipe.set_progress_bar_config(disable=lowerCamelCase       )

														__snake_case	:    List[str]											=     self.get_inputs(lowerCamelCase       )
														__snake_case	:    str											=     pipe(**lowerCamelCase       ).images
														__snake_case	:    Dict											=     image[0, -3:, -3:, -1].flatten()

														assert image.shape == (1, 256, 256, 3)
														__snake_case	:    int											=     np.array([0.5_18_25, 0.5_28_50, 0.5_25_43, 0.5_42_58, 0.5_23_04, 0.5_25_69, 0.5_43_63, 0.5_52_76, 0.5_68_78]       )
														__snake_case	:    List[str]											=     np.abs(expected_slice - image_slice       ).max()
														assert max_diff < 1E-3

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




							"""simple docstring"""


							def 					__snake_case      (							self :						List[Any]       )			->					List[Any]:
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()

							def 					__snake_case      (							self :						Any					,			lowerCamelCase :						Dict					,			lowerCamelCase :						Any=torch.floataa					,			lowerCamelCase :						List[Any]=0       )			->					Optional[Any]:
														__snake_case	:    int											=     torch.manual_seed(lowerCamelCase       )
														__snake_case	:    Tuple											=     np.random.RandomState(lowerCamelCase       ).standard_normal((1, 4, 32, 32)       )
														__snake_case	:    Optional[Any]											=     torch.from_numpy(lowerCamelCase       ).to(device=lowerCamelCase					,			dtype=lowerCamelCase       )
														__snake_case	:    List[Any]											=     {
														    "prompt": "A painting of a squirrel eating a burger",
														    "latents": latents,
														    "generator": generator,
														    "num_inference_steps": 50,
														    "guidance_scale": 6.0,
														    "output_type": "numpy",
														}
														return inputs






							def 					__snake_case      (							self :						List[str]       )			->					int:
														__snake_case	:    Union[str, Any]											=     LDMTextToImagePipeline.from_pretrained("CompVis/ldm-text2im-large-256"       ).to(lowerCamelCase       )
														pipe.set_progress_bar_config(disable=lowerCamelCase       )

														__snake_case	:    Optional[Any]											=     self.get_inputs(lowerCamelCase       )
														__snake_case	:    str											=     pipe(**lowerCamelCase       ).images[0]

														__snake_case	:    Tuple											=     load_numpy(
														    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/ldm_text2img/ldm_large_256_ddim.npy"       )
														__snake_case	:    str											=     np.abs(expected_image - image       ).max()
														assert max_diff < 1E-3

