

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






  """simple docstring"""


  a									= VideoToVideoSDPipeline
  a									= TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}	) - {"image", "width", "height"}
  a									= TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}	) - {"image"}
  a									= PipelineTesterMixin.required_optional_params - {"latents"}
  a									= False

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






  def 				lowercase_		(							self						:		str )		->							Tuple:
       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-05     ,       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 				lowercase_		(							self						:		str     ,       __lowerCamelCase						:		int     ,       __lowerCamelCase						:		Tuple=0 )		->							List[str]:
       # 3 frames
       SCREAMING_SNAKE_CASE__   =   floats_tensor((1, 3, 3, 32, 32)     ,       rng=random.Random(__lowerCamelCase ) ).to(__lowerCamelCase )

       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''',
           '''video''': video,
           '''generator''': generator,
           '''num_inference_steps''': 2,
           '''guidance_scale''': 6.0,
           '''output_type''': '''pt''',
       }
       return inputs






  def 				lowercase_		(							self						:		Optional[int] )		->							str:
       SCREAMING_SNAKE_CASE__   =   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
       SCREAMING_SNAKE_CASE__   =   self.get_dummy_components()
       SCREAMING_SNAKE_CASE__   =   VideoToVideoSDPipeline(**__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 == (32, 32, 3)
       SCREAMING_SNAKE_CASE__   =   np.array([106, 117, 113, 174, 137, 112, 148, 151, 131] )

       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						:		Tuple )		->							Tuple:
       self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__lowerCamelCase     ,       expected_max_diff=5e-3 )






  @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.''' )
  def 				lowercase_		(							self						:		Optional[int] )		->							List[str]:
       pass






  @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.''' )
  def 				lowercase_		(							self						:		List[str] )		->							Optional[int]:
       pass






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







  def 				lowercase_		(							self						:		List[Any] )		->							int:
       return super().test_progress_bar()






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






  """simple docstring"""


  def 				lowercase_		(							self						:		Dict )		->							Optional[int]:
       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, 1024, 576)     ,       generator=__lowerCamelCase )
       SCREAMING_SNAKE_CASE__   =   video.to('''cuda''' )

       SCREAMING_SNAKE_CASE__   =   '''Spiderman is surfing'''

       SCREAMING_SNAKE_CASE__   =   pipe(__lowerCamelCase     ,       video=__lowerCamelCase     ,       generator=__lowerCamelCase     ,       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





import datasets

from .evaluate import evaluate


_SCREAMING_SNAKE_CASE					:  Union[str, Any]             =      '''\
@inproceedings{Rajpurkar2016SQuAD10,
  title={SQuAD: 100, 000+ Questions for Machine Comprehension of Text},
  author={Pranav Rajpurkar and Jian Zhang and Konstantin Lopyrev and Percy Liang},
  booktitle={EMNLP},
  year={2016}
}
'''

_SCREAMING_SNAKE_CASE					:  Dict             =      '''
This metric wrap the official scoring script for version 1 of the Stanford Question Answering Dataset (SQuAD).

Stanford Question Answering Dataset (SQuAD) is a reading comprehension dataset, consisting of questions posed by
crowdworkers on a set of Wikipedia articles, where the answer to every question is a segment of text, or span,
from the corresponding reading passage, or the question might be unanswerable.
'''

_SCREAMING_SNAKE_CASE					:  str             =      '''
Computes SQuAD scores (F1 and EM).
Args:
    predictions: List of question-answers dictionaries with the following key-values:
        - \'id\': id of the question-answer pair as given in the references (see below)
        - \'prediction_text\': the text of the answer
    references: List of question-answers dictionaries with the following key-values:
        - \'id\': id of the question-answer pair (see above),
        - \'answers\': a Dict in the SQuAD dataset format
            {
                \'text\': list of possible texts for the answer, as a list of strings
                \'answer_start\': list of start positions for the answer, as a list of ints
            }
            Note that answer_start values are not taken into account to compute the metric.
Returns:
    \'exact_match\': Exact match (the normalized answer exactly match the gold answer)
    \'f1\': The F-score of predicted tokens versus the gold answer
Examples:

    >>> predictions = [{\'prediction_text\': \'1976\', \'id\': \'56e10a3be3433e1400422b22\'}]
    >>> references = [{\'answers\': {\'answer_start\': [97], \'text\': [\'1976\']}, \'id\': \'56e10a3be3433e1400422b22\'}]
    >>> squad_metric = datasets.load_metric("squad")
    >>> results = squad_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {\'exact_match\': 100.0, \'f1\': 100.0}
'''






@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION						,					_KWARGS_DESCRIPTION	)
class 	UpperCAmelCase__		(	datasets.Metric	):






  """simple docstring"""


  def 				lowercase_		(							self						:		List[Any] )		->							Optional[int]:
       return datasets.MetricInfo(
           description=_DESCRIPTION     ,       citation=_CITATION     ,       inputs_description=_KWARGS_DESCRIPTION     ,       features=datasets.Features(
               {
                   '''predictions''': {'''id''': datasets.Value('''string''' ), '''prediction_text''': datasets.Value('''string''' )},
                   '''references''': {
                       '''id''': datasets.Value('''string''' ),
                       '''answers''': datasets.features.Sequence(
                           {
                               '''text''': datasets.Value('''string''' ),
                               '''answer_start''': datasets.Value('''int32''' ),
                           } ),
                   },
               } )     ,       codebase_urls=['''https://rajpurkar.github.io/SQuAD-explorer/''']     ,       reference_urls=['''https://rajpurkar.github.io/SQuAD-explorer/''']     ,       )







  def 				lowercase_		(							self						:		Optional[Any]     ,       __lowerCamelCase						:		Dict     ,       __lowerCamelCase						:		List[Any] )		->							Optional[int]:
       SCREAMING_SNAKE_CASE__   =   {prediction['''id''']: prediction['''prediction_text'''] for prediction in predictions}
       SCREAMING_SNAKE_CASE__   =   [
           {
               '''paragraphs''': [
                   {
                       '''qas''': [
                           {
                               '''answers''': [{'''text''': answer_text} for answer_text in ref['''answers''']['''text''']],
                               '''id''': ref['''id'''],
                           }
                           for ref in references
                       ]
                   }
               ]
           }
       ]
       SCREAMING_SNAKE_CASE__   =   evaluate(dataset=__lowerCamelCase     ,       predictions=__lowerCamelCase )
       return score


