



import json
import logging
import os
import sys
from pathlib import Path

import finetune_rag

from transformers.file_utils import is_apex_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    require_ray,
    require_torch_gpu,
    require_torch_multi_gpu,
)


logging.basicConfig(level=logging.DEBUG)
SCREAMING_SNAKE_CASE__       =      logging.getLogger()

SCREAMING_SNAKE_CASE__       =      logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)


class      __lowerCAmelCase     (		UpperCAmelCase_      ):
					"""simple docstring"""






					def     _a							(     self					:       Optional[int]				,					_snake_case					:       int					):






								"""simple docstring"""


								os.makedirs(_snake_case				,					exist_ok=_snake_case					)
								A__       =    {'source': 'What is love ?', 'target': 'life'}
								A__       =    {'train': 12, 'val': 2, 'test': 2}
								for split in ["train", "test", "val"]:
											for field in ["source", "target"]:
														A__       =    '\n'.join([contents[field]] * n_lines[split]					)
														with open(os.path.join(_snake_case				,					F'''{split}.{field}'''					)				,					'w'					) as f:
																	f.write(_snake_case					)



					def     _a							(     self					:       Union[str, Any]				,					_snake_case					:       int				,					_snake_case					:       str = "pytorch"					):






								"""simple docstring"""


								A__       =    self.get_auto_remove_tmp_dir()
								A__       =    os.path.join(_snake_case				,					'output'					)
								A__       =    os.path.join(_snake_case				,					'data'					)
								self._create_dummy_data(data_dir=_snake_case					)

								A__       =    F'''
                --data_dir {data_dir} \
                --output_dir {output_dir} \
                --model_name_or_path facebook/rag-sequence-base \
                --model_type rag_sequence \
                --do_train \
                --do_predict \
                --n_val -1 \
                --val_check_interval 1.0 \
                --train_batch_size 2 \
                --eval_batch_size 1 \
                --max_source_length 25 \
                --max_target_length 25 \
                --val_max_target_length 25 \
                --test_max_target_length 25 \
                --label_smoothing 0.1 \
                --dropout 0.1 \
                --attention_dropout 0.1 \
                --weight_decay 0.001 \
                --adam_epsilon 1e-08 \
                --max_grad_norm 0.1 \
                --lr_scheduler polynomial \
                --learning_rate 3e-04 \
                --num_train_epochs 1 \
                --warmup_steps 4 \
                --gradient_accumulation_steps 1 \
                --distributed-port 8787 \
                --use_dummy_dataset 1 \
                --distributed_retriever {distributed_retriever} \
            '''.split()

								if gpus > 0:
											testargs.append(F'''--gpus={gpus}'''					)
											if is_apex_available():
														testargs.append('--fp16'					)
								else:
											testargs.append('--gpus=0'					)
											testargs.append('--distributed_backend=ddp_cpu'					)
											testargs.append('--num_processes=2'					)

								A__       =    [sys.executable, str(Path(finetune_rag.__file__					).resolve()					)] + testargs
								execute_subprocess_async(_snake_case				,					env=self.get_env()					)

								A__       =    os.path.join(_snake_case				,					'metrics.json'					)
								with open(_snake_case					) as f:
											A__       =    json.load(_snake_case					)
								return result



					@require_torch_gpu
					def     _a							(     self					:       Dict					):






								"""simple docstring"""


								A__       =    self._run_finetune(gpus=1					)
								self.assertGreaterEqual(result['test'][0]['test_avg_em']				,					0.2					)



					@require_torch_multi_gpu
					def     _a							(     self					:       Tuple					):






								"""simple docstring"""


								A__       =    self._run_finetune(gpus=2					)
								self.assertGreaterEqual(result['test'][0]['test_avg_em']				,					0.2					)



					@require_torch_gpu
					@require_ray
					def     _a							(     self					:       List[Any]					):






								"""simple docstring"""


								A__       =    self._run_finetune(gpus=1				,					distributed_retriever='ray'					)
								self.assertGreaterEqual(result['test'][0]['test_avg_em']				,					0.2					)



					@require_torch_multi_gpu
					@require_ray
					def     _a							(     self					:       Optional[int]					):






								"""simple docstring"""


								A__       =    self._run_finetune(gpus=1				,					distributed_retriever='ray'					)
								self.assertGreaterEqual(result['test'][0]['test_avg_em']				,					0.2					)









from typing import List, Optional, Union

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


class      __lowerCAmelCase     (		UpperCAmelCase_      ):
					"""simple docstring"""






					A__				:		Union[str, Any]			       =    ["image_processor", "tokenizer"]
					A__				:		Optional[Any]			       =    "BridgeTowerImageProcessor"
					A__				:		List[Any]			       =    ("RobertaTokenizer", "RobertaTokenizerFast")



					def __init__(     self					:       List[Any]				,					_snake_case					:       Optional[Any]				,					_snake_case					:       Optional[int]					):






								"""simple docstring"""


								super().__init__(_snake_case				,					_snake_case					)



					def __call__(     self					:       List[Any]				,					_snake_case					:       int				,					_snake_case					:       Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None				,					_snake_case					:       bool = True				,					_snake_case					:       Union[bool, str, PaddingStrategy] = False				,					_snake_case					:       Union[bool, str, TruncationStrategy] = None				,					_snake_case					:       Optional[int] = None				,					_snake_case					:       int = 0				,					_snake_case					:       Optional[int] = None				,					_snake_case					:       Optional[bool] = None				,					_snake_case					:       Optional[bool] = None				,					_snake_case					:       bool = False				,					_snake_case					:       bool = False				,					_snake_case					:       bool = False				,					_snake_case					:       bool = False				,					_snake_case					:       bool = True				,					_snake_case					:       Optional[Union[str, TensorType]] = None				,					**_snake_case					:       Optional[int]				,					):






								"""simple docstring"""


								A__       =    self.tokenizer(
								    text=_snake_case				,					add_special_tokens=_snake_case				,					padding=_snake_case				,					truncation=_snake_case				,					max_length=_snake_case				,					stride=_snake_case				,					pad_to_multiple_of=_snake_case				,					return_token_type_ids=_snake_case				,					return_attention_mask=_snake_case				,					return_overflowing_tokens=_snake_case				,					return_special_tokens_mask=_snake_case				,					return_offsets_mapping=_snake_case				,					return_length=_snake_case				,					verbose=_snake_case				,					return_tensors=_snake_case				,					**_snake_case				,					)
								# add pixel_values + pixel_mask
								A__       =    self.image_processor(
								    _snake_case				,					return_tensors=_snake_case				,					do_normalize=_snake_case				,					do_center_crop=_snake_case				,					**_snake_case					)
								encoding.update(_snake_case					)

								return encoding



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






								"""simple docstring"""


								return self.tokenizer.batch_decode(*_snake_case				,					**_snake_case					)



					def     _a							(     self					:       Dict				,					*_snake_case					:       Dict				,					**_snake_case					:       List[str]					):






								"""simple docstring"""


								return self.tokenizer.decode(*_snake_case				,					**_snake_case					)



					@property
					def     _a							(     self					:       Tuple					):






								"""simple docstring"""


								A__       =    self.tokenizer.model_input_names
								A__       =    self.image_processor.model_input_names
								return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names					)					)




