
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        lowercase      (	_SCREAMING_SNAKE_CASE						,     unittest.TestCase     ):
		__lowercase : Union[str, Any]			  =       VideoToVideoSDPipeline
		__lowercase : Any			  =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}     ) - {"image", "width", "height"}
		__lowercase : Optional[Any]			  =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}     ) - {"image"}
		__lowercase : str			  =       PipelineTesterMixin.required_optional_params - {"latents"}
		__lowercase : int			  =       False

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



		def __UpperCamelCase				(				self )						->      Optional[int]:
				"""simple docstring"""
				torch.manual_seed(0 )
				UpperCamelCase						=  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       ,   )
				UpperCamelCase						=  DDIMScheduler(
				    beta_start=0.0_0085       ,   beta_end=0.012       ,   beta_schedule='scaled_linear'       ,   clip_sample=A_       ,   set_alpha_to_one=A_       ,   )
				torch.manual_seed(0 )
				UpperCamelCase						=  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 )
				UpperCamelCase						=  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=1_000       ,   hidden_act='gelu'       ,   projection_dim=512       ,   )
				UpperCamelCase						=  CLIPTextModel(A_ )
				UpperCamelCase						=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

				UpperCamelCase						=  {
				    'unet': unet,
				    'scheduler': scheduler,
				    'vae': vae,
				    'text_encoder': text_encoder,
				    'tokenizer': tokenizer,
				}
				return components



		def __UpperCamelCase				(				self       ,   A_       ,   A_=0 )						->      Any:
				"""simple docstring"""
				# 3 frames
				UpperCamelCase						=  floats_tensor((1, 3, 3, 32, 32)       ,   rng=random.Random(A_ ) ).to(A_ )

				if str(A_ ).startswith('mps' ):
						UpperCamelCase						=  torch.manual_seed(A_ )
				else:
						UpperCamelCase						=  torch.Generator(device=A_ ).manual_seed(A_ )
				UpperCamelCase						=  {
				    '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 __UpperCamelCase				(				self )						->      Tuple:
				"""simple docstring"""
				UpperCamelCase						=  'cpu'  # ensure determinism for the device-dependent torch.Generator
				UpperCamelCase						=  self.get_dummy_components()
				UpperCamelCase						=  VideoToVideoSDPipeline(**A_ )
				UpperCamelCase						=  sd_pipe.to(A_ )
				sd_pipe.set_progress_bar_config(disable=A_ )

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

				assert frames[0].shape == (32, 32, 3)
				UpperCamelCase						=  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 __UpperCamelCase				(				self )						->      int:
				"""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 __UpperCamelCase				(				self )						->      Any:
				"""simple docstring"""
				pass



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



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



		def __UpperCamelCase				(				self )						->      int:
				"""simple docstring"""
				return super().test_progress_bar()






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



		def __UpperCamelCase				(				self )						->      Optional[int]:
				"""simple docstring"""
				UpperCamelCase						=  VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'       ,   torch_dtype=torch.floataa )
				pipe.enable_model_cpu_offload()

				# 10 frames
				UpperCamelCase						=  torch.Generator(device='cpu' ).manual_seed(0 )
				UpperCamelCase						=  torch.randn((1, 10, 3, 1_024, 576)       ,   generator=A_ )
				UpperCamelCase						=  video.to('cuda' )

				UpperCamelCase						=  'Spiderman is surfing'

				UpperCamelCase						=  pipe(A_       ,   video=A_       ,   generator=A_       ,   num_inference_steps=3       ,   output_type='pt' ).frames

				UpperCamelCase						=  np.array([-1.045_8984, -1.127_9297, -0.966_3086, -0.9150_3906, -0.7509_7656] )
				assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array ).sum() < 1e-2


from scipy.stats import pearsonr, spearmanr
from sklearn.metrics import fa_score, matthews_corrcoef

import datasets


_UpperCAmelCase      :				Any          = "\\n@inproceedings{wang2019glue,\n  title={{GLUE}: A Multi-Task Benchmark and Analysis Platform for Natural Language Understanding},\n  author={Wang, Alex and Singh, Amanpreet and Michael, Julian and Hill, Felix and Levy, Omer and Bowman, Samuel R.},\n  note={In the Proceedings of ICLR.},\n  year={2019}\n}\n"

_UpperCAmelCase      :				str          = "\\nGLUE, the General Language Understanding Evaluation benchmark\n(https://gluebenchmark.com/) is a collection of resources for training,\nevaluating, and analyzing natural language understanding systems.\n"

_UpperCAmelCase      :				List[str]          = "\nCompute GLUE evaluation metric associated to each GLUE dataset.\nArgs:\n    predictions: list of predictions to score.\n        Each translation should be tokenized into a list of tokens.\n    references: list of lists of references for each translation.\n        Each reference should be tokenized into a list of tokens.\nReturns: depending on the GLUE subset, one or several of:\n    \"accuracy\": Accuracy\n    \"f1\": F1 score\n    \"pearson\": Pearson Correlation\n    \"spearmanr\": Spearman Correlation\n    \"matthews_correlation\": Matthew Correlation\nExamples:\n\n    >>> glue_metric = datasets.load_metric('glue', 'sst2')  # 'sst2' or any of [\"mnli\", \"mnli_mismatched\", \"mnli_matched\", \"qnli\", \"rte\", \"wnli\", \"hans\"]\n    >>> references = [0, 1]\n    >>> predictions = [0, 1]\n    >>> results = glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {'accuracy': 1.0}\n\n    >>> glue_metric = datasets.load_metric('glue', 'mrpc')  # 'mrpc' or 'qqp'\n    >>> references = [0, 1]\n    >>> predictions = [0, 1]\n    >>> results = glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {'accuracy': 1.0, 'f1': 1.0}\n\n    >>> glue_metric = datasets.load_metric('glue', 'stsb')\n    >>> references = [0., 1., 2., 3., 4., 5.]\n    >>> predictions = [0., 1., 2., 3., 4., 5.]\n    >>> results = glue_metric.compute(predictions=predictions, references=references)\n    >>> print({\"pearson\": round(results[\"pearson\"], 2), \"spearmanr\": round(results[\"spearmanr\"], 2)})\n    {'pearson': 1.0, 'spearmanr': 1.0}\n\n    >>> glue_metric = datasets.load_metric('glue', 'cola')\n    >>> references = [0, 1]\n    >>> predictions = [0, 1]\n    >>> results = glue_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {'matthews_correlation': 1.0}\n"


def    A (      lowercase					, lowercase				)     ->						List[str]:
		'''simple docstring'''

		return float((preds == labels).mean()				)


def    A (      lowercase					, lowercase				)     ->						Tuple:
		'''simple docstring'''

		UpperCamelCase						=  simple_accuracy(lowercase					, lowercase				)
		UpperCamelCase						=  float(fa_score(y_true=lowercase					, y_pred=lowercase				)				)
		return {
		    "accuracy": acc,
		    "f1": fa,
		}


def    A (      lowercase					, lowercase				)     ->						Optional[int]:
		'''simple docstring'''

		UpperCamelCase						=  float(pearsonr(lowercase					, lowercase				)[0]				)
		UpperCamelCase						=  float(spearmanr(lowercase					, lowercase				)[0]				)
		return {
		    "pearson": pearson_corr,
		    "spearmanr": spearman_corr,
		}


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



		def __UpperCamelCase				(				self )						->      Optional[Any]:
				"""simple docstring"""
				if self.config_name not in [
				    "sst2",
				    "mnli",
				    "mnli_mismatched",
				    "mnli_matched",
				    "cola",
				    "stsb",
				    "mrpc",
				    "qqp",
				    "qnli",
				    "rte",
				    "wnli",
				    "hans",
				]:
						raise KeyError(
						    'You should supply a configuration name selected in '
						    '["sst2", "mnli", "mnli_mismatched", "mnli_matched", '
						    '"cola", "stsb", "mrpc", "qqp", "qnli", "rte", "wnli", "hans"]' )
				return datasets.MetricInfo(
				    description=_DESCRIPTION       ,   citation=_CITATION       ,   inputs_description=_KWARGS_DESCRIPTION       ,   features=datasets.Features(
				        {
				            'predictions': datasets.Value('int64' if self.config_name != 'stsb' else 'float32' ),
				            'references': datasets.Value('int64' if self.config_name != 'stsb' else 'float32' ),
				        } )       ,   codebase_urls=[]       ,   reference_urls=[]       ,   format='numpy'       ,   )



		def __UpperCamelCase				(				self       ,   A_       ,   A_ )						->      Any:
				"""simple docstring"""
				if self.config_name == "cola":
						return {"matthews_correlation": matthews_corrcoef(A_       ,   A_ )}
				elif self.config_name == "stsb":
						return pearson_and_spearman(A_       ,   A_ )
				elif self.config_name in ["mrpc", "qqp"]:
						return acc_and_fa(A_       ,   A_ )
				elif self.config_name in ["sst2", "mnli", "mnli_mismatched", "mnli_matched", "qnli", "rte", "wnli", "hans"]:
						return {"accuracy": simple_accuracy(A_       ,   A_ )}
				else:
						raise KeyError(
						    'You should supply a configuration name selected in '
						    '["sst2", "mnli", "mnli_mismatched", "mnli_matched", '
						    '"cola", "stsb", "mrpc", "qqp", "qnli", "rte", "wnli", "hans"]' )
