




import os
import posixpath
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union

import numpy as np
import pyarrow as pa

import datasets
from datasets.arrow_writer import ArrowWriter, ParquetWriter
from datasets.config import MAX_SHARD_SIZE
from datasets.filesystems import (
    is_remote_filesystem,
    rename,
)
from datasets.iterable_dataset import _BaseExamplesIterable
from datasets.utils.py_utils import convert_file_size_to_int


__A								=       datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
					import pyspark







@dataclass
class        __lowerCAmelCase					(			datasets.BuilderConfig					):

			"""simple docstring"""







			snake_case_  						=			None







def      lowerCamelCase_						(			UpperCamelCase__							: "pyspark.sql.DataFrame"					,	UpperCamelCase__							: List[int]					,	)	->			Optional[int]:





				"""simple docstring"""


				import pyspark

				def generate_fn():
								__lowerCamelCase									=       df.select('*'					,	pyspark.sql.functions.spark_partition_id().alias('part_id' ) )
								for partition_id in partition_order:
												__lowerCamelCase									=       df_with_partition_id.select('*' ).where(F"""part_id = {partition_id}""" ).drop('part_id' )
												__lowerCamelCase									=       partition_df.collect()
												__lowerCamelCase									=       0
												for row in rows:
																yield F"""{partition_id}_{row_id}""", row.asDict()
																row_id += 1

				return generate_fn







class        __lowerCAmelCase					(			_BaseExamplesIterable					):

			"""simple docstring"""







			def __init__( self     ,		lowerCamelCase__     ,		lowerCamelCase__=None     ,		)       -> Tuple:



							'''simple docstring'''
							__lowerCamelCase									=       df
							__lowerCamelCase									=       partition_order or range(self.df.rdd.getNumPartitions()				)
							__lowerCamelCase									=       _generate_iterable_examples(self.df     ,		self.partition_order				)


			def __iter__( self				)       -> Union[str, Any]:



							'''simple docstring'''
							yield from self.generate_examples_fn()


			def  lowercase_       ( self     ,		lowerCamelCase__				)       -> "SparkExamplesIterable":



							'''simple docstring'''
							__lowerCamelCase									=       list(range(self.df.rdd.getNumPartitions()				)				)
							generator.shuffle(lowerCamelCase__				)
							return SparkExamplesIterable(self.df     ,		partition_order=lowerCamelCase__				)


			def  lowercase_       ( self     ,		lowerCamelCase__     ,		lowerCamelCase__				)       -> "SparkExamplesIterable":



							'''simple docstring'''
							__lowerCamelCase									=       self.split_shard_indices_by_worker(lowerCamelCase__     ,		lowerCamelCase__				)
							return SparkExamplesIterable(self.df     ,		partition_order=lowerCamelCase__				)






			@property
			def  lowercase_       ( self				)       -> int:



							'''simple docstring'''
							return len(self.partition_order				)







class        __lowerCAmelCase					(			datasets.DatasetBuilder					):

			"""simple docstring"""







			snake_case_  						=			SparkConfig


			def __init__( self     ,		lowerCamelCase__     ,		lowerCamelCase__ = None     ,		lowerCamelCase__ = None     ,		**lowerCamelCase__     ,		)       -> List[Any]:



							'''simple docstring'''
							import pyspark

							__lowerCamelCase									=       pyspark.sql.SparkSession.builder.getOrCreate()
							__lowerCamelCase									=       df
							__lowerCamelCase									=       working_dir

							super().__init__(
							    cache_dir=lowerCamelCase__     ,		config_name=str(self.df.semanticHash()				)     ,		**lowerCamelCase__     ,		)


			def  lowercase_       ( self				)       -> Union[str, Any]:



							'''simple docstring'''
							# Returns the path of the created file.
							def create_cache_and_write_probe(lowerCamelCase__				):
											# makedirs with exist_ok will recursively create the directory. It will not throw an error if directories
											# already exist.
											os.makedirs(self._cache_dir     ,		exist_ok=lowerCamelCase__				)
											__lowerCamelCase									=       os.path.join(self._cache_dir     ,		'fs_test' + uuid.uuida().hex				)
											# Opening the file in append mode will create a new file unless it already exists, in which case it will not
											# change the file contents.
											open(lowerCamelCase__     ,		'a'				)
											return [probe_file]

							if self._spark.conf.get('spark.master'     ,		''				).startswith('local'				):
											return

							# If the cluster is multi-node, make sure that the user provided a cache_dir and that it is on an NFS
							# accessible to the driver.
							# TODO: Stream batches to the driver using ArrowCollectSerializer instead of throwing an error.
							if self._cache_dir:
											__lowerCamelCase									=       (
											    self._spark.sparkContext.parallelize(range(1				)     ,		1				).mapPartitions(lowerCamelCase__				).collect()
											)
											if os.path.isfile(probe[0]				):
															return

							raise ValueError(
							    'When using Dataset.from_spark on a multi-node cluster, the driver and all workers should be able to access cache_dir'				)


			def  lowercase_       ( self				)       -> Tuple:



							'''simple docstring'''
							return datasets.DatasetInfo(features=self.config.features				)


			def  lowercase_       ( self     ,		lowerCamelCase__				)       -> str:



							'''simple docstring'''
							return [datasets.SplitGenerator(name=datasets.Split.TRAIN				)]


			def  lowercase_       ( self     ,		lowerCamelCase__				)       -> Optional[Any]:



							'''simple docstring'''
							import pyspark

							def get_arrow_batch_size(lowerCamelCase__				):
											for batch in it:
															yield pa.RecordBatch.from_pydict({'batch_bytes': [batch.nbytes]}				)

							__lowerCamelCase									=       self.df.count()
							__lowerCamelCase									=       df_num_rows if df_num_rows <= 100 else 100
							# Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
							__lowerCamelCase									=       (
							    self.df.limit(lowerCamelCase__				)
							    .repartition(1				)
							    .mapInArrow(lowerCamelCase__     ,		'batch_bytes: long'				)
							    .agg(pyspark.sql.functions.sum('batch_bytes'				).alias('sample_bytes'				)				)
							    .collect()[0]
							    .sample_bytes
							    / sample_num_rows
							)
							__lowerCamelCase									=       approx_bytes_per_row * df_num_rows
							if approx_total_size > max_shard_size:
											# Make sure there is at least one row per partition.
											__lowerCamelCase									=       min(lowerCamelCase__     ,		int(approx_total_size / max_shard_size				)				)
											__lowerCamelCase									=       self.df.repartition(lowerCamelCase__				)


			def  lowercase_       ( self     ,		lowerCamelCase__     ,		lowerCamelCase__     ,		lowerCamelCase__     ,		)       -> Iterable[Tuple[int, bool, Union[int, tuple]]]:



							'''simple docstring'''
							import pyspark

							__lowerCamelCase									=       ParquetWriter if file_format == 'parquet' else ArrowWriter
							__lowerCamelCase									=       os.path.join(self._working_dir     ,		os.path.basename(lowerCamelCase__				)				) if self._working_dir else fpath
							__lowerCamelCase									=       file_format == 'parquet'

							# Define these so that we don't reference self in write_arrow, which will result in a pickling error due to
							# pickling the SparkContext.
							__lowerCamelCase									=       self.config.features
							__lowerCamelCase									=       self._writer_batch_size
							__lowerCamelCase									=       self._fs.storage_options

							def write_arrow(lowerCamelCase__				):
											# Within the same SparkContext, no two task attempts will share the same attempt ID.
											__lowerCamelCase									=       pyspark.TaskContext().taskAttemptId()
											__lowerCamelCase									=       next(lowerCamelCase__     ,		lowerCamelCase__				)
											if first_batch is None:
															# Some partitions might not receive any data.
															return pa.RecordBatch.from_arrays(
															    [[task_id], [0], [0]]     ,		names=['task_id', 'num_examples', 'num_bytes']     ,		)
											__lowerCamelCase									=       0
											__lowerCamelCase									=       writer_class(
											    features=lowerCamelCase__     ,		path=working_fpath.replace('SSSSS'     ,		f"""{shard_id:05d}"""				).replace('TTTTT'     ,		f"""{task_id:05d}"""				)     ,		writer_batch_size=lowerCamelCase__     ,		storage_options=lowerCamelCase__     ,		embed_local_files=lowerCamelCase__     ,		)
											__lowerCamelCase									=       pa.Table.from_batches([first_batch]				)
											writer.write_table(lowerCamelCase__				)
											for batch in it:
															if max_shard_size is not None and writer._num_bytes >= max_shard_size:
																			__lowerCamelCase			,		__lowerCamelCase									=       writer.finalize()
																			writer.close()
																			yield pa.RecordBatch.from_arrays(
																			    [[task_id], [num_examples], [num_bytes]]     ,		names=['task_id', 'num_examples', 'num_bytes']     ,		)
																			shard_id += 1
																			__lowerCamelCase									=       writer_class(
																			    features=writer._features     ,		path=working_fpath.replace('SSSSS'     ,		f"""{shard_id:05d}"""				).replace('TTTTT'     ,		f"""{task_id:05d}"""				)     ,		writer_batch_size=lowerCamelCase__     ,		storage_options=lowerCamelCase__     ,		embed_local_files=lowerCamelCase__     ,		)
															__lowerCamelCase									=       pa.Table.from_batches([batch]				)
															writer.write_table(lowerCamelCase__				)

											if writer._num_bytes > 0:
															__lowerCamelCase			,		__lowerCamelCase									=       writer.finalize()
															writer.close()
															yield pa.RecordBatch.from_arrays(
															    [[task_id], [num_examples], [num_bytes]]     ,		names=['task_id', 'num_examples', 'num_bytes']     ,		)

											if working_fpath != fpath:
															for file in os.listdir(os.path.dirname(lowerCamelCase__				)				):
																			__lowerCamelCase									=       os.path.join(os.path.dirname(lowerCamelCase__				)     ,		os.path.basename(lowerCamelCase__				)				)
																			shutil.move(lowerCamelCase__     ,		lowerCamelCase__				)

							__lowerCamelCase									=       (
							    self.df.mapInArrow(lowerCamelCase__     ,		'task_id: long, num_examples: long, num_bytes: long'				)
							    .groupBy('task_id'				)
							    .agg(
							        pyspark.sql.functions.sum('num_examples'				).alias('total_num_examples'				)     ,		pyspark.sql.functions.sum('num_bytes'				).alias('total_num_bytes'				)     ,		pyspark.sql.functions.count('num_bytes'				).alias('num_shards'				)     ,		pyspark.sql.functions.collect_list('num_examples'				).alias('shard_lengths'				)     ,		)
							    .collect()
							)
							for row in stats:
											yield row.task_id, (row.total_num_examples, row.total_num_bytes, row.num_shards, row.shard_lengths)


			def  lowercase_       ( self     ,		lowerCamelCase__     ,		lowerCamelCase__ = "arrow"     ,		lowerCamelCase__ = None     ,		lowerCamelCase__ = None     ,		**lowerCamelCase__     ,		)       -> Union[str, Any]:



							'''simple docstring'''
							self._validate_cache_dir()

							__lowerCamelCase									=       convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE				)
							self._repartition_df_if_needed(lowerCamelCase__				)
							__lowerCamelCase									=       not is_remote_filesystem(self._fs				)
							__lowerCamelCase									=       os.path.join if is_local else posixpath.join

							__lowerCamelCase									=       '-TTTTT-SSSSS-of-NNNNN'
							__lowerCamelCase									=       f"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
							__lowerCamelCase									=       path_join(self._output_dir     ,		lowerCamelCase__				)

							__lowerCamelCase									=       0
							__lowerCamelCase									=       0
							__lowerCamelCase									=       0
							__lowerCamelCase									=       []
							__lowerCamelCase									=       []

							for task_id, content in self._prepare_split_single(lowerCamelCase__     ,		lowerCamelCase__     ,		lowerCamelCase__				):
											(
											    (
											    __lowerCamelCase
											)			,		(
											    __lowerCamelCase
											)			,		(
											    __lowerCamelCase
											)			,		(
											    __lowerCamelCase
											)			,		
											)									=       content
											if num_bytes > 0:
															total_num_examples += num_examples
															total_num_bytes += num_bytes
															total_shards += num_shards
															task_id_and_num_shards.append((task_id, num_shards)				)
															all_shard_lengths.extend(lowerCamelCase__				)

							__lowerCamelCase									=       total_num_examples
							__lowerCamelCase									=       total_num_bytes

							# should rename everything at the end
							logger.debug(f"""Renaming {total_shards} shards."""				)
							if total_shards > 1:
											__lowerCamelCase									=       all_shard_lengths

											# Define fs outside of _rename_shard so that we don't reference self in the function, which will result in a
											# pickling error due to pickling the SparkContext.
											__lowerCamelCase									=       self._fs

											# use the -SSSSS-of-NNNNN pattern
											def _rename_shard(
											    lowerCamelCase__     ,		lowerCamelCase__     ,		lowerCamelCase__     ,		):
															rename(
															    lowerCamelCase__     ,		fpath.replace('SSSSS'     ,		f"""{shard_id:05d}"""				).replace('TTTTT'     ,		f"""{task_id:05d}"""				)     ,		fpath.replace('TTTTT-SSSSS'     ,		f"""{global_shard_id:05d}"""				).replace('NNNNN'     ,		f"""{total_shards:05d}"""				)     ,		)

											__lowerCamelCase									=       []
											__lowerCamelCase									=       0
											for i in range(len(lowerCamelCase__				)				):
															__lowerCamelCase			,		__lowerCamelCase									=       task_id_and_num_shards[i]
															for shard_id in range(lowerCamelCase__				):
																			args.append([task_id, shard_id, global_shard_id]				)
																			global_shard_id += 1
											self._spark.sparkContext.parallelize(lowerCamelCase__     ,		len(lowerCamelCase__				)				).map(lambda lowerCamelCase__				: _rename_shard(*lowerCamelCase__				)				).collect()
							else:
											# don't use any pattern
											__lowerCamelCase									=       0
											__lowerCamelCase									=       task_id_and_num_shards[0][0]
											self._rename(
											    fpath.replace('SSSSS'     ,		f"""{shard_id:05d}"""				).replace('TTTTT'     ,		f"""{task_id:05d}"""				)     ,		fpath.replace(lowerCamelCase__     ,		''				)     ,		)






			def  lowercase_       ( self     ,		lowerCamelCase__     ,		)       -> SparkExamplesIterable:



							'''simple docstring'''
							return SparkExamplesIterable(self.df				)








import argparse
import os
import pickle
import sys

import torch

from transformers import TransfoXLConfig, TransfoXLLMHeadModel, load_tf_weights_in_transfo_xl
from transformers.models.transfo_xl import tokenization_transfo_xl as data_utils
from transformers.models.transfo_xl.tokenization_transfo_xl import CORPUS_NAME, VOCAB_FILES_NAMES
from transformers.utils import CONFIG_NAME, WEIGHTS_NAME, logging


logging.set_verbosity_info()

# We do this to be able to load python 2 datasets pickles
# See e.g. https://stackoverflow.com/questions/2121874/python-pickling-after-changing-a-modules-directory/2121918#2121918
__A								=       data_utils.TransfoXLTokenizer
__A								=       data_utils.TransfoXLCorpus
__A								=       data_utils
__A								=       data_utils







def      lowerCamelCase_						(			UpperCamelCase__							: int					,	UpperCamelCase__							: str					,	UpperCamelCase__							: Optional[Any]					,	UpperCamelCase__							: Tuple )	->			Tuple:





				"""simple docstring"""


				if transfo_xl_dataset_file:
								# Convert a pre-processed corpus (see original TensorFlow repo)
								with open(UpperCamelCase__					,	'rb' ) as fp:
												__lowerCamelCase									=       pickle.load(UpperCamelCase__					,	encoding='latin1' )
								# Save vocabulary and dataset cache as Dictionaries (should be better than pickles for the long-term)
								__lowerCamelCase									=       pytorch_dump_folder_path + '/' + VOCAB_FILES_NAMES['pretrained_vocab_file']
								print(F"""Save vocabulary to {pytorch_vocab_dump_path}""" )
								__lowerCamelCase									=       corpus.vocab.__dict__
								torch.save(UpperCamelCase__					,	UpperCamelCase__ )

								__lowerCamelCase									=       corpus.__dict__
								corpus_dict_no_vocab.pop('vocab'					,	UpperCamelCase__ )
								__lowerCamelCase									=       pytorch_dump_folder_path + '/' + CORPUS_NAME
								print(F"""Save dataset to {pytorch_dataset_dump_path}""" )
								torch.save(UpperCamelCase__					,	UpperCamelCase__ )

				if tf_checkpoint_path:
								# Convert a pre-trained TensorFlow model
								__lowerCamelCase									=       os.path.abspath(UpperCamelCase__ )
								__lowerCamelCase									=       os.path.abspath(UpperCamelCase__ )

								print(F"""Converting Transformer XL checkpoint from {tf_path} with config at {config_path}.""" )
								# Initialise PyTorch model
								if transfo_xl_config_file == "":
												__lowerCamelCase									=       TransfoXLConfig()
								else:
												__lowerCamelCase									=       TransfoXLConfig.from_json_file(UpperCamelCase__ )
								print(F"""Building PyTorch model from configuration: {config}""" )
								__lowerCamelCase									=       TransfoXLLMHeadModel(UpperCamelCase__ )

								__lowerCamelCase									=       load_tf_weights_in_transfo_xl(UpperCamelCase__					,	UpperCamelCase__					,	UpperCamelCase__ )
								# Save pytorch-model
								__lowerCamelCase									=       os.path.join(UpperCamelCase__					,	UpperCamelCase__ )
								__lowerCamelCase									=       os.path.join(UpperCamelCase__					,	UpperCamelCase__ )
								print(F"""Save PyTorch model to {os.path.abspath(UpperCamelCase__ )}""" )
								torch.save(model.state_dict()					,	UpperCamelCase__ )
								print(F"""Save configuration file to {os.path.abspath(UpperCamelCase__ )}""" )
								with open(UpperCamelCase__					,	'w'					,	encoding='utf-8' ) as f:
												f.write(config.to_json_string() )


if __name__ == "__main__":
					__A								=       argparse.ArgumentParser()
					parser.add_argument(
					    "--pytorch_dump_folder_path",
					    default=None,
					    type=str,
					    required=True,
					    help="Path to the folder to store the PyTorch model or dataset/vocab.",
					)
					parser.add_argument(
					    "--tf_checkpoint_path",
					    default="",
					    type=str,
					    help="An optional path to a TensorFlow checkpoint path to be converted.",
					)
					parser.add_argument(
					    "--transfo_xl_config_file",
					    default="",
					    type=str,
					    help=(
					        "An optional config json file corresponding to the pre-trained BERT model. \n"
					        "This specifies the model architecture."
					    ),
					)
					parser.add_argument(
					    "--transfo_xl_dataset_file",
					    default="",
					    type=str,
					    help="An optional dataset file to be converted in a vocabulary.",
					)
					__A								=       parser.parse_args()
					convert_transfo_xl_checkpoint_to_pytorch(
					    args.tf_checkpoint_path,
					    args.transfo_xl_config_file,
					    args.pytorch_dump_folder_path,
					    args.transfo_xl_dataset_file,
					)


