
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


lowerCamelCase__    :		List[str]          =   datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
     import pyspark



@dataclass
class       _snake_case   (	datasets.BuilderConfig							):
 __lowerCAmelCase  :     Tuple        =			None

def 		UpperCamelCase	(      lowercase_ , lowercase_ , )		->	Optional[int]:


  '''simple docstring'''






  import pyspark

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

  return generate_fn



class       _snake_case   (	_BaseExamplesIterable							):


 def __init__(					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_=None					,  ):


   '''simple docstring'''




   lowercase__  :   int    		=      df
   lowercase__  :   str    		=      partition_order or range(self.df.rdd.getNumPartitions())
   lowercase__  :   Tuple    		=      _generate_iterable_examples(self.df					,  self.partition_order)


 def __iter__(					self):


   '''simple docstring'''




   yield from self.generate_examples_fn()


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   lowercase__  :   Union[str, Any]    		=      list(range(self.df.rdd.getNumPartitions()))
   generator.shuffle(lowerCamelCase_)
   return SparkExamplesIterable(self.df					,  partition_order=lowerCamelCase_)


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   lowercase__  :   int    		=      self.split_shard_indices_by_worker(lowerCamelCase_					,  lowerCamelCase_)
   return SparkExamplesIterable(self.df					,  partition_order=lowerCamelCase_)



 @property
 def      lowercase__     (					self):


   '''simple docstring'''




   return len(self.partition_order)



class       _snake_case   (	datasets.DatasetBuilder							):
 __lowerCAmelCase  :     Optional[int]        =			SparkConfig


 def __init__(					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = None					,  SCREAMING_SNAKE_CASE_ = None					,  **SCREAMING_SNAKE_CASE_					,  ):


   '''simple docstring'''




   import pyspark

   lowercase__  :   Any    		=      pyspark.sql.SparkSession.builder.getOrCreate()
   lowercase__  :   List[str]    		=      df
   lowercase__  :   Optional[Any]    		=      working_dir

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


 def      lowercase__     (					self):


   '''simple docstring'''




   def create_cache_and_write_probe(SCREAMING_SNAKE_CASE_):
     # 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_)
     lowercase__  :   Union[str, Any]    		=      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:
     lowercase__  :   Optional[Any]    		=      (
         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):


   '''simple docstring'''




   return datasets.DatasetInfo(features=self.config.features)


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




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


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   import pyspark

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

   lowercase__  :   Union[str, Any]    		=      self.df.count()
   lowercase__  :   int    		=      df_num_rows if df_num_rows <= 1_00 else 1_00
   # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
   lowercase__  :   str    		=      (
       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
   )
   lowercase__  :   Dict    		=      approx_bytes_per_row * df_num_rows
   if approx_total_size > max_shard_size:
     # Make sure there is at least one row per partition.
     lowercase__  :   Union[str, Any]    		=      min(lowerCamelCase_					,  int(approx_total_size / max_shard_size))
     lowercase__  :   Union[str, Any]    		=      self.df.repartition(lowerCamelCase_)


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  ):


   '''simple docstring'''




   import pyspark

   lowercase__  :   List[str]    		=      ParquetWriter if file_format == """parquet""" else ArrowWriter
   lowercase__  :   Tuple    		=      os.path.join(self._working_dir					,  os.path.basename(lowerCamelCase_)) if self._working_dir else fpath
   lowercase__  :   List[Any]    		=      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.
   lowercase__  :   int    		=      self.config.features
   lowercase__  :   Dict    		=      self._writer_batch_size
   lowercase__  :   Tuple    		=      self._fs.storage_options

   def write_arrow(SCREAMING_SNAKE_CASE_):
     # Within the same SparkContext, no two task attempts will share the same attempt ID.
     lowercase__  :   Any    		=      pyspark.TaskContext().taskAttemptId()
     lowercase__  :   str    		=      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"""]					,  )
     lowercase__  :   Any    		=      0
     lowercase__  :   Any    		=      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_					,  )
     lowercase__  :   Tuple    		=      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:
         lowercase__  :   Tuple    		=      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
         lowercase__  :   List[str]    		=      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_					,  )
       lowercase__  :   Optional[int]    		=      pa.Table.from_batches([batch])
       writer.write_table(lowerCamelCase_)

     if writer._num_bytes > 0:
       lowercase__  :   str    		=      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_)):
         lowercase__  :   Optional[int]    		=      os.path.join(os.path.dirname(lowerCamelCase_)					,  os.path.basename(lowerCamelCase_))
         shutil.move(lowerCamelCase_					,  lowerCamelCase_)

   lowercase__  :   Dict    		=      (
       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					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = "arrow"					,  SCREAMING_SNAKE_CASE_ = None					,  SCREAMING_SNAKE_CASE_ = None					,  **SCREAMING_SNAKE_CASE_					,  ):


   '''simple docstring'''




   self._validate_cache_dir()

   lowercase__  :   Any    		=      convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE)
   self._repartition_df_if_needed(lowerCamelCase_)
   lowercase__  :   Dict    		=      not is_remote_filesystem(self._fs)
   lowercase__  :   Any    		=      os.path.join if is_local else posixpath.join

   lowercase__  :   Optional[int]    		=      """-TTTTT-SSSSS-of-NNNNN"""
   lowercase__  :   Optional[int]    		=      f'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
   lowercase__  :   List[str]    		=      path_join(self._output_dir					,  lowerCamelCase_)

   lowercase__  :   str    		=      0
   lowercase__  :   str    		=      0
   lowercase__  :   Any    		=      0
   lowercase__  :   Optional[int]    		=      []
   lowercase__  :   Optional[int]    		=      []

   for task_id, content in self._prepare_split_single(lowerCamelCase_					,  lowerCamelCase_					,  lowerCamelCase_):
     (
         lowercase__       
     )  :   Union[str, Any]    		=      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_)

   lowercase__  :   List[Any]    		=      total_num_examples
   lowercase__  :   Optional[Any]    		=      total_num_bytes

   # should rename everything at the end
   logger.debug(f'Renaming {total_shards} shards.')
   if total_shards > 1:
     lowercase__  :   List[str]    		=      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.
     lowercase__  :   List[str]    		=      self._fs

     # use the -SSSSS-of-NNNNN pattern
     def _rename_shard(
         SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  ):
       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}')					,  )

     lowercase__  :   Optional[Any]    		=      []
     lowercase__  :   Optional[int]    		=      0
     for i in range(len(lowerCamelCase_)):
       lowercase__  :   List[str]    		=      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 SCREAMING_SNAKE_CASE_: _rename_shard(*lowerCamelCase_)).collect()
   else:
     # don't use any pattern
     lowercase__  :   str    		=      0
     lowercase__  :   int    		=      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					,  SCREAMING_SNAKE_CASE_					,  ):


   '''simple docstring'''




   return SparkExamplesIterable(self.df)






import unittest
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory

from transformers import BertConfig, BertTokenizerFast, FeatureExtractionPipeline
from transformers.convert_graph_to_onnx import (
    convert,
    ensure_valid_input,
    generate_identified_filename,
    infer_shapes,
    quantize,
)
from transformers.testing_utils import require_tf, require_tokenizers, require_torch, slow



class       _snake_case   :


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   return None



class       _snake_case   :


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   return None



class       _snake_case   (	unittest.TestCase							):
 __lowerCAmelCase  :     Optional[Any]        =			[
     # (model_name, model_kwargs)
     ('bert-base-cased', {}),
     ('gpt2', {'use_cache': False}),  # We don't support exporting GPT2 past keys anymore
 ]


 @require_tf
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
     self._test_export(SCREAMING_SNAKE_CASE_					,  """tf"""					,  12					,  **SCREAMING_SNAKE_CASE_)


 @require_torch
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
     self._test_export(SCREAMING_SNAKE_CASE_					,  """pt"""					,  12					,  **SCREAMING_SNAKE_CASE_)


 @require_torch
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   from transformers import BertModel

   lowercase__  :   Union[str, Any]    		=      ["""[UNK]""", """[SEP]""", """[CLS]""", """[PAD]""", """[MASK]""", """some""", """other""", """words"""]
   with NamedTemporaryFile(mode="""w+t""") as vocab_file:
     vocab_file.write("""\n""".join(SCREAMING_SNAKE_CASE_))
     vocab_file.flush()
     lowercase__  :   Optional[int]    		=      BertTokenizerFast(vocab_file.name)

   with TemporaryDirectory() as bert_save_dir:
     lowercase__  :   Union[str, Any]    		=      BertModel(BertConfig(vocab_size=len(SCREAMING_SNAKE_CASE_)))
     model.save_pretrained(SCREAMING_SNAKE_CASE_)
     self._test_export(SCREAMING_SNAKE_CASE_					,  """pt"""					,  12					,  SCREAMING_SNAKE_CASE_)


 @require_tf
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
     lowercase__  :   Any    		=      self._test_export(SCREAMING_SNAKE_CASE_					,  """tf"""					,  12					,  **SCREAMING_SNAKE_CASE_)
     lowercase__  :   Optional[Any]    		=      quantize(Path(SCREAMING_SNAKE_CASE_))

     # Ensure the actual quantized model is not bigger than the original one
     if quantized_path.stat().st_size >= Path(SCREAMING_SNAKE_CASE_).stat().st_size:
       self.fail("""Quantized model is bigger than initial ONNX model""")


 @require_torch
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
     lowercase__  :   Tuple    		=      self._test_export(SCREAMING_SNAKE_CASE_					,  """pt"""					,  12					,  **SCREAMING_SNAKE_CASE_)
     lowercase__  :   Dict    		=      quantize(SCREAMING_SNAKE_CASE_)

     # Ensure the actual quantized model is not bigger than the original one
     if quantized_path.stat().st_size >= Path(SCREAMING_SNAKE_CASE_).stat().st_size:
       self.fail("""Quantized model is bigger than initial ONNX model""")


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_=None					,  **SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   try:
     # Compute path
     with TemporaryDirectory() as tempdir:
       lowercase__  :   List[Any]    		=      Path(SCREAMING_SNAKE_CASE_).joinpath("""model.onnx""")

     # Remove folder if exists
     if path.parent.exists():
       path.parent.rmdir()

     # Export
     convert(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  **SCREAMING_SNAKE_CASE_)

     return path
   except Exception as e:
     self.fail(SCREAMING_SNAKE_CASE_)


 @require_torch
 @require_tokenizers
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   from transformers import BertModel

   lowercase__  :   Dict    		=      BertModel(BertConfig.from_pretrained("""lysandre/tiny-bert-random"""))
   lowercase__  :   Union[str, Any]    		=      BertTokenizerFast.from_pretrained("""lysandre/tiny-bert-random""")
   self._test_infer_dynamic_axis(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  """pt""")


 @require_tf
 @require_tokenizers
 @slow
 def      lowercase__     (					self):


   '''simple docstring'''




   from transformers import TFBertModel

   lowercase__  :   Union[str, Any]    		=      TFBertModel(BertConfig.from_pretrained("""lysandre/tiny-bert-random"""))
   lowercase__  :   Optional[int]    		=      BertTokenizerFast.from_pretrained("""lysandre/tiny-bert-random""")
   self._test_infer_dynamic_axis(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  """tf""")


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   lowercase__  :   Dict    		=      FeatureExtractionPipeline(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_)

   lowercase__  :   Union[str, Any]    		=      ["""input_ids""", """token_type_ids""", """attention_mask""", """output_0""", """output_1"""]
   lowercase__		,    lowercase__		,    lowercase__		,    lowercase__  :   List[str]    		=      infer_shapes(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_)

   # Assert all variables are present
   self.assertEqual(len(SCREAMING_SNAKE_CASE_)					,  len(SCREAMING_SNAKE_CASE_))
   self.assertTrue(all(var_name in shapes for var_name in variable_names))
   self.assertSequenceEqual(variable_names[:3]					,  SCREAMING_SNAKE_CASE_)
   self.assertSequenceEqual(variable_names[3:]					,  SCREAMING_SNAKE_CASE_)

   # Assert inputs are {0: batch, 1: sequence}
   for var_name in ["input_ids", "token_type_ids", "attention_mask"]:
     self.assertDictEqual(shapes[var_name]					,  {0: """batch""", 1: """sequence"""})

   # Assert outputs are {0: batch, 1: sequence} and {0: batch}
   self.assertDictEqual(shapes["""output_0"""]					,  {0: """batch""", 1: """sequence"""})
   self.assertDictEqual(shapes["""output_1"""]					,  {0: """batch"""})


 def      lowercase__     (					self):


   '''simple docstring'''




   lowercase__  :   Any    		=      ["""input_ids""", """attention_mask""", """token_type_ids"""]
   lowercase__  :   Tuple    		=      {"""input_ids""": [1, 2, 3, 4], """attention_mask""": [0, 0, 0, 0], """token_type_ids""": [1, 1, 1, 1]}
   lowercase__		,    lowercase__  :   Optional[Any]    		=      ensure_valid_input(FuncContiguousArgs()					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_)

   # Should have exactly the same number of args (all are valid)
   self.assertEqual(len(SCREAMING_SNAKE_CASE_)					,  3)

   # Should have exactly the same input names
   self.assertEqual(set(SCREAMING_SNAKE_CASE_)					,  set(SCREAMING_SNAKE_CASE_))

   # Parameter should be reordered according to their respective place in the function:
   # (input_ids, token_type_ids, attention_mask)
   self.assertEqual(SCREAMING_SNAKE_CASE_					,  (tokens["""input_ids"""], tokens["""token_type_ids"""], tokens["""attention_mask"""]))

   # Generated args are interleaved with another args (for instance parameter "past" in GPT2)
   lowercase__		,    lowercase__  :   Any    		=      ensure_valid_input(FuncNonContiguousArgs()					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_)

   # Should have exactly the one arg (all before the one not provided "some_other_args")
   self.assertEqual(len(SCREAMING_SNAKE_CASE_)					,  1)
   self.assertEqual(len(SCREAMING_SNAKE_CASE_)					,  1)

   # Should have only "input_ids"
   self.assertEqual(inputs_args[0]					,  tokens["""input_ids"""])
   self.assertEqual(ordered_input_names[0]					,  """input_ids""")



 def      lowercase__     (					self):


   '''simple docstring'''




   lowercase__  :   List[str]    		=      generate_identified_filename(Path("""/home/something/my_fake_model.onnx""")					,  """-test""")
   self.assertEqual("""/home/something/my_fake_model-test.onnx"""					,  generated.as_posix())






