
import doctest
import glob
import importlib
import inspect
import os
import re
from contextlib import contextmanager
from functools import wraps
from unittest.mock import patch

import numpy as np
import pytest
from absl.testing import parameterized

import datasets
from datasets import load_metric

from .utils import for_all_test_methods, local, slow


# mark all tests as integration
_lowercase    =		pytest.mark.integration


_lowercase    =		{"""comet"""}
_lowercase    =		importlib.util.find_spec("""fairseq""") is not None

_lowercase    =		{"""code_eval"""}
_lowercase    =		os.name == """nt"""

_lowercase    =		{"""bertscore""", """frugalscore""", """perplexity"""}
_lowercase    =		importlib.util.find_spec("""transformers""") is not None


def A				(__lowerCamelCase				:Dict		):
						@wraps(lowercase__		)
						def wrapper(self				:Optional[Any]			,     __lowerCamelCase				:Optional[Any]		):
												if not _has_fairseq and metric_name in REQUIRE_FAIRSEQ:
																		self.skipTest("""\"test requires Fairseq\""""		)
												else:
																		test_case(self			,     lowercase__		)

						return wrapper


def A				(__lowerCamelCase				:Optional[int]		):
						@wraps(lowercase__		)
						def wrapper(self				:Tuple			,     __lowerCamelCase				:Optional[Any]		):
												if not _has_transformers and metric_name in REQUIRE_TRANSFORMERS:
																		self.skipTest("""\"test requires transformers\""""		)
												else:
																		test_case(self			,     lowercase__		)

						return wrapper


def A				(__lowerCamelCase				:Tuple		):
						@wraps(lowercase__		)
						def wrapper(self				:List[Any]			,     __lowerCamelCase				:Tuple		):
												if _on_windows and metric_name in UNSUPPORTED_ON_WINDOWS:
																		self.skipTest("""\"test not supported on Windows\""""		)
												else:
																		test_case(self			,     lowercase__		)

						return wrapper


def A				():
						_lowerCAmelCase  						=			[metric_dir.split(os.sep		)[-2] for metric_dir in glob.glob("""./metrics/*/"""		)]
						return [{"testcase_name": x, "metric_name": x} for x in metrics if x != "gleu"]  # gleu is unfinished




@parameterized.named_parameters(get_local_metric_names()			)
@for_all_test_methods(
    _SCREAMING_SNAKE_CASE	,							_SCREAMING_SNAKE_CASE	,							_SCREAMING_SNAKE_CASE			)
@local
class   UpperCAmelCase_					(		parameterized.TestCase			):






						'''simple docstring'''






						_lowercase		:				Tuple									=       {}
						_lowercase		:				List[Any]									=       None




						@pytest.mark.filterwarnings("""ignore:metric_module_factory is deprecated:FutureWarning"""				)
						@pytest.mark.filterwarnings("""ignore:load_metric is deprecated:FutureWarning"""				)
						def  _lowercase     ( self   ,       _lowercase				):

												"""simple docstring"""



												_lowerCAmelCase  						=			'[...]'
												_lowerCAmelCase  						=			importlib.import_module(
												    datasets.load.metric_module_factory(os.path.join("""metrics"""   ,       _lowercase				)				).module_path				)
												_lowerCAmelCase  						=			datasets.load.import_main_class(metric_module.__name__   ,       dataset=_lowercase				)
												# check parameters
												_lowerCAmelCase  						=			inspect.signature(metric._compute				).parameters
												self.assertTrue(all(p.kind != p.VAR_KEYWORD for p in parameters.values()				)				)  # no **kwargs
												# run doctest
												with self.patch_intensive_calls(_lowercase   ,       metric_module.__name__				):
																		with self.use_local_metrics():
																								try:
																														_lowerCAmelCase  						=			doctest.testmod(_lowercase   ,       verbose=_lowercase   ,       raise_on_error=_lowercase				)
																								except doctest.UnexpectedException as e:
																														raise e.exc_info[1]  # raise the exception that doctest caught
												self.assertEqual(results.failed   ,       0				)
												self.assertGreater(results.attempted   ,       1				)




						@slow
						def  _lowercase     ( self   ,       _lowercase				):

												"""simple docstring"""



												_lowerCAmelCase  						=			'[...]'
												_lowerCAmelCase  						=			importlib.import_module(
												    datasets.load.metric_module_factory(os.path.join("""metrics"""   ,       _lowercase				)				).module_path				)
												# run doctest
												with self.use_local_metrics():
																		_lowerCAmelCase  						=			doctest.testmod(_lowercase   ,       verbose=_lowercase   ,       raise_on_error=_lowercase				)
												self.assertEqual(results.failed   ,       0				)
												self.assertGreater(results.attempted   ,       1				)




						@contextmanager
						def  _lowercase     ( self   ,       _lowercase   ,       _lowercase				):

												"""simple docstring"""



												if metric_name in self.INTENSIVE_CALLS_PATCHER:
																		with self.INTENSIVE_CALLS_PATCHER[metric_name](_lowercase				):
																								yield
												else:
																		yield




						@contextmanager
						def  _lowercase     ( self				):

												"""simple docstring"""



												def load_local_metric(_lowercase   ,       *_lowercase   ,       **_lowercase				):
																		return load_metric(os.path.join("""metrics"""   ,       _lowercase				)   ,       *_lowercase   ,       **_lowercase				)

												with patch("""datasets.load_metric"""				) as mock_load_metric:
																		_lowerCAmelCase  						=			load_local_metric
																		yield




						@classmethod
						def  _lowercase     ( cls   ,       _lowercase				):

												"""simple docstring"""



												def wrapper(_lowercase				):
																		_lowerCAmelCase  						=			contextmanager(_lowercase				)
																		_lowerCAmelCase  						=			patcher
																		return patcher

												return wrapper






@LocalMetricTest.register_intensive_calls_patcher("""bleurt"""		)
def A				(__lowerCamelCase				:int		):
						import tensorflow.compat.va as tf
						from bleurt.score import Predictor

						tf.flags.DEFINE_string("""sv"""			,     """"""			,     """"""		)  # handle pytest cli flags




						class   UpperCAmelCase_					(		_SCREAMING_SNAKE_CASE			):






												'''simple docstring'''






												def  _lowercase     ( self   ,       _lowercase				):

																		"""simple docstring"""



																		assert len(input_dict["""input_ids"""]				) == 2
																		return np.array([1.03, 1.04]				)

    # mock predict_fn which is supposed to do a forward pass with a bleurt model
						with patch("""bleurt.score._create_predictor"""		) as mock_create_predictor:
												_lowerCAmelCase  						=			MockedPredictor()
												yield


@LocalMetricTest.register_intensive_calls_patcher("""bertscore"""		)
def A				(__lowerCamelCase				:Optional[Any]		):
						import torch

						def bert_cos_score_idf(__lowerCamelCase				:int			,     __lowerCamelCase				:Tuple			,     *__lowerCamelCase				:List[Any]			,     **__lowerCamelCase				:Any		):
												return torch.tensor([[1.0, 1.0, 1.0]] * len(lowercase__		)		)

						# mock get_model which is supposed to do download a bert model
						# mock bert_cos_score_idf which is supposed to do a forward pass with a bert model
						with patch("""bert_score.scorer.get_model"""		), patch(
						    """bert_score.scorer.bert_cos_score_idf"""		) as mock_bert_cos_score_idf:
												_lowerCAmelCase  						=			bert_cos_score_idf
												yield


@LocalMetricTest.register_intensive_calls_patcher("""comet"""		)
def A				(__lowerCamelCase				:List[Any]		):
						def load_from_checkpoint(__lowerCamelCase				:Tuple		):




												class   UpperCAmelCase_					:






																		'''simple docstring'''






																		def  _lowercase     ( self   ,       _lowercase   ,       *_lowercase   ,       **_lowercase				):

																								"""simple docstring"""



																								assert len(_lowercase				) == 2
																								_lowerCAmelCase  						=			[0.19, 0.92]
																								return scores, sum(_lowercase				) / len(_lowercase				)






												return Model()

						# mock load_from_checkpoint which is supposed to do download a bert model
						# mock load_from_checkpoint which is supposed to do download a bert model
						with patch("""comet.download_model"""		) as mock_download_model:
												_lowerCAmelCase  						=			None
												with patch("""comet.load_from_checkpoint"""		) as mock_load_from_checkpoint:
																		_lowerCAmelCase  						=			load_from_checkpoint
																		yield





def A				():
						_lowerCAmelCase  						=			load_metric(os.path.join("""metrics"""			,     """seqeval"""		)		)
						_lowerCAmelCase  						=			'ERROR'
						_lowerCAmelCase  						=			f'Scheme should be one of [IOB1, IOB2, IOE1, IOE2, IOBES, BILOU], got {wrong_scheme}'
						with pytest.raises(lowercase__			,     match=re.escape(lowercase__		)		):
												metric.compute(predictions=[]			,     references=[]			,     scheme=lowercase__		)


'''simple docstring'''



from typing import Callable, List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModel, CLIPTokenizer

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin, TransformeraDModel, VQModel
from ...schedulers import VQDiffusionScheduler
from ...utils import logging
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


_lowercase    =		logging.get_logger(__name__)  # pylint: disable=invalid-name




class   UpperCAmelCase_					(		_SCREAMING_SNAKE_CASE	,							_SCREAMING_SNAKE_CASE			):






						'''simple docstring'''






						@register_to_config
						def __init__( self   ,       _lowercase   ,       _lowercase = None   ,       _lowercase = None				):

												"""simple docstring"""



												super().__init__()

												_lowerCAmelCase  						=			learnable

												if self.learnable:
																		assert hidden_size is not None, "learnable=True requires `hidden_size` to be set"
																		assert length is not None, "learnable=True requires `length` to be set"

																		_lowerCAmelCase  						=			torch.zeros(_lowercase   ,       _lowercase				)
												else:
																		_lowerCAmelCase  						=			None

												_lowerCAmelCase  						=			torch.nn.Parameter(_lowercase				)






class   UpperCAmelCase_					(		_SCREAMING_SNAKE_CASE			):






						'''simple docstring'''






						_lowercase		:				VQModel
						_lowercase		:				CLIPTextModel
						_lowercase		:				CLIPTokenizer
						_lowercase		:				TransformeraDModel
						_lowercase		:				LearnedClassifierFreeSamplingEmbeddings
						_lowercase		:				VQDiffusionScheduler




						def __init__( self   ,       _lowercase   ,       _lowercase   ,       _lowercase   ,       _lowercase   ,       _lowercase   ,       _lowercase   ,       ):

												"""simple docstring"""



												super().__init__()

												self.register_modules(
												    vqvae=_lowercase   ,       transformer=_lowercase   ,       text_encoder=_lowercase   ,       tokenizer=_lowercase   ,       scheduler=_lowercase   ,       learned_classifier_free_sampling_embeddings=_lowercase   ,       )




						def  _lowercase     ( self   ,       _lowercase   ,       _lowercase   ,       _lowercase				):

												"""simple docstring"""



												_lowerCAmelCase  						=			len(_lowercase				) if isinstance(_lowercase   ,       _lowercase				) else 1

												# get prompt text embeddings
												_lowerCAmelCase  						=			self.tokenizer(
												    _lowercase   ,       padding="""max_length"""   ,       max_length=self.tokenizer.model_max_length   ,       return_tensors="""pt"""   ,       )
												_lowerCAmelCase  						=			text_inputs.input_ids

												if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
																		_lowerCAmelCase  						=			self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]				)
																		logger.warning(
																		    """The following part of your input was truncated because CLIP can only handle sequences up to"""
																		    F' {self.tokenizer.model_max_length} tokens: {removed_text}'				)
																		_lowerCAmelCase  						=			text_input_ids[:, : self.tokenizer.model_max_length]
												_lowerCAmelCase  						=			self.text_encoder(text_input_ids.to(self.device				)				)[0]

												# NOTE: This additional step of normalizing the text embeddings is from VQ-Diffusion.
												# While CLIP does normalize the pooled output of the text transformer when combining
												# the image and text embeddings, CLIP does not directly normalize the last hidden state.
												#
												# CLIP normalizing the pooled output.
												# https://github.com/huggingface/transformers/blob/d92e22d1f28324f513f3080e5c47c071a3916721/src/transformers/models/clip/modeling_clip.py#L1052-L1053
												_lowerCAmelCase  						=			prompt_embeds / prompt_embeds.norm(dim=-1   ,       keepdim=_lowercase				)

												# duplicate text embeddings for each generation per prompt
												_lowerCAmelCase  						=			prompt_embeds.repeat_interleave(_lowercase   ,       dim=0				)

												if do_classifier_free_guidance:
																		if self.learned_classifier_free_sampling_embeddings.learnable:
																								_lowerCAmelCase  						=			self.learned_classifier_free_sampling_embeddings.embeddings
																								_lowerCAmelCase  						=			negative_prompt_embeds.unsqueeze(0				).repeat(_lowercase   ,       1   ,       1				)
																		else:
																								_lowerCAmelCase  						=			[""""""] * batch_size

																								_lowerCAmelCase  						=			text_input_ids.shape[-1]
																								_lowerCAmelCase  						=			self.tokenizer(
																								    _lowercase   ,       padding="""max_length"""   ,       max_length=_lowercase   ,       truncation=_lowercase   ,       return_tensors="""pt"""   ,       )
																								_lowerCAmelCase  						=			self.text_encoder(uncond_input.input_ids.to(self.device				)				)[0]
																								# See comment for normalizing text embeddings
																								_lowerCAmelCase  						=			negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1   ,       keepdim=_lowercase				)

																		# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
																		_lowerCAmelCase  						=			negative_prompt_embeds.shape[1]
																		_lowerCAmelCase  						=			negative_prompt_embeds.repeat(1   ,       _lowercase   ,       1				)
																		_lowerCAmelCase  						=			negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,       _lowercase   ,       -1				)

																		# For classifier free guidance, we need to do two forward passes.
																		# Here we concatenate the unconditional and text embeddings into a single batch
																		# to avoid doing two forward passes
																		_lowerCAmelCase  						=			torch.cat([negative_prompt_embeds, prompt_embeds]				)

												return prompt_embeds




						@torch.no_grad()
						def __call__( self   ,       _lowercase   ,       _lowercase = 100   ,       _lowercase = 5.0   ,       _lowercase = 1.0   ,       _lowercase = 1   ,       _lowercase = None   ,       _lowercase = None   ,       _lowercase = "pil"   ,       _lowercase = True   ,       _lowercase = None   ,       _lowercase = 1   ,       ):

												"""simple docstring"""



												if isinstance(_lowercase   ,       _lowercase				):
																		_lowerCAmelCase  						=			1
												elif isinstance(_lowercase   ,       _lowercase				):
																		_lowerCAmelCase  						=			len(_lowercase				)
												else:
																		raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(_lowercase				)}'				)

												_lowerCAmelCase  						=			batch_size * num_images_per_prompt

												_lowerCAmelCase  						=			guidance_scale > 1.0

												_lowerCAmelCase  						=			self._encode_prompt(_lowercase   ,       _lowercase   ,       _lowercase				)

												if (callback_steps is None) or (
												    callback_steps is not None and (not isinstance(_lowercase   ,       _lowercase				) or callback_steps <= 0)
												):
																		raise ValueError(
																		    F'`callback_steps` has to be a positive integer but is {callback_steps} of type'
																		    F' {type(_lowercase				)}.'				)

												# get the initial completely masked latents unless the user supplied it

												_lowerCAmelCase  						=			(batch_size, self.transformer.num_latent_pixels)
												if latents is None:
																		_lowerCAmelCase  						=			self.transformer.num_vector_embeds - 1
																		_lowerCAmelCase  						=			torch.full(_lowercase   ,       _lowercase				).to(self.device				)
												else:
																		if latents.shape != latents_shape:
																								raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'				)
																		if (latents < 0).any() or (latents >= self.transformer.num_vector_embeds).any():
																								raise ValueError(
																								    """Unexpected latents value(s). All latents be valid embedding indices i.e. in the range 0,"""
																								    F' {self.transformer.num_vector_embeds - 1} (inclusive).'				)
																		_lowerCAmelCase  						=			latents.to(self.device				)

												# set timesteps
												self.scheduler.set_timesteps(_lowercase   ,       device=self.device				)

												_lowerCAmelCase  						=			self.scheduler.timesteps.to(self.device				)

												_lowerCAmelCase  						=			latents

												for i, t in enumerate(self.progress_bar(_lowercase				)				):
																		# expand the sample if we are doing classifier free guidance
																		_lowerCAmelCase  						=			torch.cat([sample] * 2				) if do_classifier_free_guidance else sample

																		# predict the un-noised image
																		# model_output == `log_p_x_0`
																		_lowerCAmelCase  						=			self.transformer(_lowercase   ,       encoder_hidden_states=_lowercase   ,       timestep=_lowercase				).sample

																		if do_classifier_free_guidance:
																								_lowerCAmelCase						,   _lowerCAmelCase  						=			model_output.chunk(2				)
																								_lowerCAmelCase  						=			model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
																								model_output -= torch.logsumexp(_lowercase   ,       dim=1   ,       keepdim=_lowercase				)

																		_lowerCAmelCase  						=			self.truncate(_lowercase   ,       _lowercase				)

																		# remove `log(0)`'s (`-inf`s)
																		_lowerCAmelCase  						=			model_output.clamp(-70				)

																		# compute the previous noisy sample x_t -> x_t-1
																		_lowerCAmelCase  						=			self.scheduler.step(_lowercase   ,       timestep=_lowercase   ,       sample=_lowercase   ,       generator=_lowercase				).prev_sample

																		# call the callback, if provided
																		if callback is not None and i % callback_steps == 0:
																								callback(_lowercase   ,       _lowercase   ,       _lowercase				)

												_lowerCAmelCase  						=			self.vqvae.config.vq_embed_dim
												_lowerCAmelCase  						=			(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
												_lowerCAmelCase  						=			self.vqvae.quantize.get_codebook_entry(_lowercase   ,       shape=_lowercase				)
												_lowerCAmelCase  						=			self.vqvae.decode(_lowercase   ,       force_not_quantize=_lowercase				).sample

												_lowerCAmelCase  						=			(image / 2 + 0.5).clamp(0   ,       1				)
												_lowerCAmelCase  						=			image.cpu().permute(0   ,       2   ,       3   ,       1				).numpy()

												if output_type == "pil":
																		_lowerCAmelCase  						=			self.numpy_to_pil(_lowercase				)

												if not return_dict:
																		return (image,)

												return ImagePipelineOutput(images=_lowercase				)




						def  _lowercase     ( self   ,       _lowercase   ,       _lowercase				):

												"""simple docstring"""



												_lowerCAmelCase						,   _lowerCAmelCase  						=			torch.sort(_lowercase   ,       1   ,       descending=_lowercase				)
												_lowerCAmelCase  						=			torch.exp(_lowercase				)
												_lowerCAmelCase  						=			sorted_p_x_0.cumsum(dim=1				) < truncation_rate

												# Ensure that at least the largest probability is not zeroed out
												_lowerCAmelCase  						=			torch.full_like(keep_mask[:, 0:1, :]   ,       _lowercase				)
												_lowerCAmelCase  						=			torch.cat((all_true, keep_mask)   ,       dim=1				)
												_lowerCAmelCase  						=			keep_mask[:, :-1, :]

												_lowerCAmelCase  						=			keep_mask.gather(1   ,       indices.argsort(1				)				)

												_lowerCAmelCase  						=			log_p_x_0.clone()

												_lowerCAmelCase  						=			-torch.inf  # -inf = log(0)

												return rv
