
"""simple docstring"""


import json
import os
from typing import Dict, List, Optional, Tuple

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


_lowercase				 =					logging.get_logger(__name__)


_lowercase				 =					{
    '''vocab_file''': '''vocab.json''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
    '''merges_file''': '''merges.txt''',
}

_lowercase				 =					{
    '''vocab_file''': {
        '''facebook/s2t-wav2vec2-large-en-de''': (
            '''https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/vocab.json'''
        ),
    },
    '''tokenizer_config_file''': {
        '''facebook/s2t-wav2vec2-large-en-de''': (
            '''https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/tokenizer_config.json'''
        ),
    },
    '''merges_file''': {
        '''facebook/s2t-wav2vec2-large-en-de''': (
            '''https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/merges.txt'''
        ),
    },
}

_lowercase				 =					'''</w>'''
_lowercase				 =					'''@@ '''






def    _snake_case					(		snake_case__ :  Tuple  ):
	A     = set()
	A     = word[0]
	for char in word[1:]:
		pairs.add((prev_char, char)  )
		A     = char
	return pairs


# Speech2Text2 has no max input length
_lowercase				 =					{'''facebook/s2t-wav2vec2-large-en-de''': 10_24}







class 					lowerCAmelCase_					(   _lowercase     ):



						'''simple docstring'''


						_lowerCamelCase:     Dict   =     VOCAB_FILES_NAMES
						_lowerCamelCase:     Tuple   =     PRETRAINED_VOCAB_FILES_MAP
						_lowerCamelCase:     List[Any]   =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_lowerCamelCase:     Union[str, Any]   =     ['''input_ids''', '''attention_mask''']


						def __init__(			self  :   Any	,A_  :   List[str]	,A_  :   str="<s>"	,A_  :   Optional[int]="<pad>"	,A_  :   List[str]="</s>"	,A_  :   List[Any]="<unk>"	,A_  :   List[str]=False	,A_  :   int=None	,**A_  :   Union[str, Any]	,)  -> Optional[int]:
							super().__init__(
							    unk_token=A_	,bos_token=A_	,eos_token=A_	,pad_token=A_	,do_lower_case=A_	,**A_	,)

							A     = do_lower_case

							with open(A_	,encoding='utf-8'   ) as vocab_handle:
								A     = json.load(A_   )
							A     = {v: k for k, v in self.encoder.items()}

							if merges_file is None:
								logger.info(F'No merges files provided. {self.__class__.__name__} can only be used for decoding.'   )

								A     = None
								A     = None
							else:
								with open(A_	,encoding='utf-8'   ) as merges_handle:
									A     = merges_handle.read().split('\n'   )[:-1]

								A     = [tuple(merge.split()[:2]   ) for merge in merges]
								A     = dict(zip(A_	,range(len(A_   )   )   )   )
								A     = {}


						@property
						def     _SCREAMING_SNAKE_CASE   (			self  :   str   )  -> int:
							return len(self.decoder   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   List[Any]   )  -> Dict:
							return dict(self.encoder	,**self.added_tokens_encoder   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   Any	,A_  :   Optional[Any]   )  -> Tuple:
							A     = tuple(token[:-1]   ) + (token[-1] + BPE_TOKEN_MERGES,)
							if token in self.cache:
								return self.cache[token]
							A     = get_pairs(A_   )

							if not pairs:
								return token

							while True:
								A     = min(A_	,key=lambda A_   : self.bpe_ranks.get(A_	,float('inf'   )   )   )
								if bigram not in self.bpe_ranks:
									break
								A ,   A     = bigram
								A     = []
								A     = 0
								while i < len(A_   ):
									try:
										A     = word.index(A_	,A_   )
									except ValueError:
										new_word.extend(word[i:]   )
										break
									else:
										new_word.extend(word[i:j]   )
										A     = j

									if word[i] == first and i < len(A_   ) - 1 and word[i + 1] == second:
										new_word.append(first + second   )
										i += 2
									else:
										new_word.append(word[i]   )
										i += 1
								A     = tuple(A_   )
								A     = new_word
								if len(A_   ) == 1:
									break
								else:
									A     = get_pairs(A_   )
							A     = ' '.join(A_   )
							if word == "\n  " + BPE_TOKEN_MERGES:
								A     = '\n' + BPE_TOKEN_MERGES

							if word.endswith(A_   ):
								A     = word.replace(A_	,''   )

							A     = word.replace(' '	,A_   )
							A     = word
							return word


						def     _SCREAMING_SNAKE_CASE   (			self  :   Any	,A_  :   str   )  -> List[str]:

							if self.bpe_ranks is None:
								raise ValueError(
								    'This tokenizer was instantiated without a `merges.txt` file, so'
								    ' that it can only be used for decoding, not for encoding.'
								    'Make sure to provide `merges.txt` file at instantiation to enable '
								    'encoding.'   )

							if self.do_lower_case:
								A     = text.lower()

							A     = text.split()

							A     = []
							for token in text:
								if token:
									split_tokens.extend(list(self.bpe(A_   ).split(' '   )   )   )

							return split_tokens


						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[int]	,A_  :   str   )  -> int:
							return self.encoder.get(A_	,self.encoder.get(self.unk_token   )   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[Any]	,A_  :   int   )  -> str:
							A     = self.decoder.get(A_	,self.unk_token   )
							return result


						def     _SCREAMING_SNAKE_CASE   (			self  :   str	,A_  :   List[str]   )  -> str:
							A     = ' '.join(A_   )

							# make sure @@ tokens are concatenated
							A     = ''.join(string.split(A_   )   )

							return string




						def     _SCREAMING_SNAKE_CASE   (			self  :   str	,A_  :   str	,A_  :   Optional[str] = None   )  -> Tuple[str]:
							if not os.path.isdir(A_   ):
								logger.error(F'Vocabulary path ({save_directory}) should be a directory'   )
								return
							A     = os.path.join(
							    A_	,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']   )
							A     = os.path.join(
							    A_	,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']   )

							with open(A_	,'w'	,encoding='utf-8'   ) as f:
								f.write(json.dumps(self.encoder	,indent=2	,sort_keys=A_	,ensure_ascii=A_   ) + '\n'   )

							A     = 0
							if self.bpe_ranks is None:
								return (vocab_file,)

							with open(A_	,'w'	,encoding='utf-8'   ) as writer:
								for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	,key=lambda A_   : kv[1]   ):
									if index != token_index:
										logger.warning(
										    F'Saving vocabulary to {merges_file}: BPE merge indices are not consecutive.'
										    ' Please check that the tokenizer is not corrupted!'   )
										A     = token_index
									writer.write(' '.join(A_   ) + '\n'   )
									index += 1

							return (vocab_file, merges_file)




"""simple docstring"""







import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

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()

@skip_mps
class       _a  ( SCREAMING_SNAKE_CASE__     ,	SCREAMING_SNAKE_CASE__     ,	unittest.TestCase):
			__magic_name__     					=     StableDiffusionPanoramaPipeline
			__magic_name__     					=     TEXT_TO_IMAGE_PARAMS
			__magic_name__     					=     TEXT_TO_IMAGE_BATCH_PARAMS
			__magic_name__     					=     TEXT_TO_IMAGE_IMAGE_PARAMS
			__magic_name__     					=     TEXT_TO_IMAGE_IMAGE_PARAMS



			def 				__lowercase					(			self							:				Optional[int]		)							->					int:
									torch.manual_seed(0		)
									snake_case   :			Optional[Any]						    = UNetaDConditionModel(
									    block_out_channels=(32, 64)   ,	layers_per_block=1   ,	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   :			Tuple						    = DDIMScheduler()
									torch.manual_seed(0		)
									snake_case   :			List[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   :			str						    = 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   :			int						    = CLIPTextModel(_lowercase		)
									snake_case   :			Union[str, Any]						    = CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

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



			def 				__lowercase					(			self							:				List[str]   ,	_lowercase							:				int   ,	_lowercase							:				Dict=0		)							->					Optional[int]:
									snake_case   :			Union[str, Any]						    = torch.manual_seed(_lowercase		)
									snake_case   :			str						    = {
									    "prompt": "a photo of the dolomites",
									    "generator": generator,
									    # Setting height and width to None to prevent OOMs on CPU.
									    "height": None,
									    "width": None,
									    "num_inference_steps": 1,
									    "guidance_scale": 6.0,
									    "output_type": "numpy",
									}
									return inputs



			def 				__lowercase					(			self							:				str		)							->					List[str]:
									snake_case   :			Any						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
									snake_case   :			Dict						    = self.get_dummy_components()
									snake_case   :			Any						    = StableDiffusionPanoramaPipeline(**_lowercase		)
									snake_case   :			List[Any]						    = sd_pipe.to(_lowercase		)
									sd_pipe.set_progress_bar_config(disable=_lowercase		)

									snake_case   :			Optional[Any]						    = self.get_dummy_inputs(_lowercase		)
									snake_case   :			Union[str, Any]						    = sd_pipe(**_lowercase		).images
									snake_case   :			List[Any]						    = image[0, -3:, -3:, -1]
									assert image.shape == (1, 64, 64, 3)

									snake_case   :			List[Any]						    = np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757]		)

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



			def 				__lowercase					(			self							:				int		)							->					Union[str, Any]:
									super().test_inference_batch_consistent(batch_sizes=[1, 2]		)



			def 				__lowercase					(			self							:				Tuple		)							->					Tuple:
									super().test_inference_batch_single_identical(batch_size=2   ,	expected_max_diff=3.25E-3		)



			def 				__lowercase					(			self							:				Any		)							->					List[Any]:
									snake_case   :			Any						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
									snake_case   :			Union[str, Any]						    = self.get_dummy_components()
									snake_case   :			Tuple						    = StableDiffusionPanoramaPipeline(**_lowercase		)
									snake_case   :			Tuple						    = sd_pipe.to(_lowercase		)
									sd_pipe.set_progress_bar_config(disable=_lowercase		)

									snake_case   :			List[str]						    = self.get_dummy_inputs(_lowercase		)
									snake_case   :			int						    = "french fries"
									snake_case   :			Union[str, Any]						    = sd_pipe(**_lowercase   ,	negative_prompt=_lowercase		)
									snake_case   :			str						    = output.images
									snake_case   :			Optional[Any]						    = image[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)

									snake_case   :			Union[str, Any]						    = np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]		)

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



			def 				__lowercase					(			self							:				str		)							->					Any:
									snake_case   :			Optional[Any]						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
									snake_case   :			List[Any]						    = self.get_dummy_components()
									snake_case   :			List[Any]						    = StableDiffusionPanoramaPipeline(**_lowercase		)
									snake_case   :			int						    = sd_pipe.to(_lowercase		)
									sd_pipe.set_progress_bar_config(disable=_lowercase		)

									snake_case   :			Tuple						    = self.get_dummy_inputs(_lowercase		)
									snake_case   :			str						    = sd_pipe(**_lowercase   ,	view_batch_size=2		)
									snake_case   :			Optional[Any]						    = output.images
									snake_case   :			int						    = image[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)

									snake_case   :			str						    = np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]		)

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



			def 				__lowercase					(			self							:				int		)							->					Optional[Any]:
									snake_case   :			Dict						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
									snake_case   :			List[str]						    = self.get_dummy_components()
									snake_case   :			Any						    = EulerAncestralDiscreteScheduler(
									    beta_start=0.00085   ,	beta_end=0.012   ,	beta_schedule="scaled_linear"		)
									snake_case   :			List[str]						    = StableDiffusionPanoramaPipeline(**_lowercase		)
									snake_case   :			List[str]						    = sd_pipe.to(_lowercase		)
									sd_pipe.set_progress_bar_config(disable=_lowercase		)

									snake_case   :			List[Any]						    = self.get_dummy_inputs(_lowercase		)
									snake_case   :			Optional[Any]						    = sd_pipe(**_lowercase		).images
									snake_case   :			int						    = image[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)

									snake_case   :			Tuple						    = np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952]		)

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



			def 				__lowercase					(			self							:				Tuple		)							->					Union[str, Any]:
									snake_case   :			str						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
									snake_case   :			str						    = self.get_dummy_components()
									snake_case   :			Optional[int]						    = PNDMScheduler(
									    beta_start=0.00085   ,	beta_end=0.012   ,	beta_schedule="scaled_linear"   ,	skip_prk_steps=_lowercase		)
									snake_case   :			Dict						    = StableDiffusionPanoramaPipeline(**_lowercase		)
									snake_case   :			Dict						    = sd_pipe.to(_lowercase		)
									sd_pipe.set_progress_bar_config(disable=_lowercase		)

									snake_case   :			Dict						    = self.get_dummy_inputs(_lowercase		)
									snake_case   :			Optional[int]						    = sd_pipe(**_lowercase		).images
									snake_case   :			str						    = image[0, -3:, -3:, -1]

									assert image.shape == (1, 64, 64, 3)

									snake_case   :			Optional[int]						    = np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539]		)

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



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



			def 				__lowercase					(			self							:				Union[str, Any]		)							->					str:
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()



			def 				__lowercase					(			self							:				Dict   ,	_lowercase							:				Any=0		)							->					Optional[Any]:
									snake_case   :			Any						    = torch.manual_seed(_lowercase		)
									snake_case   :			Any						    = {
									    "prompt": "a photo of the dolomites",
									    "generator": generator,
									    "num_inference_steps": 3,
									    "guidance_scale": 7.5,
									    "output_type": "numpy",
									}
									return inputs



			def 				__lowercase					(			self							:				Optional[Any]		)							->					Union[str, Any]:
									snake_case   :			List[Any]						    = "stabilityai/stable-diffusion-2-base"
									snake_case   :			Dict						    = DDIMScheduler.from_pretrained(_lowercase   ,	subfolder="scheduler"		)
									snake_case   :			Any						    = StableDiffusionPanoramaPipeline.from_pretrained(_lowercase   ,	scheduler=_lowercase   ,	safety_checker=_lowercase		)
									pipe.to(_lowercase		)
									pipe.set_progress_bar_config(disable=_lowercase		)
									pipe.enable_attention_slicing()

									snake_case   :			str						    = self.get_inputs()
									snake_case   :			List[Any]						    = pipe(**_lowercase		).images
									snake_case   :			Optional[int]						    = image[0, -3:, -3:, -1].flatten()

									assert image.shape == (1, 512, 2048, 3)

									snake_case   :			List[str]						    = np.array(
									    [
									        0.36968392,
									        0.27025372,
									        0.32446766,
									        0.28379387,
									        0.36363274,
									        0.30733347,
									        0.27100027,
									        0.27054125,
									        0.25536096,
									    ]		)

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



			def 				__lowercase					(			self							:				Tuple		)							->					List[str]:
									snake_case   :			List[str]						    = StableDiffusionPanoramaPipeline.from_pretrained(
									    "stabilityai/stable-diffusion-2-base"   ,	safety_checker=_lowercase		)
									snake_case   :			int						    = LMSDiscreteScheduler.from_config(pipe.scheduler.config		)
									pipe.to(_lowercase		)
									pipe.set_progress_bar_config(disable=_lowercase		)
									pipe.enable_attention_slicing()

									snake_case   :			Dict						    = self.get_inputs()
									snake_case   :			int						    = pipe(**_lowercase		).images
									snake_case   :			Optional[Any]						    = image[0, -3:, -3:, -1].flatten()

									assert image.shape == (1, 512, 2048, 3)

									snake_case   :			List[str]						    = np.array(
									    [
									        [
									            0.0,
									            0.0,
									            0.0,
									            0.0,
									            0.0,
									            0.0,
									            0.0,
									            0.0,
									            0.0,
									        ]
									    ]		)

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



			def 				__lowercase					(			self							:				str		)							->					Any:
									snake_case   :			Any						    = 0

									def callback_fn(_lowercase							:				int   ,	_lowercase							:				int   ,	_lowercase							:				torch.FloatTensor		) -> None:
															snake_case   :			List[Any]						    = True
															nonlocal number_of_steps
															number_of_steps += 1
															if step == 1:
																					snake_case   :			List[Any]						    = latents.detach().cpu().numpy()
																					assert latents.shape == (1, 4, 64, 256)
																					snake_case   :			int						    = latents[0, -3:, -3:, -1]

																					snake_case   :			Tuple						    = np.array(
																					    [
																					        0.18681869,
																					        0.33907816,
																					        0.5361276,
																					        0.14432865,
																					        -0.02856611,
																					        -0.73941123,
																					        0.23397987,
																					        0.47322682,
																					        -0.37823164,
																					    ]		)
																					assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2
															elif step == 2:
																					snake_case   :			Union[str, Any]						    = latents.detach().cpu().numpy()
																					assert latents.shape == (1, 4, 64, 256)
																					snake_case   :			Tuple						    = latents[0, -3:, -3:, -1]

																					snake_case   :			Tuple						    = np.array(
																					    [
																					        0.18539645,
																					        0.33987248,
																					        0.5378559,
																					        0.14437142,
																					        -0.02455261,
																					        -0.7338317,
																					        0.23990755,
																					        0.47356272,
																					        -0.3786505,
																					    ]		)

																					assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2

									snake_case   :			Dict						    = False

									snake_case   :			List[Any]						    = "stabilityai/stable-diffusion-2-base"
									snake_case   :			List[str]						    = DDIMScheduler.from_pretrained(_lowercase   ,	subfolder="scheduler"		)
									snake_case   :			Optional[Any]						    = StableDiffusionPanoramaPipeline.from_pretrained(_lowercase   ,	scheduler=_lowercase   ,	safety_checker=_lowercase		)
									snake_case   :			Any						    = pipe.to(_lowercase		)
									pipe.set_progress_bar_config(disable=_lowercase		)
									pipe.enable_attention_slicing()

									snake_case   :			Tuple						    = self.get_inputs()
									pipe(**_lowercase   ,	callback=_lowercase   ,	callback_steps=1		)
									assert callback_fn.has_been_called
									assert number_of_steps == 3



			def 				__lowercase					(			self							:				Any		)							->					Tuple:
									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									snake_case   :			Optional[int]						    = "stabilityai/stable-diffusion-2-base"
									snake_case   :			Tuple						    = DDIMScheduler.from_pretrained(_lowercase   ,	subfolder="scheduler"		)
									snake_case   :			Any						    = StableDiffusionPanoramaPipeline.from_pretrained(_lowercase   ,	scheduler=_lowercase   ,	safety_checker=_lowercase		)
									snake_case   :			Tuple						    = pipe.to(_lowercase		)
									pipe.set_progress_bar_config(disable=_lowercase		)
									pipe.enable_attention_slicing(1		)
									pipe.enable_sequential_cpu_offload()

									snake_case   :			Any						    = self.get_inputs()
									snake_case   :			Optional[int]						    = pipe(**_lowercase		)

									snake_case   :			int						    = torch.cuda.max_memory_allocated()
									# make sure that less than 5.2 GB is allocated
									assert mem_bytes < 5.5 * 10**9



