import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()







@skip_mps
class __snake_case       (		__a     ,	unittest.TestCase	):
			lowerCAmelCase_														=							VideoToVideoSDPipeline
			lowerCAmelCase_														=							TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}	) - {"image", "width", "height"}
			lowerCAmelCase_														=							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}	) - {"image"}
			lowerCAmelCase_														=							PipelineTesterMixin.required_optional_params - {"latents"}
			lowerCAmelCase_														=							False

			# No `output_type`.
			lowerCAmelCase_														=							frozenset(
			    [
			        "num_inference_steps",
			        "generator",
			        "latents",
			        "return_dict",
			        "callback",
			        "callback_steps",
			    ]	)




			def 			__a		(					self    :       str   ):



					"""simple docstring"""






					torch.manual_seed(0   )
					SCREAMING_SNAKE_CASE__    							=   UNetaDConditionModel(
					    block_out_channels=(32, 64, 64, 64) ,			layers_per_block=2 ,			sample_size=32 ,			in_channels=4 ,			out_channels=4 ,			down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""") ,			up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""") ,			cross_attention_dim=32 ,			attention_head_dim=4 ,			)
					SCREAMING_SNAKE_CASE__    							=   DDIMScheduler(
					    beta_start=0.0_00_85 ,			beta_end=0.0_12 ,			beta_schedule="""scaled_linear""" ,			clip_sample=a_ ,			set_alpha_to_one=a_ ,			)
					torch.manual_seed(0   )
					SCREAMING_SNAKE_CASE__    							=   AutoencoderKL(
					    block_out_channels=[32, 64] ,			in_channels=3 ,			out_channels=3 ,			down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""] ,			up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""] ,			latent_channels=4 ,			sample_size=1_28 ,			)
					torch.manual_seed(0   )
					SCREAMING_SNAKE_CASE__    							=   CLIPTextConfig(
					    bos_token_id=0 ,			eos_token_id=2 ,			hidden_size=32 ,			intermediate_size=37 ,			layer_norm_eps=1E-05 ,			num_attention_heads=4 ,			num_hidden_layers=5 ,			pad_token_id=1 ,			vocab_size=10_00 ,			hidden_act="""gelu""" ,			projection_dim=5_12 ,			)
					SCREAMING_SNAKE_CASE__    							=   CLIPTextModel(a_   )
					SCREAMING_SNAKE_CASE__    							=   CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""   )

					SCREAMING_SNAKE_CASE__    							=   {
					    """unet""": unet,
					    """scheduler""": scheduler,
					    """vae""": vae,
					    """text_encoder""": text_encoder,
					    """tokenizer""": tokenizer,
					}
					return components




			def 			__a		(					self    :       Dict ,			_lowercase    :       Tuple ,			_lowercase    :       List[Any]=0   ):



					"""simple docstring"""






					SCREAMING_SNAKE_CASE__    							=   floats_tensor((1, 3, 3, 32, 32) ,			rng=random.Random(a_   )   ).to(a_   )

					if str(a_   ).startswith("""mps"""   ):
							SCREAMING_SNAKE_CASE__    							=   torch.manual_seed(a_   )
					else:
							SCREAMING_SNAKE_CASE__    							=   torch.Generator(device=a_   ).manual_seed(a_   )
					SCREAMING_SNAKE_CASE__    							=   {
					    """prompt""": """A painting of a squirrel eating a burger""",
					    """video""": video,
					    """generator""": generator,
					    """num_inference_steps""": 2,
					    """guidance_scale""": 6.0,
					    """output_type""": """pt""",
					}
					return inputs




			def 			__a		(					self    :       str   ):



					"""simple docstring"""






					SCREAMING_SNAKE_CASE__    							=   """cpu"""  # ensure determinism for the device-dependent torch.Generator
					SCREAMING_SNAKE_CASE__    							=   self.get_dummy_components()
					SCREAMING_SNAKE_CASE__    							=   VideoToVideoSDPipeline(**a_   )
					SCREAMING_SNAKE_CASE__    							=   sd_pipe.to(a_   )
					sd_pipe.set_progress_bar_config(disable=a_   )

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

					assert frames[0].shape == (32, 32, 3)
					SCREAMING_SNAKE_CASE__    							=   np.array([1_06, 1_17, 1_13, 1_74, 1_37, 1_12, 1_48, 1_51, 1_31]   )

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




			@unittest.skipIf(
			    torch_device != """cuda""" or not is_xformers_available() ,			reason="""XFormers attention is only available with CUDA and `xformers` installed""" ,			)
			def 			__a		(					self    :       Tuple   ):



					"""simple docstring"""






					self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=a_ ,			expected_max_diff=5E-3   )




			@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""   )
			def 			__a		(					self    :       Any   ):



					"""simple docstring"""






					pass




			@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""   )
			def 			__a		(					self    :       List[Any]   ):



					"""simple docstring"""






					pass




			@unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""   )
			def 			__a		(					self    :       List[Any]   ):



					"""simple docstring"""






					pass




			def 			__a		(					self    :       int   ):



					"""simple docstring"""






					return super().test_progress_bar()







@slow
@skip_mps
class __snake_case       (		unittest.TestCase	):




			def 			__a		(					self    :       List[Any]   ):



					"""simple docstring"""






					SCREAMING_SNAKE_CASE__    							=   VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL""" ,			torch_dtype=torch.floataa   )
					pipe.enable_model_cpu_offload()

					# 10 frames
					SCREAMING_SNAKE_CASE__    							=   torch.Generator(device="""cpu"""   ).manual_seed(0   )
					SCREAMING_SNAKE_CASE__    							=   torch.randn((1, 10, 3, 10_24, 5_76) ,			generator=a_   )
					SCREAMING_SNAKE_CASE__    							=   video.to("""cuda"""   )

					SCREAMING_SNAKE_CASE__    							=   """Spiderman is surfing"""

					SCREAMING_SNAKE_CASE__    							=   pipe(a_ ,			video=a_ ,			generator=a_ ,			num_inference_steps=3 ,			output_type="""pt"""   ).frames

					SCREAMING_SNAKE_CASE__    							=   np.array([-1.0_45_89_84, -1.1_27_92_97, -0.9_66_30_86, -0.91_50_39_06, -0.75_09_76_56]   )
					assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array   ).sum() < 1E-2

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


__lowerCamelCase    :				Any							     =	logging.get_logger(__name__)

__lowerCamelCase    :				Optional[Any]							     =	{'''vocab_file''': '''vocab.txt'''}

__lowerCamelCase    :				Union[str, Any]							     =	{
    '''vocab_file''': {
        '''openbmb/cpm-ant-10b''': '''https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt''',
    },
}

__lowerCamelCase    :				Optional[Any]							     =	{
    '''openbmb/cpm-ant-10b''': 1024,
}



def 			__SCREAMING_SNAKE_CASE						(      __UpperCamelCase					: str			)							->      Optional[Any]:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__    							=   collections.OrderedDict()
  with open(__UpperCamelCase				,  """r"""				,  encoding="""utf-8"""			) as reader:
    SCREAMING_SNAKE_CASE__    							=   reader.readlines()
  for index, token in enumerate(__UpperCamelCase			):
    SCREAMING_SNAKE_CASE__    							=   token.rstrip("""\n"""			)
    SCREAMING_SNAKE_CASE__    							=   index
  return vocab







class __snake_case       (		lowerCamelCase_	):




   def __init__(					self    :       Optional[int] ,			_lowercase    :       Optional[Any] ,			_lowercase    :       int="<unk>" ,			_lowercase    :       int=2_00   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   vocab
     SCREAMING_SNAKE_CASE__    							=   unk_token
     SCREAMING_SNAKE_CASE__    							=   max_input_chars_per_word




   def 			__a		(					self    :       Optional[int] ,			_lowercase    :       List[Any]   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   list(_lowercase   )
     if len(_lowercase   ) > self.max_input_chars_per_word:
       return [self.unk_token]

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

     return sub_tokens







class __snake_case       (		lowerCamelCase_	):
   lowerCAmelCase_														=							VOCAB_FILES_NAMES
   lowerCAmelCase_														=							PRETRAINED_VOCAB_FILES_MAP
   lowerCAmelCase_														=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   lowerCAmelCase_														=							["input_ids", "attention_mask"]
   lowerCAmelCase_														=							False




   def __init__(					self    :       int ,			_lowercase    :       str ,			_lowercase    :       List[Any]="<d>" ,			_lowercase    :       List[Any]="</d>" ,			_lowercase    :       Union[str, Any]="<s>" ,			_lowercase    :       List[str]="</s>" ,			_lowercase    :       str="<pad>" ,			_lowercase    :       int="<unk>" ,			_lowercase    :       List[str]="</n>" ,			_lowercase    :       Tuple="</_>" ,			_lowercase    :       Any="left" ,			**_lowercase    :       Any ,			):



     """simple docstring"""






     requires_backends(self ,			["""jieba"""]   )
     super().__init__(
         bod_token=_lowercase ,			eod_token=_lowercase ,			bos_token=_lowercase ,			eos_token=_lowercase ,			pad_token=_lowercase ,			unk_token=_lowercase ,			line_token=_lowercase ,			space_token=_lowercase ,			padding_side=_lowercase ,			**_lowercase ,			)
     SCREAMING_SNAKE_CASE__    							=   bod_token
     SCREAMING_SNAKE_CASE__    							=   eod_token
     SCREAMING_SNAKE_CASE__    							=   load_vocab(_lowercase   )
     SCREAMING_SNAKE_CASE__    							=   self.encoder[space_token]
     SCREAMING_SNAKE_CASE__    							=   self.encoder[line_token]

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

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

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




   @property
   def 			__a		(					self    :       Optional[Any]   ):



     """simple docstring"""






     return self.encoder[self.bod_token]




   @property
   def 			__a		(					self    :       List[Any]   ):



     """simple docstring"""






     return self.encoder[self.eod_token]




   @property
   def 			__a		(					self    :       Any   ):



     """simple docstring"""






     return self.encoder["\n"]




   @property
   def 			__a		(					self    :       Union[str, Any]   ):



     """simple docstring"""






     return len(self.encoder   )




   def 			__a		(					self    :       int   ):



     """simple docstring"""






     return dict(self.encoder ,			**self.added_tokens_encoder   )




   def 			__a		(					self    :       Union[str, Any] ,			_lowercase    :       str   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   []
     for x in jieba.cut(_lowercase ,			cut_all=_lowercase   ):
       output_tokens.extend(self.wordpiece_tokenizer.tokenize(_lowercase   )   )
     return output_tokens




   def 			__a		(					self    :       int ,			_lowercase    :       Any ,			**_lowercase    :       Any   ):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__    							=   [i for i in token_ids if i >= 0]
     SCREAMING_SNAKE_CASE__    							=   [
         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(_lowercase ,			**_lowercase   )




   def 			__a		(					self    :       Optional[int] ,			_lowercase    :       List[Any]   ):



     """simple docstring"""






     return token in self.encoder




   def 			__a		(					self    :       List[str] ,			_lowercase    :       List[str]   ):



     """simple docstring"""






     return "".join(_lowercase   )




   def 			__a		(					self    :       Optional[int] ,			_lowercase    :       Any   ):



     """simple docstring"""






     return self.encoder.get(_lowercase ,			self.encoder.get(self.unk_token   )   )




   def 			__a		(					self    :       Tuple ,			_lowercase    :       List[Any]   ):



     """simple docstring"""






     return self.decoder.get(_lowercase ,			self.unk_token   )




   def 			__a		(					self    :       Optional[Any] ,			_lowercase    :       str ,			_lowercase    :       Optional[str] = None   ):



     """simple docstring"""






     if os.path.isdir(_lowercase   ):
       SCREAMING_SNAKE_CASE__    							=   os.path.join(
           _lowercase ,			(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]   )
     else:
       SCREAMING_SNAKE_CASE__    							=   (filename_prefix + """-""" if filename_prefix else """""") + save_directory
     SCREAMING_SNAKE_CASE__    							=   0
     if " " in self.encoder:
       SCREAMING_SNAKE_CASE__    							=   self.encoder[""" """]
       del self.encoder[" "]
     if "\n" in self.encoder:
       SCREAMING_SNAKE_CASE__    							=   self.encoder["""\n"""]
       del self.encoder["\n"]
     SCREAMING_SNAKE_CASE__    							=   collections.OrderedDict(sorted(self.encoder.items() ,			key=lambda _lowercase   : x[1]   )   )
     with open(_lowercase ,			"""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!"""   )
           SCREAMING_SNAKE_CASE__    							=   token_index
         writer.write(token + """\n"""   )
         index += 1
     return (vocab_file,)




   def 			__a		(					self    :       int ,			_lowercase    :       List[int] ,			_lowercase    :       List[int] = None   ):



     """simple docstring"""






     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 			__a		(					self    :       Union[str, Any] ,			_lowercase    :       List[int] ,			_lowercase    :       Optional[List[int]] = None ,			_lowercase    :       bool = False   ):



     """simple docstring"""






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

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