




'''simple docstring'''


import argparse
import dataclasses
import json
import logging
import os
import shutil
from typing import List, Optional

import datasets
from accelerate import Accelerator
from datasets import load_dataset
from finetuning import finetune
from tqdm.auto import tqdm

import transformers
from transformers import AutoConfig, set_seed
from transformers.trainer_utils import IntervalStrategy


_snake_case : List[Any]  				=						logging.getLogger(__name__)

_snake_case : Union[str, Any]  				=						"""pytorch_model.bin"""

@dataclasses.dataclass
class      A       :
							lowercase_							     =							dataclasses.field(
							    metadata={'help': 'Path to pretrained model or model identifier from huggingface.co/models.'}       )
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'Where do you want to store the pretrained models downloaded from huggingface.co.'}	,)



@dataclasses.dataclass
class      A       :
							lowercase_							     =							dataclasses.field(metadata={'help': 'A csv or a json file containing the training data.'}       )
							lowercase_							     =							dataclasses.field(metadata={'help': 'A csv or a json file containing the data to predict on.'}       )
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'A csv or a json file containing the validation data.'}       )
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'The name of the task to train on.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'The list of labels for the task.'}       )



@dataclasses.dataclass
class      A       :
							lowercase_							     =							dataclasses.field(
							    metadata={'help': 'The output directory where the model predictions and checkpoints will be written.'}       )
							lowercase_							     =							dataclasses.field(
							    default='accuracy'	,metadata={'help': 'The evaluation metric used for the task.'}       )
							lowercase_							     =							dataclasses.field(
							    default='no'	,metadata={
							        'help': 'The evaluation strategy to adopt during training. Possible values are: ["no", "step", "epoch]'
							    }	,)
							lowercase_							     =							dataclasses.field(
							    default=10	,metadata={'help': 'Number of evaluation calls with no improvement after which training will be stopped.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=0.0	,metadata={
							        'help': 'How much the specified evaluation metric must improve to satisfy early stopping conditions.'
							    }	,)
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'Whether to filter the pseudo-labeled data based on the confidence score.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'Whether to filter the pseudo-labeled data based on the validation performance.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'Whether to fine-tune on labeled data after pseudo training.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=0.0	,metadata={'help': 'Confidence threshold for pseudo-labeled data filtering.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=100	,metadata={'help': 'Number of evaluation calls with no improvement after which training will be stopped.'}	,)
							lowercase_							     =							dataclasses.field(
							    default=__lowercase	,metadata={'help': 'Random seed for initialization.'}	,)



def 						snake_case_       (UpperCamelCase			:    Union[str, Any]					,		UpperCamelCase			:    Optional[int]					,		UpperCamelCase			:    List[str]					,		UpperCamelCase			:    List[Any]					,		UpperCamelCase			:    Dict					,		UpperCamelCase			:    List[str] ):





							'''simple docstring'''


							_a      	=							datasets.concatenate_datasets([infer_input, infer_output]					,		axis=1 )

							if args.do_filter_by_confidence:
														_a      	=							dataset.filter(lambda UpperCamelCase : example["probability"] > args.confidence_threshold )

							if args.do_filter_by_val_performance:
														assert eval_result >= 0.0 and eval_result <= 1.0
														_a      	=							int(eval_result * len(UpperCamelCase__ ) )
														print(UpperCamelCase__ )
														_a      	=							dataset.sort('''probability'''					,		reverse=UpperCamelCase__ )
														_a      	=							dataset.select(range(UpperCamelCase__ ) )

							_a      	=							dataset.remove_columns(['''label''', '''probability'''] )
							_a      	=							dataset.rename_column('''prediction'''					,		'''label''' )
							_a      	=							dataset.map(lambda UpperCamelCase : {"label": idalabel[example["label"]]} )
							_a      	=							dataset.shuffle(seed=args.seed )

							_a      	=							os.path.join(UpperCamelCase__					,		f'train_pseudo.{args.data_file_extension}' )
							if args.data_file_extension == "csv":
														dataset.to_csv(UpperCamelCase__					,		index=UpperCamelCase__ )
							else:
														dataset.to_json(UpperCamelCase__ )

def 						snake_case_       (UpperCamelCase			:    Tuple					,		UpperCamelCase			:    Union[str, Any]					,		UpperCamelCase			:    Optional[int]					,		UpperCamelCase			:    int					,		**UpperCamelCase			:    List[Any] ):





							'''simple docstring'''


							_a      	=							Accelerator()
							# Make one log on every process with the configuration for debugging.
							logging.basicConfig(
							    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s'''					,		datefmt='''%m/%d/%Y %H:%M:%S'''					,		level=logging.INFO					,		)
							logger.info(accelerator.state )

							# Setup logging, we only want one process per machine to log things on the
							# screen. accelerator.is_local_main_process is only True for one process per
							# machine.
							logger.setLevel(logging.INFO if accelerator.is_local_main_process else logging.ERROR )

							if accelerator.is_local_main_process:
														datasets.utils.logging.set_verbosity_warning()
														transformers.utils.logging.set_verbosity_info()
							else:
														datasets.utils.logging.set_verbosity_error()
														transformers.utils.logging.set_verbosity_error()

							_a      	=							STModelArguments(model_name_or_path=UpperCamelCase__ )
							_a      	=							STDataArguments(train_file=UpperCamelCase__					,		infer_file=UpperCamelCase__ )
							_a      	=							STTrainingArguments(output_dir=UpperCamelCase__ )
							_a      	=							argparse.Namespace()

							for arg_class in (model_args, data_args, training_args):
														for key, value in vars(UpperCamelCase__ ).items():
																					setattr(UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__ )

							for key, value in kwargs.items():
														if hasattr(UpperCamelCase__					,		UpperCamelCase__ ):
																					setattr(UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__ )

    # Sanity checks
							_a      	=							{}
							_a      	=							None

							# You need to provide the training data and the data to predict on
							assert args.train_file is not None
							assert args.infer_file is not None
							_a      	=							args.train_file
							_a      	=							args.infer_file

							if args.evaluation_strategy != IntervalStrategy.NO.value:
														assert args.eval_file is not None
														_a      	=							args.eval_file

							for key in data_files:
														_a      	=							data_files[key].split('''.''' )[-1]
														assert extension in ["csv", "json"], f'`{key}_file` should be a csv or a json file.'
														if args.data_file_extension is None:
																					_a      	=							extension
														else:
																					assert extension == args.data_file_extension, f'`{key}_file` should be a {args.data_file_extension} file`.'

							assert (
							    args.eval_metric in datasets.list_metrics()
							), f'{args.eval_metric} not in the list of supported metrics {datasets.list_metrics()}.'

							# If passed along, set the training seed now.
							if args.seed is not None:
														set_seed(args.seed )

							logger.info('''Creating the initial data directory for self-training...''' )
							_a      	=							f'{args.output_dir}/self-train_iter-{{}}'.format
							_a      	=							data_dir_format(0 )

							if accelerator.is_main_process:
														if args.output_dir is not None:
																					os.makedirs(args.output_dir					,		exist_ok=UpperCamelCase__ )
																					os.makedirs(UpperCamelCase__					,		exist_ok=UpperCamelCase__ )
							accelerator.wait_for_everyone()

							_a      	=							None
							_a      	=							None
							_a      	=							0
							_a      	=							False
							# Show the progress bar
							_a      	=							tqdm(range(args.max_selftrain_iterations )					,		disable=not accelerator.is_local_main_process )

							# Self-train
							for iteration in range(0					,		int(args.max_selftrain_iterations ) ):
														_a      	=							data_dir_format(UpperCamelCase__ )
														assert os.path.exists(UpperCamelCase__ )

														# Stage 1: initial fine-tuning for iteration = 0 or pseudo-training for
														# iteration > 0
														_a      	=							os.path.join(UpperCamelCase__					,		'''stage-1''' )
														_a      	=							{
														    '''accelerator''': accelerator,
														    '''model_name_or_path''': args.model_name_or_path,
														    '''cache_dir''': args.cache_dir,
														    '''do_train''': True,
														    '''train_file''': data_files['''train'''] if iteration == 0 else data_files['''train_pseudo'''],
														    '''do_eval''': True if args.eval_file is not None else False,
														    '''eval_file''': data_files['''eval'''],
														    '''do_predict''': True,
														    '''infer_file''': data_files['''infer'''],
														    '''task_name''': args.task_name,
														    '''label_list''': args.label_list,
														    '''output_dir''': current_output_dir,
														    '''eval_metric''': args.eval_metric,
														    '''evaluation_strategy''': args.evaluation_strategy,
														    '''early_stopping_patience''': args.early_stopping_patience,
														    '''early_stopping_threshold''': args.early_stopping_threshold,
														    '''seed''': args.seed,
														}
														# Add additional training arguments
														for key, value in kwargs.items():
																					if key not in arguments_dict and not hasattr(UpperCamelCase__					,		UpperCamelCase__ ):
																												arguments_dict.update({key: value} )

														_a      	=							os.path.join(UpperCamelCase__					,		'''best-checkpoint'''					,		UpperCamelCase__ )
														if os.path.exists(UpperCamelCase__ ):
																					logger.info(
																					    '''Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 1.'''					,		UpperCamelCase__					,		UpperCamelCase__					,		)
														else:
																					logger.info('''***** Running self-training: iteration: %d, stage: 1 *****'''					,		UpperCamelCase__ )
																					finetune(**UpperCamelCase__ )
																					accelerator.wait_for_everyone()
																					assert os.path.exists(UpperCamelCase__ )
																					logger.info('''Self-training job completed: iteration: %d, stage: 1.'''					,		UpperCamelCase__ )

														if iteration > 0 and args.finetune_on_labeled_data:
																					# Stage 2 (optional): fine-tuning on the original labeled data
																					_a      	=							os.path.join(UpperCamelCase__					,		'''best-checkpoint''' )
																					_a      	=							os.path.join(UpperCamelCase__					,		'''stage-2''' )
																					# Update arguments_dict
																					_a      	=							model_path
																					_a      	=							data_files['''train''']
																					_a      	=							current_output_dir

																					_a      	=							os.path.join(UpperCamelCase__					,		'''best-checkpoint'''					,		UpperCamelCase__ )
																					if os.path.exists(UpperCamelCase__ ):
																												logger.info(
																												    '''Found existing model checkpoint at %s. Skipping self-training: iteration: %d, stage: 2.'''					,		UpperCamelCase__					,		UpperCamelCase__					,		)
																					else:
																												logger.info('''***** Running self-training: iteration: %d, stage: 2 *****'''					,		UpperCamelCase__ )
																												finetune(**UpperCamelCase__ )
																												accelerator.wait_for_everyone()
																												assert os.path.exists(UpperCamelCase__ )
																												logger.info('''Self-training job completed: iteration: %d, stage: 2.'''					,		UpperCamelCase__ )

														_a      	=							iteration
														_a      	=							data_dir_format(iteration + 1 )

														_a      	=							AutoConfig.from_pretrained(os.path.join(UpperCamelCase__					,		'''best-checkpoint''' ) )
														_a      	=							config.idalabel
														_a      	=							os.path.join(UpperCamelCase__					,		'''eval_results_best-checkpoint.json''' )
														_a      	=							os.path.join(UpperCamelCase__					,		'''test_results_best-checkpoint.json''' )
														assert os.path.exists(UpperCamelCase__ )

														with open(UpperCamelCase__					,		'''r''' ) as f:
																					_a      	=							float(json.load(UpperCamelCase__ )[args.eval_metric] )
														_a      	=							os.path.join(UpperCamelCase__					,		'''infer_output_best-checkpoint.csv''' )
														assert os.path.exists(UpperCamelCase__ )
														# Loading the dataset from local csv or json files.
														_a      	=							load_dataset(args.data_file_extension					,		data_files={'''data''': data_files['''infer''']} )['''data''']
														_a      	=							load_dataset('''csv'''					,		data_files={'''data''': infer_output_file} )['''data''']

														if accelerator.is_main_process:
																					os.makedirs(UpperCamelCase__					,		exist_ok=UpperCamelCase__ )
																					shutil.copy(UpperCamelCase__					,		os.path.join(UpperCamelCase__					,		f'eval_results_iter-{iteration}.json' ) )
																					if os.path.exists(UpperCamelCase__ ):
																												shutil.copy(UpperCamelCase__					,		os.path.join(UpperCamelCase__					,		f'test_results_iter-{iteration}.json' ) )
																					create_pseudo_labeled_data(UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__					,		UpperCamelCase__ )
														accelerator.wait_for_everyone()

														_a      	=							os.path.join(UpperCamelCase__					,		f'train_pseudo.{args.data_file_extension}' )

														if args.evaluation_strategy != IntervalStrategy.NO.value:
																					_a      	=							eval_result

																					if best_iteration is None:
																												_a      	=							new_iteration
																												_a      	=							new_eval_result
																					else:
																												if new_eval_result - best_eval_result > args.early_stopping_threshold:
																																			_a      	=							new_iteration
																																			_a      	=							new_eval_result
																																			_a      	=							0
																												else:
																																			if new_eval_result == best_eval_result:
																																										_a      	=							new_iteration
																																										_a      	=							new_eval_result
																																			early_stopping_patience_counter += 1

																												if early_stopping_patience_counter >= args.early_stopping_patience:
																																			_a      	=							True

														progress_bar.update(1 )

														if should_training_stop:
																					break

							if best_iteration is not None:
														# Save the best iteration
														logger.info('''Best iteration: %d'''					,		UpperCamelCase__ )
														logger.info('''Best evaluation result: %s = %f'''					,		args.eval_metric					,		UpperCamelCase__ )
														accelerator.wait_for_everyone()
														if accelerator.is_main_process:
																					shutil.copy(
																					    os.path.join(UpperCamelCase__					,		f'eval_results_iter-{iteration}.json' )					,		os.path.join(UpperCamelCase__					,		'''eval_results_best-iteration.json''' )					,		)
							else:
														# Assume that the last iteration is the best
														logger.info('''Best iteration: %d'''					,		args.max_selftrain_iterations - 1 )
														logger.info('''Best evaluation result: %s = %f'''					,		args.eval_metric					,		UpperCamelCase__ )
														accelerator.wait_for_everyone()
														if accelerator.is_main_process:
																					shutil.copy(
																					    os.path.join(UpperCamelCase__					,		f'eval_results_iter-{args.max_selftrain_iterations - 1}.json' )					,		os.path.join(UpperCamelCase__					,		'''eval_results_best-iteration.json''' )					,		)








from typing import Callable, Optional

from .. import Features
from ..packaged_modules.generator.generator import Generator
from .abc import AbstractDatasetInputStream





class 				a			(      __lowercase			):






				def __init__(  self			, _lowerCAmelCase			, _lowerCAmelCase = None			, _lowerCAmelCase = None			, _lowerCAmelCase = False			, _lowerCAmelCase = False			, _lowerCAmelCase = None			, _lowerCAmelCase = None			, **_lowerCAmelCase			, ):

										"""simple docstring"""




										super().__init__(
										    features=_lowerCAmelCase			, cache_dir=_lowerCAmelCase			, keep_in_memory=_lowerCAmelCase			, streaming=_lowerCAmelCase			, num_proc=_lowerCAmelCase			, **_lowerCAmelCase			, )
										__SCREAMING_SNAKE_CASE:				Any           =   Generator(
										    cache_dir=_lowerCAmelCase			, features=_lowerCAmelCase			, generator=_lowerCAmelCase			, gen_kwargs=_lowerCAmelCase			, **_lowerCAmelCase			, )






				def 			snake_case_    (  self					):

										"""simple docstring"""




										if self.streaming:
																__SCREAMING_SNAKE_CASE:				List[str]           =   self.builder.as_streaming_dataset(split='''train'''					)
										# Build regular (map-style) dataset
										else:
																__SCREAMING_SNAKE_CASE:				str           =   None
																__SCREAMING_SNAKE_CASE:				List[Any]           =   None
																__SCREAMING_SNAKE_CASE:				Tuple           =   None
																__SCREAMING_SNAKE_CASE:				Optional[Any]           =   None

																self.builder.download_and_prepare(
																    download_config=_lowerCAmelCase			, download_mode=_lowerCAmelCase			, verification_mode=_lowerCAmelCase			, base_path=_lowerCAmelCase			, num_proc=self.num_proc			, )
																__SCREAMING_SNAKE_CASE:				List[str]           =   self.builder.as_dataset(
																    split='''train'''			, verification_mode=_lowerCAmelCase			, in_memory=self.keep_in_memory					)
										return dataset



