





import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin

class 		UpperCamelCase__ (      lowerCAmelCase_   ):



     '''simple docstring'''

     __snake_case    :	Any			=			["image_processor", "tokenizer"]
     __snake_case    :	str			=			"AutoImageProcessor"
     __snake_case    :	Optional[int]			=			"AutoTokenizer"






     def __init__(							self						:    Tuple						,lowerCamelCase__						:    List[Any]=None						,lowerCamelCase__						:    Optional[int]=None						,**lowerCamelCase__						:    int       )     ->		Tuple:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						None
           if "feature_extractor" in kwargs:
                 warnings.warn(
                     """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
                     """ instead."""						,lowerCamelCase__						,)
                 SCREAMING_SNAKE_CASE		=						kwargs.pop("""feature_extractor"""       )

           SCREAMING_SNAKE_CASE		=						image_processor if image_processor is not None else feature_extractor
           if image_processor is None:
                 raise ValueError("""You need to specify an `image_processor`."""       )
           if tokenizer is None:
                 raise ValueError("""You need to specify a `tokenizer`."""       )

           super().__init__(lowerCamelCase__						,lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						self.image_processor
           SCREAMING_SNAKE_CASE		=						False






     def __call__(							self						:    Union[str, Any]						,*lowerCamelCase__						:    Tuple						,**lowerCamelCase__						:    List[Any]       )     ->		Union[str, Any]:




           '''simple docstring'''


           if self._in_target_context_manager:
                 return self.current_processor(*lowerCamelCase__						,**lowerCamelCase__       )

           SCREAMING_SNAKE_CASE		=						kwargs.pop("""images"""						,lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						kwargs.pop("""text"""						,lowerCamelCase__       )
           if len(lowerCamelCase__       ) > 0:
                 SCREAMING_SNAKE_CASE		=						args[0]
                 SCREAMING_SNAKE_CASE		=						args[1:]

           if images is None and text is None:
                 raise ValueError("""You need to specify either an `images` or `text` input to process."""       )

           if images is not None:
                 SCREAMING_SNAKE_CASE		=						self.image_processor(lowerCamelCase__						,*lowerCamelCase__						,**lowerCamelCase__       )
           if text is not None:
                 SCREAMING_SNAKE_CASE		=						self.tokenizer(lowerCamelCase__						,**lowerCamelCase__       )

           if text is None:
                 return inputs
           elif images is None:
                 return encodings
           else:
                 SCREAMING_SNAKE_CASE		=						encodings["""input_ids"""]
                 return inputs






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]						,*lowerCamelCase__						:    Union[str, Any]						,**lowerCamelCase__						:    Tuple       )     ->		List[str]:




           '''simple docstring'''


           return self.tokenizer.batch_decode(*lowerCamelCase__						,**lowerCamelCase__       )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Dict						,*lowerCamelCase__						:    Any						,**lowerCamelCase__						:    Any       )     ->		int:




           '''simple docstring'''


           return self.tokenizer.decode(*lowerCamelCase__						,**lowerCamelCase__       )






     @contextmanager
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[Any]       )     ->		Union[str, Any]:




           '''simple docstring'''


           warnings.warn(
               """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
               """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
               """your images inputs, or in a separate call."""       )
           SCREAMING_SNAKE_CASE		=						True
           SCREAMING_SNAKE_CASE		=						self.tokenizer
           yield
           SCREAMING_SNAKE_CASE		=						self.image_processor
           SCREAMING_SNAKE_CASE		=						False






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any						,lowerCamelCase__						:    Optional[int]						,lowerCamelCase__						:    Union[str, Any]=False						,lowerCamelCase__						:    Optional[int]=None       )     ->		List[str]:




           '''simple docstring'''


           if added_vocab is None:
                 SCREAMING_SNAKE_CASE		=						self.tokenizer.get_added_vocab()

           SCREAMING_SNAKE_CASE		=						{}

           while tokens:
                 SCREAMING_SNAKE_CASE		=						re.search(R"""<s_(.*?)>"""						,lowerCamelCase__						,re.IGNORECASE       )
                 if start_token is None:
                       break
                 SCREAMING_SNAKE_CASE		=						start_token.group(1       )
                 SCREAMING_SNAKE_CASE		=						re.search(RF"""</s_{key}>"""						,lowerCamelCase__						,re.IGNORECASE       )
                 SCREAMING_SNAKE_CASE		=						start_token.group()
                 if end_token is None:
                       SCREAMING_SNAKE_CASE		=						tokens.replace(lowerCamelCase__						,""""""       )
                 else:
                       SCREAMING_SNAKE_CASE		=						end_token.group()
                       SCREAMING_SNAKE_CASE		=						re.escape(lowerCamelCase__       )
                       SCREAMING_SNAKE_CASE		=						re.escape(lowerCamelCase__       )
                       SCREAMING_SNAKE_CASE		=						re.search(F"""{start_token_escaped}(.*?){end_token_escaped}"""						,lowerCamelCase__						,re.IGNORECASE       )
                       if content is not None:
                             SCREAMING_SNAKE_CASE		=						content.group(1       ).strip()
                             if r"<s_" in content and r"</s_" in content:  # non-leaf node
                                   SCREAMING_SNAKE_CASE		=						self.tokenajson(lowerCamelCase__						,is_inner_value=lowerCamelCase__						,added_vocab=lowerCamelCase__       )
                                   if value:
                                         if len(lowerCamelCase__       ) == 1:
                                               SCREAMING_SNAKE_CASE		=						value[0]
                                         SCREAMING_SNAKE_CASE		=						value
                             else:  # leaf nodes
                                   SCREAMING_SNAKE_CASE		=						[]
                                   for leaf in content.split(R"""<sep/>"""       ):
                                         SCREAMING_SNAKE_CASE		=						leaf.strip()
                                         if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
                                               SCREAMING_SNAKE_CASE		=						leaf[1:-2]  # for categorical special tokens
                                         output[key].append(lowerCamelCase__       )
                                   if len(output[key]       ) == 1:
                                         SCREAMING_SNAKE_CASE		=						output[key][0]

                       SCREAMING_SNAKE_CASE		=						tokens[tokens.find(lowerCamelCase__       ) + len(lowerCamelCase__       ) :].strip()
                       if tokens[:6] == r"<sep/>":  # non-leaf nodes
                             return [output] + self.tokenajson(tokens[6:]						,is_inner_value=lowerCamelCase__						,added_vocab=lowerCamelCase__       )

           if len(lowerCamelCase__       ):
                 return [output] if is_inner_value else output
           else:
                 return [] if is_inner_value else {"text_sequence": tokens}






     @property
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Tuple       )     ->		Optional[Any]:




           '''simple docstring'''


           warnings.warn(
               """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""						,lowerCamelCase__						,)
           return self.image_processor_class






     @property
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		List[str]:




           '''simple docstring'''


           warnings.warn(
               """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""						,lowerCamelCase__						,)
           return self.image_processor









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



     '''simple docstring'''

     __snake_case    :	List[str]			=			TextToVideoSDPipeline
     __snake_case    :	int			=			TEXT_TO_IMAGE_PARAMS
     __snake_case    :	Tuple			=			TEXT_TO_IMAGE_BATCH_PARAMS
     # No `output_type`.
     __snake_case    :	Dict			=			frozenset(
         [
             "num_inference_steps",
             "generator",
             "latents",
             "return_dict",
             "callback",
             "callback_steps",
         ]   )






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]       )     ->		Optional[int]:




           '''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=lowerCamelCase__						,set_alpha_to_one=lowerCamelCase__						,)
           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=128						,)
           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-0_5						,num_attention_heads=4						,num_hidden_layers=5						,pad_token_id=1						,vocab_size=1000						,hidden_act="""gelu"""						,projection_dim=512						,)
           SCREAMING_SNAKE_CASE		=						CLIPTextModel(lowerCamelCase__       )
           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 				SCREAMING_SNAKE_CASE__  (							self						:    int						,lowerCamelCase__						:    Optional[Any]						,lowerCamelCase__						:    int=0       )     ->		List[Any]:




           '''simple docstring'''


           if str(lowerCamelCase__       ).startswith("""mps"""       ):
                 SCREAMING_SNAKE_CASE		=						torch.manual_seed(lowerCamelCase__       )
           else:
                 SCREAMING_SNAKE_CASE		=						torch.Generator(device=lowerCamelCase__       ).manual_seed(lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						{
               """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						:    Optional[Any]       )     ->		List[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		=						TextToVideoSDPipeline(**lowerCamelCase__       )
           SCREAMING_SNAKE_CASE		=						sd_pipe.to(lowerCamelCase__       )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase__       )

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

           assert frames[0].shape == (64, 64, 3)
           SCREAMING_SNAKE_CASE		=						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						:    Dict       )     ->		int:




           '''simple docstring'''


           self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCamelCase__						,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						:    Tuple       )     ->		Any:




           '''simple docstring'''


           self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCamelCase__						,expected_max_diff=1e-2       )






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




           '''simple docstring'''


           pass






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




           '''simple docstring'''


           pass






     @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""       )
     def 				SCREAMING_SNAKE_CASE__  (							self						:    Optional[int]       )     ->		Tuple:




           '''simple docstring'''


           pass






     def 				SCREAMING_SNAKE_CASE__  (							self						:    List[str]       )     ->		Optional[Any]:




           '''simple docstring'''


           return super().test_progress_bar()

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



     '''simple docstring'''

     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"""       )

           SCREAMING_SNAKE_CASE		=						TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""       )
           SCREAMING_SNAKE_CASE		=						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config       )
           SCREAMING_SNAKE_CASE		=						pipe.to("""cuda"""       )

           SCREAMING_SNAKE_CASE		=						"""Spiderman is surfing"""
           SCREAMING_SNAKE_CASE		=						torch.Generator(device="""cpu"""       ).manual_seed(0       )

           SCREAMING_SNAKE_CASE		=						pipe(lowerCamelCase__						,generator=lowerCamelCase__						,num_inference_steps=25						,output_type="""pt"""       ).frames
           SCREAMING_SNAKE_CASE		=						video_frames.cpu().numpy()

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






     def 				SCREAMING_SNAKE_CASE__  (							self						:    Any       )     ->		Any:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE		=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"""       )

           SCREAMING_SNAKE_CASE		=						TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""       )
           SCREAMING_SNAKE_CASE		=						pipe.to("""cuda"""       )

           SCREAMING_SNAKE_CASE		=						"""Spiderman is surfing"""
           SCREAMING_SNAKE_CASE		=						torch.Generator(device="""cpu"""       ).manual_seed(0       )

           SCREAMING_SNAKE_CASE		=						pipe(lowerCamelCase__						,generator=lowerCamelCase__						,num_inference_steps=2						,output_type="""pt"""       ).frames
           SCREAMING_SNAKE_CASE		=						video_frames.cpu().numpy()

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


