"""simple docstring"""



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




    """simple docstring"""




    UpperCamelCase_	:					Union[str, Any]							=					TextToVideoSDPipeline
    UpperCamelCase_	:					List[Any]							=					TEXT_TO_IMAGE_PARAMS
    UpperCamelCase_	:					str							=					TEXT_TO_IMAGE_BATCH_PARAMS
    # No `output_type`.
    UpperCamelCase_	:					str							=					frozenset(
        [
            '''num_inference_steps''',
            '''generator''',
            '''latents''',
            '''return_dict''',
            '''callback''',
            '''callback_steps''',
        ]					)
    def  lowercase					(		self							:     Union[str, Any]    )    ->       Dict:
         torch.manual_seed(0    )
         __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	,      )
         __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    )
         __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    )
         __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=1_000	,      hidden_act='''gelu'''	,      projection_dim=512	,      )
         __snake_case				=							CLIPTextModel(A_    )
         __snake_case				=							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''    )

         __snake_case				=							{
             '''unet''': unet,
             '''scheduler''': scheduler,
             '''vae''': vae,
             '''text_encoder''': text_encoder,
             '''tokenizer''': tokenizer,
         }
         return components
    def  lowercase					(		self							:     Dict	,      A_							:     int	,      A_							:     List[Any]=0    )    ->       Any:
         if str(A_    ).startswith('''mps'''    ):
              __snake_case				=							torch.manual_seed(A_    )
         else:
              __snake_case				=							torch.Generator(device=A_    ).manual_seed(A_    )
         __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  lowercase					(		self							:     int    )    ->       Union[str, Any]:
         __snake_case				=							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
         __snake_case				=							self.get_dummy_components()
         __snake_case				=							TextToVideoSDPipeline(**A_    )
         __snake_case				=							sd_pipe.to(A_    )
         sd_pipe.set_progress_bar_config(disable=A_    )

         __snake_case				=							self.get_dummy_inputs(A_    )
         __snake_case				=							'''np'''
         __snake_case				=							sd_pipe(**A_    ).frames
         __snake_case				=							frames[0][-3:, -3:, -1]

         assert frames[0].shape == (64, 64, 3)
         __snake_case				=							np.array([1_58.0, 1_60.0, 1_53.0, 1_25.0, 1_00.0, 1_21.0, 1_11.0, 93.0, 1_13.0]    )

         assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1E-2
    def  lowercase					(		self							:     Optional[int]    )    ->       Dict:
         self._test_attention_slicing_forward_pass(test_mean_pixel_difference=A_	,      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  lowercase					(		self							:     str    )    ->       Optional[int]:
         self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A_	,      expected_max_diff=1E-2    )
    @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''    )
    def  lowercase					(		self							:     Dict    )    ->       int:
         pass
    @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''    )
    def  lowercase					(		self							:     int    )    ->       List[Any]:
         pass
    @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''    )
    def  lowercase					(		self							:     Dict    )    ->       Any:
         pass







    def  lowercase					(		self							:     Optional[Any]    )    ->       Dict:
         return super().test_progress_bar()






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




    """simple docstring"""




    def  lowercase					(		self							:     List[str]    )    ->       Any:
         __snake_case				=							load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''    )

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

         __snake_case				=							'''Spiderman is surfing'''
         __snake_case				=							torch.Generator(device='''cpu'''    ).manual_seed(0    )

         __snake_case				=							pipe(A_	,      generator=A_	,      num_inference_steps=25	,      output_type='''pt'''    ).frames
         __snake_case				=							video_frames.cpu().numpy()

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







    def  lowercase					(		self							:     List[Any]    )    ->       Optional[int]:
         __snake_case				=							load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'''    )

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

         __snake_case				=							'''Spiderman is surfing'''
         __snake_case				=							torch.Generator(device='''cpu'''    ).manual_seed(0    )

         __snake_case				=							pipe(A_	,      generator=A_	,      num_inference_steps=2	,      output_type='''pt'''    ).frames
         __snake_case				=							video_frames.cpu().numpy()

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



import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_blenderbot import BlenderbotTokenizer


if TYPE_CHECKING:
       from transformers.pipelines.conversational import Conversation

__lowercase		:		int														=							logging.get_logger(__name__)


__lowercase		:		Tuple														=							{
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
    "tokenizer_config_file": "tokenizer_config.json",
}

__lowercase		:		List[Any]														=							{
    "vocab_file": {"facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json"},
    "merges_file": {"facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt"},
    "tokenizer_config_file": {
        "facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json"
    },
}

__lowercase		:		List[str]														=							{"facebook/blenderbot-3B": 128}



class      _A	(						_UpperCAmelCase					):




    """simple docstring"""




    UpperCamelCase_	:					Optional[Any]							=					VOCAB_FILES_NAMES
    UpperCamelCase_	:					int							=					PRETRAINED_VOCAB_FILES_MAP
    UpperCamelCase_	:					Tuple							=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    UpperCamelCase_	:					int							=					['''input_ids''', '''attention_mask''']
    UpperCamelCase_	:					str							=					BlenderbotTokenizer
    def __init__(		self							:     Union[str, Any]	,      A_							:     Any=None	,      A_							:     Optional[int]=None	,      A_							:     Optional[Any]=None	,      A_							:     Union[str, Any]="replace"	,      A_							:     Union[str, Any]="<s>"	,      A_							:     Union[str, Any]="</s>"	,      A_							:     Optional[int]="</s>"	,      A_							:     List[Any]="<s>"	,      A_							:     Union[str, Any]="<unk>"	,      A_							:     Any="<pad>"	,      A_							:     List[str]="<mask>"	,      A_							:     int=False	,      A_							:     Tuple=True	,      **A_							:     Optional[Any]	,      )    ->       int:
         super().__init__(
             A_	,      A_	,      tokenizer_file=A_	,      errors=A_	,      bos_token=A_	,      eos_token=A_	,      sep_token=A_	,      cls_token=A_	,      unk_token=A_	,      pad_token=A_	,      mask_token=A_	,      add_prefix_space=A_	,      trim_offsets=A_	,      **A_	,      )

         __snake_case				=							json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()    )
         if pre_tok_state.get('''add_prefix_space'''	,      A_    ) != add_prefix_space:
              __snake_case				=							getattr(A_	,      pre_tok_state.pop('''type'''    )    )
              __snake_case				=							add_prefix_space
              __snake_case				=							pre_tok_class(**A_    )

         __snake_case				=							add_prefix_space

         __snake_case				=							'''post_processor'''
         __snake_case				=							getattr(self.backend_tokenizer	,      A_	,      A_    )
         if tokenizer_component_instance:
              __snake_case				=							json.loads(tokenizer_component_instance.__getstate__()    )

              # The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
              if "sep" in state:
                   __snake_case				=							tuple(state['''sep''']    )
              if "cls" in state:
                   __snake_case				=							tuple(state['''cls''']    )

              __snake_case				=							False

              if state.get('''add_prefix_space'''	,      A_    ) != add_prefix_space:
                   __snake_case				=							add_prefix_space
                   __snake_case				=							True

              if state.get('''trim_offsets'''	,      A_    ) != trim_offsets:
                   __snake_case				=							trim_offsets
                   __snake_case				=							True

              if changes_to_apply:
                   __snake_case				=							getattr(A_	,      state.pop('''type'''    )    )
                   __snake_case				=							component_class(**A_    )
                   setattr(self.backend_tokenizer	,      A_	,      A_    )
    @property
    # Copied from transformers.models.roberta.tokenization_roberta_fast.RobertaTokenizerFast.mask_token with Roberta->Blenderbot, RoBERTa->Blenderbot
    def  lowercase					(		self							:     List[Any]    )    ->       str:
         if self._mask_token is None:
              if self.verbose:
                   logger.error('''Using mask_token, but it is not set yet.'''    )
              return None
         return str(self._mask_token    )
    @mask_token.setter
    def  lowercase					(		self							:     int	,      A_							:     Union[str, Any]    )    ->       List[Any]:
         __snake_case				=							AddedToken(A_	,      lstrip=A_	,      rstrip=A_    ) if isinstance(A_	,      A_    ) else value
         __snake_case				=							value
    def  lowercase					(		self							:     Any	,      *A_							:     List[str]	,      **A_							:     Optional[int]    )    ->       BatchEncoding:
         __snake_case				=							kwargs.get('''is_split_into_words'''	,      A_    )
         assert self.add_prefix_space or not is_split_into_words, (
             f"You need to instantiate {self.__class__.__name__} with add_prefix_space=True "
             "to use it with pretokenized inputs."
         )

         return super()._batch_encode_plus(*A_	,      **A_    )
    def  lowercase					(		self							:     str	,      *A_							:     Tuple	,      **A_							:     str    )    ->       BatchEncoding:
         __snake_case				=							kwargs.get('''is_split_into_words'''	,      A_    )

         assert self.add_prefix_space or not is_split_into_words, (
             f"You need to instantiate {self.__class__.__name__} with add_prefix_space=True "
             "to use it with pretokenized inputs."
         )

         return super()._encode_plus(*A_	,      **A_    )
    def  lowercase					(		self							:     str	,      A_							:     str	,      A_							:     Optional[str] = None    )    ->       Tuple[str]:
         __snake_case				=							self._tokenizer.model.save(A_	,      name=A_    )
         return tuple(A_    )
    def  lowercase					(		self							:     Tuple	,      A_							:     List[int]	,      A_							:     Optional[List[int]] = None    )    ->       List[int]:
         __snake_case				=							[self.sep_token_id]
         __snake_case				=							[self.cls_token_id]

         if token_ids_a is None:
              return len(cls + token_ids_a + sep    ) * [0]
         return len(cls + token_ids_a + sep + sep + token_ids_a + sep    ) * [0]
    def  lowercase					(		self							:     Tuple	,      A_							:     List[int]	,      A_							:     Optional[List[int]] = None    )    ->       Any:
         return token_ids_a + [self.eos_token_id]







    def  lowercase					(		self							:     Optional[Any]	,      A_							:     "Conversation"    )    ->       List[int]:
         __snake_case				=							[]
         for is_user, text in conversation.iter_texts():
              if is_user:
                   # We need to space prefix as it's being done within blenderbot
                   inputs.append(''' ''' + text    )
              else:
                   # Generated responses should contain them already.
                   inputs.append(A_    )

         __snake_case				=							'''  '''.join(A_    )
         __snake_case				=							self.encode(A_    )
         if len(A_    ) > self.model_max_length:
              __snake_case				=							input_ids[-self.model_max_length :]
              logger.warning(f"Trimmed input from conversation as it was longer than {self.model_max_length} tokens."    )
         return input_ids