


'''simple docstring'''





import pyarrow.parquet as pq
import pytest

from datasets import Audio, Dataset, DatasetDict, Features, NamedSplit, Sequence, Value, config
from datasets.features.image import Image
from datasets.io.parquet import ParquetDatasetReader, ParquetDatasetWriter, get_writer_batch_size

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases


def        a__     (					UpperCamelCase_						:      Optional[Any],   UpperCamelCase_						:      List[Any]						):
   assert isinstance(UpperCamelCase_,   UpperCamelCase_						)
   assert dataset.num_rows == 4
   assert dataset.num_columns == 3
   assert dataset.column_names == ["col_1", "col_2", "col_3"]
   for feature, expected_dtype in expected_features.items():
      assert dataset.features[feature].dtype == expected_dtype


@pytest.mark.parametrize('''keep_in_memory''',   [False, True]						)
def        a__     (					UpperCamelCase_						:      Any,   UpperCamelCase_						:      Optional[int],   UpperCamelCase_						:      Any						):
   UpperCAmelCase__	:int      =      tmp_path / '''cache'''
   UpperCAmelCase__	:Optional[Any]      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      UpperCAmelCase__	:Dict      =      ParquetDatasetReader(UpperCamelCase_,   cache_dir=UpperCamelCase_,   keep_in_memory=UpperCamelCase_						).read()
   _check_parquet_dataset(UpperCamelCase_,   UpperCamelCase_						)


@pytest.mark.parametrize(
    '''features''',   [
        None,
        {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''},
        {'''col_1''': '''string''', '''col_2''': '''string''', '''col_3''': '''string'''},
        {'''col_1''': '''int32''', '''col_2''': '''int32''', '''col_3''': '''int32'''},
        {'''col_1''': '''float32''', '''col_2''': '''float32''', '''col_3''': '''float32'''},
    ],   )
def        a__     (					UpperCamelCase_						:      Optional[int],   UpperCamelCase_						:      List[Any],   UpperCamelCase_						:      str						):
   UpperCAmelCase__	:Optional[int]      =      tmp_path / '''cache'''
   UpperCAmelCase__	:Any      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   UpperCAmelCase__	:Optional[Any]      =      features.copy() if features else default_expected_features
   UpperCAmelCase__	:Dict      =      (
       Features({feature: Value(UpperCamelCase_						) for feature, dtype in features.items()}						) if features is not None else None
   )
   UpperCAmelCase__	:Any      =      ParquetDatasetReader(UpperCamelCase_,   features=UpperCamelCase_,   cache_dir=UpperCamelCase_						).read()
   _check_parquet_dataset(UpperCamelCase_,   UpperCamelCase_						)


@pytest.mark.parametrize('''split''',   [None, NamedSplit('''train'''						), '''train''', '''test''']						)
def        a__     (					UpperCamelCase_						:      Union[str, Any],   UpperCamelCase_						:      Any,   UpperCamelCase_						:      Union[str, Any]						):
   UpperCAmelCase__	:Any      =      tmp_path / '''cache'''
   UpperCAmelCase__	:str      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   UpperCAmelCase__	:Optional[int]      =      ParquetDatasetReader(UpperCamelCase_,   cache_dir=UpperCamelCase_,   split=UpperCamelCase_						).read()
   _check_parquet_dataset(UpperCamelCase_,   UpperCamelCase_						)
   assert dataset.split == split if split else "train"


@pytest.mark.parametrize('''path_type''',   [str, list]						)
def        a__     (					UpperCamelCase_						:      Any,   UpperCamelCase_						:      str,   UpperCamelCase_						:      Dict						):
   if issubclass(UpperCamelCase_,   UpperCamelCase_						):
      UpperCAmelCase__	:Any      =      parquet_path
   elif issubclass(UpperCamelCase_,   UpperCamelCase_						):
      UpperCAmelCase__	:Union[str, Any]      =      [parquet_path]
   UpperCAmelCase__	:int      =      tmp_path / '''cache'''
   UpperCAmelCase__	:Optional[int]      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   UpperCAmelCase__	:Any      =      ParquetDatasetReader(UpperCamelCase_,   cache_dir=UpperCamelCase_						).read()
   _check_parquet_dataset(UpperCamelCase_,   UpperCamelCase_						)


def        a__     (					UpperCamelCase_						:      List[str],   UpperCamelCase_						:      Optional[int],   UpperCamelCase_						:      List[Any]=("train",)						):
   assert isinstance(UpperCamelCase_,   UpperCamelCase_						)
   for split in splits:
      UpperCAmelCase__	:List[str]      =      dataset_dict[split]
      assert dataset.num_rows == 4
      assert dataset.num_columns == 3
      assert dataset.column_names == ["col_1", "col_2", "col_3"]
      for feature, expected_dtype in expected_features.items():
         assert dataset.features[feature].dtype == expected_dtype


@pytest.mark.parametrize('''keep_in_memory''',   [False, True]						)
def        a__     (					UpperCamelCase_						:      Union[str, Any],   UpperCamelCase_						:      int,   UpperCamelCase_						:      Tuple						):
   UpperCAmelCase__	:Tuple      =      tmp_path / '''cache'''
   UpperCAmelCase__	:Optional[int]      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      UpperCAmelCase__	:Any      =      ParquetDatasetReader(
          {'''train''': parquet_path},   cache_dir=UpperCamelCase_,   keep_in_memory=UpperCamelCase_						).read()
   _check_parquet_datasetdict(UpperCamelCase_,   UpperCamelCase_						)


@pytest.mark.parametrize(
    '''features''',   [
        None,
        {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''},
        {'''col_1''': '''string''', '''col_2''': '''string''', '''col_3''': '''string'''},
        {'''col_1''': '''int32''', '''col_2''': '''int32''', '''col_3''': '''int32'''},
        {'''col_1''': '''float32''', '''col_2''': '''float32''', '''col_3''': '''float32'''},
    ],   )
def        a__     (					UpperCamelCase_						:      str,   UpperCamelCase_						:      Dict,   UpperCamelCase_						:      Union[str, Any]						):
   UpperCAmelCase__	:Union[str, Any]      =      tmp_path / '''cache'''
   UpperCAmelCase__	:Optional[int]      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   UpperCAmelCase__	:List[Any]      =      features.copy() if features else default_expected_features
   UpperCAmelCase__	:Optional[int]      =      (
       Features({feature: Value(UpperCamelCase_						) for feature, dtype in features.items()}						) if features is not None else None
   )
   UpperCAmelCase__	:Optional[Any]      =      ParquetDatasetReader({'''train''': parquet_path},   features=UpperCamelCase_,   cache_dir=UpperCamelCase_						).read()
   _check_parquet_datasetdict(UpperCamelCase_,   UpperCamelCase_						)


@pytest.mark.parametrize('''split''',   [None, NamedSplit('''train'''						), '''train''', '''test''']						)
def        a__     (					UpperCamelCase_						:      List[Any],   UpperCamelCase_						:      Union[str, Any],   UpperCamelCase_						:      int						):
   if split:
      UpperCAmelCase__	:Union[str, Any]      =      {split: parquet_path}
   else:
      UpperCAmelCase__	:int      =      '''train'''
      UpperCAmelCase__	:int      =      {'''train''': parquet_path, '''test''': parquet_path}
   UpperCAmelCase__	:Dict      =      tmp_path / '''cache'''
   UpperCAmelCase__	:Dict      =      {'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
   UpperCAmelCase__	:Any      =      ParquetDatasetReader(UpperCamelCase_,   cache_dir=UpperCamelCase_						).read()
   _check_parquet_datasetdict(UpperCamelCase_,   UpperCamelCase_,   splits=list(path.keys()						)						)
   assert all(dataset[split].split == split for split in path.keys()						)


def        a__     (					UpperCamelCase_						:      Tuple,   UpperCamelCase_						:      Optional[int]						):
   UpperCAmelCase__	:List[str]      =      ParquetDatasetWriter(UpperCamelCase_,   tmp_path / '''foo.parquet'''						)
   assert writer.write() > 0
   UpperCAmelCase__	:List[str]      =      pq.ParquetFile(tmp_path / '''foo.parquet'''						)
   UpperCAmelCase__	:int      =      pf.read()
   assert dataset.data.table == output_table


def        a__     (					UpperCamelCase_						:      Any,   UpperCamelCase_						:      str						):
   UpperCAmelCase__	:str      =      str(shared_datadir / '''test_image_rgb.jpg'''						)
   UpperCAmelCase__	:List[Any]      =      {'''image''': [image_path]}
   UpperCAmelCase__	:Any      =      Features({'''image''': Image()}						)
   UpperCAmelCase__	:Tuple      =      Dataset.from_dict(UpperCamelCase_,   features=UpperCamelCase_						)
   UpperCAmelCase__	:Any      =      ParquetDatasetWriter(UpperCamelCase_,   tmp_path / '''foo.parquet'''						)
   assert writer.write() > 0

   UpperCAmelCase__	:int      =      Dataset.from_parquet(str(tmp_path / '''foo.parquet'''						)						)
   assert dataset.features == reloaded_dataset.features

   UpperCAmelCase__	:Optional[Any]      =      ParquetDatasetReader(str(tmp_path / '''foo.parquet'''						),   streaming=UpperCamelCase_						).read()
   assert dataset.features == reloaded_iterable_dataset.features


@pytest.mark.parametrize(
    '''feature, expected''',   [
        (Features({'''foo''': Value('''int32'''						)}						), None),
        (Features({'''image''': Image(), '''foo''': Value('''int32'''						)}						), config.PARQUET_ROW_GROUP_SIZE_FOR_IMAGE_DATASETS),
        (Features({'''nested''': Sequence(Audio()						)}						), config.PARQUET_ROW_GROUP_SIZE_FOR_AUDIO_DATASETS),
    ],   )
def        a__     (					UpperCamelCase_						:      Optional[int],   UpperCamelCase_						:      List[Any]						):
   assert get_writer_batch_size(UpperCamelCase_						) == expected







'''simple docstring'''





import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__lowerCamelCase												=						logging.get_logger(__name__)


def        a__     (					):
   # Get the sagemaker specific mp parameters from smp_options variable.
   UpperCAmelCase__	:Any      =      os.getenv('''SM_HP_MP_PARAMETERS''',   '''{}'''						)
   try:
      # Parse it and check the field "partitions" is included, it is required for model parallel.
      UpperCAmelCase__	:Any      =      json.loads(UpperCamelCase_						)
      if "partitions" not in smp_options:
         return False
   except json.JSONDecodeError:
      return False

   # Get the sagemaker specific framework parameters from mpi_options variable.
   UpperCAmelCase__	:Optional[Any]      =      os.getenv('''SM_FRAMEWORK_PARAMS''',   '''{}'''						)
   try:
      # Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
      UpperCAmelCase__	:str      =      json.loads(UpperCamelCase_						)
      if not mpi_options.get('''sagemaker_mpi_enabled''',   UpperCamelCase_						):
         return False
   except json.JSONDecodeError:
      return False
   # Lastly, check if the `smdistributed` module is present.
   return importlib.util.find_spec('''smdistributed'''						) is not None


if is_sagemaker_model_parallel_available():
    import smdistributed.modelparallel.torch as smp

    smp.init()




@dataclass
class 				UpperCAmelCase			(			_snake_case						):
 UpperCAmelCase				=					field(
     default=""						, metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}						, )


 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    super().__post_init__()
    warnings.warn(
        '''`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use '''
        '''`TrainingArguments` instead.'''			,       __lowerCamelCase			,       )


 @cached_property
 def        __SCREAMING_SNAKE_CASE					(      self							:	List[Any] ):
    logger.info('''PyTorch: setting up devices''' )
    if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
       logger.warning(
           '''torch.distributed process group is initialized, but local_rank == -1. '''
           '''In order to use Torch DDP, launch your script with `python -m torch.distributed.launch''' )
    if self.no_cuda:
       UpperCAmelCase__	:Union[str, Any]      =      torch.device('''cpu''' )
       UpperCAmelCase__	:Dict      =      0
    elif is_sagemaker_model_parallel_available():
       UpperCAmelCase__	:Dict      =      smp.local_rank()
       UpperCAmelCase__	:List[str]      =      torch.device('''cuda'''			,       __lowerCamelCase )
       UpperCAmelCase__	:Optional[int]      =      1
    elif is_sagemaker_dp_enabled():
       import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

       torch.distributed.init_process_group(backend='''smddp'''			,       timeout=self.ddp_timeout_delta )
       UpperCAmelCase__	:int      =      int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK''' ) )
       UpperCAmelCase__	:int      =      torch.device('''cuda'''			,       self.local_rank )
       UpperCAmelCase__	:Union[str, Any]      =      1
    elif self.local_rank == -1:
       # if n_gpu is > 1 we'll use nn.DataParallel.
       # If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
       # Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
       # trigger an error that a device index is missing. Index 0 takes into account the
       # GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
       # will use the first GPU in that env, i.e. GPU#1
       UpperCAmelCase__	:str      =      torch.device('''cuda:0''' if torch.cuda.is_available() else '''cpu''' )
       # Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
       # the default value.
       UpperCAmelCase__	:str      =      torch.cuda.device_count()
    else:
       # Here, we'll use torch.distributed.
       # Initializes the distributed backend which will take care of synchronizing nodes/GPUs
       if not torch.distributed.is_initialized():
          torch.distributed.init_process_group(backend='''nccl'''			,       timeout=self.ddp_timeout_delta )
       UpperCAmelCase__	:str      =      torch.device('''cuda'''			,       self.local_rank )
       UpperCAmelCase__	:Union[str, Any]      =      1

    if device.type == "cuda":
       torch.cuda.set_device(__lowerCamelCase )

    return device


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	List[str] ):
    if is_sagemaker_model_parallel_available():
       return smp.dp_size()

    return super().world_size


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict ):
    return not is_sagemaker_model_parallel_available()





 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	int ):
    return False



