



'''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 __snake_case(				_lowerCAmelCase ):


						'''simple docstring'''



						def __init__( self				)     ->     int:
							lowerCAmelCase   =     []


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     Dict:
							self.events.append("""on_init_end"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     List[str]:
							self.events.append("""on_train_begin"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     str:
							self.events.append("""on_train_end"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     List[Any]:
							self.events.append("""on_epoch_begin"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     List[Any]:
							self.events.append("""on_epoch_end"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     Optional[int]:
							self.events.append("""on_step_begin"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     Optional[Any]:
							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_				)     ->     Dict:
							self.events.append("""on_predict"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     int:
							self.events.append("""on_save"""				)


						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     List[str]:
							self.events.append("""on_log"""				)




						def     __snake_case   ( self     ,	A_     ,	A_     ,	A_     ,	**A_				)     ->     int:
							self.events.append("""on_prediction_step"""				)







@require_torch
class __snake_case(				unittest.TestCase ):


						'''simple docstring'''



						def     __snake_case   ( self				)     ->     List[Any]:
							lowerCAmelCase   =     tempfile.mkdtemp()


						def     __snake_case   ( self				)     ->     Union[str, Any]:
							shutil.rmtree(self.output_dir				)


						def     __snake_case   ( self     ,	A_=0     ,	A_=0     ,	A_=64     ,	A_=64     ,	A_=None     ,	A_=False     ,	**A_				)     ->     List[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.
							lowerCAmelCase   =     RegressionDataset(length=A_				)
							lowerCAmelCase   =     RegressionDataset(length=A_				)
							lowerCAmelCase   =     RegressionModelConfig(a=A_     ,	b=A_				)
							lowerCAmelCase   =     RegressionPreTrainedModel(A_				)

							lowerCAmelCase   =     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_				)     ->     Optional[int]:
							self.assertEqual(len(A_				)     ,	len(A_				)				)

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

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


						def     __snake_case   ( self				)     ->     Tuple:
							lowerCAmelCase   =     DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							lowerCAmelCase   =     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_				)

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

							lowerCAmelCase   =     self.get_trainer()
							lowerCAmelCase   =     trainer.callback_handler.callbacks[0]
							lowerCAmelCase   =     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				)     ->     Optional[int]:
							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_				)

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

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

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

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

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

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




'''simple docstring'''

from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType

class __snake_case(				_lowerCAmelCase ):


						'''simple docstring'''



						UpperCAmelCase							:   Union[str, Any]									=		["image_processor", "tokenizer"]
						UpperCAmelCase							:   Tuple									=		"BlipImageProcessor"
						UpperCAmelCase							:   Union[str, Any]									=		("BertTokenizer", "BertTokenizerFast")


						def __init__( self     ,	A_     ,	A_				)     ->     Dict:
							lowerCAmelCase   =     False
							super().__init__(A_     ,	A_				)
							lowerCAmelCase   =     self.image_processor


						def __call__( self     ,	A_ = None     ,	A_ = None     ,	A_ = True     ,	A_ = False     ,	A_ = None     ,	A_ = None     ,	A_ = 0     ,	A_ = None     ,	A_ = None     ,	A_ = False     ,	A_ = False     ,	A_ = False     ,	A_ = False     ,	A_ = False     ,	A_ = True     ,	A_ = None     ,	**A_     ,	)     ->     BatchEncoding:
							if images is None and text is None:
								raise ValueError("""You have to specify either images or text."""				)

							# Get only text
							if images is None:
								lowerCAmelCase   =     self.tokenizer
								lowerCAmelCase   =     self.tokenizer(
								    text=A_     ,	add_special_tokens=A_     ,	padding=A_     ,	truncation=A_     ,	max_length=A_     ,	stride=A_     ,	pad_to_multiple_of=A_     ,	return_attention_mask=A_     ,	return_overflowing_tokens=A_     ,	return_special_tokens_mask=A_     ,	return_offsets_mapping=A_     ,	return_token_type_ids=A_     ,	return_length=A_     ,	verbose=A_     ,	return_tensors=A_     ,	**A_     ,	)
								return text_encoding

							# add pixel_values
							lowerCAmelCase   =     self.image_processor(A_     ,	return_tensors=A_				)

							if text is not None:
								lowerCAmelCase   =     self.tokenizer(
								    text=A_     ,	add_special_tokens=A_     ,	padding=A_     ,	truncation=A_     ,	max_length=A_     ,	stride=A_     ,	pad_to_multiple_of=A_     ,	return_attention_mask=A_     ,	return_overflowing_tokens=A_     ,	return_special_tokens_mask=A_     ,	return_offsets_mapping=A_     ,	return_token_type_ids=A_     ,	return_length=A_     ,	verbose=A_     ,	return_tensors=A_     ,	**A_     ,	)
							else:
								lowerCAmelCase   =     None

							if text_encoding is not None:
								encoding_image_processor.update(A_				)

							return encoding_image_processor


						def     __snake_case   ( self     ,	*A_     ,	**A_				)     ->     Union[str, Any]:
							return self.tokenizer.batch_decode(*A_     ,	**A_				)


						def     __snake_case   ( self     ,	*A_     ,	**A_				)     ->     Tuple:
							return self.tokenizer.decode(*A_     ,	**A_				)




						@property
						def     __snake_case   ( self				)     ->     str:
							lowerCAmelCase   =     self.tokenizer.model_input_names
							lowerCAmelCase   =     self.image_processor.model_input_names
							return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names				)				)