






import os
import textwrap

import pyarrow as pa
import pytest

from datasets import ClassLabel, Features, Image
from datasets.packaged_modules.csv.csv import Csv

from ..utils import require_pil



@pytest.fixture
def  a__ (__lowercase							:Union[str, Any]   )		-> Optional[Any]:
				_A	:				Any   =      tmp_path / '''file.csv'''
				_A	:				Optional[int]   =      textwrap.dedent(
				    '''\
        header1,header2
        1,2
        10,20
        '''   )
				with open(__lowercase					,  '''w'''   ) as f:
								f.write(__lowercase   )
				return str(__lowercase   )



@pytest.fixture
def  a__ (__lowercase							:Tuple   )		-> int:
				_A	:				str   =      tmp_path / '''malformed_file.csv'''
				_A	:				List[str]   =      textwrap.dedent(
				    '''\
        header1,header2
        1,2
        10,20,
        '''   )
				with open(__lowercase					,  '''w'''   ) as f:
								f.write(__lowercase   )
				return str(__lowercase   )



@pytest.fixture
def  a__ (__lowercase							:Union[str, Any]					,  __lowercase							:Tuple   )		-> Optional[int]:
				_A	:				Optional[int]   =      tmp_path / '''csv_with_image.csv'''
				_A	:				Dict   =      textwrap.dedent(
				    f"""\
        image
        {image_file}
        """   )
				with open(__lowercase					,  '''w'''   ) as f:
								f.write(__lowercase   )
				return str(__lowercase   )



@pytest.fixture
def  a__ (__lowercase							:List[str]   )		-> Union[str, Any]:
				_A	:				Dict   =      tmp_path / '''csv_with_label.csv'''
				_A	:				str   =      textwrap.dedent(
				    '''\
        label
        good
        bad
        good
        '''   )
				with open(__lowercase					,  '''w'''   ) as f:
								f.write(__lowercase   )
				return str(__lowercase   )



@pytest.fixture
def  a__ (__lowercase							:Optional[Any]   )		-> int:
				_A	:				Optional[int]   =      tmp_path / '''csv_with_int_list.csv'''
				_A	:				Union[str, Any]   =      textwrap.dedent(
				    '''\
        int_list
        1 2 3
        4 5 6
        7 8 9
        '''   )
				with open(__lowercase					,  '''w'''   ) as f:
								f.write(__lowercase   )
				return str(__lowercase   )



def  a__ (__lowercase							:Dict					,  __lowercase							:Any					,  __lowercase							:Union[str, Any]   )		-> int:
				_A	:				Tuple   =      Csv()
				_A	:				int   =      csv._generate_tables([[csv_file, malformed_csv_file]]   )
				with pytest.raises(__lowercase					,  match='''Error tokenizing data'''   ):
								for _ in generator:
												pass
				assert any(
				    record.levelname == '''ERROR'''
				    and '''Failed to read file''' in record.message
				    and os.path.basename(__lowercase   ) in record.message
				    for record in caplog.records   )



@require_pil
def  a__ (__lowercase							:Tuple   )		-> List[str]:
				with open(__lowercase					,  encoding='''utf-8'''   ) as f:
								_A	:				List[str]   =      f.read().splitlines()[1]
				_A	:				Dict   =      Csv(encoding='''utf-8'''					,  features=Features({'''image''': Image()}   )   )
				_A	:				List[Any]   =      csv._generate_tables([[csv_file_with_image]]   )
				_A	:				Optional[Any]   =      pa.concat_tables([table for _, table in generator]   )
				assert pa_table.schema.field('''image'''   ).type == Image()()
				_A	:				List[str]   =      pa_table.to_pydict()['''image''']
				assert generated_content == [{"path": image_file, "bytes": None}]



def  a__ (__lowercase							:List[str]   )		-> List[Any]:
				with open(__lowercase					,  encoding='''utf-8'''   ) as f:
								_A	:				Union[str, Any]   =      f.read().splitlines()[1:]
				_A	:				Tuple   =      Csv(encoding='''utf-8'''					,  features=Features({'''label''': ClassLabel(names=['''good''', '''bad''']   )}   )   )
				_A	:				Any   =      csv._generate_tables([[csv_file_with_label]]   )
				_A	:				Tuple   =      pa.concat_tables([table for _, table in generator]   )
				assert pa_table.schema.field('''label'''   ).type == ClassLabel(names=['''good''', '''bad''']   )()
				_A	:				Union[str, Any]   =      pa_table.to_pydict()['''label''']
				assert generated_content == [ClassLabel(names=['''good''', '''bad''']   ).straint(__lowercase   ) for label in labels]






def  a__ (__lowercase							:Optional[int]   )		-> List[str]:
				_A	:				Tuple   =      Csv(encoding='''utf-8'''					,  sep=''','''					,  converters={'''int_list''': lambda __lowercase   : [int(__lowercase   ) for i in x.split()]}   )
				_A	:				Union[str, Any]   =      csv._generate_tables([[csv_file_with_int_list]]   )
				_A	:				int   =      pa.concat_tables([table for _, table in generator]   )
				assert pa.types.is_list(pa_table.schema.field('''int_list'''   ).type   )
				_A	:				Any   =      pa_table.to_pydict()['''int_list''']
				assert generated_content == [[1, 2, 3], [4, 5, 6], [7, 8, 9]]











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


_UpperCamelCase			:	Optional[Any]								=logging.get_logger(__name__)



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

				# Get the sagemaker specific framework parameters from mpi_options variable.
				_A	:				Optional[int]   =      os.getenv('''SM_FRAMEWORK_PARAMS'''					,  '''{}'''   )
				try:
								# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
								_A	:				Any   =      json.loads(__lowercase   )
								if not mpi_options.get('''sagemaker_mpi_enabled'''					,  __lowercase   ):
												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							):
		__snake_case      :       str     =   field(
		    default=""     ,				metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}     ,				)







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







		@cached_property
		def  A__      (					self		):
						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:
										_A	:				Union[str, Any]   =      torch.device('''cpu'''		)
										_A	:				List[str]   =      0
						elif is_sagemaker_model_parallel_available():
										_A	:				Dict   =      smp.local_rank()
										_A	:				Union[str, Any]   =      torch.device('''cuda'''       ,A__		)
										_A	:				Tuple   =      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		)
										_A	:				int   =      int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK'''		)		)
										_A	:				int   =      torch.device('''cuda'''       ,self.local_rank		)
										_A	:				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
										_A	:				Dict   =      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.
										_A	:				Union[str, Any]   =      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		)
										_A	:				Union[str, Any]   =      torch.device('''cuda'''       ,self.local_rank		)
										_A	:				List[Any]   =      1

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

						return device







		@property
		def  A__      (					self		):
						if is_sagemaker_model_parallel_available():
										return smp.dp_size()

						return super().world_size







		@property
		def  A__      (					self		):
						return not is_sagemaker_model_parallel_available()







		@property
		def  A__      (					self		):
						return False



