



'''simple docstring'''





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


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

if TYPE_CHECKING:
  import pyspark






@dataclass
class __A    (						datasets.BuilderConfig		):



 '''simple docstring'''







 __lowerCamelCase					: Union[str, Any]    = None





def lowerCAmelCase						(__A						,			__A						,			):
     import pyspark

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

     return generate_fn






class __A    (						_BaseExamplesIterable		):



 '''simple docstring'''







 def __init__(self				,     A				,     A=None				,     )					->  Dict:


      """simple docstring"""



      _a											=						df
      _a											=						partition_order or range(self.df.rdd.getNumPartitions() )
      _a											=						_generate_iterable_examples(self.df				,     self.partition_order )



 def __iter__(self )					->  Optional[Any]:


      """simple docstring"""



      yield from self.generate_examples_fn()



 def       a__						(self				,     A )					->  int:


      """simple docstring"""



      _a											=						list(range(self.df.rdd.getNumPartitions() ) )
      generator.shuffle(__lowerCAmelCase )
      return SparkExamplesIterable(self.df				,     partition_order=__lowerCAmelCase )



 def       a__						(self				,     A				,     A )					->  Tuple:


      """simple docstring"""



      _a											=						self.split_shard_indices_by_worker(__lowerCAmelCase				,     __lowerCAmelCase )
      return SparkExamplesIterable(self.df				,     partition_order=__lowerCAmelCase )



 @property
 def       a__						(self )					->  Union[str, Any]:


      """simple docstring"""



      return len(self.partition_order )






class __A    (						datasets.DatasetBuilder		):



 '''simple docstring'''







 __lowerCamelCase					: List[Any]    = SparkConfig



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


      """simple docstring"""



      import pyspark

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

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



 def       a__						(self )					->  Optional[Any]:


      """simple docstring"""



      def create_cache_and_write_probe(A ):
           # 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 )
           _a											=						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:
           _a											=						(
               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       a__						(self )					->  int:


      """simple docstring"""



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



 def       a__						(self				,     A )					->  int:


      """simple docstring"""



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



 def       a__						(self				,     A )					->  int:


      """simple docstring"""



      import pyspark

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

      _a											=						self.df.count()
      _a											=						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.
      _a											=						(
          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
      )
      _a											=						approx_bytes_per_row * df_num_rows
      if approx_total_size > max_shard_size:
           # Make sure there is at least one row per partition.
           _a											=						min(__lowerCAmelCase				,     int(approx_total_size / max_shard_size ) )
           _a											=						self.df.repartition(__lowerCAmelCase )



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


      """simple docstring"""



      import pyspark

      _a											=						ParquetWriter if file_format == '''parquet''' else ArrowWriter
      _a											=						os.path.join(self._working_dir				,     os.path.basename(__lowerCAmelCase ) ) if self._working_dir else fpath
      _a											=						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.
      _a											=						self.config.features
      _a											=						self._writer_batch_size
      _a											=						self._fs.storage_options

      def write_arrow(A ):
           # Within the same SparkContext, no two task attempts will share the same attempt ID.
           _a											=						pyspark.TaskContext().taskAttemptId()
           _a											=						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''']				,     )
           _a											=						0
           _a											=						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				,     )
           _a											=						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:
                     _a						,					_a											=						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
                     _a											=						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				,     )
                _a											=						pa.Table.from_batches([batch] )
                writer.write_table(__lowerCAmelCase )

           if writer._num_bytes > 0:
                _a						,					_a											=						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 ) ):
                     _a											=						os.path.join(os.path.dirname(__lowerCAmelCase )				,     os.path.basename(__lowerCAmelCase ) )
                     shutil.move(__lowerCAmelCase				,     __lowerCAmelCase )

      _a											=						(
          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       a__						(self				,     A				,     A = "arrow"				,     A = None				,     A = None				,     **A				,     )					->  Optional[Any]:


      """simple docstring"""



      self._validate_cache_dir()

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

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

      _a											=						0
      _a											=						0
      _a											=						0
      _a											=						[]
      _a											=						[]

      for task_id, content in self._prepare_split_single(__lowerCAmelCase				,     __lowerCAmelCase				,     __lowerCAmelCase ):
           (
               (
               _a			
           )						,					(
               _a			
           )						,					(
               _a			
           )						,					(
               _a			
           )						,								
           )											=						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 )

      _a											=						total_num_examples
      _a											=						total_num_bytes

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

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

           _a											=						[]
           _a											=						0
           for i in range(len(__lowerCAmelCase ) ):
                _a						,					_a											=						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 A : _rename_shard(*__lowerCAmelCase ) ).collect()
      else:
           # don't use any pattern
           _a											=						0
           _a											=						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       a__						(self				,     A				,     )					->  Any:


      """simple docstring"""



      return SparkExamplesIterable(self.df )








'''simple docstring'''





from __future__ import annotations

import unittest

from transformers import EsmConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
  import numpy
  import tensorflow as tf

  from transformers.models.esm.modeling_tf_esm import (
      TF_ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFEsmForMaskedLM,
      TFEsmForSequenceClassification,
      TFEsmForTokenClassification,
      TFEsmModel,
  )






class __A    :



 '''simple docstring'''







 def __init__(self				,     A				,     )					->  Optional[int]:


      """simple docstring"""



      _a											=						parent
      _a											=						13
      _a											=						7
      _a											=						True
      _a											=						True
      _a											=						True
      _a											=						99
      _a											=						32
      _a											=						2
      _a											=						4
      _a											=						37
      _a											=						'''gelu'''
      _a											=						0.1
      _a											=						0.1
      _a											=						512
      _a											=						16
      _a											=						2
      _a											=						0.02
      _a											=						3
      _a											=						4
      _a											=						None



 def       a__						(self )					->  Optional[Any]:


      """simple docstring"""



      _a											=						ids_tensor([self.batch_size, self.seq_length]				,     self.vocab_size )

      _a											=						None
      if self.use_input_mask:
           _a											=						random_attention_mask([self.batch_size, self.seq_length] )

      _a											=						None
      _a											=						None
      _a											=						None
      if self.use_labels:
           _a											=						ids_tensor([self.batch_size]				,     self.type_sequence_label_size )
           _a											=						ids_tensor([self.batch_size, self.seq_length]				,     self.num_labels )
           _a											=						ids_tensor([self.batch_size]				,     self.num_choices )

      _a											=						EsmConfig(
          vocab_size=self.vocab_size				,     hidden_size=self.hidden_size				,     num_hidden_layers=self.num_hidden_layers				,     pad_token_id=1				,     num_attention_heads=self.num_attention_heads				,     intermediate_size=self.intermediate_size				,     hidden_act=self.hidden_act				,     hidden_dropout_prob=self.hidden_dropout_prob				,     attention_probs_dropout_prob=self.attention_probs_dropout_prob				,     max_position_embeddings=self.max_position_embeddings				,     type_vocab_size=self.type_vocab_size				,     initializer_range=self.initializer_range				,     )

      return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels



 def       a__						(self )					->  str:


      """simple docstring"""



      (
          (
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					
      )											=						self.prepare_config_and_inputs()

      _a											=						True
      _a											=						floats_tensor([self.batch_size, self.seq_length, self.hidden_size] )
      _a											=						ids_tensor([self.batch_size, self.seq_length]				,     vocab_size=2 )

      return (
          config,
          input_ids,
          input_mask,
          sequence_labels,
          token_labels,
          choice_labels,
          encoder_hidden_states,
          encoder_attention_mask,
      )



 def       a__						(self				,     A				,     A				,     A				,     A				,     A				,     A )					->  str:


      """simple docstring"""



      _a											=						TFEsmModel(config=A )
      _a											=						{'''input_ids''': input_ids, '''attention_mask''': input_mask}
      _a											=						model(A )

      _a											=						[input_ids, input_mask]
      _a											=						model(A )

      _a											=						model(A )

      self.parent.assertEqual(result.last_hidden_state.shape				,     (self.batch_size, self.seq_length, self.hidden_size) )



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


      """simple docstring"""



      _a											=						True

      _a											=						TFEsmModel(config=A )
      _a											=						{
          '''input_ids''': input_ids,
          '''attention_mask''': input_mask,
          '''encoder_hidden_states''': encoder_hidden_states,
          '''encoder_attention_mask''': encoder_attention_mask,
      }
      _a											=						model(A )

      _a											=						[input_ids, input_mask]
      _a											=						model(A				,     encoder_hidden_states=A )

      # Also check the case where encoder outputs are not passed
      _a											=						model(A				,     attention_mask=A )

      self.parent.assertEqual(result.last_hidden_state.shape				,     (self.batch_size, self.seq_length, self.hidden_size) )



 def       a__						(self				,     A				,     A				,     A				,     A				,     A				,     A )					->  Optional[Any]:


      """simple docstring"""



      _a											=						TFEsmForMaskedLM(config=A )
      _a											=						model([input_ids, input_mask] )
      self.parent.assertEqual(result.logits.shape				,     (self.batch_size, self.seq_length, self.vocab_size) )



 def       a__						(self				,     A				,     A				,     A				,     A				,     A				,     A )					->  Any:


      """simple docstring"""



      _a											=						self.num_labels
      _a											=						TFEsmForTokenClassification(config=A )
      _a											=						{'''input_ids''': input_ids, '''attention_mask''': input_mask}
      _a											=						model(A )
      self.parent.assertEqual(result.logits.shape				,     (self.batch_size, self.seq_length, self.num_labels) )



 def       a__						(self )					->  int:


      """simple docstring"""



      _a											=						self.prepare_config_and_inputs()
      (
          (
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					(
          _a
      )						,					
      )											=						config_and_inputs
      _a											=						{'''input_ids''': input_ids, '''attention_mask''': input_mask}
      return config, inputs_dict






@require_tf
class __A    (						A     ,     A     ,     unittest.TestCase		):



 '''simple docstring'''







 __lowerCamelCase					: str    = (
     (
         TFEsmModel,
         TFEsmForMaskedLM,
         TFEsmForSequenceClassification,
         TFEsmForTokenClassification,
     )
     if is_tf_available()
     else ()
 )
 __lowerCamelCase					: Union[str, Any]    = (
     {
         'feature-extraction': TFEsmModel,
         'fill-mask': TFEsmForMaskedLM,
         'text-classification': TFEsmForSequenceClassification,
         'token-classification': TFEsmForTokenClassification,
         'zero-shot': TFEsmForSequenceClassification,
     }
     if is_tf_available()
     else {}
 )
 __lowerCamelCase					: Dict    = False
 __lowerCamelCase					: Tuple    = False



 def       a__						(self )					->  Optional[Any]:


      """simple docstring"""



      _a											=						TFEsmModelTester(self )
      _a											=						ConfigTester(self				,     config_class=A				,     hidden_size=37 )



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


      """simple docstring"""



      self.config_tester.run_common_tests()



 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*A )



 def       a__						(self )					->  Dict:


      """simple docstring"""



      _a											=						self.model_tester.prepare_config_and_inputs_for_decoder()
      self.model_tester.create_and_check_model_as_decoder(*A )



 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_masked_lm(*A )



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


      """simple docstring"""



      _a											=						self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_token_classification(*A )



 @slow
 def       a__						(self )					->  Tuple:


      """simple docstring"""



      for model_name in TF_ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           _a											=						TFEsmModel.from_pretrained(A )
           self.assertIsNotNone(A )



 @unittest.skip('''Protein models do not support embedding resizing.''' )
 def       a__						(self )					->  str:


      """simple docstring"""



      pass



 @unittest.skip('''Protein models do not support embedding resizing.''' )
 def       a__						(self )					->  Tuple:


      """simple docstring"""



      pass



 def       a__						(self )					->  Dict:


      """simple docstring"""



      _a						,					_a											=						self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
           _a											=						model_class(A )
           assert isinstance(model.get_input_embeddings()				,     tf.keras.layers.Layer )
           if model_class is TFEsmForMaskedLM:
                # Output embedding test differs from the main test because they're a matrix, not a layer
                _a											=						model.get_bias()
                assert isinstance(A				,     A )
                for k, v in name.items():
                     assert isinstance(A				,     tf.Variable )
           else:
                _a											=						model.get_output_embeddings()
                assert x is None
                _a											=						model.get_bias()
                assert name is None






@require_tf
class __A    (						unittest.TestCase		):



 '''simple docstring'''







 @slow
 def       a__						(self )					->  List[str]:


      """simple docstring"""



      _a											=						TFEsmForMaskedLM.from_pretrained('''facebook/esm2_t6_8M_UR50D''' )

      _a											=						tf.constant([[0, 1, 2, 3, 4, 5]] )
      _a											=						model(A )[0]
      _a											=						[1, 6, 33]
      self.assertEqual(list(output.numpy().shape )				,     A )
      # compare the actual values for a slice.
      _a											=						tf.constant(
          [
              [
                  [8.921518, -10.589814, -6.4671307],
                  [-6.3967156, -13.911377, -1.1211915],
                  [-7.781247, -13.951557, -3.740592],
              ]
          ] )
      self.assertTrue(numpy.allclose(output[:, :3, :3].numpy()				,     expected_slice.numpy()				,     atol=1E-2 ) )



 @slow
 def       a__						(self )					->  Tuple:


      """simple docstring"""



      _a											=						TFEsmModel.from_pretrained('''facebook/esm2_t6_8M_UR50D''' )

      _a											=						tf.constant([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]] )
      _a											=						model(A )[0]
      # compare the actual values for a slice.
      _a											=						tf.constant(
          [
              [
                  [0.14443092, 0.54125327, 0.3247739],
                  [0.30340484, 0.00526676, 0.31077722],
                  [0.32278043, -0.24987096, 0.3414628],
              ]
          ] )
      self.assertTrue(numpy.allclose(output[:, :3, :3].numpy()				,     expected_slice.numpy()				,     atol=1E-4 ) )



