





import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__A             =       logging.get_logger(__name__)

__A             =       {"vocab_file": "sentencepiece.bpe.model"}

__A             =       {
    "vocab_file": {
        "moussaKam/mbarthez": "https://huggingface.co/moussaKam/mbarthez/resolve/main/sentencepiece.bpe.model",
        "moussaKam/barthez": "https://huggingface.co/moussaKam/barthez/resolve/main/sentencepiece.bpe.model",
        "moussaKam/barthez-orangesum-title": (
            "https://huggingface.co/moussaKam/barthez-orangesum-title/resolve/main/sentencepiece.bpe.model"
        ),
    },
}

__A             =       {
    "moussaKam/mbarthez": 1024,
    "moussaKam/barthez": 1024,
    "moussaKam/barthez-orangesum-title": 1024,
}

__A             =       "▁"



class      _SCREAMING_SNAKE_CASE		(			__SCREAMING_SNAKE_CASE					):







 '''simple docstring'''
 lowercase_	=    VOCAB_FILES_NAMES
 lowercase_	=    PRETRAINED_VOCAB_FILES_MAP
 lowercase_	=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 lowercase_	=    ["input_ids", "attention_mask"]

 def __init__(self       :     str ,				UpperCAmelCase_       :     Tuple ,				UpperCAmelCase_       :     List[Any]="<s>" ,				UpperCAmelCase_       :     Dict="</s>" ,				UpperCAmelCase_       :     Union[str, Any]="</s>" ,				UpperCAmelCase_       :     Tuple="<s>" ,				UpperCAmelCase_       :     int="<unk>" ,				UpperCAmelCase_       :     Union[str, Any]="<pad>" ,				UpperCAmelCase_       :     Any="<mask>" ,				UpperCAmelCase_       :     Optional[Dict[str, Any]] = None ,				**UpperCAmelCase_       :     Optional[int] ,				) ->Tuple:



   '''simple docstring'''



   lowerCamelCase__:   Optional[Any]    						=AddedToken(_a ,				lstrip=_a ,				rstrip=_a) if isinstance(_a ,				_a) else mask_token

   lowerCamelCase__:   Any    						={} if sp_model_kwargs is None else sp_model_kwargs

   super().__init__(
       bos_token=_a ,				eos_token=_a ,				unk_token=_a ,				sep_token=_a ,				cls_token=_a ,				pad_token=_a ,				mask_token=_a ,				sp_model_kwargs=self.sp_model_kwargs ,				**_a ,				)

   lowerCamelCase__:   Any    						=vocab_file
   lowerCamelCase__:   Optional[int]    						=spm.SentencePieceProcessor(**self.sp_model_kwargs)
   self.sp_model.Load(str(_a))

   lowerCamelCase__:   str    						={"""<s>""": 0, """<pad>""": 1, """</s>""": 2, """<unk>""": 3}

   lowerCamelCase__:   Optional[Any]    						=len(self.sp_model) - 1
   lowerCamelCase__:   Optional[int]    						={v: k for k, v in self.fairseq_tokens_to_ids.items()}

 def 							SCREAMING_SNAKE_CASE_      (self       :     Optional[Any] ,				UpperCAmelCase_       :     List[int] ,				UpperCAmelCase_       :     Optional[List[int]] = None) ->List[Any]:



   '''simple docstring'''



   if token_ids_a is None:
     return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
   lowerCamelCase__:   Tuple    						=[self.cls_token_id]
   lowerCamelCase__:   List[str]    						=[self.sep_token_id]
   return cls + token_ids_a + sep + sep + token_ids_a + sep

 def 							SCREAMING_SNAKE_CASE_      (self       :     Any ,				UpperCAmelCase_       :     List[int] ,				UpperCAmelCase_       :     Optional[List[int]] = None ,				UpperCAmelCase_       :     bool = False) ->Union[str, Any]:



   '''simple docstring'''



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

   if token_ids_a is None:
     return [1] + ([0] * len(_a)) + [1]
   return [1] + ([0] * len(_a)) + [1, 1] + ([0] * len(_a)) + [1]

 def 							SCREAMING_SNAKE_CASE_      (self       :     Any ,				UpperCAmelCase_       :     List[int] ,				UpperCAmelCase_       :     Optional[List[int]] = None) ->int:



   '''simple docstring'''



   lowerCamelCase__:   str    						=[self.sep_token_id]
   lowerCamelCase__:   List[str]    						=[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]

 @property
 def 							SCREAMING_SNAKE_CASE_      (self       :     Optional[int]) ->int:



   '''simple docstring'''



   return len(self.sp_model)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Any) ->int:



   '''simple docstring'''



   lowerCamelCase__:   Tuple    						={self.convert_ids_to_tokens(_a): i for i in range(self.vocab_size)}
   vocab.update(self.added_tokens_encoder)
   return vocab

 def 							SCREAMING_SNAKE_CASE_      (self       :     str ,				UpperCAmelCase_       :     str) ->str:



   '''simple docstring'''



   return self.sp_model.encode(_a ,				out_type=_a)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Dict ,				UpperCAmelCase_       :     List[str]) ->Union[str, Any]:



   '''simple docstring'''



   if token in self.fairseq_tokens_to_ids:
     return self.fairseq_tokens_to_ids[token]
   lowerCamelCase__:   str    						=self.sp_model.PieceToId(_a)

   return spm_id if spm_id else self.unk_token_id

 def 							SCREAMING_SNAKE_CASE_      (self       :     Optional[int] ,				UpperCAmelCase_       :     List[Any]) ->int:



   '''simple docstring'''



   if index in self.fairseq_ids_to_tokens:
     return self.fairseq_ids_to_tokens[index]
   return self.sp_model.IdToPiece(_a)

 def 							SCREAMING_SNAKE_CASE_      (self       :     Dict ,				UpperCAmelCase_       :     Optional[Any]) ->Any:



   '''simple docstring'''



   lowerCamelCase__:   Optional[Any]    						=[]
   lowerCamelCase__:   int    						=""""""
   lowerCamelCase__:   int    						=False
   for token in tokens:
     # make sure that special tokens are not decoded using sentencepiece model
     if token in self.all_special_tokens:
       if not prev_is_special:
         out_string += " "
       out_string += self.sp_model.decode(_a) + token
       lowerCamelCase__:   int    						=True
       lowerCamelCase__:   Optional[int]    						=[]
     else:
       current_sub_tokens.append(_a)
       lowerCamelCase__:   Optional[Any]    						=False
   out_string += self.sp_model.decode(_a)
   return out_string.strip()

 def __getstate__(self       :     List[str]) ->str:



   '''simple docstring'''



   lowerCamelCase__:   str    						=self.__dict__.copy()
   lowerCamelCase__:   Any    						=None
   return state

 def __setstate__(self       :     Union[str, Any] ,				UpperCAmelCase_       :     Union[str, Any]) ->Optional[int]:



   '''simple docstring'''



   lowerCamelCase__:   Tuple    						=d

   # for backward compatibility
   if not hasattr(self ,				"sp_model_kwargs"):
     lowerCamelCase__:   Optional[int]    						={}

   lowerCamelCase__:   Optional[int]    						=spm.SentencePieceProcessor(**self.sp_model_kwargs)
   self.sp_model.Load(self.vocab_file)







 def 							SCREAMING_SNAKE_CASE_      (self       :     List[Any] ,				UpperCAmelCase_       :     str ,				UpperCAmelCase_       :     Optional[str] = None) ->List[Any]:



   '''simple docstring'''



   if not os.path.isdir(_a):
     logger.error(F"""Vocabulary path ({save_directory}) should be a directory""")
     return
   lowerCamelCase__:   Dict    						=os.path.join(
       _a ,				(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])

   if os.path.abspath(self.vocab_file) != os.path.abspath(_a) and os.path.isfile(self.vocab_file):
     copyfile(self.vocab_file ,				_a)
   elif not os.path.isfile(self.vocab_file):
     with open(_a ,				"wb") as fi:
       lowerCamelCase__:   int    						=self.sp_model.serialized_model_proto()
       fi.write(_a)

   return (out_vocab_file,)








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







 '''simple docstring'''
 lowercase_	=    TextToVideoSDPipeline
 lowercase_	=    TEXT_TO_IMAGE_PARAMS
 lowercase_	=    TEXT_TO_IMAGE_BATCH_PARAMS
 # No `output_type`.
 lowercase_	=    frozenset(
     [
         "num_inference_steps",
         "generator",
         "latents",
         "return_dict",
         "callback",
         "callback_steps",
     ]					)

 def 							SCREAMING_SNAKE_CASE_      (self       :     List[str]) ->List[str]:



   '''simple docstring'''



   torch.manual_seed(0)
   lowerCamelCase__:   Optional[int]    						=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 ,				)
   lowerCamelCase__:   Union[str, Any]    						=DDIMScheduler(
       beta_start=0.0_0085 ,				beta_end=0.012 ,				beta_schedule="scaled_linear" ,				clip_sample=UpperCAmelCase_ ,				set_alpha_to_one=UpperCAmelCase_ ,				)
   torch.manual_seed(0)
   lowerCamelCase__:   List[str]    						=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)
   lowerCamelCase__:   Optional[int]    						=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 ,				)
   lowerCamelCase__:   Optional[Any]    						=CLIPTextModel(UpperCAmelCase_)
   lowerCamelCase__:   Dict    						=CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

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

 def 							SCREAMING_SNAKE_CASE_      (self       :     str ,				UpperCAmelCase_       :     Dict ,				UpperCAmelCase_       :     List[Any]=0) ->Union[str, Any]:



   '''simple docstring'''



   if str(UpperCAmelCase_).startswith("mps"):
     lowerCamelCase__:   Optional[int]    						=torch.manual_seed(UpperCAmelCase_)
   else:
     lowerCamelCase__:   Any    						=torch.Generator(device=UpperCAmelCase_).manual_seed(UpperCAmelCase_)
   lowerCamelCase__:   List[Any]    						={
       "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       :     List[Any]) ->Optional[int]:



   '''simple docstring'''



   lowerCamelCase__:   List[str]    						="cpu"  # ensure determinism for the device-dependent torch.Generator
   lowerCamelCase__:   Optional[Any]    						=self.get_dummy_components()
   lowerCamelCase__:   List[Any]    						=TextToVideoSDPipeline(**UpperCAmelCase_)
   lowerCamelCase__:   int    						=sd_pipe.to(UpperCAmelCase_)
   sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_)

   lowerCamelCase__:   Any    						=self.get_dummy_inputs(UpperCAmelCase_)
   lowerCamelCase__:   List[Any]    						="np"
   lowerCamelCase__:   Optional[int]    						=sd_pipe(**UpperCAmelCase_).frames
   lowerCamelCase__:   Dict    						=frames[0][-3:, -3:, -1]

   assert frames[0].shape == (64, 64, 3)
   lowerCamelCase__:   Optional[int]    						=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       :     Tuple) ->str:



   '''simple docstring'''



   self._test_attention_slicing_forward_pass(test_mean_pixel_difference=UpperCAmelCase_ ,				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       :     Optional[Any]) ->List[Any]:



   '''simple docstring'''



   self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=UpperCAmelCase_ ,				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[int]) ->str:



   '''simple docstring'''



   pass

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



   '''simple docstring'''



   pass

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



   '''simple docstring'''



   pass







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



   '''simple docstring'''



   return super().test_progress_bar()



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







 '''simple docstring'''

 def 							SCREAMING_SNAKE_CASE_      (self       :     List[Any]) ->int:



   '''simple docstring'''



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

   lowerCamelCase__:   Optional[int]    						=TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b")
   lowerCamelCase__:   str    						=DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
   lowerCamelCase__:   Tuple    						=pipe.to("cuda")

   lowerCamelCase__:   List[Any]    						="Spiderman is surfing"
   lowerCamelCase__:   Dict    						=torch.Generator(device="cpu").manual_seed(0)

   lowerCamelCase__:   Optional[int]    						=pipe(UpperCAmelCase_ ,				generator=UpperCAmelCase_ ,				num_inference_steps=25 ,				output_type="pt").frames
   lowerCamelCase__:   str    						=video_frames.cpu().numpy()

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







 def 							SCREAMING_SNAKE_CASE_      (self       :     List[str]) ->Optional[int]:



   '''simple docstring'''



   lowerCamelCase__:   Optional[Any]    						=load_numpy(
       "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy")

   lowerCamelCase__:   List[str]    						=TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b")
   lowerCamelCase__:   Any    						=pipe.to("cuda")

   lowerCamelCase__:   Dict    						="Spiderman is surfing"
   lowerCamelCase__:   Dict    						=torch.Generator(device="cpu").manual_seed(0)

   lowerCamelCase__:   Optional[int]    						=pipe(UpperCAmelCase_ ,				generator=UpperCAmelCase_ ,				num_inference_steps=2 ,				output_type="pt").frames
   lowerCamelCase__:   List[Any]    						=video_frames.cpu().numpy()

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

