






import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


UpperCamelCase        =				"\\n    Text data.\n    Second line of data."

UpperCamelCase        =				"file"




@pytest.fixture(scope="""session"""  )
def 				A		(					lowercase__  :  List[str]  )		-> Union[str, Any]:
	UpperCamelCase__ :Optional[int]						=      tmp_path_factory.mktemp("""data"""  ) / (FILE_PATH + """.zstd""")
	UpperCamelCase__ :Optional[Any]						=      bytes(lowercase__ ,		"""utf-8"""  )
	with zstd.open(lowercase__ ,		"""wb"""  ) as f:
		f.write(lowercase__  )
	return path




@pytest.fixture
def 				A		(					lowercase__  :  str  )		-> int:
	with open(os.path.join(tmpfs.local_root_dir ,		lowercase__  ) ,		"""w"""  ) as f:
		f.write(lowercase__  )
	return FILE_PATH




@pytest.mark.parametrize("""compression_format""" ,		["""gzip""", """xz""", """zstd"""]  )
def 				A		(					lowercase__  :  Optional[Any] ,		lowercase__  :  Dict ,		lowercase__  :  int ,		lowercase__  :  List[str] ,		lowercase__  :  Optional[int] ,		lowercase__  :  Any  )		-> Union[str, Any]:
	UpperCamelCase__ :Optional[int]						=      {"""gzip""": gz_file, """xz""": xz_file, """zstd""": zstd_path}
	UpperCamelCase__ :List[Any]						=      input_paths[compression_format]
	UpperCamelCase__ :Tuple						=      tmp_path / """cache"""
	UpperCamelCase__ :Dict						=      DownloadConfig(cache_dir=lowercase__ ,		extract_compressed_file=lowercase__  )
	UpperCamelCase__ :int						=      cached_path(lowercase__ ,		download_config=lowercase__  )
	with open(lowercase__  ) as f:
		UpperCamelCase__ :int						=      f.read()
	with open(lowercase__  ) as f:
		UpperCamelCase__ :Tuple						=      f.read()
	assert extracted_file_content == expected_file_content




@pytest.mark.parametrize("""default_extracted""" ,		[True, False]  )
@pytest.mark.parametrize("""default_cache_dir""" ,		[True, False]  )
def 				A		(					lowercase__  :  Union[str, Any] ,		lowercase__  :  Dict ,		lowercase__  :  Dict ,		lowercase__  :  Any ,		lowercase__  :  List[Any]  )		-> List[str]:
	UpperCamelCase__ :Dict						=      """custom_cache"""
	UpperCamelCase__ :Union[str, Any]						=      """custom_extracted_dir"""
	UpperCamelCase__ :Optional[int]						=      tmp_path / """custom_extracted_path"""
	if default_extracted:
		UpperCamelCase__ :Union[str, Any]						=      ("""downloads""" if default_cache_dir else custom_cache_dir, """extracted""")
	else:
		monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_DIR""" ,		lowercase__  )
		monkeypatch.setattr("""datasets.config.EXTRACTED_DATASETS_PATH""" ,		str(lowercase__  )  )
		UpperCamelCase__ :Dict						=      custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

	UpperCamelCase__ :Optional[int]						=      xz_file
	UpperCamelCase__ :Optional[Any]						=      (
	    DownloadConfig(extract_compressed_file=lowercase__  )
	    if default_cache_dir
	    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir ,		extract_compressed_file=lowercase__  )
	)
	UpperCamelCase__ :str						=      cached_path(lowercase__ ,		download_config=lowercase__  )
	assert Path(lowercase__  ).parent.parts[-2:] == expected




def 				A		(					lowercase__  :  List[str]  )		-> Dict:
	# absolute path
	UpperCamelCase__ :Optional[Any]						=      str(Path(lowercase__  ).resolve()  )
	assert cached_path(lowercase__  ) == text_file
	# relative path
	UpperCamelCase__ :str						=      str(Path(lowercase__  ).resolve().relative_to(Path(os.getcwd()  )  )  )
	assert cached_path(lowercase__  ) == text_file




def 				A		(					lowercase__  :  Optional[Any]  )		-> Tuple:
	# absolute path
	UpperCamelCase__ :Tuple						=      str(tmp_path.resolve() / """__missing_file__.txt"""  )
	with pytest.raises(lowercase__  ):
		cached_path(lowercase__  )
	# relative path
	UpperCamelCase__ :Tuple						=      """./__missing_file__.txt"""
	with pytest.raises(lowercase__  ):
		cached_path(lowercase__  )




def 				A		(					lowercase__  :  str  )		-> Optional[int]:
	UpperCamelCase__ :Any						=      get_from_cache(f"""tmp://{tmpfs_file}"""  )
	with open(lowercase__  ) as f:
		UpperCamelCase__ :Tuple						=      f.read()
	assert output_file_content == FILE_CONTENT




@patch("""datasets.config.HF_DATASETS_OFFLINE""" ,		lowercase__  )
def 				A		(					)		-> Tuple:
	with pytest.raises(lowercase__  ):
		cached_path("""https://huggingface.co"""  )




@patch("""datasets.config.HF_DATASETS_OFFLINE""" ,		lowercase__  )
def 				A		(					lowercase__  :  str  )		-> Optional[int]:
	UpperCamelCase__ :Optional[Any]						=      tmp_path_factory.mktemp("""data"""  ) / """file.html"""
	with pytest.raises(lowercase__  ):
		http_get("""https://huggingface.co""" ,		temp_file=lowercase__  )
	with pytest.raises(lowercase__  ):
		http_head("""https://huggingface.co"""  )




@patch("""datasets.config.HF_DATASETS_OFFLINE""" ,		lowercase__  )
def 				A		(					lowercase__  :  List[Any]  )		-> int:
	UpperCamelCase__ :List[str]						=      tmp_path_factory.mktemp("""data"""  ) / """file.html"""
	with pytest.raises(lowercase__  ):
		ftp_get("""ftp://huggingface.co""" ,		temp_file=lowercase__  )
	with pytest.raises(lowercase__  ):
		ftp_head("""ftp://huggingface.co"""  )




@patch("""datasets.config.HF_DATASETS_OFFLINE""" ,		lowercase__  )
def 				A		(					lowercase__  :  Optional[int]  )		-> List[Any]:
	UpperCamelCase__ :Optional[Any]						=      tmp_path_factory.mktemp("""data"""  ) / """file.html"""
	with pytest.raises(lowercase__  ):
		fsspec_get("""s3://huggingface.co""" ,		temp_file=lowercase__  )
	with pytest.raises(lowercase__  ):
		fsspec_head("""s3://huggingface.co"""  )
'''simple docstring'''

import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment







def __UpperCamelCase( _A       :					Any	,      _A       :					List[str]=()	,      _A       :					List[str]=None	,      _A       :					Dict="no"	,      _A       :					List[str]="29500"     ):
   '''simple docstring'''
   UpperCAmelCase__				:					int 					=	False
   UpperCAmelCase__				:					List[str] 					=	False
   if any(key.startswith('''KAGGLE'''     ) for key in os.environ.keys()     ):
      UpperCAmelCase__				:					List[str] 					=	True
   elif "IPython" in sys.modules:
      UpperCAmelCase__				:					List[Any] 					=	'''google.colab''' in str(sys.modules['''IPython'''].get_ipython()     )

   try:
      UpperCAmelCase__				:					Tuple 					=	PrecisionType(mixed_precision.lower()     )
   except ValueError:
      raise ValueError(
          F'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.'''     )

   if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''	,      _A     ) is not None):
      # TPU launch
      import torch_xla.distributed.xla_multiprocessing as xmp

      if len(AcceleratorState._shared_state     ) > 0:
         raise ValueError(
             '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
             '''your training function. Restart your notebook and make sure no cells initializes an '''
             '''`Accelerator`.'''     )
      if num_processes is None:
         UpperCAmelCase__				:					Tuple 					=	8

      UpperCAmelCase__				:					Optional[Any] 					=	PrepareForLaunch(_A	,      distributed_type='''TPU'''     )
      print(F'''Launching a training on {num_processes} TPU cores.'''     )
      xmp.spawn(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )
   elif in_colab:
      # No need for a distributed launch otherwise as it's either CPU or one GPU.
      if torch.cuda.is_available():
         print('''Launching training on one GPU.'''     )
      else:
         print('''Launching training on one CPU.'''     )
      function(*_A     )
   else:
      if num_processes is None:
         raise ValueError(
             '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.'''     )

      if num_processes > 1:
         # Multi-GPU launch
         from torch.multiprocessing import start_processes
         from torch.multiprocessing.spawn import ProcessRaisedException

         if len(AcceleratorState._shared_state     ) > 0:
            raise ValueError(
                '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
                '''inside your training function. Restart your notebook and make sure no cells initializes an '''
                '''`Accelerator`.'''     )

         if torch.cuda.is_initialized():
            raise ValueError(
                '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
                '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
                '''function.'''     )

         # torch.distributed will expect a few environment variable to be here. We set the ones common to each
         # process here (the other ones will be set be the launcher).
         with patch_environment(
             world_size=_A	,      master_addr='''127.0.01'''	,      master_port=_A	,      mixed_precision=_A     ):
            UpperCAmelCase__				:					str 					=	PrepareForLaunch(_A	,      distributed_type='''MULTI_GPU'''     )
            print(F'''Launching training on {num_processes} GPUs.'''     )
            try:
               start_processes(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )
            except ProcessRaisedException as e:
               if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
                  raise RuntimeError(
                      '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
                      '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
                      '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
                      '''which one is problematic.'''     ) from e

      else:
         # No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
         if is_mps_available():
            UpperCAmelCase__				:					Union[str, Any] 					=	'''1'''
            print('''Launching training on MPS.'''     )
         elif torch.cuda.is_available():
            print('''Launching training on one GPU.'''     )
         else:
            print('''Launching training on CPU.'''     )
         function(*_A     )







def __UpperCamelCase( _A       :					List[str]	,      _A       :					Optional[Any]=()	,      _A       :					str=2     ):
   '''simple docstring'''
   from torch.multiprocessing import start_processes

   with tempfile.NamedTemporaryFile() as tmp_file:
      # torch.distributed will expect a few environment variable to be here. We set the ones common to each
      # process here (the other ones will be set be the launcher).
      with patch_environment(
          world_size=_A	,      master_addr='''127.0.01'''	,      master_port='''29500'''	,      accelerate_mixed_precision='''no'''	,      accelerate_debug_rdv_file=tmp_file.name	,      accelerate_use_cpu='''yes'''	,      ):
         UpperCAmelCase__				:					Optional[int] 					=	PrepareForLaunch(_A	,      debug=_A     )
         start_processes(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )




