





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	(  __SCREAMING_SNAKE_CASE			,     unittest.TestCase    ):


    '''simple docstring'''







    lowerCamelCase__    :		Any				=	VideoToVideoSDPipeline
    lowerCamelCase__    :		List[Any]				=	TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}    ) - {"""image""", """width""", """height"""}
    lowerCamelCase__    :		List[str]				=	TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}    ) - {"""image"""}
    lowerCamelCase__    :		List[str]				=	PipelineTesterMixin.required_optional_params - {"""latents"""}
    lowerCamelCase__    :		int				=	False

    # No `output_type`.
    lowerCamelCase__    :		List[Any]				=	frozenset(
        [
            """num_inference_steps""",
            """generator""",
            """latents""",
            """return_dict""",
            """callback""",
            """callback_steps""",
        ]    )



    def  lowercase_					(       self     ):





        '''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.00085   ,				beta_end=0.012   ,				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  lowercase_					(       self   ,				A_   ,				A_=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  lowercase_					(       self     ):





        '''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  lowercase_					(       self     ):





        '''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  lowercase_					(       self     ):





        '''simple docstring'''

        pass



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





        '''simple docstring'''

        pass



    @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''     )
    def  lowercase_					(       self     ):





        '''simple docstring'''

        pass




    def  lowercase_					(       self     ):





        '''simple docstring'''

        return super().test_progress_bar()





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


    '''simple docstring'''







    def  lowercase_					(       self     ):





        '''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.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656]     )
        assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array     ).sum() < 1E-2






"""simple docstring"""




import json
import os
from typing import Optional, Tuple

import regex as re

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


lowerCAmelCase				:     Union[str, Any]					=				logging.get_logger(__name__)

lowerCAmelCase				:     Optional[int]					=				{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

lowerCAmelCase				:     Optional[Any]					=				{
    """vocab_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"""},
    """merges_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"""},
}

lowerCAmelCase				:     Optional[Any]					=				{
    """ctrl""": 256,
}

lowerCAmelCase				:     int					=				{
    """Pregnancy""": 16_8629,
    """Christianity""": 7675,
    """Explain""": 10_6423,
    """Fitness""": 6_3440,
    """Saving""": 6_3163,
    """Ask""": 2_7171,
    """Ass""": 9_5985,
    """Joke""": 16_3509,
    """Questions""": 4_5622,
    """Thoughts""": 4_9605,
    """Retail""": 5_2342,
    """Feminism""": 16_4338,
    """Writing""": 1_1992,
    """Atheism""": 19_2263,
    """Netflix""": 4_8616,
    """Computing""": 3_9639,
    """Opinion""": 4_3213,
    """Alone""": 4_4967,
    """Funny""": 5_8917,
    """Gaming""": 4_0358,
    """Human""": 4088,
    """India""": 1331,
    """Joker""": 7_7138,
    """Diet""": 3_6206,
    """Legal""": 1_1859,
    """Norman""": 4939,
    """Tip""": 7_2689,
    """Weight""": 5_2343,
    """Movies""": 4_6273,
    """Running""": 2_3425,
    """Science""": 2090,
    """Horror""": 3_7793,
    """Confession""": 6_0572,
    """Finance""": 1_2250,
    """Politics""": 1_6360,
    """Scary""": 19_1985,
    """Support""": 1_2654,
    """Technologies""": 3_2516,
    """Teenage""": 6_6160,
    """Event""": 3_2769,
    """Learned""": 6_7460,
    """Notion""": 18_2770,
    """Wikipedia""": 3_7583,
    """Books""": 6665,
    """Extract""": 7_6050,
    """Confessions""": 10_2701,
    """Conspiracy""": 7_5932,
    """Links""": 6_3674,
    """Narcissus""": 15_0425,
    """Relationship""": 5_4766,
    """Relationships""": 13_4796,
    """Reviews""": 4_1671,
    """News""": 4256,
    """Translation""": 2_6820,
    """multilingual""": 12_8406,
}



def        a__					(							snake_case__							)		->   Optional[Any]:
			lowerCamelCase      =   set()
			lowerCamelCase      =   word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)							)
						lowerCamelCase      =   char

			lowerCamelCase      =   set(snake_case__							)
			return pairs






class        __magic_name__				(			UpperCAmelCase__							):






				'''simple docstring'''
				__UpperCamelCase         =		VOCAB_FILES_NAMES
				__UpperCamelCase         =		PRETRAINED_VOCAB_FILES_MAP
				__UpperCamelCase         =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__UpperCamelCase         =		CONTROL_CODES





				def __init__(			self						,					_a						,					_a						,					_a="<unk>"						,					**_a					):





							"""simple docstring"""
							super().__init__(unk_token=_a						,					**_a					)

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





				@property
				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return len(self.encoder					)





				def   _lowerCAmelCase							(			self					):





							"""simple docstring"""
							return dict(self.encoder						,					**self.added_tokens_encoder					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							if token in self.cache:
										return self.cache[token]
							lowerCamelCase      =   tuple(_a					)
							lowerCamelCase      =   tuple(list(word[:-1]					) + [word[-1] + """</w>"""]					)
							lowerCamelCase      =   get_pairs(_a					)

							if not pairs:
										return token

							while True:
										lowerCamelCase      =   min(_a						,					key=lambda _a					: self.bpe_ranks.get(_a						,					float("""inf"""					)					)					)
										if bigram not in self.bpe_ranks:
													break
										lowerCamelCase ,  lowerCamelCase      =   bigram
										lowerCamelCase      =   []
										lowerCamelCase      =   0
										while i < len(_a					):
													try:
																lowerCamelCase      =   word.index(_a						,					_a					)
													except ValueError:
																new_word.extend(word[i:]					)
																break
													else:
																new_word.extend(word[i:j]					)
																lowerCamelCase      =   j

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





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   []

							lowerCamelCase      =   re.findall(r"""\S+\n?"""						,					_a					)

							for token in words:
										split_tokens.extend(list(self.bpe(_a					).split(""" """					)					)					)
							return split_tokens





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.encoder.get(_a						,					self.encoder.get(self.unk_token					)					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.decoder.get(_a						,					self.unk_token					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   """ """.join(_a					).replace("""@@ """						,					""""""					).strip()
							return out_string





				def   _lowerCAmelCase							(			self						,					_a						,					_a = None					):





							"""simple docstring"""
							if not os.path.isdir(_a					):
										logger.error(f'Vocabulary path ({save_directory}) should be a directory'					)
										return
							lowerCamelCase      =   os.path.join(
							    _a						,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]					)
							lowerCamelCase      =   os.path.join(
							    _a						,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]					)

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

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

							return vocab_file, merge_file

				# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
				#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
				#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
				#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
				#     return ''.join(tokens_generated_so_far)



