

import argparse
import json
import re
import string
import sys

import numpy as np


_UpperCAmelCase     :   int    	=			0.5


def 			__UpperCamelCase    (lowerCAmelCase					:  Tuple,    lowerCAmelCase					:  List[str]     )     ->  int:
      A    =				['.', ',', ';', ':']

      for token in remove_tokens:
            A    =				ground_truth.replace(lowerCAmelCase,    ''     )
            A    =				prediction.replace(lowerCAmelCase,    ''     )

      A ,	A    =				ground_truth.lower(), prediction.lower()
      A ,	A    =				ground_truth.replace('/',    ' '     ), prediction.replace('/',    ' '     )
      A ,	A    =				set(ground_truth.split(' '     )     ), set(prediction.split(' '     )     )

      A    =				ground_truth.intersection(lowerCAmelCase     )
      A    =				ground_truth.union(lowerCAmelCase     )
      A    =				len(lowerCAmelCase     ) / len(lowerCAmelCase     )
      return jaccard


def 			__UpperCamelCase    (lowerCAmelCase					:  str     )     ->  List[Any]:

      def remove_articles(lowerCAmelCase					:  Tuple     ):
            return re.sub(r'\b(a|an|the)\b',    ' ',    lowerCAmelCase     )

      def white_space_fix(lowerCAmelCase					:  Union[str, Any]     ):
            return " ".join(text.split()     )

      def remove_punc(lowerCAmelCase					:  Optional[int]     ):
            A    =				set(string.punctuation     )
            return "".join(ch for ch in text if ch not in exclude     )

      def lower(lowerCAmelCase					:  Optional[int]     ):
            return text.lower()

      return white_space_fix(remove_articles(remove_punc(lower(lowerCAmelCase     )     )     )     )


def 			__UpperCamelCase    (lowerCAmelCase					:  Dict,    lowerCAmelCase					:  str,    lowerCAmelCase					:  Optional[Any]     )     ->  Union[str, Any]:
      A ,	A ,	A    =				0, 0, 0

      A    =				'Parties' in qa_id

      # first check if ground truth is empty
      if len(lowerCAmelCase     ) == 0:
            if len(lowerCAmelCase     ) > 0:
                  fp += len(lowerCAmelCase     )  # false positive for each one
      else:
            for ground_truth in ground_truths:
                  assert len(lowerCAmelCase     ) > 0
                  # check if there is a match
                  A    =				False
                  for pred in predictions:
                        if substr_ok:
                              A    =				get_jaccard(lowerCAmelCase,    lowerCAmelCase     ) >= IOU_THRESH or ground_truth in pred
                        else:
                              A    =				get_jaccard(lowerCAmelCase,    lowerCAmelCase     ) >= IOU_THRESH
                        if is_match:
                              A    =				True

                  if match_found:
                        tp += 1
                  else:
                        fn += 1

        # now also get any fps by looping through preds
            for pred in predictions:
                  # Check if there's a match. if so, don't count (don't want to double count based on the above)
                  # but if there's no match, then this is a false positive.
                  # (Note: we get the true positives in the above loop instead of this loop so that we don't double count
                  # multiple predictions that are matched with the same answer.)
                  A    =				False
                  for ground_truth in ground_truths:
                        assert len(lowerCAmelCase     ) > 0
                        if substr_ok:
                              A    =				get_jaccard(lowerCAmelCase,    lowerCAmelCase     ) >= IOU_THRESH or ground_truth in pred
                        else:
                              A    =				get_jaccard(lowerCAmelCase,    lowerCAmelCase     ) >= IOU_THRESH
                        if is_match:
                              A    =				True

                  if not match_found:
                        fp += 1

      A    =				tp / (tp + fp) if tp + fp > 0 else np.nan
      A    =				tp / (tp + fn) if tp + fn > 0 else np.nan

      return precision, recall


def 			__UpperCamelCase    (lowerCAmelCase					:  int     )     ->  Optional[int]:
      A    =				precisions[::-1]
      for i in range(1,    len(lowerCAmelCase     )     ):
            A    =				max(precision_best[i - 1],    precision_best[i]     )
      A    =				precision_best[::-1]
      return precisions


def 			__UpperCamelCase    (lowerCAmelCase					:  int,    lowerCAmelCase					:  Tuple     )     ->  Optional[int]:
      A    =				process_precisions(lowerCAmelCase     )
      A    =				np.trapz(lowerCAmelCase,    lowerCAmelCase     )
      if np.isnan(lowerCAmelCase     ):
            return 0
      return aupr


def 			__UpperCamelCase    (lowerCAmelCase					:  int,    lowerCAmelCase					:  Optional[int],    lowerCAmelCase					:  List[Any]     )     ->  List[str]:
      A    =				process_precisions(lowerCAmelCase     )
      A    =				0
      for prec, recall in zip(lowerCAmelCase,    lowerCAmelCase     ):
            if recall >= recall_thresh:
                  A    =				prec
                  break
      return prec_at_recall


def 			__UpperCamelCase    (lowerCAmelCase					:  Optional[Any],    lowerCAmelCase					:  Dict     )     ->  int:
      return normalize_answer(lowerCAmelCase     ) == normalize_answer(lowerCAmelCase     )


def 			__UpperCamelCase    (lowerCAmelCase					:  Any,    lowerCAmelCase					:  str,    lowerCAmelCase					:  Optional[int]     )     ->  Union[str, Any]:
      A    =				0
      for pred in predictions:
            for ground_truth in ground_truths:
                  A    =				metric_fn(lowerCAmelCase,    lowerCAmelCase     )
                  if score == 1:  # break the loop when one prediction matches the ground truth
                        break
            if score == 1:
                  break
      return score


def 			__UpperCamelCase    (lowerCAmelCase					:  List[Any],    lowerCAmelCase					:  Optional[int]     )     ->  Optional[Any]:
      A    =				A    =				A    =				0
      A    =				[]
      A    =				[]
      for article in dataset:
            for paragraph in article["paragraphs"]:
                  for qa in paragraph["qas"]:
                        total += 1
                        if qa["id"] not in predictions:
                              A    =				'Unanswered question ' + qa['id'] + ' will receive score 0.'
                              print(lowerCAmelCase,    file=sys.stderr     )
                              continue
                        A    =				[x['text'] for x in qa['answers']]
                        A    =				predictions[qa['id']]
                        A ,	A    =				compute_precision_recall(lowerCAmelCase,    lowerCAmelCase,    qa['id']     )

                        precisions.append(lowerCAmelCase     )
                        recalls.append(lowerCAmelCase     )

                        if precision == 0 and recall == 0:
                              fa += 0
                        else:
                              fa += 2 * (precision * recall) / (precision + recall)

                        exact_match += metric_max_over_ground_truths(lowerCAmelCase,    lowerCAmelCase,    lowerCAmelCase     )

      A    =				[x for _, x in sorted(zip(lowerCAmelCase,    lowerCAmelCase     )     )]
      recalls.sort()

      A    =				100.0 * fa / total
      A    =				100.0 * exact_match / total
      A    =				get_aupr(lowerCAmelCase,    lowerCAmelCase     )

      A    =				get_prec_at_recall(lowerCAmelCase,    lowerCAmelCase,    recall_thresh=0.9     )
      A    =				get_prec_at_recall(lowerCAmelCase,    lowerCAmelCase,    recall_thresh=0.8     )

      return {
          "exact_match": exact_match,
          "f1": fa,
          "aupr": aupr,
          "prec_at_80_recall": prec_at_aa_recall,
          "prec_at_90_recall": prec_at_aa_recall,
      }


if __name__ == "__main__":
  _UpperCAmelCase     :   str    	=			argparse.ArgumentParser(description="Evaluation for CUAD")
  parser.add_argument("dataset_file", help="Dataset file")
  parser.add_argument("prediction_file", help="Prediction File")
  _UpperCAmelCase     :   List[str]    	=			parser.parse_args()
  with open(args.dataset_file) as dataset_file:
    _UpperCAmelCase     :   Optional[int]    	=			json.load(dataset_file)
    _UpperCAmelCase     :   str    	=			dataset_json["data"]
  with open(args.prediction_file) as prediction_file:
    _UpperCAmelCase     :   List[Any]    	=			json.load(prediction_file)
  print(json.dumps(evaluate(dataset, predictions)))






import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import PriorTransformer, UnCLIPPipeline, UnCLIPScheduler, UNetaDConditionModel, UNetaDModel
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class        _UpperCAmelCase (   __lowercase     ,				unittest.TestCase			):
  '''simple docstring'''







  SCREAMING_SNAKE_CASE							:				List[str]					      =				UnCLIPPipeline
  SCREAMING_SNAKE_CASE							:				List[str]					      =				TEXT_TO_IMAGE_PARAMS - {
      '''negative_prompt''',
      '''height''',
      '''width''',
      '''negative_prompt_embeds''',
      '''guidance_scale''',
      '''prompt_embeds''',
      '''cross_attention_kwargs''',
  }
  SCREAMING_SNAKE_CASE							:				int					      =				TEXT_TO_IMAGE_BATCH_PARAMS
  SCREAMING_SNAKE_CASE							:				Optional[int]					      =				[
      '''generator''',
      '''return_dict''',
      '''prior_num_inference_steps''',
      '''decoder_num_inference_steps''',
      '''super_res_num_inference_steps''',
  ]
  SCREAMING_SNAKE_CASE							:				str					      =				False







  @property
  def 	UpperCamelCase       ( self	: List[Any] ):
        return 32







  @property
  def 	UpperCamelCase       ( self	: int ):
        return 32







  @property
  def 	UpperCamelCase       ( self	: Tuple ):
        return self.time_input_dim







  @property
  def 	UpperCamelCase       ( self	: Union[str, Any] ):
        return self.time_input_dim * 4







  @property
  def 	UpperCamelCase       ( self	: List[Any] ):
        return 100







  @property
  def 	UpperCamelCase       ( self	: int ):
        A    =				CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )
        return tokenizer







  @property
  def 	UpperCamelCase       ( self	: List[Any] ):
        torch.manual_seed(0 )
        A    =				CLIPTextConfig(
            bos_token_id=0		,						eos_token_id=2		,						hidden_size=self.text_embedder_hidden_size		,						projection_dim=self.text_embedder_hidden_size		,						intermediate_size=37		,						layer_norm_eps=1e-0_5		,						num_attention_heads=4		,						num_hidden_layers=5		,						pad_token_id=1		,						vocab_size=1000		,						)
        return CLIPTextModelWithProjection(UpperCamelCase__ )







  @property
  def 	UpperCamelCase       ( self	: List[str] ):
        torch.manual_seed(0 )

        A    =				{
            'num_attention_heads': 2,
            'attention_head_dim': 12,
            'embedding_dim': self.text_embedder_hidden_size,
            'num_layers': 1,
        }

        A    =				PriorTransformer(**UpperCamelCase__ )
        return model







  @property
  def 	UpperCamelCase       ( self	: Optional[int] ):
        torch.manual_seed(0 )

        A    =				{
            'clip_embeddings_dim': self.text_embedder_hidden_size,
            'time_embed_dim': self.time_embed_dim,
            'cross_attention_dim': self.cross_attention_dim,
        }

        A    =				UnCLIPTextProjModel(**UpperCamelCase__ )
        return model







  @property
  def 	UpperCamelCase       ( self	: List[Any] ):
        torch.manual_seed(0 )

        A    =				{
            'sample_size': 32,
            # RGB in channels
            'in_channels': 3,
            # Out channels is double in channels because predicts mean and variance
            'out_channels': 6,
            'down_block_types': ('ResnetDownsampleBlock2D', 'SimpleCrossAttnDownBlock2D'),
            'up_block_types': ('SimpleCrossAttnUpBlock2D', 'ResnetUpsampleBlock2D'),
            'mid_block_type': 'UNetMidBlock2DSimpleCrossAttn',
            'block_out_channels': (self.block_out_channels_a, self.block_out_channels_a * 2),
            'layers_per_block': 1,
            'cross_attention_dim': self.cross_attention_dim,
            'attention_head_dim': 4,
            'resnet_time_scale_shift': 'scale_shift',
            'class_embed_type': 'identity',
        }

        A    =				UNetaDConditionModel(**UpperCamelCase__ )
        return model







  @property
  def 	UpperCamelCase       ( self	: Any ):
        return {
            "sample_size": 64,
            "layers_per_block": 1,
            "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
            "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
            "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
            "in_channels": 6,
            "out_channels": 3,
        }







  @property
  def 	UpperCamelCase       ( self	: str ):
        torch.manual_seed(0 )

        A    =				UNetaDModel(**self.dummy_super_res_kwargs )
        return model







  @property
  def 	UpperCamelCase       ( self	: Any ):
        # seeded differently to get different unet than `self.dummy_super_res_first`
        torch.manual_seed(1 )

        A    =				UNetaDModel(**self.dummy_super_res_kwargs )
        return model







  def 	UpperCamelCase       ( self	: List[Any] ):
        A    =				self.dummy_prior
        A    =				self.dummy_decoder
        A    =				self.dummy_text_proj
        A    =				self.dummy_text_encoder
        A    =				self.dummy_tokenizer
        A    =				self.dummy_super_res_first
        A    =				self.dummy_super_res_last

        A    =				UnCLIPScheduler(
            variance_type='fixed_small_log'		,						prediction_type='sample'		,						num_train_timesteps=1000		,						clip_sample_range=5.0		,						)

        A    =				UnCLIPScheduler(
            variance_type='learned_range'		,						prediction_type='epsilon'		,						num_train_timesteps=1000		,						)

        A    =				UnCLIPScheduler(
            variance_type='fixed_small_log'		,						prediction_type='epsilon'		,						num_train_timesteps=1000		,						)

        A    =				{
            'prior': prior,
            'decoder': decoder,
            'text_proj': text_proj,
            'text_encoder': text_encoder,
            'tokenizer': tokenizer,
            'super_res_first': super_res_first,
            'super_res_last': super_res_last,
            'prior_scheduler': prior_scheduler,
            'decoder_scheduler': decoder_scheduler,
            'super_res_scheduler': super_res_scheduler,
        }

        return components







  def 	UpperCamelCase       ( self	: List[Any]		,						UpperCamelCase__	: Optional[Any]		,						UpperCamelCase__	: Optional[int]=0 ):
        if str(UpperCamelCase__ ).startswith('mps' ):
              A    =				torch.manual_seed(UpperCamelCase__ )
        else:
              A    =				torch.Generator(device=UpperCamelCase__ ).manual_seed(UpperCamelCase__ )
        A    =				{
            'prompt': 'horse',
            'generator': generator,
            'prior_num_inference_steps': 2,
            'decoder_num_inference_steps': 2,
            'super_res_num_inference_steps': 2,
            'output_type': 'numpy',
        }
        return inputs







  def 	UpperCamelCase       ( self	: int ):
        A    =				'cpu'

        A    =				self.get_dummy_components()

        A    =				self.pipeline_class(**UpperCamelCase__ )
        A    =				pipe.to(UpperCamelCase__ )

        pipe.set_progress_bar_config(disable=UpperCamelCase__ )

        A    =				pipe(**self.get_dummy_inputs(UpperCamelCase__ ) )
        A    =				output.images

        A    =				pipe(
            **self.get_dummy_inputs(UpperCamelCase__ )		,						return_dict=UpperCamelCase__		,						)[0]

        A    =				image[0, -3:, -3:, -1]
        A    =				image_from_tuple[0, -3:, -3:, -1]

        assert image.shape == (1, 64, 64, 3)

        A    =				np.array(
            [
                0.9_997,
                0.9_988,
                0.0_028,
                0.9_997,
                0.9_984,
                0.9_965,
                0.0_029,
                0.9_986,
                0.0_025,
            ] )

        assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2
        assert np.abs(image_from_tuple_slice.flatten() - expected_slice ).max() < 1e-2







  def 	UpperCamelCase       ( self	: Any ):
        A    =				torch.device('cpu' )



        class        _UpperCAmelCase :
          '''simple docstring'''







          SCREAMING_SNAKE_CASE							:				Tuple					      =				1
        A    =				self.get_dummy_components()

        A    =				self.pipeline_class(**UpperCamelCase__ )
        A    =				pipe.to(UpperCamelCase__ )

        A    =				components['prior']
        A    =				components['decoder']
        A    =				components['super_res_first']
        A    =				components['tokenizer']
        A    =				components['text_encoder']

        A    =				torch.Generator(device=UpperCamelCase__ ).manual_seed(0 )
        A    =				prior.dtype
        A    =				1

        A    =				(batch_size, prior.config.embedding_dim)
        A    =				pipe.prepare_latents(
            UpperCamelCase__		,						dtype=UpperCamelCase__		,						device=UpperCamelCase__		,						generator=UpperCamelCase__		,						latents=UpperCamelCase__		,						scheduler=DummyScheduler() )
        A    =				(batch_size, decoder.config.in_channels, decoder.config.sample_size, decoder.config.sample_size)
        A    =				pipe.prepare_latents(
            UpperCamelCase__		,						dtype=UpperCamelCase__		,						device=UpperCamelCase__		,						generator=UpperCamelCase__		,						latents=UpperCamelCase__		,						scheduler=DummyScheduler() )

        A    =				(
            batch_size,
            super_res_first.config.in_channels // 2,
            super_res_first.config.sample_size,
            super_res_first.config.sample_size,
        )
        A    =				pipe.prepare_latents(
            UpperCamelCase__		,						dtype=UpperCamelCase__		,						device=UpperCamelCase__		,						generator=UpperCamelCase__		,						latents=UpperCamelCase__		,						scheduler=DummyScheduler() )

        pipe.set_progress_bar_config(disable=UpperCamelCase__ )

        A    =				'this is a prompt example'

        A    =				torch.Generator(device=UpperCamelCase__ ).manual_seed(0 )
        A    =				pipe(
            [prompt]		,						generator=UpperCamelCase__		,						prior_num_inference_steps=2		,						decoder_num_inference_steps=2		,						super_res_num_inference_steps=2		,						prior_latents=UpperCamelCase__		,						decoder_latents=UpperCamelCase__		,						super_res_latents=UpperCamelCase__		,						output_type='np'		,						)
        A    =				output.images

        A    =				tokenizer(
            UpperCamelCase__		,						padding='max_length'		,						max_length=tokenizer.model_max_length		,						return_tensors='pt'		,						)
        A    =				text_encoder(text_inputs.input_ids )
        A    =				text_inputs.attention_mask

        A    =				torch.Generator(device=UpperCamelCase__ ).manual_seed(0 )
        A    =				pipe(
            generator=UpperCamelCase__		,						prior_num_inference_steps=2		,						decoder_num_inference_steps=2		,						super_res_num_inference_steps=2		,						prior_latents=UpperCamelCase__		,						decoder_latents=UpperCamelCase__		,						super_res_latents=UpperCamelCase__		,						text_model_output=UpperCamelCase__		,						text_attention_mask=UpperCamelCase__		,						output_type='np'		,						)[0]

        # make sure passing text embeddings manually is identical
        assert np.abs(image - image_from_text ).max() < 1e-4







  @skip_mps
  def 	UpperCamelCase       ( self	: Any ):
        A    =				torch_device == 'cpu'

        self._test_attention_slicing_forward_pass(test_max_difference=UpperCamelCase__		,						expected_max_diff=0.01 )







  @skip_mps
  def 	UpperCamelCase       ( self	: Optional[Any] ):
        A    =				torch_device == 'cpu'
        A    =				True
        A    =				[
            'prior_num_inference_steps',
            'decoder_num_inference_steps',
            'super_res_num_inference_steps',
        ]

        self._test_inference_batch_single_identical(
            test_max_difference=UpperCamelCase__		,						relax_max_difference=UpperCamelCase__		,						additional_params_copy_to_batched_inputs=UpperCamelCase__		,						)







  def 	UpperCamelCase       ( self	: List[str] ):
        A    =				[
            'prior_num_inference_steps',
            'decoder_num_inference_steps',
            'super_res_num_inference_steps',
        ]

        if torch_device == "mps":
              # TODO: MPS errors with larger batch sizes
              A    =				[2, 3]
              self._test_inference_batch_consistent(
                  batch_sizes=UpperCamelCase__		,						additional_params_copy_to_batched_inputs=UpperCamelCase__		,						)
        else:
              self._test_inference_batch_consistent(
                  additional_params_copy_to_batched_inputs=UpperCamelCase__ )







  @skip_mps
  def 	UpperCamelCase       ( self	: int ):
        return super().test_dict_tuple_outputs_equivalent()







  @skip_mps
  def 	UpperCamelCase       ( self	: Optional[Any] ):
        return super().test_save_load_local()







  @skip_mps
  def 	UpperCamelCase       ( self	: List[str] ):
        return super().test_save_load_optional_components()



@nightly
class        _UpperCAmelCase (   unittest.TestCase			):
  '''simple docstring'''







  def 	UpperCamelCase       ( self	: Any ):
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()







  def 	UpperCamelCase       ( self	: List[Any] ):
        A    =				load_numpy(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/unclip/karlo_v1_alpha_horse_cpu.npy' )

        A    =				UnCLIPPipeline.from_pretrained('kakaobrain/karlo-v1-alpha' )
        pipeline.set_progress_bar_config(disable=UpperCamelCase__ )

        A    =				torch.manual_seed(0 )
        A    =				pipeline(
            'horse'		,						num_images_per_prompt=1		,						generator=UpperCamelCase__		,						output_type='np'		,						)

        A    =				output.images[0]

        assert image.shape == (256, 256, 3)
        assert np.abs(expected_image - image ).max() < 1e-1



@slow
@require_torch_gpu
class        _UpperCAmelCase (   unittest.TestCase			):
  '''simple docstring'''







  def 	UpperCamelCase       ( self	: List[str] ):
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()







  def 	UpperCamelCase       ( self	: List[Any] ):
        A    =				load_numpy(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/unclip/karlo_v1_alpha_horse_fp16.npy' )

        A    =				UnCLIPPipeline.from_pretrained('kakaobrain/karlo-v1-alpha'		,						torch_dtype=torch.floataa )
        A    =				pipeline.to(UpperCamelCase__ )
        pipeline.set_progress_bar_config(disable=UpperCamelCase__ )

        A    =				torch.Generator(device='cpu' ).manual_seed(0 )
        A    =				pipeline(
            'horse'		,						generator=UpperCamelCase__		,						output_type='np'		,						)

        A    =				output.images[0]

        assert image.shape == (256, 256, 3)

        assert_mean_pixel_difference(UpperCamelCase__		,						UpperCamelCase__ )







  def 	UpperCamelCase       ( self	: Dict ):
        torch.cuda.empty_cache()
        torch.cuda.reset_max_memory_allocated()
        torch.cuda.reset_peak_memory_stats()

        A    =				UnCLIPPipeline.from_pretrained('kakaobrain/karlo-v1-alpha'		,						torch_dtype=torch.floataa )
        A    =				pipe.to(UpperCamelCase__ )
        pipe.set_progress_bar_config(disable=UpperCamelCase__ )
        pipe.enable_attention_slicing()
        pipe.enable_sequential_cpu_offload()

        A    =				pipe(
            'horse'		,						num_images_per_prompt=1		,						prior_num_inference_steps=2		,						decoder_num_inference_steps=2		,						super_res_num_inference_steps=2		,						output_type='np'		,						)

        A    =				torch.cuda.max_memory_allocated()
        # make sure that less than 7 GB is allocated
        assert mem_bytes < 7 * 10**9



