
import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
		from transformers.trainer import DEFAULT_CALLBACKS

		from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel




class 				lowerCamelCase__							(       __lowercase):




	'''simple docstring'''




	def __init__(      self	:Union[str, Any]		)	->				str:
								__UpperCamelCase			:     Optional[Any]         =      []







	def      _lowerCamelCase       (      self	:Optional[int]  ,     a	:Tuple  ,     a	:Any  ,     a	:List[str]  ,     **a	:List[Any]		)	->				Union[str, Any]:
								self.events.append("on_init_end"		)







	def      _lowerCamelCase       (      self	:Any  ,     a	:Dict  ,     a	:Optional[int]  ,     a	:int  ,     **a	:int		)	->				Any:
								self.events.append("on_train_begin"		)







	def      _lowerCamelCase       (      self	:List[str]  ,     a	:Optional[int]  ,     a	:Any  ,     a	:Optional[Any]  ,     **a	:List[str]		)	->				List[str]:
								self.events.append("on_train_end"		)







	def      _lowerCamelCase       (      self	:str  ,     a	:Tuple  ,     a	:Optional[Any]  ,     a	:Union[str, Any]  ,     **a	:Tuple		)	->				Tuple:
								self.events.append("on_epoch_begin"		)







	def      _lowerCamelCase       (      self	:Tuple  ,     a	:str  ,     a	:List[str]  ,     a	:Union[str, Any]  ,     **a	:Tuple		)	->				List[str]:
								self.events.append("on_epoch_end"		)







	def      _lowerCamelCase       (      self	:Optional[int]  ,     a	:Tuple  ,     a	:Any  ,     a	:int  ,     **a	:Optional[Any]		)	->				Union[str, Any]:
								self.events.append("on_step_begin"		)







	def      _lowerCamelCase       (      self	:Optional[Any]  ,     a	:Optional[int]  ,     a	:Dict  ,     a	:Union[str, Any]  ,     **a	:int		)	->				int:
								self.events.append("on_step_end"		)







	def      _lowerCamelCase       (      self	:str  ,     a	:Union[str, Any]  ,     a	:Union[str, Any]  ,     a	:List[Any]  ,     **a	:int		)	->				List[Any]:
								self.events.append("on_evaluate"		)







	def      _lowerCamelCase       (      self	:Any  ,     a	:Dict  ,     a	:Optional[int]  ,     a	:Union[str, Any]  ,     **a	:Any		)	->				Any:
								self.events.append("on_predict"		)







	def      _lowerCamelCase       (      self	:Any  ,     a	:Any  ,     a	:Any  ,     a	:Optional[Any]  ,     **a	:Optional[Any]		)	->				Optional[Any]:
								self.events.append("on_save"		)







	def      _lowerCamelCase       (      self	:int  ,     a	:int  ,     a	:List[Any]  ,     a	:Any  ,     **a	:Dict		)	->				int:
								self.events.append("on_log"		)







	def      _lowerCamelCase       (      self	:Optional[Any]  ,     a	:Union[str, Any]  ,     a	:List[Any]  ,     a	:Tuple  ,     **a	:Optional[int]		)	->				List[Any]:
								self.events.append("on_prediction_step"		)




@require_torch
class 				lowerCamelCase__							(       unittest.TestCase):




	'''simple docstring'''




	def      _lowerCamelCase       (      self	:str		)	->				Optional[int]:
								__UpperCamelCase			:     List[str]         =      tempfile.mkdtemp()







	def      _lowerCamelCase       (      self	:List[Any]		)	->				List[Any]:
								shutil.rmtree(self.output_dir		)







	def      _lowerCamelCase       (      self	:Dict  ,     a	:Dict=0  ,     a	:Tuple=0  ,     a	:Tuple=6_4  ,     a	:List[Any]=6_4  ,     a	:Union[str, Any]=None  ,     a	:str=False  ,     **a	:Tuple		)	->				str:
								# disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
								# its set to False since the tests later on depend on its value.
								__UpperCamelCase			:     Optional[Any]         =      RegressionDataset(length=a		)
								__UpperCamelCase			:     Dict         =      RegressionDataset(length=a		)
								__UpperCamelCase			:     Any         =      RegressionModelConfig(a=a  ,     b=a		)
								__UpperCamelCase			:     List[Any]         =      RegressionPreTrainedModel(a		)

								__UpperCamelCase			:     int         =      TrainingArguments(self.output_dir  ,     disable_tqdm=a  ,     report_to=[]  ,     **a		)
								return Trainer(
								    a  ,     a  ,     train_dataset=a  ,     eval_dataset=a  ,     callbacks=a  ,     )







	def      _lowerCamelCase       (      self	:str  ,     a	:int  ,     a	:Dict		)	->				int:
								self.assertEqual(len(a		)  ,     len(a		)		)

								# Order doesn't matter
								__UpperCamelCase			:     str         =      sorted(a  ,     key=lambda a		: cb.__name__ if isinstance(a  ,     a		) else cb.__class__.__name__		)
								__UpperCamelCase			:     str         =      sorted(a  ,     key=lambda a		: cb.__name__ if isinstance(a  ,     a		) else cb.__class__.__name__		)

								for cba, cba in zip(a  ,     a		):
															if isinstance(a  ,     a		) and isinstance(a  ,     a		):
																						self.assertEqual(a  ,     a		)
															elif isinstance(a  ,     a		) and not isinstance(a  ,     a		):
																						self.assertEqual(a  ,     cba.__class__		)
															elif not isinstance(a  ,     a		) and isinstance(a  ,     a		):
																						self.assertEqual(cba.__class__  ,     a		)
															else:
																						self.assertEqual(a  ,     a		)







	def      _lowerCamelCase       (      self	:Tuple  ,     a	:Any		)	->				Tuple:
								__UpperCamelCase			:     str         =      ["on_init_end", "on_train_begin"]
								__UpperCamelCase			:     Any         =      0
								__UpperCamelCase			:     Optional[int]         =      len(trainer.get_eval_dataloader()		)
								__UpperCamelCase			:     Optional[int]         =      ["on_prediction_step"] * len(trainer.get_eval_dataloader()		) + ["on_log", "on_evaluate"]
								for _ in range(trainer.state.num_train_epochs		):
															expected_events.append("on_epoch_begin"		)
															for _ in range(a		):
																						step += 1
																						expected_events += ["on_step_begin", "on_step_end"]
																						if step % trainer.args.logging_steps == 0:
																													expected_events.append("on_log"		)
																						if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
																													expected_events += evaluation_events.copy()
																						if step % trainer.args.save_steps == 0:
																													expected_events.append("on_save"		)
															expected_events.append("on_epoch_end"		)
															if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
																						expected_events += evaluation_events.copy()
								expected_events += ["on_log", "on_train_end"]
								return expected_events







	def      _lowerCamelCase       (      self	:Union[str, Any]		)	->				Optional[Any]:
								__UpperCamelCase			:     List[str]         =      self.get_trainer()
								__UpperCamelCase			:     Tuple         =      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								# Callbacks passed at init are added to the default callbacks
								__UpperCamelCase			:     Any         =      self.get_trainer(callbacks=[MyTestTrainerCallback]		)
								expected_callbacks.append(a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								# TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
								__UpperCamelCase			:     Optional[Any]         =      self.get_trainer(disable_tqdm=a		)
								__UpperCamelCase			:     Optional[Any]         =      DEFAULT_CALLBACKS.copy() + [PrinterCallback]
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)







	def      _lowerCamelCase       (      self	:str		)	->				str:
								__UpperCamelCase			:     Optional[Any]         =      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
								__UpperCamelCase			:     Optional[int]         =      self.get_trainer()

								# We can add, pop, or remove by class name
								trainer.remove_callback(a		)
								expected_callbacks.remove(a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								__UpperCamelCase			:     Tuple         =      self.get_trainer()
								__UpperCamelCase			:     str         =      trainer.pop_callback(a		)
								self.assertEqual(cb.__class__  ,     a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								trainer.add_callback(a		)
								expected_callbacks.insert(0  ,     a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								# We can also add, pop, or remove by instance
								__UpperCamelCase			:     Union[str, Any]         =      self.get_trainer()
								__UpperCamelCase			:     List[Any]         =      trainer.callback_handler.callbacks[0]
								trainer.remove_callback(a		)
								expected_callbacks.remove(a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								__UpperCamelCase			:     Optional[Any]         =      self.get_trainer()
								__UpperCamelCase			:     Any         =      trainer.callback_handler.callbacks[0]
								__UpperCamelCase			:     List[str]         =      trainer.pop_callback(a		)
								self.assertEqual(a  ,     a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)

								trainer.add_callback(a		)
								expected_callbacks.insert(0  ,     a		)
								self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     a		)







	def      _lowerCamelCase       (      self	:Union[str, Any]		)	->				List[Any]:
								import warnings

								# XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
								warnings.simplefilter(action="ignore"  ,     category=a		)

								__UpperCamelCase			:     Optional[Any]         =      self.get_trainer(callbacks=[MyTestTrainerCallback]		)
								trainer.train()
								__UpperCamelCase			:     Optional[Any]         =      trainer.callback_handler.callbacks[-2].events
								self.assertEqual(a  ,     self.get_expected_events(a		)		)

								# Independent log/save/eval
								__UpperCamelCase			:     Dict         =      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     logging_steps=5		)
								trainer.train()
								__UpperCamelCase			:     Optional[Any]         =      trainer.callback_handler.callbacks[-2].events
								self.assertEqual(a  ,     self.get_expected_events(a		)		)

								__UpperCamelCase			:     int         =      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     save_steps=5		)
								trainer.train()
								__UpperCamelCase			:     List[str]         =      trainer.callback_handler.callbacks[-2].events
								self.assertEqual(a  ,     self.get_expected_events(a		)		)

								__UpperCamelCase			:     str         =      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     eval_steps=5  ,     evaluation_strategy="steps"		)
								trainer.train()
								__UpperCamelCase			:     List[Any]         =      trainer.callback_handler.callbacks[-2].events
								self.assertEqual(a  ,     self.get_expected_events(a		)		)

								__UpperCamelCase			:     Optional[Any]         =      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     evaluation_strategy="epoch"		)
								trainer.train()
								__UpperCamelCase			:     int         =      trainer.callback_handler.callbacks[-2].events
								self.assertEqual(a  ,     self.get_expected_events(a		)		)

								# A bit of everything
								__UpperCamelCase			:     List[str]         =      self.get_trainer(
								    callbacks=[MyTestTrainerCallback]  ,     logging_steps=3  ,     save_steps=1_0  ,     eval_steps=5  ,     evaluation_strategy="steps"  ,     )
								trainer.train()
								__UpperCamelCase			:     Union[str, Any]         =      trainer.callback_handler.callbacks[-2].events
								self.assertEqual(a  ,     self.get_expected_events(a		)		)

								# warning should be emitted for duplicated callbacks
								with patch("transformers.trainer_callback.logger.warning"		) as warn_mock:
															__UpperCamelCase			:     Tuple         =      self.get_trainer(
															    callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,     )
															assert str(a		) in warn_mock.call_args[0][0]

# 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.
import torch

from ..models.auto import AutoModelForSequenceClassification, AutoTokenizer
from .base import PipelineTool




class 				lowerCamelCase__							(       __lowercase):




	'''simple docstring'''




	_A			=  'facebook/bart-large-mnli'
	_A			=  (
	    'This is a tool that classifies an English text using provided labels. It takes two inputs: `text`, which '
	    'should be the text to classify, and `labels`, which should be the list of labels to use for classification. '
	    'It returns the most likely label in the list of provided `labels` for the input text.'
	)
	_A			=  'text_classifier'
	_A			=  AutoTokenizer
	_A			=  AutoModelForSequenceClassification

	_A			=  ['text', ['text']]
	_A			=  ['text']







	def      _lowerCamelCase       (      self	:str		)	->				Tuple:
								super().setup()
								__UpperCamelCase			:     Any         =      self.model.config
								__UpperCamelCase			:     List[Any]         =      -1
								for idx, label in config.idalabel.items():
															if label.lower().startswith("entail"		):
																						__UpperCamelCase			:     Optional[Any]         =      int(a		)
								if self.entailment_id == -1:
															raise ValueError("Could not determine the entailment ID from the model config, please pass it at init."		)







	def      _lowerCamelCase       (      self	:Optional[int]  ,     a	:Dict  ,     a	:Optional[Any]		)	->				Optional[int]:
								__UpperCamelCase			:     str         =      labels
								return self.pre_processor(
								    [text] * len(a		)  ,     [f'This example is {label}' for label in labels]  ,     return_tensors="pt"  ,     padding="max_length"  ,     )







	def      _lowerCamelCase       (      self	:Any  ,     a	:Any		)	->				Optional[int]:
								__UpperCamelCase			:     Optional[int]         =      outputs.logits
								__UpperCamelCase			:     Dict         =      torch.argmax(logits[:, 2]		).item()
								return self._labels[label_id]