




"""simple docstring"""


import itertools
import json
import linecache
import os
import pickle
import re
import socket
import string
from collections import Counter
from logging import getLogger
from pathlib import Path
from typing import Callable, Dict, Iterable, List

import git
import torch
from torch.utils.data import Dataset

from transformers import BartTokenizer, RagTokenizer, TaTokenizer


def       UpperCAmelCase__ (snake_case__							:      Any	,     snake_case__							:      Tuple	,     snake_case__							:      Optional[int]	,     snake_case__							:      Optional[int]	,     snake_case__							:      Union[str, Any]=True	,     snake_case__							:      int="pt"					):
  """simple docstring"""

  _snake_case							:	Dict			=						{"""add_prefix_space""": True} if isinstance(snake_case__	,     snake_case__					) and not line.startswith(""" """					) else {}
  _snake_case							:	Dict			=						padding_side
  return tokenizer(
      [line]	,     max_length=snake_case__	,     padding="""max_length""" if pad_to_max_length else None	,     truncation=snake_case__	,     return_tensors=snake_case__	,     add_special_tokens=snake_case__	,     **snake_case__	,     )


def       UpperCAmelCase__ (snake_case__							:      List[Any]	,     snake_case__							:      Union[str, Any]	,     snake_case__							:      int=None	,     ):
  """simple docstring"""

  _snake_case							:	str			=						input_ids.ne(snake_case__					).any(dim=0					)
  if attention_mask is None:
    return input_ids[:, keep_column_mask]
  else:
    return (input_ids[:, keep_column_mask], attention_mask[:, keep_column_mask])






class       lowercase(  __a				):







 '''simple docstring'''





 def __init__(       self:	Union[str, Any],   a_:	Optional[int],   a_:	List[Any],   a_:	Optional[Any],   a_:	List[Any],   a_:	Dict="train",   a_:	List[Any]=None,   a_:	Tuple=None,   a_:	Any=None,   a_:	Tuple="",   ):




   '''simple docstring'''



   super().__init__()
   _snake_case							:	Union[str, Any]			=						Path(a_	).joinpath(type_path + """.source"""	)
   _snake_case							:	int			=						Path(a_	).joinpath(type_path + """.target"""	)
   _snake_case							:	int			=						self.get_char_lens(self.src_file	)
   _snake_case							:	List[Any]			=						max_source_length
   _snake_case							:	Union[str, Any]			=						max_target_length
   assert min(self.src_lens	) > 0, f"found empty line in {self.src_file}"
   _snake_case							:	Optional[Any]			=						tokenizer
   _snake_case							:	Tuple			=						prefix
   if n_obs is not None:
     _snake_case							:	Any			=						self.src_lens[:n_obs]
   _snake_case							:	Union[str, Any]			=						src_lang
   _snake_case							:	str			=						tgt_lang


 def __len__(       self:	Dict	):




   '''simple docstring'''



   return len(self.src_lens	)


 def __getitem__(       self:	Union[str, Any],   a_:	Union[str, Any]	):




   '''simple docstring'''



   _snake_case							:	Any			=						index + 1  # linecache starts at 1
   _snake_case							:	Tuple			=						self.prefix + linecache.getline(str(self.src_file	),   a_	).rstrip("""\n"""	)
   _snake_case							:	Tuple			=						linecache.getline(str(self.tgt_file	),   a_	).rstrip("""\n"""	)
   assert source_line, f"empty source line for index {index}"
   assert tgt_line, f"empty tgt line for index {index}"

   # Need to add eos token manually for T5
   if isinstance(self.tokenizer,   a_	):
     source_line += self.tokenizer.eos_token
     tgt_line += self.tokenizer.eos_token

   # Pad source and target to the right
   _snake_case							:	List[str]			=						(
       self.tokenizer.question_encoder if isinstance(self.tokenizer,   a_	) else self.tokenizer
   )
   _snake_case							:	Union[str, Any]			=						self.tokenizer.generator if isinstance(self.tokenizer,   a_	) else self.tokenizer

   _snake_case							:	List[str]			=						encode_line(a_,   a_,   self.max_source_length,   """right"""	)
   _snake_case							:	Optional[Any]			=						encode_line(a_,   a_,   self.max_target_length,   """right"""	)

   _snake_case							:	Dict			=						source_inputs["""input_ids"""].squeeze()
   _snake_case							:	int			=						target_inputs["""input_ids"""].squeeze()
   _snake_case							:	List[Any]			=						source_inputs["""attention_mask"""].squeeze()
   return {
       "input_ids": source_ids,
       "attention_mask": src_mask,
       "decoder_input_ids": target_ids,
   }


 @staticmethod
 def 				UpperCamelCase_     (       a_:	List[str]	):




   '''simple docstring'''



   return [len(a_	) for x in Path(a_	).open().readlines()]


 def 				UpperCamelCase_     (       self:	str,   a_:	Dict	):




   '''simple docstring'''



   _snake_case							:	List[str]			=						torch.stack([x["""input_ids"""] for x in batch]	)
   _snake_case							:	List[Any]			=						torch.stack([x["""attention_mask"""] for x in batch]	)
   _snake_case							:	Optional[int]			=						torch.stack([x["""decoder_input_ids"""] for x in batch]	)
   _snake_case							:	Optional[int]			=						(
       self.tokenizer.generator.pad_token_id
       if isinstance(self.tokenizer,   a_	)
       else self.tokenizer.pad_token_id
   )
   _snake_case							:	int			=						(
       self.tokenizer.question_encoder.pad_token_id
       if isinstance(self.tokenizer,   a_	)
       else self.tokenizer.pad_token_id
   )
   _snake_case							:	Tuple			=						trim_batch(a_,   a_	)
   _snake_case	,      _snake_case							:	List[str]			=						trim_batch(a_,   a_,   attention_mask=a_	)
   _snake_case							:	Optional[Any]			=						{
       """input_ids""": source_ids,
       """attention_mask""": source_mask,
       """decoder_input_ids""": y,
   }
   return batch







A_      			=       getLogger(__name__)


def       UpperCAmelCase__ (snake_case__							:      List[List]					):
  """simple docstring"""

  return list(itertools.chain.from_iterable(snake_case__					)					)


def       UpperCAmelCase__ (snake_case__							:      str					):
  """simple docstring"""

  _snake_case							:	List[str]			=						get_git_info()
  save_json(snake_case__	,     os.path.join(snake_case__	,     """git_log.json"""					)					)


def       UpperCAmelCase__ (snake_case__							:      Union[str, Any]	,     snake_case__							:      Optional[Any]	,     snake_case__							:      Tuple=4	,     **snake_case__							:      List[Any]					):
  """simple docstring"""

  with open(snake_case__	,     """w"""					) as f:
    json.dump(snake_case__	,     snake_case__	,     indent=snake_case__	,     **snake_case__					)


def       UpperCAmelCase__ (snake_case__							:      Any					):
  """simple docstring"""

  with open(snake_case__					) as f:
    return json.load(snake_case__					)


def       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Dict			=						git.Repo(search_parent_directories=snake_case__					)
  _snake_case							:	Any			=						{
      """repo_id""": str(snake_case__					),
      """repo_sha""": str(repo.head.object.hexsha					),
      """repo_branch""": str(repo.active_branch					),
      """hostname""": str(socket.gethostname()					),
  }
  return repo_infos


def       UpperCAmelCase__ (snake_case__							:      Callable	,     snake_case__							:      Iterable					):
  """simple docstring"""

  return list(map(snake_case__	,     snake_case__					)					)


def       UpperCAmelCase__ (snake_case__							:      Union[str, Any]	,     snake_case__							:      Any					):
  """simple docstring"""

  with open(snake_case__	,     """wb"""					) as f:
    return pickle.dump(snake_case__	,     snake_case__					)


def       UpperCAmelCase__ (snake_case__							:      int					):
  """simple docstring"""

  def remove_articles(snake_case__							:      Union[str, Any]					):
    return re.sub(R"""\b(a|an|the)\b"""	,     """ """	,     snake_case__					)

  def white_space_fix(snake_case__							:      List[Any]					):
    return " ".join(text.split()					)

  def remove_punc(snake_case__							:      Any					):
    _snake_case							:	List[str]			=						set(string.punctuation					)
    return "".join(ch for ch in text if ch not in exclude					)

  def lower(snake_case__							:      int					):
    return text.lower()

  return white_space_fix(remove_articles(remove_punc(lower(snake_case__					)					)					)					)


def       UpperCAmelCase__ (snake_case__							:      Dict	,     snake_case__							:      Dict					):
  """simple docstring"""

  _snake_case							:	List[str]			=						normalize_answer(snake_case__					).split()
  _snake_case							:	int			=						normalize_answer(snake_case__					).split()
  _snake_case							:	List[Any]			=						Counter(snake_case__					) & Counter(snake_case__					)
  _snake_case							:	Union[str, Any]			=						sum(common.values()					)
  if num_same == 0:
    return 0
  _snake_case							:	Dict			=						1.0 * num_same / len(snake_case__					)
  _snake_case							:	List[Any]			=						1.0 * num_same / len(snake_case__					)
  _snake_case							:	Optional[int]			=						(2 * precision * recall) / (precision + recall)
  return fa


def       UpperCAmelCase__ (snake_case__							:      str	,     snake_case__							:      Any					):
  """simple docstring"""

  return normalize_answer(snake_case__					) == normalize_answer(snake_case__					)


def       UpperCAmelCase__ (snake_case__							:      List[str]	,     snake_case__							:      List[str]					):
  """simple docstring"""

  assert len(snake_case__					) == len(snake_case__					)
  _snake_case							:	Optional[Any]			=						0
  for hypo, pred in zip(snake_case__	,     snake_case__					):
    em += exact_match_score(snake_case__	,     snake_case__					)
  if len(snake_case__					) > 0:
    em /= len(snake_case__					)
  return {"em": em}


def       UpperCAmelCase__ (snake_case__							:      Tuple					):
  """simple docstring"""

  return model_prefix.startswith("""rag"""					)






def       UpperCAmelCase__ (snake_case__							:      Any	,     snake_case__							:      Union[str, Any]	,     snake_case__							:      Tuple					):
  """simple docstring"""

  _snake_case							:	str			=						{p: p for p in extra_params}
  # T5 models don't have `dropout` param, they have `dropout_rate` instead
  _snake_case							:	List[str]			=						"""dropout_rate"""
  for p in extra_params:
    if getattr(snake_case__	,     snake_case__	,     snake_case__					):
      if not hasattr(snake_case__	,     snake_case__					) and not hasattr(snake_case__	,     equivalent_param[p]					):
        logger.info("""config doesn't have a `{}` attribute""".format(snake_case__					)					)
        delattr(snake_case__	,     snake_case__					)
        continue
      _snake_case							:	Optional[int]			=						p if hasattr(snake_case__	,     snake_case__					) else equivalent_param[p]
      setattr(snake_case__	,     snake_case__	,     getattr(snake_case__	,     snake_case__					)					)
      delattr(snake_case__	,     snake_case__					)
  return hparams, config








"""simple docstring"""


import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ConvNextConfig, UperNetConfig
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch

     from transformers import UperNetForSemanticSegmentation
     from transformers.models.upernet.modeling_upernet import UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from PIL import Image

     from transformers import AutoImageProcessor






class       lowercase:







 '''simple docstring'''





 def __init__(       self:	Dict,   a_:	Union[str, Any],   a_:	Tuple=13,   a_:	Dict=32,   a_:	Optional[Any]=3,   a_:	Optional[Any]=4,   a_:	Optional[int]=[10, 20, 30, 40],   a_:	Any=[2, 2, 3, 2],   a_:	Dict=True,   a_:	Dict=True,   a_:	List[str]=37,   a_:	Dict="gelu",   a_:	List[str]=10,   a_:	Union[str, Any]=0.02,   a_:	Any=["stage2", "stage3", "stage4"],   a_:	Optional[int]=3,   a_:	Tuple=None,   ):




   '''simple docstring'''



   _snake_case							:	Dict			=						parent
   _snake_case							:	Dict			=						batch_size
   _snake_case							:	Optional[Any]			=						image_size
   _snake_case							:	int			=						num_channels
   _snake_case							:	Tuple			=						num_stages
   _snake_case							:	int			=						hidden_sizes
   _snake_case							:	List[str]			=						depths
   _snake_case							:	str			=						is_training
   _snake_case							:	Dict			=						use_labels
   _snake_case							:	List[str]			=						intermediate_size
   _snake_case							:	Optional[int]			=						hidden_act
   _snake_case							:	Any			=						type_sequence_label_size
   _snake_case							:	List[str]			=						initializer_range
   _snake_case							:	Union[str, Any]			=						out_features
   _snake_case							:	Dict			=						num_labels
   _snake_case							:	int			=						scope
   _snake_case							:	Dict			=						num_stages


 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   _snake_case							:	Optional[Any]			=						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]	)

   _snake_case							:	Optional[int]			=						None
   if self.use_labels:
     _snake_case							:	Optional[Any]			=						ids_tensor([self.batch_size],   self.type_sequence_label_size	)

   _snake_case							:	Tuple			=						self.get_config()

   return config, pixel_values, labels


 def 				UpperCamelCase_     (       self:	Tuple	):




   '''simple docstring'''



   return ConvNextConfig(
       num_channels=self.num_channels,   num_stages=self.num_stages,   hidden_sizes=self.hidden_sizes,   depths=self.depths,   is_training=self.is_training,   intermediate_size=self.intermediate_size,   hidden_act=self.hidden_act,   out_features=self.out_features,   )


 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   return UperNetConfig(
       backbone_config=self.get_backbone_config(),   hidden_size=512,   pool_scales=[1, 2, 3, 6],   use_auxiliary_head=a_,   auxiliary_loss_weight=0.4,   auxiliary_in_channels=40,   auxiliary_channels=256,   auxiliary_num_convs=1,   auxiliary_concat_input=a_,   loss_ignore_index=255,   num_labels=self.num_labels,   )


 def 				UpperCamelCase_     (       self:	Tuple,   a_:	List[Any],   a_:	Dict,   a_:	Tuple	):




   '''simple docstring'''



   _snake_case							:	List[Any]			=						UperNetForSemanticSegmentation(config=a_	)
   model.to(a_	)
   model.eval()
   _snake_case							:	Tuple			=						model(a_	)
   self.parent.assertEqual(
       result.logits.shape,   (self.batch_size, self.num_labels, self.image_size, self.image_size)	)


 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   _snake_case							:	Any			=						self.prepare_config_and_inputs()
   (
       (
       _snake_case
   )	,      (
       _snake_case
   )	,      (
       _snake_case
   )	,      
   )							:	List[Any]			=						config_and_inputs
   _snake_case							:	Any			=						{"""pixel_values""": pixel_values}
   return config, inputs_dict







@require_torch
class       lowercase(  __a ,    __a ,    unittest.TestCase				):







 '''simple docstring'''





 lowercase__				      =      (UperNetForSemanticSegmentation,) if is_torch_available() else ()
 lowercase__				      =      {"image-segmentation": UperNetForSemanticSegmentation} if is_torch_available() else {}
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False
 lowercase__				      =      False


 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   _snake_case							:	List[str]			=						UperNetModelTester(self	)
   _snake_case							:	Dict			=						ConfigTester(self,   config_class=a_,   has_text_modality=a_,   hidden_size=37	)


 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   self.create_and_test_config_common_properties()
   self.config_tester.create_and_test_config_to_json_string()
   self.config_tester.create_and_test_config_to_json_file()
   self.config_tester.create_and_test_config_from_and_save_pretrained()
   self.config_tester.create_and_test_config_with_num_labels()
   self.config_tester.check_config_can_be_init_without_params()
   self.config_tester.check_config_arguments_init()


 def 				UpperCamelCase_     (       self:	List[Any]	):




   '''simple docstring'''



   return


 def 				UpperCamelCase_     (       self:	Union[str, Any]	):




   '''simple docstring'''



   _snake_case	,      _snake_case							:	Any			=						self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
     _snake_case							:	Dict			=						model_class(a_	)
     _snake_case							:	Optional[int]			=						inspect.signature(model.forward	)
     # signature.parameters is an OrderedDict => so arg_names order is deterministic
     _snake_case							:	Tuple			=						[*signature.parameters.keys()]

     _snake_case							:	Any			=						["""pixel_values"""]
     self.assertListEqual(arg_names[:1],   a_	)


 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   _snake_case							:	Union[str, Any]			=						self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_for_semantic_segmentation(*a_	)


 @unittest.skip(reason="""UperNet does not use inputs_embeds"""	)
 def 				UpperCamelCase_     (       self:	Any	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""UperNet does not support input and output embeddings"""	)
 def 				UpperCamelCase_     (       self:	Dict	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""UperNet does not have a base model"""	)
 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   pass


 @unittest.skip(reason="""UperNet does not have a base model"""	)
 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   pass


 @require_torch_multi_gpu
 @unittest.skip(reason="""UperNet has some layers using `add_module` which doesn't work well with `nn.DataParallel`"""	)
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   pass


 @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""	)
 def 				UpperCamelCase_     (       self:	Dict	):




   '''simple docstring'''



   pass


 def 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   def check_hidden_states_output(a_:	Dict,   a_:	List[str],   a_:	Optional[int]	):
     _snake_case							:	Optional[Any]			=						model_class(a_	)
     model.to(a_	)
     model.eval()

     with torch.no_grad():
       _snake_case							:	Any			=						model(**self._prepare_for_class(a_,   a_	)	)

     _snake_case							:	Any			=						outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

     _snake_case							:	List[str]			=						self.model_tester.num_stages
     self.assertEqual(len(a_	),   expected_num_stages + 1	)

     # ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
     self.assertListEqual(
         list(hidden_states[0].shape[-2:]	),   [self.model_tester.image_size // 4, self.model_tester.image_size // 4],   )

   _snake_case	,      _snake_case							:	str			=						self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
     _snake_case							:	int			=						True
     check_hidden_states_output(a_,   a_,   a_	)

     # check that output_hidden_states also work using config
     del inputs_dict["output_hidden_states"]
     _snake_case							:	Optional[int]			=						True

     check_hidden_states_output(a_,   a_,   a_	)


 def 				UpperCamelCase_     (       self:	str	):




   '''simple docstring'''



   _snake_case	,      _snake_case							:	int			=						self.model_tester.prepare_config_and_inputs_for_common()

   _snake_case							:	Tuple			=						_config_zero_init(a_	)
   _snake_case							:	Dict			=						_config_zero_init(configs_no_init.backbone_config	)
   for model_class in self.all_model_classes:
     _snake_case							:	Optional[int]			=						model_class(config=a_	)
     for name, param in model.named_parameters():
       if param.requires_grad:
         self.assertIn(
             ((param.data.mean() * 1E9).round() / 1E9).item(),   [0.0, 1.0],   msg=f"Parameter {name} of model {model_class} seems not properly initialized",   )


 @unittest.skip(reason="""UperNet does not have tied weights"""	)
 def 				UpperCamelCase_     (       self:	Union[str, Any]	):




   '''simple docstring'''



   pass


 @slow
 def 				UpperCamelCase_     (       self:	List[str]	):




   '''simple docstring'''



   for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
     _snake_case							:	int			=						UperNetForSemanticSegmentation.from_pretrained(a_	)
     self.assertIsNotNone(a_	)







def       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Union[str, Any]			=						hf_hub_download(
      repo_id="""hf-internal-testing/fixtures_ade20k"""	,     repo_type="""dataset"""	,     filename="""ADE_val_00000001.jpg"""					)
  _snake_case							:	List[Any]			=						Image.open(snake_case__					).convert("""RGB"""					)
  return image






@require_torch
@require_vision
@slow
class       lowercase(  unittest.TestCase				):







 '''simple docstring'''





 def 				UpperCamelCase_     (       self:	Tuple	):




   '''simple docstring'''



   _snake_case							:	str			=						AutoImageProcessor.from_pretrained("""openmmlab/upernet-swin-tiny"""	)
   _snake_case							:	Any			=						UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-swin-tiny"""	).to(a_	)

   _snake_case							:	Dict			=						prepare_img()
   _snake_case							:	str			=						processor(images=a_,   return_tensors="""pt"""	).to(a_	)

   with torch.no_grad():
     _snake_case							:	Tuple			=						model(**a_	)

   _snake_case							:	Tuple			=						torch.Size((1, model.config.num_labels, 512, 512)	)
   self.assertEqual(outputs.logits.shape,   a_	)

   _snake_case							:	int			=						torch.tensor(
       [[-7.5_958, -7.5_958, -7.4_302], [-7.5_958, -7.5_958, -7.4_302], [-7.4_797, -7.4_797, -7.3_068]]	).to(a_	)
   self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3],   a_,   atol=1E-4	)	)


 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   _snake_case							:	Optional[Any]			=						AutoImageProcessor.from_pretrained("""openmmlab/upernet-convnext-tiny"""	)
   _snake_case							:	Optional[int]			=						UperNetForSemanticSegmentation.from_pretrained("""openmmlab/upernet-convnext-tiny"""	).to(a_	)

   _snake_case							:	List[str]			=						prepare_img()
   _snake_case							:	Tuple			=						processor(images=a_,   return_tensors="""pt"""	).to(a_	)

   with torch.no_grad():
     _snake_case							:	Optional[Any]			=						model(**a_	)

   _snake_case							:	Union[str, Any]			=						torch.Size((1, model.config.num_labels, 512, 512)	)
   self.assertEqual(outputs.logits.shape,   a_	)

   _snake_case							:	Optional[Any]			=						torch.tensor(
       [[-8.8_110, -8.8_110, -8.6_521], [-8.8_110, -8.8_110, -8.6_521], [-8.7_746, -8.7_746, -8.6_130]]	).to(a_	)
   self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3],   a_,   atol=1E-4	)	)


