






"""simple docstring"""
import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


_lowerCAmelCase         =							"http://www.mocksite.com/file1.txt"
_lowerCAmelCase         =							"\"text\": [\"foo\", \"foo\"]"
_lowerCAmelCase         =							"6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8"



class      SCREAMING_SNAKE_CASE  :





	"""simple docstring"""




	_A       :				Optional[int]    =				200
	_A       :				Dict    =				{"""Content-Length""": """100"""}
	_A       :				List[Any]    =				{}





	def    lowerCamelCase(self     ,     **lowerCAmelCase_							):
		return [bytes(lowerCAmelCase_     ,     """utf-8"""							)]






def 			__UpperCamelCase	(		*snake_case__   ,   **snake_case__					):
	return MockResponse()






@pytest.mark.parametrize("""urls_type"""   ,   [str, list, dict]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__					):
	import requests

	monkeypatch.setattr(snake_case__   ,   """request"""   ,   snake_case__					)

	A_		:						Optional[int]	    =   URL
	if issubclass(snake_case__   ,   snake_case__					):
		A_		:						str	    =   url
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						Union[str, Any]	    =   [url]
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						List[Any]	    =   {"""train""": url}
	A_		:						Tuple	    =   """dummy"""
	A_		:						List[str]	    =   """downloads"""
	A_		:						List[str]	    =   tmp_path
	A_		:						str	    =   DownloadConfig(
	    cache_dir=os.path.join(snake_case__   ,   snake_case__					)   ,   use_etag=snake_case__   ,   )
	A_		:						Any	    =   DownloadManager(dataset_name=snake_case__   ,   download_config=snake_case__					)
	A_		:						Dict	    =   dl_manager.download(snake_case__					)
	A_		:						List[str]	    =   urls
	for downloaded_paths in [downloaded_paths]:
		if isinstance(snake_case__   ,   snake_case__					):
			A_		:						Optional[Any]	    =   [downloaded_paths]
			A_		:						Tuple	    =   [urls]
		elif isinstance(snake_case__   ,   snake_case__					):
			assert "train" in downloaded_paths.keys()
			A_		:						Union[str, Any]	    =   downloaded_paths.values()
			A_		:						Tuple	    =   urls.values()
		assert downloaded_paths
		for downloaded_path, input_url in zip(snake_case__   ,   snake_case__					):
			assert downloaded_path == dl_manager.downloaded_paths[input_url]
			A_		:						Union[str, Any]	    =   Path(snake_case__					)
			A_		:						str	    =   downloaded_path.parts
			assert parts[-1] == HASH
			assert parts[-2] == cache_subdir
			assert downloaded_path.exists()
			A_		:						List[str]	    =   downloaded_path.read_text()
			assert content == CONTENT
			A_		:						Union[str, Any]	    =   downloaded_path.with_suffix(""".json"""					)
			assert metadata_downloaded_path.exists()
			A_		:						List[str]	    =   json.loads(metadata_downloaded_path.read_text()					)
			assert metadata_content == {"url": URL, "etag": None}






@pytest.mark.parametrize("""paths_type"""   ,   [str, list, dict]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__					):
	A_		:						str	    =   str(snake_case__					)
	if issubclass(snake_case__   ,   snake_case__					):
		A_		:						List[str]	    =   filename
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						Tuple	    =   [filename]
	elif issubclass(snake_case__   ,   snake_case__					):
		A_		:						Dict	    =   {"""train""": filename}
	A_		:						Union[str, Any]	    =   """dummy"""
	A_		:						Dict	    =   xz_file.parent
	A_		:						List[str]	    =   """extracted"""
	A_		:						int	    =   DownloadConfig(
	    cache_dir=snake_case__   ,   use_etag=snake_case__   ,   )
	A_		:						str	    =   DownloadManager(dataset_name=snake_case__   ,   download_config=snake_case__					)
	A_		:						int	    =   dl_manager.extract(snake_case__					)
	A_		:						int	    =   paths
	for extracted_paths in [extracted_paths]:
		if isinstance(snake_case__   ,   snake_case__					):
			A_		:						Dict	    =   [extracted_paths]
			A_		:						str	    =   [paths]
		elif isinstance(snake_case__   ,   snake_case__					):
			assert "train" in extracted_paths.keys()
			A_		:						Any	    =   extracted_paths.values()
			A_		:						str	    =   paths.values()
		assert extracted_paths
		for extracted_path, input_path in zip(snake_case__   ,   snake_case__					):
			assert extracted_path == dl_manager.extracted_paths[input_path]
			A_		:						str	    =   Path(snake_case__					)
			A_		:						List[str]	    =   extracted_path.parts
			assert parts[-1] == hash_url_to_filename(snake_case__   ,   etag=snake_case__					)
			assert parts[-2] == extracted_subdir
			assert extracted_path.exists()
			A_		:						Optional[Any]	    =   extracted_path.read_text()
			A_		:						Optional[int]	    =   text_file.read_text()
			assert extracted_file_content == expected_file_content






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__					):
	assert path.endswith(""".jsonl"""					)
	for num_items, line in enumerate(snake_case__   ,   start=1					):
		A_		:						List[str]	    =   json.loads(line.decode("""utf-8"""					)					)
		assert item.keys() == {"col_1", "col_2", "col_3"}
	assert num_items == 4






@pytest.mark.parametrize("""archive_jsonl"""   ,   ["""tar_jsonl_path""", """zip_jsonl_path"""]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__					):
	A_		:						Optional[int]	    =   request.getfixturevalue(snake_case__					)
	A_		:						List[str]	    =   DownloadManager()
	for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(snake_case__					)   ,   start=1					):
		_test_jsonl(snake_case__   ,   snake_case__					)
	assert num_jsonl == 2






@pytest.mark.parametrize("""archive_nested_jsonl"""   ,   ["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""]					)
def 			__UpperCamelCase	(		snake_case__   ,   snake_case__					):
	A_		:						int	    =   request.getfixturevalue(snake_case__					)
	A_		:						Union[str, Any]	    =   DownloadManager()
	for num_tar, (path, file) in enumerate(dl_manager.iter_archive(snake_case__					)   ,   start=1					):
		for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(snake_case__					)   ,   start=1					):
			_test_jsonl(snake_case__   ,   snake_case__					)
	assert num_tar == 1
	assert num_jsonl == 2






def 			__UpperCamelCase	(		snake_case__					):
	A_		:						int	    =   DownloadManager()
	for num_file, file in enumerate(dl_manager.iter_files(snake_case__					)   ,   start=1					):
		assert os.path.basename(snake_case__					) == ("test.txt" if num_file == 1 else "train.txt")
	assert num_file == 2









"""simple docstring"""
import os

import torch

from ..logging import get_logger
from .constants import FSDP_PYTORCH_VERSION, MODEL_NAME, OPTIMIZER_NAME
from .versions import is_torch_version


if is_torch_version(">=", FSDP_PYTORCH_VERSION):
							import torch.distributed.checkpoint as dist_cp
							from torch.distributed.checkpoint.default_planner import DefaultLoadPlanner, DefaultSavePlanner
							from torch.distributed.checkpoint.optimizer import load_sharded_optimizer_state_dict
							from torch.distributed.fsdp.fully_sharded_data_parallel import FullyShardedDataParallel as FSDP
							from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType


_lowerCAmelCase         =							get_logger(__name__)






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__=0					):
	os.makedirs(snake_case__   ,   exist_ok=snake_case__					)
	with FSDP.state_dict_type(
	    snake_case__   ,   fsdp_plugin.state_dict_type   ,   fsdp_plugin.state_dict_config   ,   fsdp_plugin.optim_state_dict_config					):
		A_		:						Optional[int]	    =   model.state_dict()
		if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
			A_		:						int	    =   F"""{MODEL_NAME}.bin""" if model_index == 0 else F"""{MODEL_NAME}_{model_index}.bin"""
			A_		:						Union[str, Any]	    =   os.path.join(snake_case__   ,   snake_case__					)
			if accelerator.process_index == 0:
				logger.info(F"""Saving model to {output_model_file}"""					)
				torch.save(snake_case__   ,   snake_case__					)
				logger.info(F"""Model saved to {output_model_file}"""					)
		elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
			A_		:						List[str]	    =   (
			    F"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
			    if model_index == 0
			    else F"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
			)
			A_		:						List[str]	    =   os.path.join(snake_case__   ,   snake_case__					)
			logger.info(F"""Saving model to {output_model_file}"""					)
			torch.save(snake_case__   ,   snake_case__					)
			logger.info(F"""Model saved to {output_model_file}"""					)
		elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
			A_		:						str	    =   os.path.join(snake_case__   ,   F"""{MODEL_NAME}_{model_index}"""					)
			os.makedirs(snake_case__   ,   exist_ok=snake_case__					)
			logger.info(F"""Saving model to {ckpt_dir}"""					)
			A_		:						Dict	    =   {"""model""": state_dict}

			dist_cp.save_state_dict(
			    state_dict=snake_case__   ,   storage_writer=dist_cp.FileSystemWriter(snake_case__					)   ,   planner=DefaultSavePlanner()   ,   )
			logger.info(F"""Model saved to {ckpt_dir}"""					)






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__=0					):
	accelerator.wait_for_everyone()
	with FSDP.state_dict_type(
	    snake_case__   ,   fsdp_plugin.state_dict_type   ,   fsdp_plugin.state_dict_config   ,   fsdp_plugin.optim_state_dict_config					):
		if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
			if type(snake_case__					) != FSDP and accelerator.process_index != 0:
				if not fsdp_plugin.sync_module_states:
					raise ValueError(
					    """Set the `sync_module_states` flag to `True` so that model states are synced across processes when """
					    """initializing FSDP object"""					)
				return
			A_		:						Tuple	    =   F"""{MODEL_NAME}.bin""" if model_index == 0 else F"""{MODEL_NAME}_{model_index}.bin"""
			A_		:						Any	    =   os.path.join(snake_case__   ,   snake_case__					)
			logger.info(F"""Loading model from {input_model_file}"""					)
			A_		:						List[Any]	    =   torch.load(snake_case__					)
			logger.info(F"""Model loaded from {input_model_file}"""					)
		elif fsdp_plugin.state_dict_type == StateDictType.LOCAL_STATE_DICT:
			A_		:						List[str]	    =   (
			    F"""{MODEL_NAME}_rank{accelerator.process_index}.bin"""
			    if model_index == 0
			    else F"""{MODEL_NAME}_{model_index}_rank{accelerator.process_index}.bin"""
			)
			A_		:						Any	    =   os.path.join(snake_case__   ,   snake_case__					)
			logger.info(F"""Loading model from {input_model_file}"""					)
			A_		:						List[str]	    =   torch.load(snake_case__					)
			logger.info(F"""Model loaded from {input_model_file}"""					)
		elif fsdp_plugin.state_dict_type == StateDictType.SHARDED_STATE_DICT:
			A_		:						Optional[Any]	    =   (
			    os.path.join(snake_case__   ,   F"""{MODEL_NAME}_{model_index}"""					)
			    if F"""{MODEL_NAME}""" not in input_dir
			    else input_dir
			)
			logger.info(F"""Loading model from {ckpt_dir}"""					)
			A_		:						List[str]	    =   {"""model""": model.state_dict()}
			dist_cp.load_state_dict(
			    state_dict=snake_case__   ,   storage_reader=dist_cp.FileSystemReader(snake_case__					)   ,   planner=DefaultLoadPlanner()   ,   )
			A_		:						Tuple	    =   state_dict["""model"""]
			logger.info(F"""Model loaded from {ckpt_dir}"""					)
		model.load_state_dict(snake_case__					)






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__=0					):
	os.makedirs(snake_case__   ,   exist_ok=snake_case__					)
	with FSDP.state_dict_type(
	    snake_case__   ,   fsdp_plugin.state_dict_type   ,   fsdp_plugin.state_dict_config   ,   fsdp_plugin.optim_state_dict_config					):
		A_		:						Optional[Any]	    =   FSDP.optim_state_dict(snake_case__   ,   snake_case__					)
		if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
			if accelerator.process_index == 0:
				A_		:						Any	    =   (
				    F"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else F"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
				)
				A_		:						Optional[int]	    =   os.path.join(snake_case__   ,   snake_case__					)
				logger.info(F"""Saving Optimizer state to {output_optimizer_file}"""					)
				torch.save(snake_case__   ,   snake_case__					)
				logger.info(F"""Optimizer state saved in {output_optimizer_file}"""					)
		else:
			A_		:						Union[str, Any]	    =   os.path.join(snake_case__   ,   F"""{OPTIMIZER_NAME}_{optimizer_index}"""					)
			os.makedirs(snake_case__   ,   exist_ok=snake_case__					)
			logger.info(F"""Saving Optimizer state to {ckpt_dir}"""					)
			dist_cp.save_state_dict(
			    state_dict={"""optimizer""": optim_state}   ,   storage_writer=dist_cp.FileSystemWriter(snake_case__					)   ,   planner=DefaultSavePlanner()   ,   )
			logger.info(F"""Optimizer state saved in {ckpt_dir}"""					)






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__   ,   snake_case__=0					):
	accelerator.wait_for_everyone()
	with FSDP.state_dict_type(
	    snake_case__   ,   fsdp_plugin.state_dict_type   ,   fsdp_plugin.state_dict_config   ,   fsdp_plugin.optim_state_dict_config					):
		if fsdp_plugin.state_dict_type == StateDictType.FULL_STATE_DICT:
			A_		:						List[str]	    =   None
			# below check should work but currently it isn't working (mostly opytorch issue),
			# in the meantime disabling it at the cost of excess memory usage
			# if accelerator.process_index == 0 or not fsdp_plugin.optim_state_dict_config.rank0_only:
			A_		:						Tuple	    =   (
			    F"""{OPTIMIZER_NAME}.bin""" if optimizer_index == 0 else F"""{OPTIMIZER_NAME}_{optimizer_index}.bin"""
			)
			A_		:						int	    =   os.path.join(snake_case__   ,   snake_case__					)
			logger.info(F"""Loading Optimizer state from {input_optimizer_file}"""					)
			A_		:						Tuple	    =   torch.load(snake_case__					)
			logger.info(F"""Optimizer state loaded from {input_optimizer_file}"""					)
		else:
			A_		:						str	    =   (
			    os.path.join(snake_case__   ,   F"""{OPTIMIZER_NAME}_{optimizer_index}"""					)
			    if F"""{OPTIMIZER_NAME}""" not in input_dir
			    else input_dir
			)
			logger.info(F"""Loading Optimizer from {ckpt_dir}"""					)
			A_		:						Union[str, Any]	    =   load_sharded_optimizer_state_dict(
			    model_state_dict=model.state_dict()   ,   optimizer_key="""optimizer"""   ,   storage_reader=dist_cp.FileSystemReader(snake_case__					)   ,   )
			A_		:						Tuple	    =   optim_state["""optimizer"""]
			logger.info(F"""Optimizer loaded from {ckpt_dir}"""					)
		A_		:						List[Any]	    =   FSDP.optim_state_dict_to_load(snake_case__   ,   snake_case__   ,   snake_case__					)
		optimizer.load_state_dict(snake_case__					)

