



# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

import gc
import random
import tempfile
import unittest

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

from diffusers import (
    AutoencoderKL,
    ControlNetModel,
    DDIMScheduler,
    StableDiffusionControlNetImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()
class    __UpperCamelCase			(       _lowercase	,	_lowercase	,	_lowercase	,	unittest.TestCase				):

 """simple docstring"""



 _lowercase   :    List[str]      						=					StableDiffusionControlNetImgaImgPipeline
 _lowercase   :    Any      						=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
 _lowercase   :    Tuple      						=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 _lowercase   :    Optional[Any]      						=					IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'''control_image'''}				)
 _lowercase   :    Tuple      						=					IMAGE_TO_IMAGE_IMAGE_PARAMS

 def 						_UpperCAmelCase       (	self      ) ->			Any:
  torch.manual_seed(0      )
  a__												= UNetaDConditionModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=2		,				sample_size=3_2		,				in_channels=4		,				out_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,				cross_attention_dim=3_2		,				)
  torch.manual_seed(0      )
  a__												= ControlNetModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=2		,				in_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				cross_attention_dim=3_2		,				conditioning_embedding_out_channels=(1_6, 3_2)		,				)
  torch.manual_seed(0      )
  a__												= DDIMScheduler(
      beta_start=0.0_00_85		,				beta_end=0.0_12		,				beta_schedule='''scaled_linear'''		,				clip_sample=SCREAMING_SNAKE_CASE		,				set_alpha_to_one=SCREAMING_SNAKE_CASE		,				)
  torch.manual_seed(0      )
  a__												= AutoencoderKL(
      block_out_channels=[3_2, 6_4]		,				in_channels=3		,				out_channels=3		,				down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,				up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,				latent_channels=4		,				)
  torch.manual_seed(0      )
  a__												= CLIPTextConfig(
      bos_token_id=0		,				eos_token_id=2		,				hidden_size=3_2		,				intermediate_size=3_7		,				layer_norm_eps=1e-0_5		,				num_attention_heads=4		,				num_hidden_layers=5		,				pad_token_id=1		,				vocab_size=1_0_0_0		,				)
  a__												= CLIPTextModel(SCREAMING_SNAKE_CASE      )
  a__												= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

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

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=0      ) ->			Optional[Any]:
  if str(SCREAMING_SNAKE_CASE      ).startswith('''mps'''      ):
   a__												= torch.manual_seed(SCREAMING_SNAKE_CASE      )
  else:
   a__												= torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(SCREAMING_SNAKE_CASE      )

  a__												= 2
  a__												= randn_tensor(
      (1, 3, 3_2 * controlnet_embedder_scale_factor, 3_2 * controlnet_embedder_scale_factor)		,				generator=SCREAMING_SNAKE_CASE		,				device=torch.device(SCREAMING_SNAKE_CASE      )		,				)
  a__												= floats_tensor(control_image.shape		,				rng=random.Random(SCREAMING_SNAKE_CASE      )      ).to(SCREAMING_SNAKE_CASE      )
  a__												= image.cpu().permute(0		,				2		,				3		,				1      )[0]
  a__												= Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE      )      ).convert('''RGB'''      ).resize((6_4, 6_4)      )
  a__												= {
      '''prompt''': '''A painting of a squirrel eating a burger''',
      '''generator''': generator,
      '''num_inference_steps''': 2,
      '''guidance_scale''': 6.0,
      '''output_type''': '''numpy''',
      '''image''': image,
      '''control_image''': control_image,
  }

  return inputs

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3      )

 @unittest.skipIf(
     torch_device != '''cuda''' or not is_xformers_available()		,				reason='''XFormers attention is only available with CUDA and `xformers` installed'''		,				)
 def 						_UpperCAmelCase       (	self      ) ->			List[str]:
  self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3      )



 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  self._test_inference_batch_single_identical(expected_max_diff=2e-3      )





class    __UpperCamelCase			(       _lowercase	,	_lowercase	,	unittest.TestCase				):

 """simple docstring"""



 _lowercase   :    Dict      						=					StableDiffusionControlNetImgaImgPipeline
 _lowercase   :    Tuple      						=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
 _lowercase   :    List[str]      						=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 _lowercase   :    Union[str, Any]      						=					frozenset([]				)  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess

 def 						_UpperCAmelCase       (	self      ) ->			str:
  torch.manual_seed(0      )
  a__												= UNetaDConditionModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=2		,				sample_size=3_2		,				in_channels=4		,				out_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,				cross_attention_dim=3_2		,				)
  torch.manual_seed(0      )

  def init_weights(SCREAMING_SNAKE_CASE      ):
   if isinstance(SCREAMING_SNAKE_CASE		,				torch.nn.Convad      ):
    torch.nn.init.normal(m.weight      )
    m.bias.data.fill_(1.0      )

  a__												= ControlNetModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=2		,				in_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				cross_attention_dim=3_2		,				conditioning_embedding_out_channels=(1_6, 3_2)		,				)
  controlneta.controlnet_down_blocks.apply(SCREAMING_SNAKE_CASE      )

  torch.manual_seed(0      )
  a__												= ControlNetModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=2		,				in_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				cross_attention_dim=3_2		,				conditioning_embedding_out_channels=(1_6, 3_2)		,				)
  controlneta.controlnet_down_blocks.apply(SCREAMING_SNAKE_CASE      )

  torch.manual_seed(0      )
  a__												= DDIMScheduler(
      beta_start=0.0_00_85		,				beta_end=0.0_12		,				beta_schedule='''scaled_linear'''		,				clip_sample=SCREAMING_SNAKE_CASE		,				set_alpha_to_one=SCREAMING_SNAKE_CASE		,				)
  torch.manual_seed(0      )
  a__												= AutoencoderKL(
      block_out_channels=[3_2, 6_4]		,				in_channels=3		,				out_channels=3		,				down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,				up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,				latent_channels=4		,				)
  torch.manual_seed(0      )
  a__												= CLIPTextConfig(
      bos_token_id=0		,				eos_token_id=2		,				hidden_size=3_2		,				intermediate_size=3_7		,				layer_norm_eps=1e-0_5		,				num_attention_heads=4		,				num_hidden_layers=5		,				pad_token_id=1		,				vocab_size=1_0_0_0		,				)
  a__												= CLIPTextModel(SCREAMING_SNAKE_CASE      )
  a__												= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

  a__												= MultiControlNetModel([controlneta, controlneta]      )

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

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=0      ) ->			Optional[Any]:
  if str(SCREAMING_SNAKE_CASE      ).startswith('''mps'''      ):
   a__												= torch.manual_seed(SCREAMING_SNAKE_CASE      )
  else:
   a__												= torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(SCREAMING_SNAKE_CASE      )

  a__												= 2

  a__												= [
      randn_tensor(
          (1, 3, 3_2 * controlnet_embedder_scale_factor, 3_2 * controlnet_embedder_scale_factor)		,				generator=SCREAMING_SNAKE_CASE		,				device=torch.device(SCREAMING_SNAKE_CASE      )		,				),
      randn_tensor(
          (1, 3, 3_2 * controlnet_embedder_scale_factor, 3_2 * controlnet_embedder_scale_factor)		,				generator=SCREAMING_SNAKE_CASE		,				device=torch.device(SCREAMING_SNAKE_CASE      )		,				),
  ]

  a__												= floats_tensor(control_image[0].shape		,				rng=random.Random(SCREAMING_SNAKE_CASE      )      ).to(SCREAMING_SNAKE_CASE      )
  a__												= image.cpu().permute(0		,				2		,				3		,				1      )[0]
  a__												= Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE      )      ).convert('''RGB'''      ).resize((6_4, 6_4)      )
  a__												= {
      '''prompt''': '''A painting of a squirrel eating a burger''',
      '''generator''': generator,
      '''num_inference_steps''': 2,
      '''guidance_scale''': 6.0,
      '''output_type''': '''numpy''',
      '''image''': image,
      '''control_image''': control_image,
  }

  return inputs

 def 						_UpperCAmelCase       (	self      ) ->			int:
  a__												= self.get_dummy_components()
  a__												= self.pipeline_class(**SCREAMING_SNAKE_CASE      )
  pipe.to(SCREAMING_SNAKE_CASE      )

  a__												= 10.0
  a__												= 4

  a__												= self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )
  a__												= steps
  a__												= scale
  a__												= pipe(**SCREAMING_SNAKE_CASE      )[0]

  a__												= self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )
  a__												= steps
  a__												= scale
  a__												= pipe(**SCREAMING_SNAKE_CASE		,				control_guidance_start=0.1		,				control_guidance_end=0.2      )[0]

  a__												= self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )
  a__												= steps
  a__												= scale
  a__												= pipe(**SCREAMING_SNAKE_CASE		,				control_guidance_start=[0.1, 0.3]		,				control_guidance_end=[0.2, 0.7]      )[0]

  a__												= self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )
  a__												= steps
  a__												= scale
  a__												= pipe(**SCREAMING_SNAKE_CASE		,				control_guidance_start=0.4		,				control_guidance_end=[0.5, 0.8]      )[0]

  # make sure that all outputs are different
  assert np.sum(np.abs(output_a - output_a      )      ) > 1e-3
  assert np.sum(np.abs(output_a - output_a      )      ) > 1e-3
  assert np.sum(np.abs(output_a - output_a      )      ) > 1e-3

 def 						_UpperCAmelCase       (	self      ) ->			int:
  return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3      )

 @unittest.skipIf(
     torch_device != '''cuda''' or not is_xformers_available()		,				reason='''XFormers attention is only available with CUDA and `xformers` installed'''		,				)
 def 						_UpperCAmelCase       (	self      ) ->			str:
  self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  self._test_inference_batch_single_identical(expected_max_diff=2e-3      )



 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  a__												= self.get_dummy_components()
  a__												= self.pipeline_class(**SCREAMING_SNAKE_CASE      )
  pipe.to(SCREAMING_SNAKE_CASE      )
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )
  with tempfile.TemporaryDirectory() as tmpdir:
   try:
    # save_pretrained is not implemented for Multi-ControlNet
    pipe.save_pretrained(SCREAMING_SNAKE_CASE      )
   except NotImplementedError:
    pass





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

 """simple docstring"""



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



 def 						_UpperCAmelCase       (	self      ) ->			Union[str, Any]:
  a__												= ControlNetModel.from_pretrained('''lllyasviel/sd-controlnet-canny'''      )

  a__												= StableDiffusionControlNetImgaImgPipeline.from_pretrained(
      '''runwayml/stable-diffusion-v1-5'''		,				safety_checker=SCREAMING_SNAKE_CASE		,				controlnet=SCREAMING_SNAKE_CASE      )
  pipe.enable_model_cpu_offload()
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

  a__												= torch.Generator(device='''cpu'''      ).manual_seed(0      )
  a__												= '''evil space-punk bird'''
  a__												= load_image(
      '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png'''      ).resize((5_1_2, 5_1_2)      )
  a__												= load_image(
      '''https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png'''      ).resize((5_1_2, 5_1_2)      )

  a__												= pipe(
      SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				control_image=SCREAMING_SNAKE_CASE		,				generator=SCREAMING_SNAKE_CASE		,				output_type='''np'''		,				num_inference_steps=5_0		,				strength=0.6		,				)

  a__												= output.images[0]

  assert image.shape == (5_1_2, 5_1_2, 3)

  a__												= load_numpy(
      '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy'''      )

  assert np.abs(expected_image - image      ).max() < 9e-2










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

import regex as re

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


a_      :      Dict    =     logging.get_logger(__name__)


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

a_      :      Union[str, Any]    =     {
    'vocab_file': {
        'facebook/blenderbot_small-90M': 'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/vocab.json'
    },
    'merges_file': {
        'facebook/blenderbot_small-90M': 'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/merges.txt'
    },
    'tokenizer_config_file': {
        'facebook/blenderbot_small-90M': (
            'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/tokenizer_config.json'
        )
    },
}

a_      :      List[Any]    =     {'facebook/blenderbot_small-90M': 5_12}






def     __a					(		__UpperCAmelCase       ):
 a__												= set()
 a__												= word[0]
 for char in word[1:]:
  pairs.add((prev_char, char)       )
  a__												= char

 a__												= set(__UpperCAmelCase       )
 return pairs

class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



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

 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE="__start__"		,				SCREAMING_SNAKE_CASE="__end__"		,				SCREAMING_SNAKE_CASE="__unk__"		,				SCREAMING_SNAKE_CASE="__null__"		,				**SCREAMING_SNAKE_CASE		,				) ->			List[str]:
  super().__init__(unk_token=SCREAMING_SNAKE_CASE		,				bos_token=SCREAMING_SNAKE_CASE		,				eos_token=SCREAMING_SNAKE_CASE		,				pad_token=SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )

  with open(SCREAMING_SNAKE_CASE		,				encoding='''utf-8'''      ) as vocab_handle:
   a__												= json.load(SCREAMING_SNAKE_CASE      )
  a__												= {v: k for k, v in self.encoder.items()}
  with open(SCREAMING_SNAKE_CASE		,				encoding='''utf-8'''      ) as merges_handle:
   a__												= merges_handle.read().split('''\n'''      )[1:-1]
  a__												= [tuple(merge.split()      ) for merge in merges]
  a__												= dict(zip(SCREAMING_SNAKE_CASE		,				range(len(SCREAMING_SNAKE_CASE      )      )      )      )
  a__												= {}

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

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  return dict(self.encoder		,				**self.added_tokens_encoder      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  if token in self.cache:
   return self.cache[token]
  a__												= re.sub('''([.,!?()])'''		,				R''' \1'''		,				SCREAMING_SNAKE_CASE      )
  a__												= re.sub('''(\')'''		,				R''' \1 '''		,				SCREAMING_SNAKE_CASE      )
  a__												= re.sub(R'''\s{2,}'''		,				''' '''		,				SCREAMING_SNAKE_CASE      )
  if "\n" in token:
   a__												= token.replace('''\n'''		,				''' __newln__'''      )

  a__												= token.split(''' '''      )
  a__												= []
  for token in tokens:
   if not len(SCREAMING_SNAKE_CASE      ):
    continue

   a__												= token.lower()
   a__												= tuple(SCREAMING_SNAKE_CASE      )
   a__												= tuple(list(word[:-1]      ) + [word[-1] + '''</w>''']      )
   a__												= get_pairs(SCREAMING_SNAKE_CASE      )

   if not pairs:
    words.append(SCREAMING_SNAKE_CASE      )
    continue

   while True:
    a__												= min(SCREAMING_SNAKE_CASE		,				key=lambda SCREAMING_SNAKE_CASE      : self.bpe_ranks.get(SCREAMING_SNAKE_CASE		,				float('''inf'''      )      )      )
    if bigram not in self.bpe_ranks:
     break
    a__							,			a__												= bigram
    a__												= []
    a__												= 0

    while i < len(SCREAMING_SNAKE_CASE      ):
     try:
      a__												= word.index(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
      new_word.extend(word[i:j]      )
      a__												= j
     except ValueError:
      new_word.extend(word[i:]      )
      break

     if word[i] == first and i < len(SCREAMING_SNAKE_CASE      ) - 1 and word[i + 1] == second:
      new_word.append(first + second      )
      i += 2
     else:
      new_word.append(word[i]      )
      i += 1
    a__												= tuple(SCREAMING_SNAKE_CASE      )
    a__												= new_word
    if len(SCREAMING_SNAKE_CASE      ) == 1:
     break
    else:
     a__												= get_pairs(SCREAMING_SNAKE_CASE      )
   a__												= '''@@ '''.join(SCREAMING_SNAKE_CASE      )
   a__												= word[:-4]

   a__												= word
   words.append(SCREAMING_SNAKE_CASE      )
  return " ".join(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			List[str]:
  a__												= []

  a__												= re.findall(R'''\S+\n?'''		,				SCREAMING_SNAKE_CASE      )

  for token in words:
   split_tokens.extend(list(self.bpe(SCREAMING_SNAKE_CASE      ).split(''' '''      )      )      )
  return split_tokens

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			int:
  a__												= token.lower()
  return self.encoder.get(SCREAMING_SNAKE_CASE		,				self.encoder.get(self.unk_token      )      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  return self.decoder.get(SCREAMING_SNAKE_CASE		,				self.unk_token      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  a__												= ''' '''.join(SCREAMING_SNAKE_CASE      ).replace('''@@ '''		,				''''''      ).strip()
  return out_string



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None      ) ->			Tuple[str]:
  if not os.path.isdir(SCREAMING_SNAKE_CASE      ):
   logger.error(f"Vocabulary path ({save_directory}) should be a directory"      )
   return
  a__												= os.path.join(
      SCREAMING_SNAKE_CASE		,				(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']      )
  a__												= os.path.join(
      SCREAMING_SNAKE_CASE		,				(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']      )

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

  a__												= 0
  with open(SCREAMING_SNAKE_CASE		,				'''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 SCREAMING_SNAKE_CASE      : 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!'''      )
     a__												= token_index
    writer.write(''' '''.join(SCREAMING_SNAKE_CASE      ) + '''\n'''      )
    index += 1

  return vocab_file, merge_file





