






"""simple docstring"""





import logging
import os
import sys
from dataclasses import dataclass, field
from importlib import import_module
from typing import Dict, List, Optional, Tuple

import numpy as np
from seqeval.metrics import accuracy_score, fa_score, precision_score, recall_score
from torch import nn
from utils_ner import Split, TokenClassificationDataset, TokenClassificationTask

import transformers
from transformers import (
    AutoConfig,
    AutoModelForTokenClassification,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import is_main_process


__lowerCamelCase							=  logging.getLogger(__name__)



@dataclass
class       __A		:
					UpperCAmelCase__								=				field(
					    metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"}			)
					UpperCAmelCase__								=				field(
					    default=SCREAMING_SNAKE_CASE_     ,metadata={"help": "Pretrained config name or path if not the same as model_name"}			)
					UpperCAmelCase__								=				field(
					    default="NER"     ,metadata={"help": "Task type to fine tune in training (e.g. NER, POS, etc)"}			)
					UpperCAmelCase__								=				field(
					    default=SCREAMING_SNAKE_CASE_     ,metadata={"help": "Pretrained tokenizer name or path if not the same as model_name"}			)
					UpperCAmelCase__								=				field(default=SCREAMING_SNAKE_CASE_     ,metadata={"help": "Set this flag to use fast tokenization."}			)
					# If you want to tweak more attributes on your tokenizer, you should do it in a distinct script,
					# or just modify its tokenizer_config.json.
					UpperCAmelCase__								=				field(
					    default=SCREAMING_SNAKE_CASE_     ,metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}     ,)



@dataclass
class       __A		:
					UpperCAmelCase__								=				field(
					    metadata={"help": "The input data dir. Should contain the .txt files for a CoNLL-2003-formatted task."}			)
					UpperCAmelCase__								=				field(
					    default=SCREAMING_SNAKE_CASE_     ,metadata={"help": "Path to a file containing all labels. If not specified, CoNLL-2003 labels are used."}     ,)
					UpperCAmelCase__								=				field(
					    default=1_2_8     ,metadata={
					        "help": (
					            "The maximum total input sequence length after tokenization. Sequences longer "
					            "than this will be truncated, sequences shorter will be padded."
					        )
					    }     ,)
					UpperCAmelCase__								=				field(
					    default=SCREAMING_SNAKE_CASE_     ,metadata={"help": "Overwrite the cached training and evaluation sets"}			)



def   a    (     )    ->							Optional[int]:
						# See all possible arguments in src/transformers/training_args.py
						# or by passing the --help flag to this script.
						# We now keep distinct sets of args, for a cleaner separation of concerns.

						__magic_name__:	Optional[Any]							=       HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)  )
						if len(sys.argv  ) == 2 and sys.argv[1].endswith(""".json"""  ):
												# If we pass only one argument to the script and it's the path to a json file,
												# let's parse it to get our arguments.
												__magic_name__,			__magic_name__,			__magic_name__:	Optional[Any]							=       parser.parse_json_file(json_file=os.path.abspath(sys.argv[1]  )  )
						else:
												__magic_name__,			__magic_name__,			__magic_name__:	Union[str, Any]							=       parser.parse_args_into_dataclasses()

						if (
						    os.path.exists(training_args.output_dir  )
						    and os.listdir(training_args.output_dir  )
						    and training_args.do_train
						    and not training_args.overwrite_output_dir
						):
												raise ValueError(
												    f'Output directory ({training_args.output_dir}) already exists and is not empty. Use'
												    """ --overwrite_output_dir to overcome."""  )

						__magic_name__:	Optional[Any]							=       import_module("""tasks"""  )
						try:
												__magic_name__:	Dict							=       getattr(__UpperCAmelCase      ,   model_args.task_type  )
												__magic_name__:	TokenClassificationTask							=       token_classification_task_clazz()
						except AttributeError:
												raise ValueError(
												    f'Task {model_args.task_type} needs to be defined as a TokenClassificationTask subclass in {module}. '
												    f'Available tasks classes are: {TokenClassificationTask.__subclasses__()}'  )

						# Setup logging
						logging.basicConfig(
						    format="""%(asctime)s - %(levelname)s - %(name)s - %(message)s"""      ,   datefmt="""%m/%d/%Y %H:%M:%S"""      ,   level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN      ,   )
						logger.warning(
						    """Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"""      ,   training_args.local_rank      ,   training_args.device      ,   training_args.n_gpu      ,   bool(training_args.local_rank != -1  )      ,   training_args.fpaa      ,   )
						# Set the verbosity to info of the Transformers logger (on main process only):
						if is_main_process(training_args.local_rank  ):
												transformers.utils.logging.set_verbosity_info()
												transformers.utils.logging.enable_default_handler()
												transformers.utils.logging.enable_explicit_format()
						logger.info("""Training/evaluation parameters %s"""      ,   __UpperCAmelCase  )

						# Set seed
						set_seed(training_args.seed  )

						# Prepare CONLL-2003 task
						__magic_name__:	Optional[int]							=       token_classification_task.get_labels(data_args.labels  )
						__magic_name__:	Dict[int, str]							=       dict(enumerate(__UpperCAmelCase  )  )
						__magic_name__:	int							=       len(__UpperCAmelCase  )

						# Load pretrained model and tokenizer
						#
						# Distributed training:
						# The .from_pretrained methods guarantee that only one local process can concurrently
						# download model & vocab.

						__magic_name__:	List[Any]							=       AutoConfig.from_pretrained(
						    model_args.config_name if model_args.config_name else model_args.model_name_or_path      ,   num_labels=__UpperCAmelCase      ,   idalabel=__UpperCAmelCase      ,   labelaid={label: i for i, label in enumerate(__UpperCAmelCase  )}      ,   cache_dir=model_args.cache_dir      ,   )
						__magic_name__:	Any							=       AutoTokenizer.from_pretrained(
						    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path      ,   cache_dir=model_args.cache_dir      ,   use_fast=model_args.use_fast      ,   )
						__magic_name__:	Union[str, Any]							=       AutoModelForTokenClassification.from_pretrained(
						    model_args.model_name_or_path      ,   from_tf=bool(""".ckpt""" in model_args.model_name_or_path  )      ,   config=__UpperCAmelCase      ,   cache_dir=model_args.cache_dir      ,   )

						# Get datasets
						__magic_name__:	int							=       (
						    TokenClassificationDataset(
						        token_classification_task=__UpperCAmelCase      ,   data_dir=data_args.data_dir      ,   tokenizer=__UpperCAmelCase      ,   labels=__UpperCAmelCase      ,   model_type=config.model_type      ,   max_seq_length=data_args.max_seq_length      ,   overwrite_cache=data_args.overwrite_cache      ,   mode=Split.train      ,   )
						    if training_args.do_train
						    else None
						)
						__magic_name__:	Tuple							=       (
						    TokenClassificationDataset(
						        token_classification_task=__UpperCAmelCase      ,   data_dir=data_args.data_dir      ,   tokenizer=__UpperCAmelCase      ,   labels=__UpperCAmelCase      ,   model_type=config.model_type      ,   max_seq_length=data_args.max_seq_length      ,   overwrite_cache=data_args.overwrite_cache      ,   mode=Split.dev      ,   )
						    if training_args.do_eval
						    else None
						)

						def align_predictions(__UpperCAmelCase      :    np.ndarray      ,   __UpperCAmelCase      :    np.ndarray  ) -> Tuple[List[int], List[int]]:
												__magic_name__:	List[str]							=       np.argmax(__UpperCAmelCase      ,   axis=2  )

												__magic_name__,			__magic_name__:	Union[str, Any]							=       preds.shape

												__magic_name__:	Union[str, Any]							=       [[] for _ in range(__UpperCAmelCase  )]
												__magic_name__:	int							=       [[] for _ in range(__UpperCAmelCase  )]

												for i in range(__UpperCAmelCase  ):
																		for j in range(__UpperCAmelCase  ):
																								if label_ids[i, j] != nn.CrossEntropyLoss().ignore_index:
																														out_label_list[i].append(label_map[label_ids[i][j]]  )
																														preds_list[i].append(label_map[preds[i][j]]  )

												return preds_list, out_label_list

						def compute_metrics(__UpperCAmelCase      :    EvalPrediction  ) -> Dict:
												__magic_name__,			__magic_name__:	int							=       align_predictions(p.predictions      ,   p.label_ids  )
												return {
												    "accuracy_score": accuracy_score(__UpperCAmelCase      ,   __UpperCAmelCase  ),
												    "precision": precision_score(__UpperCAmelCase      ,   __UpperCAmelCase  ),
												    "recall": recall_score(__UpperCAmelCase      ,   __UpperCAmelCase  ),
												    "f1": fa_score(__UpperCAmelCase      ,   __UpperCAmelCase  ),
												}

						# Data collator
						__magic_name__:	Union[str, Any]							=       DataCollatorWithPadding(__UpperCAmelCase      ,   pad_to_multiple_of=8  ) if training_args.fpaa else None

						# Initialize our Trainer
						__magic_name__:	List[str]							=       Trainer(
						    model=__UpperCAmelCase      ,   args=__UpperCAmelCase      ,   train_dataset=__UpperCAmelCase      ,   eval_dataset=__UpperCAmelCase      ,   compute_metrics=__UpperCAmelCase      ,   data_collator=__UpperCAmelCase      ,   )

						# Training
						if training_args.do_train:
												trainer.train(
												    model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path  ) else None  )
												trainer.save_model()
												# For convenience, we also re-save the tokenizer to the same directory,
												# so that you can share your model easily on huggingface.co/models =)
												if trainer.is_world_process_zero():
																		tokenizer.save_pretrained(training_args.output_dir  )

    # Evaluation
						__magic_name__:	List[Any]							=       {}
						if training_args.do_eval:
												logger.info("""*** Evaluate ***"""  )

												__magic_name__:	int							=       trainer.evaluate()

												__magic_name__:	int							=       os.path.join(training_args.output_dir      ,   """eval_results.txt"""  )
												if trainer.is_world_process_zero():
																		with open(__UpperCAmelCase      ,   """w"""  ) as writer:
																								logger.info("""***** Eval results *****"""  )
																								for key, value in result.items():
																														logger.info("""  %s = %s"""      ,   __UpperCAmelCase      ,   __UpperCAmelCase  )
																														writer.write("""%s = %s\n""" % (key, value)  )

																		results.update(__UpperCAmelCase  )

    # Predict
						if training_args.do_predict:
												__magic_name__:	Optional[int]							=       TokenClassificationDataset(
												    token_classification_task=__UpperCAmelCase      ,   data_dir=data_args.data_dir      ,   tokenizer=__UpperCAmelCase      ,   labels=__UpperCAmelCase      ,   model_type=config.model_type      ,   max_seq_length=data_args.max_seq_length      ,   overwrite_cache=data_args.overwrite_cache      ,   mode=Split.test      ,   )

												__magic_name__,			__magic_name__,			__magic_name__:	str							=       trainer.predict(__UpperCAmelCase  )
												__magic_name__,			__magic_name__:	Any							=       align_predictions(__UpperCAmelCase      ,   __UpperCAmelCase  )

												__magic_name__:	str							=       os.path.join(training_args.output_dir      ,   """test_results.txt"""  )
												if trainer.is_world_process_zero():
																		with open(__UpperCAmelCase      ,   """w"""  ) as writer:
																								for key, value in metrics.items():
																														logger.info("""  %s = %s"""      ,   __UpperCAmelCase      ,   __UpperCAmelCase  )
																														writer.write("""%s = %s\n""" % (key, value)  )

        # Save predictions
												__magic_name__:	Optional[int]							=       os.path.join(training_args.output_dir      ,   """test_predictions.txt"""  )
												if trainer.is_world_process_zero():
																		with open(__UpperCAmelCase      ,   """w"""  ) as writer:
																								with open(os.path.join(data_args.data_dir      ,   """test.txt"""  )      ,   """r"""  ) as f:
																														token_classification_task.write_predictions_to_file(__UpperCAmelCase      ,   __UpperCAmelCase      ,   __UpperCAmelCase  )

						return results



def   a    (     __UpperCAmelCase      :    Dict  )    ->							Optional[Any]:
						# For xla_spawn (TPUs)
						main()


if __name__ == "__main__":
							main()











"""simple docstring"""





# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool



class       __A		(   SCREAMING_SNAKE_CASE_			):
					UpperCAmelCase__								=				"openai/whisper-base"
					UpperCAmelCase__								=				(
					    "This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the "
					    "transcribed text."
					)
					UpperCAmelCase__								=				"transcriber"
					UpperCAmelCase__								=				WhisperProcessor
					UpperCAmelCase__								=				WhisperForConditionalGeneration

					UpperCAmelCase__								=				["audio"]
					UpperCAmelCase__								=				["text"]

					def   lowerCamelCase__   (			self  :      Dict     ,		__snake_case  :      List[str]					)			->							Dict:
											return self.pre_processor(__snake_case     ,		return_tensors="""pt"""					).input_features

					def   lowerCamelCase__   (			self  :      int     ,		__snake_case  :      Union[str, Any]					)			->							List[str]:
											return self.model.generate(inputs=__snake_case					)






					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      List[Any]					)			->							Any:
											return self.pre_processor.batch_decode(__snake_case     ,		skip_special_tokens=__snake_case					)[0]



