





'''simple docstring'''

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 					a_     (				snake_case_       ):

    '''simple docstring'''


    def __init__(		self       )		->   List[str]:
        _SCREAMING_SNAKE_CASE									=			[]


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Tuple:
        self.events.append("""on_init_end"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Optional[int]:
        self.events.append("""on_train_begin"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Any:
        self.events.append("""on_train_end"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Tuple:
        self.events.append("""on_epoch_begin"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Tuple:
        self.events.append("""on_epoch_end"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   str:
        self.events.append("""on_step_begin"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Dict:
        self.events.append("""on_step_end"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   List[Any]:
        self.events.append("""on_evaluate"""       )


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Optional[Any]:
        self.events.append("""on_predict"""       )


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


    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   int:
        self.events.append("""on_log"""       )





    def 			snake_case_(		self ,		A ,		A ,		A ,		**A       )		->   Union[str, Any]:
        self.events.append("""on_prediction_step"""       )




@require_torch
class 					a_     (				unittest.TestCase       ):

    '''simple docstring'''


    def 			snake_case_(		self       )		->   Any:
        _SCREAMING_SNAKE_CASE									=			tempfile.mkdtemp()


    def 			snake_case_(		self       )		->   List[str]:
        shutil.rmtree(self.output_dir       )


    def 			snake_case_(		self ,		A=0 ,		A=0 ,		A=64 ,		A=64 ,		A=None ,		A=False ,		**A       )		->   Dict:
        # 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.
        _SCREAMING_SNAKE_CASE									=			RegressionDataset(length=A       )
        _SCREAMING_SNAKE_CASE									=			RegressionDataset(length=A       )
        _SCREAMING_SNAKE_CASE									=			RegressionModelConfig(a=A ,		b=A       )
        _SCREAMING_SNAKE_CASE									=			RegressionPreTrainedModel(A       )

        _SCREAMING_SNAKE_CASE									=			TrainingArguments(self.output_dir ,		disable_tqdm=A ,		report_to=[] ,		**A       )
        return Trainer(
            A ,		A ,		train_dataset=A ,		eval_dataset=A ,		callbacks=A ,		)


    def 			snake_case_(		self ,		A ,		A       )		->   Tuple:
        self.assertEqual(len(A       ) ,		len(A       )       )

        # Order doesn't matter
        _SCREAMING_SNAKE_CASE									=			sorted(A ,		key=lambda A       : cb.__name__ if isinstance(A ,		A       ) else cb.__class__.__name__       )
        _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self ,		A       )		->   Optional[Any]:
        _SCREAMING_SNAKE_CASE									=			["""on_init_end""", """on_train_begin"""]
        _SCREAMING_SNAKE_CASE									=			0
        _SCREAMING_SNAKE_CASE									=			len(trainer.get_eval_dataloader()       )
        _SCREAMING_SNAKE_CASE									=			["""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 			snake_case_(		self       )		->   Tuple:
        _SCREAMING_SNAKE_CASE									=			self.get_trainer()
        _SCREAMING_SNAKE_CASE									=			DEFAULT_CALLBACKS.copy() + [ProgressCallback]
        self.check_callbacks_equality(trainer.callback_handler.callbacks ,		A       )

        # Callbacks passed at init are added to the default callbacks
        _SCREAMING_SNAKE_CASE									=			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
        _SCREAMING_SNAKE_CASE									=			self.get_trainer(disable_tqdm=A       )
        _SCREAMING_SNAKE_CASE									=			DEFAULT_CALLBACKS.copy() + [PrinterCallback]
        self.check_callbacks_equality(trainer.callback_handler.callbacks ,		A       )


    def 			snake_case_(		self       )		->   Optional[int]:
        _SCREAMING_SNAKE_CASE									=			DEFAULT_CALLBACKS.copy() + [ProgressCallback]
        _SCREAMING_SNAKE_CASE									=			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       )

        _SCREAMING_SNAKE_CASE									=			self.get_trainer()
        _SCREAMING_SNAKE_CASE									=			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
        _SCREAMING_SNAKE_CASE									=			self.get_trainer()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[0]
        trainer.remove_callback(A       )
        expected_callbacks.remove(A       )
        self.check_callbacks_equality(trainer.callback_handler.callbacks ,		A       )

        _SCREAMING_SNAKE_CASE									=			self.get_trainer()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[0]
        _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self       )		->   Tuple:
        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       )

        _SCREAMING_SNAKE_CASE									=			self.get_trainer(callbacks=[MyTestTrainerCallback]       )
        trainer.train()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[-2].events
        self.assertEqual(A ,		self.get_expected_events(A       )       )

        # Independent log/save/eval
        _SCREAMING_SNAKE_CASE									=			self.get_trainer(callbacks=[MyTestTrainerCallback] ,		logging_steps=5       )
        trainer.train()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[-2].events
        self.assertEqual(A ,		self.get_expected_events(A       )       )

        _SCREAMING_SNAKE_CASE									=			self.get_trainer(callbacks=[MyTestTrainerCallback] ,		save_steps=5       )
        trainer.train()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[-2].events
        self.assertEqual(A ,		self.get_expected_events(A       )       )

        _SCREAMING_SNAKE_CASE									=			self.get_trainer(callbacks=[MyTestTrainerCallback] ,		eval_steps=5 ,		evaluation_strategy="""steps"""       )
        trainer.train()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[-2].events
        self.assertEqual(A ,		self.get_expected_events(A       )       )

        _SCREAMING_SNAKE_CASE									=			self.get_trainer(callbacks=[MyTestTrainerCallback] ,		evaluation_strategy="""epoch"""       )
        trainer.train()
        _SCREAMING_SNAKE_CASE									=			trainer.callback_handler.callbacks[-2].events
        self.assertEqual(A ,		self.get_expected_events(A       )       )

        # A bit of everything
        _SCREAMING_SNAKE_CASE									=			self.get_trainer(
            callbacks=[MyTestTrainerCallback] ,		logging_steps=3 ,		save_steps=10 ,		eval_steps=5 ,		evaluation_strategy="""steps""" ,		)
        trainer.train()
        _SCREAMING_SNAKE_CASE									=			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:
            _SCREAMING_SNAKE_CASE									=			self.get_trainer(
                callbacks=[MyTestTrainerCallback, MyTestTrainerCallback] ,		)
            assert str(A       ) in warn_mock.call_args[0][0]






import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
					import tensorflow as tf

if is_torch_available():
					import torch


_UpperCAmelCase		:   Dict										=       logging.get_logger(__name__)







class  lowercase				:




				def __init__(					self      ,   snake_case = None      ,   snake_case = None      ,   snake_case=None      ,   snake_case=None       ):
											if not conversation_id:
																		snake_case_							=    uuid.uuida()
											if past_user_inputs is None:
																		snake_case_							=    []
											if generated_responses is None:
																		snake_case_							=    []

											snake_case_							=    conversation_id
											snake_case_							=    past_user_inputs
											snake_case_							=    generated_responses
											snake_case_							=    text




				def __eq__(					self      ,   snake_case       ):
											if not isinstance(snake_case      ,   snake_case       ):
																		return False
											if self.uuid == other.uuid:
																		return True
											return (
											    self.new_user_input == other.new_user_input
											    and self.past_user_inputs == other.past_user_inputs
											    and self.generated_responses == other.generated_responses
											)




				def     a							(					self      ,   snake_case      ,   snake_case = False       ):
											if self.new_user_input:
																		if overwrite:
																									logger.warning(
																									    F'''User input added while unprocessed input was existing: "{self.new_user_input}" was overwritten '''
																									    F'''with: "{text}".'''       )
																									snake_case_							=    text
																		else:
																									logger.warning(
																									    F'''User input added while unprocessed input was existing: "{self.new_user_input}" new input '''
																									    F'''ignored: "{text}". Set `overwrite` to True to overwrite unprocessed user input'''       )
											else:
																		snake_case_							=    text




				def     a							(					self       ):
											if self.new_user_input:
																		self.past_user_inputs.append(self.new_user_input       )
											snake_case_							=    None




				def     a							(					self      ,   snake_case       ):
											self.generated_responses.append(snake_case       )




				def     a							(					self       ):
											for user_input, generated_response in zip(self.past_user_inputs      ,   self.generated_responses       ):
																		yield True, user_input
																		yield False, generated_response
											if self.new_user_input:
																		yield True, self.new_user_input




				def __repr__(					self       ):
											snake_case_							=    F'''Conversation id: {self.uuid} \n'''
											for is_user, text in self.iter_texts():
																		snake_case_							=    'user' if is_user else 'bot'
																		output += F'''{name} >> {text} \n'''
											return output







@add_end_docstrings(
    lowercase_	,  R'''
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    '''	,  )
class  lowercase				(       lowercase_			):




				def __init__(					self      ,   *snake_case      ,   **snake_case       ):
											super().__init__(*snake_case      ,   **snake_case       )
											if self.tokenizer.pad_token_id is None:
																		snake_case_							=    self.tokenizer.eos_token




				def     a							(					self      ,   snake_case=None      ,   snake_case=None      ,   snake_case=None      ,   **snake_case       ):
											snake_case_							=    {}
											snake_case_							=    {}
											snake_case_							=    {}

											if min_length_for_response is not None:
																		snake_case_							=    min_length_for_response
											if minimum_tokens is not None:
																		snake_case_							=    minimum_tokens

											if "max_length" in generate_kwargs:
																		snake_case_							=    generate_kwargs['max_length']
																		# self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
											if clean_up_tokenization_spaces is not None:
																		snake_case_							=    clean_up_tokenization_spaces

											if generate_kwargs:
																		forward_params.update(snake_case       )
											return preprocess_params, forward_params, postprocess_params




				def __call__(					self      ,   snake_case      ,   snake_case=0      ,   **snake_case       ):
											snake_case_							=    super().__call__(snake_case      ,   num_workers=snake_case      ,   **snake_case       )
											if isinstance(snake_case      ,   snake_case       ) and len(snake_case       ) == 1:
																		return outputs[0]
											return outputs




				def     a							(					self      ,   snake_case      ,   snake_case=32       ):
											if not isinstance(snake_case      ,   snake_case       ):
																		raise ValueError('ConversationalPipeline, expects Conversation as inputs'       )
											if conversation.new_user_input is None:
																		raise ValueError(
																		    F'''Conversation with UUID {type(conversation.uuid       )} does not contain new user input to process. '''
																		    'Add user inputs with the conversation\'s `add_user_input` method'       )
											if hasattr(self.tokenizer      ,   '_build_conversation_input_ids'       ):
																		snake_case_							=    self.tokenizer._build_conversation_input_ids(snake_case       )
											else:
																		# If the tokenizer cannot handle conversations, we default to only the old version
																		snake_case_							=    self._legacy_parse_and_tokenize(snake_case       )

											if self.framework == "pt":
																		snake_case_							=    torch.LongTensor([input_ids]       )
											elif self.framework == "tf":
																		snake_case_							=    tf.constant([input_ids]       )
											return {"input_ids": input_ids, "conversation": conversation}




				def     a							(					self      ,   snake_case      ,   snake_case=10      ,   **snake_case       ):
											snake_case_							=    generate_kwargs.get('max_length'      ,   self.model.config.max_length       )

											snake_case_							=    model_inputs['input_ids'].shape[1]
											if max_length - minimum_tokens < n:
																		logger.warning(F'''Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})'''       )
																		snake_case_							=    max_length - minimum_tokens
																		snake_case_							=    model_inputs['input_ids'][:, -trim:]
																		if "attention_mask" in model_inputs:
																									snake_case_							=    model_inputs['attention_mask'][:, -trim:]
											snake_case_							=    model_inputs.pop('conversation'       )
											snake_case_							=    max_length
											snake_case_							=    self.model.generate(**snake_case      ,   **snake_case       )
											if self.model.config.is_encoder_decoder:
																		snake_case_							=    1
											else:
																		snake_case_							=    n
											return {"output_ids": output_ids[:, start_position:], "conversation": conversation}




				def     a							(					self      ,   snake_case      ,   snake_case=True       ):
											snake_case_							=    model_outputs['output_ids']
											snake_case_							=    self.tokenizer.decode(
											    output_ids[0]      ,   skip_special_tokens=snake_case      ,   clean_up_tokenization_spaces=snake_case      ,   )
											snake_case_							=    model_outputs['conversation']
											conversation.mark_processed()
											conversation.append_response(snake_case       )
											return conversation




				def     a							(					self      ,   snake_case       ):
											snake_case_							=    self.tokenizer.eos_token_id
											snake_case_							=    []
											for is_user, text in conversation.iter_texts():
																		if eos_token_id is not None:
																									input_ids.extend(self.tokenizer.encode(snake_case      ,   add_special_tokens=snake_case       ) + [eos_token_id]       )
																		else:
																									input_ids.extend(self.tokenizer.encode(snake_case      ,   add_special_tokens=snake_case       )       )

											if len(snake_case       ) > self.tokenizer.model_max_length:
																		snake_case_							=    input_ids[-self.tokenizer.model_max_length :]
											return input_ids




