


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


UpperCAmelCase_      : Any  					=       logging.get_logger(__name__)

UpperCAmelCase_      : Dict  					=       {'vocab_file': 'vocab.txt'}

UpperCAmelCase_      : Optional[int]  					=       {
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

UpperCAmelCase_      : Tuple  					=       {
    'openbmb/cpm-ant-10b': 1024,
}







def        SCREAMING_SNAKE_CASE_						(			__A :			Tuple				)	->		Tuple:

       """simple docstring"""

       a_					:		Union[str, Any]							=  collections.OrderedDict()
       with open(__A  ,    'r'  ,    encoding='utf-8'				) as reader:
              a_					:		int							=  reader.readlines()
       for index, token in enumerate(__A				):
              a_					:		Union[str, Any]							=  token.rstrip('\n'				)
              a_					:		Union[str, Any]							=  index
       return vocab
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      def __init__(       self				:       int     ,							SCREAMING_SNAKE_CASE__				:       int     ,							SCREAMING_SNAKE_CASE__				:       List[str]="<unk>"     ,							SCREAMING_SNAKE_CASE__				:       Optional[Any]=2_0_0     )		->			List[str]:
             a_					:		List[Any]							=  vocab
             a_					:		Tuple							=  unk_token
             a_					:		Tuple							=  max_input_chars_per_word

      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]     )		->			Dict:
             a_					:		Any							=  list(SCREAMING_SNAKE_CASE__     )
             if len(SCREAMING_SNAKE_CASE__     ) > self.max_input_chars_per_word:
                    return [self.unk_token]

             a_					:		Tuple							=  0
             a_					:		Union[str, Any]							=  []
             while start < len(SCREAMING_SNAKE_CASE__     ):
                    a_					:		List[Any]							=  len(SCREAMING_SNAKE_CASE__     )
                    a_					:		List[Any]							=  None
                    while start < end:
                           a_					:		Dict							=  ''.join(chars[start:end]     )
                           if substr in self.vocab:
                                  a_					:		int							=  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_					:		Union[str, Any]							=  end

             return sub_tokens
class  SCREAMING_SNAKE_CASE__  (				lowercase__			):
      snake_case__				:     Optional[int]        =						VOCAB_FILES_NAMES
      snake_case__				:     Dict        =						PRETRAINED_VOCAB_FILES_MAP
      snake_case__				:     List[Any]        =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      snake_case__				:     str        =						['''input_ids''', '''attention_mask''']
      snake_case__				:     Union[str, Any]        =						False
      def __init__(       self				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       Dict="<d>"     ,							SCREAMING_SNAKE_CASE__				:       Union[str, Any]="</d>"     ,							SCREAMING_SNAKE_CASE__				:       List[Any]="<s>"     ,							SCREAMING_SNAKE_CASE__				:       Optional[int]="</s>"     ,							SCREAMING_SNAKE_CASE__				:       Tuple="<pad>"     ,							SCREAMING_SNAKE_CASE__				:       str="<unk>"     ,							SCREAMING_SNAKE_CASE__				:       str="</n>"     ,							SCREAMING_SNAKE_CASE__				:       Any="</_>"     ,							SCREAMING_SNAKE_CASE__				:       Tuple="left"     ,							**SCREAMING_SNAKE_CASE__				:       Dict     ,							)		->			Union[str, Any]:
             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_					:		Tuple							=  bod_token
             a_					:		str							=  eod_token
             a_					:		Optional[int]							=  load_vocab(SCREAMING_SNAKE_CASE__     )
             a_					:		Tuple							=  self.encoder[space_token]
             a_					:		Any							=  self.encoder[line_token]

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

             a_					:		str							=  collections.OrderedDict(sorted(self.encoder.items()     ,							key=lambda SCREAMING_SNAKE_CASE__     : x[1]     )     )
             a_					:		List[Any]							=  {v: k for k, v in self.encoder.items()}

             a_					:		str							=  WordpieceTokenizer(vocab=self.encoder     ,							unk_token=self.unk_token     )
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			int:
             return self.encoder[self.bod_token]
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     )		->			Optional[Any]:
             return self.encoder[self.eod_token]
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     )		->			List[Any]:
             return self.encoder["\n"]
      @property
      def        SCREAMING_SNAKE_CASE							(       self				:       List[str]     )		->			int:
             return len(self.encoder     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             return dict(self.encoder     ,							**self.added_tokens_encoder     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       Optional[Any]     )		->			Dict:
             a_					:		Union[str, Any]							=  []
             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        SCREAMING_SNAKE_CASE							(       self				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[str]     ,							**SCREAMING_SNAKE_CASE__				:       str     )		->			List[Any]:
             a_					:		Optional[Any]							=  [i for i in token_ids if i >= 0]
             a_					:		int							=  [
                 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        SCREAMING_SNAKE_CASE							(       self				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       List[Any]     )		->			int:
             return token in self.encoder
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       List[str]     )		->			str:
             return "".join(SCREAMING_SNAKE_CASE__     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     ,							SCREAMING_SNAKE_CASE__				:       Dict     )		->			Union[str, Any]:
             return self.encoder.get(SCREAMING_SNAKE_CASE__     ,							self.encoder.get(self.unk_token     )     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[int]     ,							SCREAMING_SNAKE_CASE__				:       List[str]     )		->			Dict:
             return self.decoder.get(SCREAMING_SNAKE_CASE__     ,							self.unk_token     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       Optional[str] = None     )		->			Tuple[str]:
             if os.path.isdir(SCREAMING_SNAKE_CASE__     ):
                    a_					:		str							=  os.path.join(
                        SCREAMING_SNAKE_CASE__     ,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']     )
             else:
                    a_					:		str							=  (filename_prefix + '-' if filename_prefix else '') + save_directory
             a_					:		int							=  0
             if " " in self.encoder:
                    a_					:		List[str]							=  self.encoder[' ']
                    del self.encoder[" "]
             if "\n" in self.encoder:
                    a_					:		Union[str, Any]							=  self.encoder['\n']
                    del self.encoder["\n"]
             a_					:		List[Any]							=  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_					:		Optional[Any]							=  token_index
                           writer.write(token + '\n'     )
                           index += 1
             return (vocab_file,)
      def        SCREAMING_SNAKE_CASE							(       self				:       Tuple     ,							SCREAMING_SNAKE_CASE__				:       List[int]     ,							SCREAMING_SNAKE_CASE__				:       List[int] = 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        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       List[int]     ,							SCREAMING_SNAKE_CASE__				:       Optional[List[int]] = None     ,							SCREAMING_SNAKE_CASE__				:       bool = 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 unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

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


enable_full_determinism()
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				lowercase__						,    unittest.TestCase			):
      snake_case__				:     Optional[Any]        =						TextToVideoSDPipeline
      snake_case__				:     Optional[int]        =						TEXT_TO_IMAGE_PARAMS
      snake_case__				:     str        =						TEXT_TO_IMAGE_BATCH_PARAMS
      # No `output_type`.
      snake_case__				:     Optional[Any]        =						frozenset(
          [
              '''num_inference_steps''',
              '''generator''',
              '''latents''',
              '''return_dict''',
              '''callback''',
              '''callback_steps''',
          ]			)
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			Union[str, Any]:
             torch.manual_seed(0     )
             a_					:		Optional[int]							=  UNetaDConditionModel(
                 block_out_channels=(3_2, 6_4, 6_4, 6_4)     ,							layers_per_block=2     ,							sample_size=3_2     ,							in_channels=4     ,							out_channels=4     ,							down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D')     ,							up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D')     ,							cross_attention_dim=3_2     ,							attention_head_dim=4     ,							)
             a_					:		int							=  DDIMScheduler(
                 beta_start=0.00085     ,							beta_end=0.012     ,							beta_schedule='scaled_linear'     ,							clip_sample=SCREAMING_SNAKE_CASE__     ,							set_alpha_to_one=SCREAMING_SNAKE_CASE__     ,							)
             torch.manual_seed(0     )
             a_					:		int							=  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_					:		Optional[int]							=  CLIPTextConfig(
                 bos_token_id=0     ,							eos_token_id=2     ,							hidden_size=3_2     ,							intermediate_size=3_7     ,							layer_norm_eps=1E-05     ,							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_					:		Dict							=  CLIPTextModel(SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

             a_					:		Union[str, Any]							=  {
                 'unet': unet,
                 'scheduler': scheduler,
                 'vae': vae,
                 'text_encoder': text_encoder,
                 'tokenizer': tokenizer,
             }
             return components
      def        SCREAMING_SNAKE_CASE							(       self				:       List[Any]     ,							SCREAMING_SNAKE_CASE__				:       str     ,							SCREAMING_SNAKE_CASE__				:       List[Any]=0     )		->			List[str]:
             if str(SCREAMING_SNAKE_CASE__     ).startswith('mps'     ):
                    a_					:		Dict							=  torch.manual_seed(SCREAMING_SNAKE_CASE__     )
             else:
                    a_					:		Optional[Any]							=  torch.Generator(device=SCREAMING_SNAKE_CASE__     ).manual_seed(SCREAMING_SNAKE_CASE__     )
             a_					:		int							=  {
                 'prompt': 'A painting of a squirrel eating a burger',
                 'generator': generator,
                 'num_inference_steps': 2,
                 'guidance_scale': 6.0,
                 'output_type': 'pt',
             }
             return inputs
      def        SCREAMING_SNAKE_CASE							(       self				:       Union[str, Any]     )		->			Tuple:
             a_					:		List[Any]							=  'cpu'  # ensure determinism for the device-dependent torch.Generator
             a_					:		Dict							=  self.get_dummy_components()
             a_					:		str							=  TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE__     )
             a_					:		Union[str, Any]							=  sd_pipe.to(SCREAMING_SNAKE_CASE__     )
             sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__     )

             a_					:		Tuple							=  self.get_dummy_inputs(SCREAMING_SNAKE_CASE__     )
             a_					:		Dict							=  'np'
             a_					:		Dict							=  sd_pipe(**SCREAMING_SNAKE_CASE__     ).frames
             a_					:		int							=  frames[0][-3:, -3:, -1]

             assert frames[0].shape == (6_4, 6_4, 3)
             a_					:		Union[str, Any]							=  np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]     )

             assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			List[Any]:
             self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=3E-3     )
      @unittest.skipIf(
          torch_device != 'cuda' or not is_xformers_available()     ,							reason='XFormers attention is only available with CUDA and `xformers` installed'     ,							)
      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			str:
             self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE__     ,							expected_max_diff=1E-2     )
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             pass
      @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       str     )		->			List[str]:
             pass
      @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'     )
      def        SCREAMING_SNAKE_CASE							(       self				:       int     )		->			List[Any]:
             pass

      def        SCREAMING_SNAKE_CASE							(       self				:       Dict     )		->			Union[str, Any]:
             return super().test_progress_bar()
@slow
@skip_mps
class  SCREAMING_SNAKE_CASE__  (				unittest.TestCase			):
      def        SCREAMING_SNAKE_CASE							(       self				:       Optional[Any]     )		->			List[Any]:
             a_					:		str							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'     )

             a_					:		Any							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Optional[Any]							=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config     )
             a_					:		Optional[Any]							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[Any]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		Optional[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2_5     ,							output_type='pt'     ).frames
             a_					:		str							=  video_frames.cpu().numpy()

             assert np.abs(expected_video - video     ).mean() < 5E-2

      def        SCREAMING_SNAKE_CASE							(       self				:       Any     )		->			Any:
             a_					:		Dict							=  load_numpy(
                 'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'     )

             a_					:		Tuple							=  TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'     )
             a_					:		Tuple							=  pipe.to('cuda'     )

             a_					:		Any							=  'Spiderman is surfing'
             a_					:		List[str]							=  torch.Generator(device='cpu'     ).manual_seed(0     )

             a_					:		List[Any]							=  pipe(SCREAMING_SNAKE_CASE__     ,							generator=SCREAMING_SNAKE_CASE__     ,							num_inference_steps=2     ,							output_type='pt'     ).frames
             a_					:		List[str]							=  video_frames.cpu().numpy()

             assert np.abs(expected_video - video     ).mean() < 5E-2


