



import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
     import jieba

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


a_      :      Any    =     logging.get_logger(__name__)

a_      :      Union[str, Any]    =     {'vocab_file': 'vocab.txt'}

a_      :      Any    =     {
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

a_      :      Dict    =     {
    'openbmb/cpm-ant-10b': 10_24,
}






def     __a					(		__UpperCAmelCase       ):
 a__												= collections.OrderedDict()
 with open(__UpperCAmelCase							,  '''r'''							,  encoding='''utf-8'''       ) as reader:
  a__												= reader.readlines()
 for index, token in enumerate(__UpperCAmelCase       ):
  a__												= token.rstrip('''\n'''       )
  a__												= index
 return vocab
class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE="<unk>"		,				SCREAMING_SNAKE_CASE=2_0_0      ) ->			Optional[Any]:
  a__												= vocab
  a__												= unk_token
  a__												= max_input_chars_per_word



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			int:
  a__												= list(SCREAMING_SNAKE_CASE      )
  if len(SCREAMING_SNAKE_CASE      ) > self.max_input_chars_per_word:
   return [self.unk_token]

  a__												= 0
  a__												= []
  while start < len(SCREAMING_SNAKE_CASE      ):
   a__												= len(SCREAMING_SNAKE_CASE      )
   a__												= None
   while start < end:
    a__												= ''''''.join(chars[start:end]      )
    if substr in self.vocab:
     a__												= substr
     break
    end -= 1
   if cur_substr is None:
    sub_tokens.append(self.unk_token      )
    start += 1
   else:
    sub_tokens.append(SCREAMING_SNAKE_CASE      )
    a__												= end

  return sub_tokens





class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 _lowercase   :    Dict      						=					VOCAB_FILES_NAMES
 _lowercase   :    int      						=					PRETRAINED_VOCAB_FILES_MAP
 _lowercase   :    Dict      						=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowercase   :    Optional[Any]      						=					['''input_ids''', '''attention_mask''']
 _lowercase   :    List[Any]      						=					False

 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE="<d>"		,				SCREAMING_SNAKE_CASE="</d>"		,				SCREAMING_SNAKE_CASE="<s>"		,				SCREAMING_SNAKE_CASE="</s>"		,				SCREAMING_SNAKE_CASE="<pad>"		,				SCREAMING_SNAKE_CASE="<unk>"		,				SCREAMING_SNAKE_CASE="</n>"		,				SCREAMING_SNAKE_CASE="</_>"		,				SCREAMING_SNAKE_CASE="left"		,				**SCREAMING_SNAKE_CASE		,				) ->			Tuple:
  requires_backends(self		,				['''jieba''']      )
  super().__init__(
      bod_token=SCREAMING_SNAKE_CASE		,				eod_token=SCREAMING_SNAKE_CASE		,				bos_token=SCREAMING_SNAKE_CASE		,				eos_token=SCREAMING_SNAKE_CASE		,				pad_token=SCREAMING_SNAKE_CASE		,				unk_token=SCREAMING_SNAKE_CASE		,				line_token=SCREAMING_SNAKE_CASE		,				space_token=SCREAMING_SNAKE_CASE		,				padding_side=SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE		,				)
  a__												= bod_token
  a__												= eod_token
  a__												= load_vocab(SCREAMING_SNAKE_CASE      )
  a__												= self.encoder[space_token]
  a__												= self.encoder[line_token]

  del self.encoder[space_token]
  del self.encoder[line_token]

  a__												= collections.OrderedDict(sorted(self.encoder.items()		,				key=lambda SCREAMING_SNAKE_CASE      : x[1]      )      )
  a__												= {v: k for k, v in self.encoder.items()}

  a__												= WordpieceTokenizer(vocab=self.encoder		,				unk_token=self.unk_token      )

 @property
 def 						_UpperCAmelCase       (	self      ) ->			int:
  return self.encoder[self.bod_token]

 @property
 def 						_UpperCAmelCase       (	self      ) ->			List[str]:
  return self.encoder[self.eod_token]

 @property
 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  return self.encoder["\n"]

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

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

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			int:
  a__												= []
  for x in jieba.cut(SCREAMING_SNAKE_CASE		,				cut_all=SCREAMING_SNAKE_CASE      ):
   output_tokens.extend(self.wordpiece_tokenizer.tokenize(SCREAMING_SNAKE_CASE      )      )
  return output_tokens

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      ) ->			List[Any]:
  a__												= [i for i in token_ids if i >= 0]
  a__												= [
      x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
  ]
  return super()._decode(SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Any:
  return token in self.encoder

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  return "".join(SCREAMING_SNAKE_CASE      )

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

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

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None      ) ->			Tuple[str]:
  if os.path.isdir(SCREAMING_SNAKE_CASE      ):
   a__												= os.path.join(
       SCREAMING_SNAKE_CASE		,				(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']      )
  else:
   a__												= (filename_prefix + '''-''' if filename_prefix else '''''') + save_directory
  a__												= 0
  if " " in self.encoder:
   a__												= self.encoder[''' ''']
   del self.encoder[" "]
  if "\n" in self.encoder:
   a__												= self.encoder['''\n''']
   del self.encoder["\n"]
  a__												= collections.OrderedDict(sorted(self.encoder.items()		,				key=lambda SCREAMING_SNAKE_CASE      : x[1]      )      )
  with open(SCREAMING_SNAKE_CASE		,				'''w'''		,				encoding='''utf-8'''      ) as writer:
   for token, token_index in self.encoder.items():
    if index != token_index:
     logger.warning(
         f"Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."
         ''' Please check that the vocabulary is not corrupted!'''      )
     a__												= token_index
    writer.write(token + '''\n'''      )
    index += 1
  return (vocab_file,)

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None      ) ->			List[int]:
  if token_ids_a is None:
   return [self.bos_token_id] + token_ids_a
  return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = False      ) ->			List[int]:

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

  if token_ids_a is not None:
   return [1] + ([0] * len(SCREAMING_SNAKE_CASE      )) + [1] + ([0] * len(SCREAMING_SNAKE_CASE      ))
  return [1] + ([0] * len(SCREAMING_SNAKE_CASE      ))










import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

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


a_      :      Optional[int]    =     False
@skip_mps
class    __UpperCamelCase			(       _lowercase	,	_lowercase	,	_lowercase	,	unittest.TestCase				):

 """simple docstring"""



 _lowercase   :    List[Any]      						=					StableDiffusionAttendAndExcitePipeline
 _lowercase   :    str      						=					False
 _lowercase   :    List[str]      						=					TEXT_TO_IMAGE_PARAMS
 _lowercase   :    Optional[int]      						=					TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}				)
 _lowercase   :    Union[str, Any]      						=					TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowercase   :    List[Any]      						=					TEXT_TO_IMAGE_IMAGE_PARAMS

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Dict:
  super().setUpClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Tuple:
  super().tearDownClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  torch.manual_seed(0      )
  a__												= UNetaDConditionModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=1		,				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		,				attention_head_dim=(2, 4)		,				use_linear_projection=SCREAMING_SNAKE_CASE		,				)
  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		,				sample_size=1_2_8		,				)
  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		,				hidden_act='''gelu'''		,				projection_dim=5_1_2		,				)
  a__												= CLIPTextModel(SCREAMING_SNAKE_CASE      )
  a__												= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

  a__												= {
      '''unet''': unet,
      '''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      ) ->			Dict:
  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__												= a__												= {
      '''prompt''': '''a cat and a frog''',
      '''token_indices''': [2, 5],
      '''generator''': generator,
      '''num_inference_steps''': 1,
      '''guidance_scale''': 6.0,
      '''output_type''': '''numpy''',
      '''max_iter_to_alter''': 2,
      '''thresholds''': {0: 0.7},
  }
  return inputs

 def 						_UpperCAmelCase       (	self      ) ->			str:
  a__												= '''cpu'''

  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      )

  a__												= self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )
  a__												= pipe(**SCREAMING_SNAKE_CASE      ).images
  a__												= image[0, -3:, -3:, -1]

  self.assertEqual(image.shape		,				(1, 6_4, 6_4, 3)      )
  a__												= np.array(
      [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96]      )
  a__												= np.abs(image_slice.flatten() - expected_slice      ).max()
  self.assertLessEqual(SCREAMING_SNAKE_CASE		,				1e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			str:
  super().test_cpu_offload_forward_pass(expected_max_diff=5e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
  self._test_inference_batch_consistent(batch_sizes=[1, 2]      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  self._test_inference_batch_single_identical(batch_size=2		,				expected_max_diff=7e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			int:
  super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  super().test_save_load_local(expected_max_difference=5e-4      )



 def 						_UpperCAmelCase       (	self      ) ->			str:
  super().test_save_load_optional_components(expected_max_difference=4e-4      )





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

 """simple docstring"""



 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Optional[int]:
  super().setUpClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			str:
  super().tearDownClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

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



 def 						_UpperCAmelCase       (	self      ) ->			str:
  a__												= torch.manual_seed(5_1      )

  a__												= StableDiffusionAttendAndExcitePipeline.from_pretrained(
      '''CompVis/stable-diffusion-v1-4'''		,				safety_checker=SCREAMING_SNAKE_CASE		,				torch_dtype=torch.floataa      )
  pipe.to('''cuda'''      )

  a__												= '''a painting of an elephant with glasses'''
  a__												= [5, 7]

  a__												= pipe(
      prompt=SCREAMING_SNAKE_CASE		,				token_indices=SCREAMING_SNAKE_CASE		,				guidance_scale=7.5		,				generator=SCREAMING_SNAKE_CASE		,				num_inference_steps=5		,				max_iter_to_alter=5		,				output_type='''numpy'''		,				).images[0]

  a__												= load_numpy(
      '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'''      )
  assert np.abs((expected_image - image).max()      ) < 5e-1





