


'''simple docstring'''


import argparse
import glob
import logging
import os
from argparse import Namespace
from importlib import import_module

import numpy as np
import torch
from lightning_base import BaseTransformer, add_generic_args, generic_train
from seqeval.metrics import accuracy_score, fa_score, precision_score, recall_score
from torch.nn import CrossEntropyLoss
from torch.utils.data import DataLoader, TensorDataset
from utils_ner import TokenClassificationTask


__magic_name__			=      logging.getLogger(__name__)



class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							"""token-classification"""






  def __init__(							self :       Tuple	,_a :       int ):

        '''simple docstring'''




        if type(_a ) == dict:
              A_						:  str          =				Namespace(**_a )
        A_						:  Dict          =				import_module("""tasks""" )
        try:
              A_						:  Optional[Any]          =				getattr(_a	,hparams.task_type )
              A_						:  TokenClassificationTask          =				token_classification_task_clazz()
        except AttributeError:
              raise ValueError(
                  f'Task {hparams.task_type} needs to be defined as a TokenClassificationTask subclass in {module}. '
                  f'Available tasks classes are: {TokenClassificationTask.__subclasses__()}' )
        A_						:  Tuple          =				self.token_classification_task.get_labels(hparams.labels )
        A_						:  str          =				CrossEntropyLoss().ignore_index
        super().__init__(_a	,len(self.labels )	,self.mode )






  def        _a							(							self :       int	,**_a :       Any ):

        '''simple docstring'''




        return self.model(**_a )






  def        _a							(							self :       Optional[Any]	,_a :       Union[str, Any]	,_a :       int ):

        '''simple docstring'''




        A_						:  str          =				{"""input_ids""": batch[0], """attention_mask""": batch[1], """labels""": batch[3]}
        if self.config.model_type != "distilbert":
              A_						:  str          =				(
                  batch[2] if self.config.model_type in ["""bert""", """xlnet"""] else None
              )  # XLM and RoBERTa don"t use token_type_ids

        A_						:  str          =				self(**_a )
        A_						:  Dict          =				outputs[0]
        # tensorboard_logs = {"loss": loss, "rate": self.lr_scheduler.get_last_lr()[-1]}
        return {"loss": loss}






  def        _a							(							self :       Dict ):

        '''simple docstring'''




        A_						:  Union[str, Any]          =				self.hparams
        for mode in ["train", "dev", "test"]:
              A_						:  Union[str, Any]          =				self._feature_file(_a )
              if os.path.exists(_a ) and not args.overwrite_cache:
                    logger.info("""Loading features from cached file %s"""	,_a )
                    A_						:  Any          =				torch.load(_a )
              else:
                    logger.info("""Creating features from dataset file at %s"""	,args.data_dir )
                    A_						:  List[str]          =				self.token_classification_task.read_examples_from_file(args.data_dir	,_a )
                    A_						:  Optional[Any]          =				self.token_classification_task.convert_examples_to_features(
                        _a	,self.labels	,args.max_seq_length	,self.tokenizer	,cls_token_at_end=bool(self.config.model_type in ["""xlnet"""] )	,cls_token=self.tokenizer.cls_token	,cls_token_segment_id=2 if self.config.model_type in ["""xlnet"""] else 0	,sep_token=self.tokenizer.sep_token	,sep_token_extra=_a	,pad_on_left=bool(self.config.model_type in ["""xlnet"""] )	,pad_token=self.tokenizer.pad_token_id	,pad_token_segment_id=self.tokenizer.pad_token_type_id	,pad_token_label_id=self.pad_token_label_id	,)
                    logger.info("""Saving features into cached file %s"""	,_a )
                    torch.save(_a	,_a )






  def        _a							(							self :       Dict	,_a :       int	,_a :       int	,_a :       bool = False ):

        '''simple docstring'''




        A_						:  int          =				self._feature_file(_a )
        logger.info("""Loading features from cached file %s"""	,_a )
        A_						:  str          =				torch.load(_a )
        A_						:  Union[str, Any]          =				torch.tensor([f.input_ids for f in features]	,dtype=torch.long )
        A_						:  List[Any]          =				torch.tensor([f.attention_mask for f in features]	,dtype=torch.long )
        if features[0].token_type_ids is not None:
              A_						:  int          =				torch.tensor([f.token_type_ids for f in features]	,dtype=torch.long )
        else:
              A_						:  Any          =				torch.tensor([0 for f in features]	,dtype=torch.long )
              # HACK(we will not use this anymore soon)
        A_						:  List[Any]          =				torch.tensor([f.label_ids for f in features]	,dtype=torch.long )
        return DataLoader(
            TensorDataset(_a	,_a	,_a	,_a )	,batch_size=_a )






  def        _a							(							self :       Optional[int]	,_a :       List[str]	,_a :       Tuple ):

        '''simple docstring'''




        """Compute validation""" ""
        A_						:  List[Any]          =				{"""input_ids""": batch[0], """attention_mask""": batch[1], """labels""": batch[3]}
        if self.config.model_type != "distilbert":
              A_						:  List[str]          =				(
                  batch[2] if self.config.model_type in ["""bert""", """xlnet"""] else None
              )  # XLM and RoBERTa don"t use token_type_ids
        A_						:  Union[str, Any]          =				self(**_a )
        A_   ,					A_						:  int          =				outputs[:2]
        A_						:  List[Any]          =				logits.detach().cpu().numpy()
        A_						:  int          =				inputs["""labels"""].detach().cpu().numpy()
        return {"val_loss": tmp_eval_loss.detach().cpu(), "pred": preds, "target": out_label_ids}






  def        _a							(							self :       Optional[Any]	,_a :       List[str] ):

        '''simple docstring'''




        A_						:  Tuple          =				torch.stack([x["""val_loss"""] for x in outputs] ).mean()
        A_						:  Optional[Any]          =				np.concatenate([x["""pred"""] for x in outputs]	,axis=0 )
        A_						:  List[str]          =				np.argmax(_a	,axis=2 )
        A_						:  Dict          =				np.concatenate([x["""target"""] for x in outputs]	,axis=0 )

        A_						:  Tuple          =				dict(enumerate(self.labels ) )
        A_						:  Optional[int]          =				[[] for _ in range(out_label_ids.shape[0] )]
        A_						:  List[Any]          =				[[] for _ in range(out_label_ids.shape[0] )]

        for i in range(out_label_ids.shape[0] ):
              for j in range(out_label_ids.shape[1] ):
                    if out_label_ids[i, j] != self.pad_token_label_id:
                          out_label_list[i].append(label_map[out_label_ids[i][j]] )
                          preds_list[i].append(label_map[preds[i][j]] )

        A_						:  Union[str, Any]          =				{
            """val_loss""": val_loss_mean,
            """accuracy_score""": accuracy_score(_a	,_a ),
            """precision""": precision_score(_a	,_a ),
            """recall""": recall_score(_a	,_a ),
            """f1""": fa_score(_a	,_a ),
        }

        A_						:  Any          =				dict(results.items() )
        A_						:  Any          =				results
        return ret, preds_list, out_label_list






  def        _a							(							self :       Optional[Any]	,_a :       Any ):

        '''simple docstring'''




        A_   ,					A_   ,					A_						:  Tuple          =				self._eval_end(_a )
        A_						:  int          =				ret["""log"""]
        return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






  def        _a							(							self :       Tuple	,_a :       List[Any] ):

        '''simple docstring'''




        A_   ,					A_   ,					A_						:  Tuple          =				self._eval_end(_a )

        # Converting to the dict required by pl
        # https://github.com/PyTorchLightning/pytorch-lightning/blob/master/\
        # pytorch_lightning/trainer/logging.py#L139
        A_						:  str          =				ret["""log"""]
        # `val_loss` is the key returned by `self._eval_end()` but actually refers to `test_loss`
        return {"avg_test_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






  @staticmethod
  def        _a							(							_a :       Tuple	,_a :       str ):

        '''simple docstring'''




        BaseTransformer.add_model_specific_args(_a	,_a )
        parser.add_argument(
            """--task_type"""	,default="""NER"""	,type=_a	,help="""Task type to fine tune in training (e.g. NER, POS, etc)""" )
        parser.add_argument(
            """--max_seq_length"""	,default=128	,type=_a	,help=(
                """The maximum total input sequence length after tokenization. Sequences longer """
                """than this will be truncated, sequences shorter will be padded."""
            )	,)

        parser.add_argument(
            """--labels"""	,default=""""""	,type=_a	,help="""Path to a file containing all labels. If not specified, CoNLL-2003 labels are used."""	,)
        parser.add_argument(
            """--gpus"""	,default=0	,type=_a	,help="""The number of GPUs allocated for this, it is by default 0 meaning none"""	,)

        parser.add_argument(
            """--overwrite_cache"""	,action="""store_true"""	,help="""Overwrite the cached training and evaluation sets""" )

        return parser


if __name__ == "__main__":
     __magic_name__			=      argparse.ArgumentParser()
     add_generic_args(parser, os.getcwd())
     __magic_name__			=      NERTransformer.add_model_specific_args(parser, os.getcwd())
     __magic_name__			=      parser.parse_args()
     __magic_name__			=      NERTransformer(args)
     __magic_name__			=      generic_train(model, args)

     if args.do_predict:
          # See https://github.com/huggingface/transformers/issues/3159
          # pl use this default format to create a checkpoint:
          # https://github.com/PyTorchLightning/pytorch-lightning/blob/master\
          # /pytorch_lightning/callbacks/model_checkpoint.py#L322
          __magic_name__			=      sorted(glob.glob(os.path.join(args.output_dir, 'checkpoint-epoch=*.ckpt'), recursive=True))
          __magic_name__			=      model.load_from_checkpoint(checkpoints[-1])
          trainer.test(model)




'''simple docstring'''


import copy
import os
from collections import OrderedDict
from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union


if TYPE_CHECKING:
     from ...processing_utils import ProcessorMixin
     from ...utils import TensorType

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


__magic_name__			=      logging.get_logger(__name__)

__magic_name__			=      {
    'google/owlvit-base-patch32': 'https://huggingface.co/google/owlvit-base-patch32/resolve/main/config.json',
    'google/owlvit-base-patch16': 'https://huggingface.co/google/owlvit-base-patch16/resolve/main/config.json',
    'google/owlvit-large-patch14': 'https://huggingface.co/google/owlvit-large-patch14/resolve/main/config.json',
}



class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							"""owlvit_text_model"""






  def __init__(							self :       Union[str, Any]	,_a :       Any=49408	,_a :       Any=512	,_a :       Tuple=2048	,_a :       Dict=12	,_a :       Optional[int]=8	,_a :       Tuple=16	,_a :       Tuple="quick_gelu"	,_a :       Optional[Any]=1e-5	,_a :       List[Any]=0.0	,_a :       Optional[int]=0.02	,_a :       Dict=1.0	,_a :       Dict=0	,_a :       Any=49406	,_a :       Tuple=49407	,**_a :       List[Any]	,):

        '''simple docstring'''




        super().__init__(pad_token_id=_a	,bos_token_id=_a	,eos_token_id=_a	,**_a )

        A_						:  Tuple          =				vocab_size
        A_						:  int          =				hidden_size
        A_						:  Optional[int]          =				intermediate_size
        A_						:  Optional[int]          =				num_hidden_layers
        A_						:  Union[str, Any]          =				num_attention_heads
        A_						:  int          =				max_position_embeddings
        A_						:  str          =				hidden_act
        A_						:  Union[str, Any]          =				layer_norm_eps
        A_						:  Tuple          =				attention_dropout
        A_						:  Union[str, Any]          =				initializer_range
        A_						:  List[Any]          =				initializer_factor






  @classmethod
  def        _a							(							cls :       List[str]	,_a :       Union[str, os.PathLike]	,**_a :       str ):

        '''simple docstring'''




        cls._set_token_in_kwargs(_a )

        A_   ,					A_						:  int          =				cls.get_config_dict(_a	,**_a )

        # get the text config dict if we are loading from OwlViTConfig
        if config_dict.get("""model_type""" ) == "owlvit":
              A_						:  Union[str, Any]          =				config_dict["""text_config"""]

        if "model_type" in config_dict and hasattr(cls	,"""model_type""" ) and config_dict["model_type"] != cls.model_type:
              logger.warning(
                  f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
                  f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.' )

        return cls.from_dict(_a	,**_a )







class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							"""owlvit_vision_model"""






  def __init__(							self :       List[Any]	,_a :       Optional[Any]=768	,_a :       Tuple=3072	,_a :       Dict=12	,_a :       int=12	,_a :       Dict=3	,_a :       Tuple=768	,_a :       int=32	,_a :       int="quick_gelu"	,_a :       List[Any]=1e-5	,_a :       Tuple=0.0	,_a :       List[Any]=0.02	,_a :       str=1.0	,**_a :       int	,):

        '''simple docstring'''




        super().__init__(**_a )

        A_						:  List[str]          =				hidden_size
        A_						:  Union[str, Any]          =				intermediate_size
        A_						:  Union[str, Any]          =				num_hidden_layers
        A_						:  Optional[Any]          =				num_attention_heads
        A_						:  int          =				num_channels
        A_						:  str          =				image_size
        A_						:  List[Any]          =				patch_size
        A_						:  int          =				hidden_act
        A_						:  List[Any]          =				layer_norm_eps
        A_						:  List[str]          =				attention_dropout
        A_						:  str          =				initializer_range
        A_						:  str          =				initializer_factor






  @classmethod
  def        _a							(							cls :       List[Any]	,_a :       Union[str, os.PathLike]	,**_a :       str ):

        '''simple docstring'''




        cls._set_token_in_kwargs(_a )

        A_   ,					A_						:  Optional[int]          =				cls.get_config_dict(_a	,**_a )

        # get the vision config dict if we are loading from OwlViTConfig
        if config_dict.get("""model_type""" ) == "owlvit":
              A_						:  List[str]          =				config_dict["""vision_config"""]

        if "model_type" in config_dict and hasattr(cls	,"""model_type""" ) and config_dict["model_type"] != cls.model_type:
              logger.warning(
                  f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
                  f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.' )

        return cls.from_dict(_a	,**_a )







class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							"""owlvit"""
  a_  					=							True






  def __init__(							self :       Union[str, Any]	,_a :       List[str]=None	,_a :       List[str]=None	,_a :       Dict=512	,_a :       List[Any]=2.6592	,_a :       Optional[Any]=True	,**_a :       Optional[int]	,):

        '''simple docstring'''




        super().__init__(**_a )

        if text_config is None:
              A_						:  List[Any]          =				{}
              logger.info("""text_config is None. Initializing the OwlViTTextConfig with default values.""" )

        if vision_config is None:
              A_						:  Tuple          =				{}
              logger.info("""vision_config is None. initializing the OwlViTVisionConfig with default values.""" )

        A_						:  Dict          =				OwlViTTextConfig(**_a )
        A_						:  Dict          =				OwlViTVisionConfig(**_a )

        A_						:  Any          =				projection_dim
        A_						:  Optional[int]          =				logit_scale_init_value
        A_						:  Optional[int]          =				return_dict
        A_						:  Dict          =				1.0






  @classmethod
  def        _a							(							cls :       Union[str, Any]	,_a :       Union[str, os.PathLike]	,**_a :       Optional[int] ):

        '''simple docstring'''




        cls._set_token_in_kwargs(_a )

        A_   ,					A_						:  List[Any]          =				cls.get_config_dict(_a	,**_a )

        if "model_type" in config_dict and hasattr(cls	,"""model_type""" ) and config_dict["model_type"] != cls.model_type:
              logger.warning(
                  f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
                  f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.' )

        return cls.from_dict(_a	,**_a )






  @classmethod
  def        _a							(							cls :       int	,_a :       Dict	,_a :       Dict	,**_a :       List[str] ):

        '''simple docstring'''




        A_						:  str          =				{}
        A_						:  int          =				text_config
        A_						:  Union[str, Any]          =				vision_config

        return cls.from_dict(_a	,**_a )






  def        _a							(							self :       Optional[int] ):

        '''simple docstring'''




        A_						:  Dict          =				copy.deepcopy(self.__dict__ )
        A_						:  str          =				self.text_config.to_dict()
        A_						:  Optional[int]          =				self.vision_config.to_dict()
        A_						:  List[Any]          =				self.__class__.model_type
        return output







class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  @property
  def        _a							(							self :       int ):

        '''simple docstring'''




        return OrderedDict(
            [
                ("""input_ids""", {0: """batch""", 1: """sequence"""}),
                ("""pixel_values""", {0: """batch""", 1: """num_channels""", 2: """height""", 3: """width"""}),
                ("""attention_mask""", {0: """batch""", 1: """sequence"""}),
            ] )






  @property
  def        _a							(							self :       str ):

        '''simple docstring'''




        return OrderedDict(
            [
                ("""logits_per_image""", {0: """batch"""}),
                ("""logits_per_text""", {0: """batch"""}),
                ("""text_embeds""", {0: """batch"""}),
                ("""image_embeds""", {0: """batch"""}),
            ] )






  @property
  def        _a							(							self :       Optional[Any] ):

        '''simple docstring'''




        return 1e-4






  def        _a							(							self :       int	,_a :       "ProcessorMixin"	,_a :       int = -1	,_a :       int = -1	,_a :       Optional["TensorType"] = None	,):

        '''simple docstring'''




        A_						:  Any          =				super().generate_dummy_inputs(
            processor.tokenizer	,batch_size=_a	,seq_length=_a	,framework=_a )
        A_						:  Any          =				super().generate_dummy_inputs(
            processor.image_processor	,batch_size=_a	,framework=_a )
        return {**text_input_dict, **image_input_dict}






  @property
  def        _a							(							self :       Optional[Any] ):

        '''simple docstring'''




        return 14
