code: from typing import List, Optional, Union
  
  import numpy as np
  import PIL
  import torch
  from PIL import Image
  
  from ...models import UNetaDConditionModel, VQModel
  from ...pipelines import DiffusionPipeline
  from ...pipelines.pipeline_utils import ImagePipelineOutput
  from ...schedulers import DDPMScheduler
  from ...utils import (
      is_accelerate_available,
      is_accelerate_version,
      logging,
      randn_tensor,
      replace_example_docstring,
  )
  
  
  UpperCAmelCase					=				logging.get_logger(__name__)  # pylint: disable=invalid-name
  
  UpperCAmelCase					=				'''
      Examples:
          ```py
          >>> import torch
          >>> import numpy as np
  
          >>> from diffusers import KandinskyV22PriorEmb2EmbPipeline, KandinskyV22ControlnetImg2ImgPipeline
          >>> from transformers import pipeline
          >>> from diffusers.utils import load_image
  
  
          >>> def make_hint(image, depth_estimator):
          ...     image = depth_estimator(image)["depth"]
          ...     image = np.array(image)
          ...     image = image[:, :, None]
          ...     image = np.concatenate([image, image, image], axis=2)
          ...     detected_map = torch.from_numpy(image).float() / 255.0
          ...     hint = detected_map.permute(2, 0, 1)
          ...     return hint
  
  
          >>> depth_estimator = pipeline("depth-estimation")
  
          >>> pipe_prior = KandinskyV22PriorEmb2EmbPipeline.from_pretrained(
          ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16
          ... )
          >>> pipe_prior = pipe_prior.to("cuda")
  
          >>> pipe = KandinskyV22ControlnetImg2ImgPipeline.from_pretrained(
          ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16
          ... )
          >>> pipe = pipe.to("cuda")
  
          >>> img = load_image(
          ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
          ...     "/kandinsky/cat.png"
          ... ).resize((768, 768))
  
  
          >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")
  
          >>> prompt = "A robot, 4k photo"
          >>> negative_prior_prompt = "lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature"
  
          >>> generator = torch.Generator(device="cuda").manual_seed(43)
  
          >>> img_emb = pipe_prior(prompt=prompt, image=img, strength=0.85, generator=generator)
          >>> negative_emb = pipe_prior(prompt=negative_prior_prompt, image=img, strength=1, generator=generator)
  
          >>> images = pipe(
          ...     image=img,
          ...     strength=0.5,
          ...     image_embeds=img_emb.image_embeds,
          ...     negative_image_embeds=negative_emb.image_embeds,
          ...     hint=hint,
          ...     num_inference_steps=50,
          ...     generator=generator,
          ...     height=768,
          ...     width=768,
          ... ).images
  
          >>> images[0].save("robot_cat.png")
          ```
  '''
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Tuple		,						lowercase_:	List[str]		,						lowercase_:	Optional[Any]=8				):
         _UpperCamelCase     :       Optional[int]  						=  height // scale_factor**2
         if height % scale_factor**2 != 0:
                new_height += 1
         _UpperCamelCase     :       Union[str, Any]  						=  width // scale_factor**2
         if width % scale_factor**2 != 0:
                new_width += 1
         return new_height * scale_factor, new_width * scale_factor
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Optional[int]		,						lowercase_:	Optional[int]=512		,						lowercase_:	str=512				):
         _UpperCamelCase     :       str  						=  pil_image.resize((w, h)		,						resample=Image.BICUBIC		,						reducing_gap=1				)
         _UpperCamelCase     :       List[str]  						=  np.array(pil_image.convert('RGB'				)				)
         _UpperCamelCase     :       Tuple  						=  arr.astype(np.floataa				) / 127.5 - 1
         _UpperCamelCase     :       List[Any]  						=  np.transpose(lowercase_		,						[2, 0, 1]				)
         _UpperCamelCase     :       Optional[int]  						=  torch.from_numpy(lowercase_				).unsqueeze(0				)
         return image
  
  
  
  
  
  class    __lowerCAmelCase  (       a_						):
  
  
  
   '''simple docstring'''
  
   def __init__(			self       :		Tuple       ,			lowerCamelCase       :		UNetaDConditionModel       ,			lowerCamelCase       :		DDPMScheduler       ,			lowerCamelCase       :		VQModel       ,			) ->							str:
          super().__init__()
  
          self.register_modules(
              unet=lowerCamelCase       ,			scheduler=lowerCamelCase       ,			movq=lowerCamelCase       ,			)
          _UpperCamelCase     :       Dict  						=  2 ** (len(self.movq.config.block_out_channels  ) - 1)
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		Optional[int]       ,			lowerCamelCase       :		Any       ,			lowerCamelCase       :		Union[str, Any]  ) ->							int:
          # get the original timestep using init_timestep
          _UpperCamelCase     :       Dict  						=  min(int(num_inference_steps * strength  )       ,			lowerCamelCase  )
  
          _UpperCamelCase     :       Tuple  						=  max(num_inference_steps - init_timestep       ,			0  )
          _UpperCamelCase     :       List[Any]  						=  self.scheduler.timesteps[t_start:]
  
          return timesteps, num_inference_steps - t_start
  
   def 					__snake_case					(			self       :		Dict       ,			lowerCamelCase       :		Union[str, Any]       ,			lowerCamelCase       :		Any       ,			lowerCamelCase       :		Union[str, Any]       ,			lowerCamelCase       :		Union[str, Any]       ,			lowerCamelCase       :		Dict       ,			lowerCamelCase       :		Optional[Any]       ,			lowerCamelCase       :		str=None  ) ->							Any:
          if not isinstance(lowerCamelCase       ,			(torch.Tensor, PIL.Image.Image, list)  ):
                 raise ValueError(
                     f"`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(lowerCamelCase  )}"  )
  
          _UpperCamelCase     :       Union[str, Any]  						=  image.to(device=lowerCamelCase       ,			dtype=lowerCamelCase  )
  
          _UpperCamelCase     :       Dict  						=  batch_size * num_images_per_prompt
  
          if image.shape[1] == 4:
                 _UpperCamelCase     :       int  						=  image
  
          else:
                 if isinstance(lowerCamelCase       ,			lowerCamelCase  ) and len(lowerCamelCase  ) != batch_size:
                        raise ValueError(
                            f"You have passed a list of generators of length {len(lowerCamelCase  )}, but requested an effective batch"
                            f" size of {batch_size}. Make sure the batch size matches the length of the generators."  )
  
                 elif isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                        _UpperCamelCase     :       Union[str, Any]  						=  [
                            self.movq.encode(image[i : i + 1]  ).latent_dist.sample(generator[i]  ) for i in range(lowerCamelCase  )
                        ]
                        _UpperCamelCase     :       Optional[int]  						=  torch.cat(lowerCamelCase       ,			dim=0  )
                 else:
                        _UpperCamelCase     :       int  						=  self.movq.encode(lowerCamelCase  ).latent_dist.sample(lowerCamelCase  )
  
                 _UpperCamelCase     :       Any  						=  self.movq.config.scaling_factor * init_latents
  
          _UpperCamelCase     :       List[str]  						=  torch.cat([init_latents]       ,			dim=0  )
  
          _UpperCamelCase     :       Union[str, Any]  						=  init_latents.shape
          _UpperCamelCase     :       Any  						=  randn_tensor(lowerCamelCase       ,			generator=lowerCamelCase       ,			device=lowerCamelCase       ,			dtype=lowerCamelCase  )
  
          # get latents
          _UpperCamelCase     :       Tuple  						=  self.scheduler.add_noise(lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase  )
  
          _UpperCamelCase     :       Tuple  						=  init_latents
  
          return latents
  
   def 					__snake_case					(			self       :		Any       ,			lowerCamelCase       :		int=0  ) ->							Any:
          if is_accelerate_available():
                 from accelerate import cpu_offload
          else:
                 raise ImportError('Please install accelerate via `pip install accelerate`'  )
  
          _UpperCamelCase     :       Tuple  						=  torch.device(f"cuda:{gpu_id}"  )
  
          _UpperCamelCase     :       int  						=  [
              self.unet,
              self.movq,
          ]
          for cpu_offloaded_model in models:
                 if cpu_offloaded_model is not None:
                        cpu_offload(lowerCamelCase       ,			lowerCamelCase  )
  
   def 					__snake_case					(			self       :		int       ,			lowerCamelCase       :		Any=0  ) ->							Union[str, Any]:
          if is_accelerate_available() and is_accelerate_version('>='       ,			'0.17.0.dev0'  ):
                 from accelerate import cpu_offload_with_hook
          else:
                 raise ImportError('`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'  )
  
          _UpperCamelCase     :       Union[str, Any]  						=  torch.device(f"cuda:{gpu_id}"  )
  
          if self.device.type != "cpu":
                 self.to('cpu'       ,			silence_dtype_warnings=lowerCamelCase  )
                 torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)
  
          _UpperCamelCase     :       Union[str, Any]  						=  None
          for cpu_offloaded_model in [self.unet, self.movq]:
                 _UpperCamelCase						,      _UpperCamelCase     :       Dict  						=  cpu_offload_with_hook(lowerCamelCase       ,			lowerCamelCase       ,			prev_module_hook=lowerCamelCase  )
  
          # We'll offload the last model manually.
          _UpperCamelCase     :       List[str]  						=  hook
  
   @property
   # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
   def 					__snake_case					(			self       :		List[str]  ) ->							Any:
          if not hasattr(self.unet       ,			'_hf_hook'  ):
                 return self.device
          for module in self.unet.modules():
                 if (
                     hasattr(lowerCamelCase       ,			'_hf_hook'  )
                     and hasattr(module._hf_hook       ,			'execution_device'  )
                     and module._hf_hook.execution_device is not None
                 ):
                        return torch.device(module._hf_hook.execution_device  )
          return self.device
  
   @torch.no_grad()
   @replace_example_docstring(lowerCamelCase  )
   def __call__(			self       :		Optional[Any]       ,			lowerCamelCase       :		Union[torch.FloatTensor, List[torch.FloatTensor]]       ,			lowerCamelCase       :		Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]]       ,			lowerCamelCase       :		Union[torch.FloatTensor, List[torch.FloatTensor]]       ,			lowerCamelCase       :		torch.FloatTensor       ,			lowerCamelCase       :		int = 5_1_2       ,			lowerCamelCase       :		int = 5_1_2       ,			lowerCamelCase       :		int = 1_0_0       ,			lowerCamelCase       :		float = 4.0       ,			lowerCamelCase       :		float = 0.3       ,			lowerCamelCase       :		int = 1       ,			lowerCamelCase       :		Optional[Union[torch.Generator, List[torch.Generator]]] = None       ,			lowerCamelCase       :		Optional[str] = "pil"       ,			lowerCamelCase       :		bool = True       ,			) ->							Optional[int]:
          _UpperCamelCase     :       Optional[Any]  						=  self._execution_device
  
          _UpperCamelCase     :       Optional[Any]  						=  guidance_scale > 1.0
  
          if isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                 _UpperCamelCase     :       List[Any]  						=  torch.cat(lowerCamelCase       ,			dim=0  )
          if isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                 _UpperCamelCase     :       List[str]  						=  torch.cat(lowerCamelCase       ,			dim=0  )
          if isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                 _UpperCamelCase     :       Optional[Any]  						=  torch.cat(lowerCamelCase       ,			dim=0  )
  
          _UpperCamelCase     :       Optional[int]  						=  image_embeds.shape[0]
  
          if do_classifier_free_guidance:
                 _UpperCamelCase     :       List[str]  						=  image_embeds.repeat_interleave(lowerCamelCase       ,			dim=0  )
                 _UpperCamelCase     :       List[Any]  						=  negative_image_embeds.repeat_interleave(lowerCamelCase       ,			dim=0  )
                 _UpperCamelCase     :       Union[str, Any]  						=  hint.repeat_interleave(lowerCamelCase       ,			dim=0  )
  
          _UpperCamelCase     :       List[Any]  						=  torch.cat([negative_image_embeds, image_embeds]       ,			dim=0  ).to(dtype=self.unet.dtype       ,			device=lowerCamelCase  )
          _UpperCamelCase     :       int  						=  torch.cat([hint, hint]       ,			dim=0  ).to(dtype=self.unet.dtype       ,			device=lowerCamelCase  )
  
          if not isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                 _UpperCamelCase     :       int  						=  [image]
          if not all(isinstance(lowerCamelCase       ,			(PIL.Image.Image, torch.Tensor)  ) for i in image  ):
                 raise ValueError(
                     f"Input is in incorrect format: {[type(lowerCamelCase  ) for i in image]}. Currently, we only support  PIL image and pytorch tensor"  )
  
          _UpperCamelCase     :       Any  						=  torch.cat([prepare_image(lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase  ) for i in image]       ,			dim=0  )
          _UpperCamelCase     :       Any  						=  image.to(dtype=image_embeds.dtype       ,			device=lowerCamelCase  )
  
          _UpperCamelCase     :       int  						=  self.movq.encode(lowerCamelCase  )['latents']
          _UpperCamelCase     :       Optional[Any]  						=  latents.repeat_interleave(lowerCamelCase       ,			dim=0  )
          self.scheduler.set_timesteps(lowerCamelCase       ,			device=lowerCamelCase  )
          _UpperCamelCase						,      _UpperCamelCase     :       Tuple  						=  self.get_timesteps(lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase  )
          _UpperCamelCase     :       Optional[int]  						=  timesteps[:1].repeat(batch_size * num_images_per_prompt  )
          _UpperCamelCase						,      _UpperCamelCase     :       List[Any]  						=  downscale_height_and_width(lowerCamelCase       ,			lowerCamelCase       ,			self.movq_scale_factor  )
          _UpperCamelCase     :       Any  						=  self.prepare_latents(
              lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase       ,			image_embeds.dtype       ,			lowerCamelCase       ,			lowerCamelCase  )
          for i, t in enumerate(self.progress_bar(lowerCamelCase  )  ):
                 # expand the latents if we are doing classifier free guidance
                 _UpperCamelCase     :       List[Any]  						=  torch.cat([latents] * 2  ) if do_classifier_free_guidance else latents
  
                 _UpperCamelCase     :       str  						=  {'image_embeds': image_embeds, 'hint': hint}
                 _UpperCamelCase     :       Any  						=  self.unet(
                     sample=lowerCamelCase       ,			timestep=lowerCamelCase       ,			encoder_hidden_states=lowerCamelCase       ,			added_cond_kwargs=lowerCamelCase       ,			return_dict=lowerCamelCase       ,			)[0]
  
                 if do_classifier_free_guidance:
                        _UpperCamelCase						,      _UpperCamelCase     :       Optional[int]  						=  noise_pred.split(latents.shape[1]       ,			dim=1  )
                        _UpperCamelCase						,      _UpperCamelCase     :       Any  						=  noise_pred.chunk(2  )
                        _UpperCamelCase						,      _UpperCamelCase     :       int  						=  variance_pred.chunk(2  )
                        _UpperCamelCase     :       int  						=  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                        _UpperCamelCase     :       Tuple  						=  torch.cat([noise_pred, variance_pred_text]       ,			dim=1  )
  
                 if not (
                     hasattr(self.scheduler.config       ,			'variance_type'  )
                     and self.scheduler.config.variance_type in ["learned", "learned_range"]
                 ):
                        _UpperCamelCase						,      _UpperCamelCase     :       List[Any]  						=  noise_pred.split(latents.shape[1]       ,			dim=1  )
  
                 # compute the previous noisy sample x_t -> x_t-1
  
                 _UpperCamelCase     :       List[Any]  						=  self.scheduler.step(
                     lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase       ,			generator=lowerCamelCase       ,			)[0]
  
          # post-processing
          _UpperCamelCase     :       Optional[int]  						=  self.movq.decode(lowerCamelCase       ,			force_not_quantize=lowerCamelCase  )['sample']
  
          if output_type not in ["pt", "np", "pil"]:
                 raise ValueError(f"Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}"  )
  
          if output_type in ["np", "pil"]:
                 _UpperCamelCase     :       Any  						=  image * 0.5 + 0.5
                 _UpperCamelCase     :       List[str]  						=  image.clamp(0       ,			1  )
                 _UpperCamelCase     :       str  						=  image.cpu().permute(0       ,			2       ,			3       ,			1  ).float().numpy()
  
          if output_type == "pil":
                 _UpperCamelCase     :       Optional[Any]  						=  self.numpy_to_pil(lowerCamelCase  )
  
          if not return_dict:
                 return (image,)
  
          return ImagePipelineOutput(images=lowerCamelCase  )
  
  
  
  
style_context: import json
  import os
  import re
  import unicodedata
  from typing import Optional, Tuple
  
  from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
  from ...utils import logging
  
  
  UpperCAmelCase					=				logging.get_logger(__name__)
  
  UpperCAmelCase					=				{
      '''vocab_file''': '''vocab.json''',
      '''merges_file''': '''merges.txt''',
  }
  
  UpperCAmelCase					=				{
      '''vocab_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/vocab.json'''},
      '''merges_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/merges.txt'''},
  }
  
  UpperCAmelCase					=				{
      '''openai-gpt''': 5_12,
  }
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	int				):
         _UpperCamelCase     :       Optional[Any]  						=  text.strip()
         if not text:
                return []
         _UpperCamelCase     :       Optional[Any]  						=  text.split()
         return tokens
  
  
  
  
  class    __lowerCAmelCase  (       a_						):
  
  
  
   '''simple docstring'''
  
   def __init__(			self       :		str       ,			lowerCamelCase       :		Union[str, Any]=True       ,			lowerCamelCase       :		Optional[Any]=None       ,			lowerCamelCase       :		Dict=True       ,			lowerCamelCase       :		List[Any]=None       ,			lowerCamelCase       :		Optional[Any]=True       ,			) ->							Optional[Any]:
          if never_split is None:
                 _UpperCamelCase     :       Union[str, Any]  						=  []
          _UpperCamelCase     :       List[Any]  						=  do_lower_case
          _UpperCamelCase     :       List[Any]  						=  set(lowerCamelCase  )
          _UpperCamelCase     :       int  						=  tokenize_chinese_chars
          _UpperCamelCase     :       Dict  						=  strip_accents
          _UpperCamelCase     :       List[Any]  						=  do_split_on_punc
  
   def 					__snake_case					(			self       :		int       ,			lowerCamelCase       :		List[str]       ,			lowerCamelCase       :		List[Any]=None  ) ->							List[str]:
          _UpperCamelCase     :       Any  						=  self.never_split.union(set(lowerCamelCase  )  ) if never_split else self.never_split
          _UpperCamelCase     :       int  						=  self._clean_text(lowerCamelCase  )
  
          # This was added on November 1st, 2018 for the multilingual and Chinese
          # models. This is also applied to the English models now, but it doesn't
          # matter since the English models were not trained on any Chinese data
          # and generally don't have any Chinese data in them (there are Chinese
          # characters in the vocabulary because Wikipedia does have some Chinese
          # words in the English Wikipedia.).
          if self.tokenize_chinese_chars:
                 _UpperCamelCase     :       Optional[int]  						=  self._tokenize_chinese_chars(lowerCamelCase  )
          # prevents treating the same character with different unicode codepoints as different characters
          _UpperCamelCase     :       int  						=  unicodedata.normalize('NFC'       ,			lowerCamelCase  )
          _UpperCamelCase     :       Tuple  						=  whitespace_tokenize(lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  []
          for token in orig_tokens:
                 if token not in never_split:
                        if self.do_lower_case:
                               _UpperCamelCase     :       Tuple  						=  token.lower()
                               if self.strip_accents is not False:
                                      _UpperCamelCase     :       List[Any]  						=  self._run_strip_accents(lowerCamelCase  )
                        elif self.strip_accents:
                               _UpperCamelCase     :       List[str]  						=  self._run_strip_accents(lowerCamelCase  )
                 split_tokens.extend(self._run_split_on_punc(lowerCamelCase       ,			lowerCamelCase  )  )
  
          _UpperCamelCase     :       str  						=  whitespace_tokenize(' '.join(lowerCamelCase  )  )
          return output_tokens
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		Union[str, Any]  ) ->							str:
          _UpperCamelCase     :       Optional[Any]  						=  unicodedata.normalize('NFD'       ,			lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  []
          for char in text:
                 _UpperCamelCase     :       Tuple  						=  unicodedata.category(lowerCamelCase  )
                 if cat == "Mn":
                        continue
                 output.append(lowerCamelCase  )
          return "".join(lowerCamelCase  )
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		List[Any]       ,			lowerCamelCase       :		int=None  ) ->							List[Any]:
          if not self.do_split_on_punc or (never_split is not None and text in never_split):
                 return [text]
          _UpperCamelCase     :       Optional[int]  						=  list(lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  0
          _UpperCamelCase     :       Union[str, Any]  						=  True
          _UpperCamelCase     :       int  						=  []
          while i < len(lowerCamelCase  ):
                 _UpperCamelCase     :       List[str]  						=  chars[i]
                 if _is_punctuation(lowerCamelCase  ):
                        output.append([char]  )
                        _UpperCamelCase     :       str  						=  True
                 else:
                        if start_new_word:
                               output.append([]  )
                        _UpperCamelCase     :       List[str]  						=  False
                        output[-1].append(lowerCamelCase  )
                 i += 1
  
          return ["".join(lowerCamelCase  ) for x in output]
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		int  ) ->							Optional[Any]:
          _UpperCamelCase     :       Dict  						=  []
          for char in text:
                 _UpperCamelCase     :       Optional[int]  						=  ord(lowerCamelCase  )
                 if self._is_chinese_char(lowerCamelCase  ):
                        output.append(' '  )
                        output.append(lowerCamelCase  )
                        output.append(' '  )
                 else:
                        output.append(lowerCamelCase  )
          return "".join(lowerCamelCase  )
  
   def 					__snake_case					(			self       :		Union[str, Any]       ,			lowerCamelCase       :		int  ) ->							Optional[Any]:
          # This defines a "chinese character" as anything in the CJK Unicode block:
          #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
          #
          # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
          # despite its name. The modern Korean Hangul alphabet is a different block,
          # as is Japanese Hiragana and Katakana. Those alphabets are used to write
          # space-separated words, so they are not treated specially and handled
          # like the all of the other languages.
          if (
              (cp >= 0x4_E00 and cp <= 0x9_FFF)
              or (cp >= 0x3_400 and cp <= 0x4_DBF)  #
              or (cp >= 0x20_000 and cp <= 0x2A_6DF)  #
              or (cp >= 0x2A_700 and cp <= 0x2B_73F)  #
              or (cp >= 0x2B_740 and cp <= 0x2B_81F)  #
              or (cp >= 0x2B_820 and cp <= 0x2C_EAF)  #
              or (cp >= 0xF_900 and cp <= 0xF_AFF)
              or (cp >= 0x2F_800 and cp <= 0x2F_A1F)  #
          ):  #
                 return True
  
          return False
  
   def 					__snake_case					(			self       :		Any       ,			lowerCamelCase       :		int  ) ->							Dict:
          _UpperCamelCase     :       int  						=  []
          for char in text:
                 _UpperCamelCase     :       str  						=  ord(lowerCamelCase  )
                 if cp == 0 or cp == 0xF_FFD or _is_control(lowerCamelCase  ):
                        continue
                 if _is_whitespace(lowerCamelCase  ):
                        output.append(' '  )
                 else:
                        output.append(lowerCamelCase  )
          return "".join(lowerCamelCase  )
  
  
  
  
  
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Optional[Any]				):
         _UpperCamelCase     :       List[Any]  						=  set()
         _UpperCamelCase     :       Tuple  						=  word[0]
         for char in word[1:]:
                pairs.add((prev_char, char)				)
                _UpperCamelCase     :       Tuple  						=  char
         return pairs
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Dict				):
         _UpperCamelCase     :       Union[str, Any]  						=  text.replace('—'		,						'-'				)
         _UpperCamelCase     :       int  						=  text.replace('–'		,						'-'				)
         _UpperCamelCase     :       Dict  						=  text.replace('―'		,						'-'				)
         _UpperCamelCase     :       Any  						=  text.replace('…'		,						'...'				)
         _UpperCamelCase     :       Dict  						=  text.replace('´'		,						'\''				)
         _UpperCamelCase     :       Union[str, Any]  						=  re.sub(R'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'		,						R' \1 '		,						lowercase_				)
         _UpperCamelCase     :       Dict  						=  re.sub(R'\s*\n\s*'		,						' \n '		,						lowercase_				)
         _UpperCamelCase     :       Union[str, Any]  						=  re.sub(R'[^\S\n]+'		,						' '		,						lowercase_				)
         return text.strip()
  
  
  
  
  
  class    __lowerCAmelCase  (       a_						):
  
  
  
   '''simple docstring'''
   _lowerCamelCase						:	int				    = VOCAB_FILES_NAMES
   _lowerCamelCase						:	Any				    = PRETRAINED_VOCAB_FILES_MAP
   _lowerCamelCase						:	Optional[int]				    = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   _lowerCamelCase						:	List[str]				    = ['input_ids', 'attention_mask']
  
   def __init__(			self       :		str       ,			lowerCamelCase       :		Optional[Any]       ,			lowerCamelCase       :		Tuple       ,			lowerCamelCase       :		Union[str, Any]="<unk>"       ,			**lowerCamelCase       :		Any  ) ->							Dict:
          super().__init__(unk_token=lowerCamelCase       ,			**lowerCamelCase  )
  
          try:
                 import ftfy
                 from spacy.lang.en import English
  
                 _UpperCamelCase     :       Any  						=  English()
                 _UpperCamelCase     :       Any  						=  _nlp.tokenizer
                 _UpperCamelCase     :       Dict  						=  ftfy.fix_text
          except ImportError:
                 logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'  )
                 _UpperCamelCase     :       int  						=  BasicTokenizer(do_lower_case=lowerCamelCase  )
                 _UpperCamelCase     :       List[Any]  						=  None
  
          with open(lowerCamelCase       ,			encoding='utf-8'  ) as vocab_handle:
                 _UpperCamelCase     :       Tuple  						=  json.load(lowerCamelCase  )
          _UpperCamelCase     :       List[Any]  						=  {v: k for k, v in self.encoder.items()}
          with open(lowerCamelCase       ,			encoding='utf-8'  ) as merges_handle:
                 _UpperCamelCase     :       List[str]  						=  merges_handle.read().split('\n'  )[1:-1]
          _UpperCamelCase     :       Optional[int]  						=  [tuple(merge.split()  ) for merge in merges]
          _UpperCamelCase     :       Dict  						=  dict(zip(lowerCamelCase       ,			range(len(lowerCamelCase  )  )  )  )
          _UpperCamelCase     :       Optional[Any]  						=  {}
  
   @property
   def 					__snake_case					(			self       :		str  ) ->							int:
          return True
  
   @property
   def 					__snake_case					(			self       :		int  ) ->							Dict:
          return len(self.encoder  )
  
   def 					__snake_case					(			self       :		Optional[Any]  ) ->							Union[str, Any]:
          return dict(self.encoder       ,			**self.added_tokens_encoder  )
  
   def 					__snake_case					(			self       :		Union[str, Any]       ,			lowerCamelCase       :		Optional[int]  ) ->							Union[str, Any]:
          _UpperCamelCase     :       int  						=  tuple(token[:-1]  ) + (token[-1] + '</w>',)
          if token in self.cache:
                 return self.cache[token]
          _UpperCamelCase     :       Dict  						=  get_pairs(lowerCamelCase  )
  
          if not pairs:
                 return token + "</w>"
  
          while True:
                 _UpperCamelCase     :       List[str]  						=  min(lowerCamelCase       ,			key=lambda lowerCamelCase  : self.bpe_ranks.get(lowerCamelCase       ,			float('inf'  )  )  )
                 if bigram not in self.bpe_ranks:
                        break
                 _UpperCamelCase						,      _UpperCamelCase     :       Any  						=  bigram
                 _UpperCamelCase     :       Optional[Any]  						=  []
                 _UpperCamelCase     :       Optional[int]  						=  0
                 while i < len(lowerCamelCase  ):
                        try:
                               _UpperCamelCase     :       int  						=  word.index(lowerCamelCase       ,			lowerCamelCase  )
                        except ValueError:
                               new_word.extend(word[i:]  )
                               break
                        else:
                               new_word.extend(word[i:j]  )
                               _UpperCamelCase     :       Union[str, Any]  						=  j
  
                        if word[i] == first and i < len(lowerCamelCase  ) - 1 and word[i + 1] == second:
                               new_word.append(first + second  )
                               i += 2
                        else:
                               new_word.append(word[i]  )
                               i += 1
                 _UpperCamelCase     :       str  						=  tuple(lowerCamelCase  )
                 _UpperCamelCase     :       Optional[int]  						=  new_word
                 if len(lowerCamelCase  ) == 1:
                        break
                 else:
                        _UpperCamelCase     :       List[Any]  						=  get_pairs(lowerCamelCase  )
          _UpperCamelCase     :       Any  						=  ' '.join(lowerCamelCase  )
          if word == "\n  </w>":
                 _UpperCamelCase     :       Union[str, Any]  						=  '\n</w>'
          _UpperCamelCase     :       Tuple  						=  word
          return word
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		Optional[Any]  ) ->							List[Any]:
          _UpperCamelCase     :       Optional[int]  						=  []
          if self.fix_text is None:
                 # Using BERT's BasicTokenizer
                 _UpperCamelCase     :       int  						=  self.nlp.tokenize(lowerCamelCase  )
                 for token in text:
                        split_tokens.extend(list(self.bpe(lowerCamelCase  ).split(' '  )  )  )
          else:
                 # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
                 _UpperCamelCase     :       Optional[Any]  						=  self.nlp(text_standardize(self.fix_text(lowerCamelCase  )  )  )
                 for token in text:
                        split_tokens.extend(list(self.bpe(token.text.lower()  ).split(' '  )  )  )
          return split_tokens
  
   def 					__snake_case					(			self       :		int       ,			lowerCamelCase       :		Optional[int]  ) ->							int:
          return self.encoder.get(lowerCamelCase       ,			self.encoder.get(self.unk_token  )  )
  
   def 					__snake_case					(			self       :		Dict       ,			lowerCamelCase       :		List[str]  ) ->							str:
          return self.decoder.get(lowerCamelCase       ,			self.unk_token  )
  
   def 					__snake_case					(			self       :		Union[str, Any]       ,			lowerCamelCase       :		Dict  ) ->							Tuple:
          _UpperCamelCase     :       str  						=  ''.join(lowerCamelCase  ).replace('</w>'       ,			' '  ).strip()
          return out_string
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		str       ,			lowerCamelCase       :		Optional[str] = None  ) ->							Tuple[str]:
          if not os.path.isdir(lowerCamelCase  ):
                 logger.error(f"Vocabulary path ({save_directory}) should be a directory"  )
                 return
          _UpperCamelCase     :       str  						=  os.path.join(
              lowerCamelCase       ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
          _UpperCamelCase     :       List[str]  						=  os.path.join(
              lowerCamelCase       ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']  )
  
          with open(lowerCamelCase       ,			'w'       ,			encoding='utf-8'  ) as f:
                 f.write(json.dumps(self.encoder       ,			indent=2       ,			sort_keys=lowerCamelCase       ,			ensure_ascii=lowerCamelCase  ) + '\n'  )
  
          _UpperCamelCase     :       List[Any]  						=  0
          with open(lowerCamelCase       ,			'w'       ,			encoding='utf-8'  ) as writer:
                 writer.write('#version: 0.2\n'  )
                 for bpe_tokens, token_index in sorted(self.bpe_ranks.items()       ,			key=lambda lowerCamelCase  : kv[1]  ):
                        if index != token_index:
                               logger.warning(
                                   f"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
                                   ' Please check that the tokenizer is not corrupted!'  )
                               _UpperCamelCase     :       str  						=  token_index
                        writer.write(' '.join(lowerCamelCase  ) + '\n'  )
                        index += 1
  
          return vocab_file, merge_file
  
  
  
  