




"""simple docstring"""





import importlib
import os

import fsspec
import pytest
from fsspec import register_implementation
from fsspec.registry import _registry as _fsspec_registry

from datasets.filesystems import COMPRESSION_FILESYSTEMS, HfFileSystem, extract_path_from_uri, is_remote_filesystem

from .utils import require_lza, require_zstandard

def 					_UpperCamelCase    (			UpperCamelCase							)			->			Dict:


			"""simple docstring"""
			assert "mock" in _fsspec_registry
			assert "bz2" in _fsspec_registry

def 					_UpperCamelCase    (			)			->			Dict:


			"""simple docstring"""
			assert "mock" not in _fsspec_registry
			assert "bz2" in _fsspec_registry

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


			"""simple docstring"""
			__UpperCAmelCase	: Union[str, Any]		=					"mock-s3-bucket"
			__UpperCAmelCase	: Union[str, Any]		=					f"s3://{mock_bucket}"
			__UpperCAmelCase	: List[str]		=					extract_path_from_uri(UpperCamelCase							)
			assert dataset_path.startswith("s3://"							) is False

			__UpperCAmelCase	: int		=					"./local/path"
			__UpperCAmelCase	: str		=					extract_path_from_uri(UpperCamelCase							)
			assert dataset_path == new_dataset_path

def 					_UpperCamelCase    (			UpperCamelCase							)			->			Optional[Any]:


			"""simple docstring"""
			__UpperCAmelCase	: Tuple		=					is_remote_filesystem(UpperCamelCase							)
			assert is_remote is True

			__UpperCAmelCase	: List[Any]		=					fsspec.filesystem("file"							)

			__UpperCAmelCase	: List[str]		=					is_remote_filesystem(UpperCamelCase							)
			assert is_remote is False

@pytest.mark.parametrize("compression_fs_class"				,       UpperCamelCase							)
def 					_UpperCamelCase    (			UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase							)			->			Optional[Any]:


			"""simple docstring"""
			__UpperCAmelCase	: Optional[int]		=					{"gzip": gz_file, "xz": xz_file, "zstd": zstd_file, "bz2": bza_file, "lz4": lza_file}
			__UpperCAmelCase	: Tuple		=					input_paths[compression_fs_class.protocol]
			if input_path is None:
						__UpperCAmelCase	: Dict		=					f"for '{compression_fs_class.protocol}' compression protocol, "
						if compression_fs_class.protocol == "lz4":
									reason += require_lza.kwargs["reason"]
						elif compression_fs_class.protocol == "zstd":
									reason += require_zstandard.kwargs["reason"]
						pytest.skip(UpperCamelCase							)
			__UpperCAmelCase	: Union[str, Any]		=					fsspec.filesystem(compression_fs_class.protocol				,       fo=UpperCamelCase							)
			assert isinstance(UpperCamelCase				,       UpperCamelCase							)
			__UpperCAmelCase	: Tuple		=					os.path.basename(UpperCamelCase							)
			__UpperCAmelCase	: Optional[Any]		=					expected_filename[: expected_filename.rindex("."							)]
			assert fs.glob("*"							) == [expected_filename]
			with fs.open(UpperCamelCase				,       "r"				,       encoding="utf-8"							) as f, open(UpperCamelCase				,       encoding="utf-8"							) as expected_file:
						assert f.read() == expected_file.read()

@pytest.mark.parametrize("protocol"				,       ["zip", "gzip"]							)
def 					_UpperCamelCase    (			UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase							)			->			Optional[int]:


			"""simple docstring"""
			__UpperCAmelCase	: Dict		=					{"zip": zip_jsonl_path, "gzip": jsonl_gz_path}
			__UpperCAmelCase	: Dict		=					compressed_file_paths[protocol]
			__UpperCAmelCase	: Any		=					"dataset.jsonl"
			__UpperCAmelCase	: Any		=					f"{protocol}://{member_file_path}::{compressed_file_path}"
			__UpperCAmelCase	,     *__UpperCAmelCase	: Dict		=					fsspec.get_fs_token_paths(UpperCamelCase							)
			assert fs.isfile(UpperCamelCase							)
			assert not fs.isfile("non_existing_" + member_file_path							)

@pytest.mark.integration
def 					_UpperCamelCase    (			UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase				,       UpperCamelCase							)			->			Optional[Any]:


			"""simple docstring"""
			__UpperCAmelCase	: List[str]		=					hf_api.dataset_info(UpperCamelCase				,       token=UpperCamelCase							)
			__UpperCAmelCase	: Any		=					HfFileSystem(repo_info=UpperCamelCase				,       token=UpperCamelCase							)
			assert sorted(hffs.glob("*"							)							) == [".gitattributes", "data"]
			assert hffs.isdir("data"							)
			assert hffs.isfile(".gitattributes"							) and hffs.isfile("data/text_data.txt"							)
			with open(UpperCamelCase							) as f:
						assert hffs.open("data/text_data.txt"				,       "r"							).read() == f.read()





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


			"""simple docstring"""
			__UpperCAmelCase	: int		=					"bz2"

			# Import module
			import datasets.filesystems

			# Overwrite protocol and reload
			register_implementation(UpperCamelCase				,       UpperCamelCase				,       clobber=UpperCamelCase							)
			with pytest.warns(UpperCamelCase							) as warning_info:
						importlib.reload(datasets.filesystems							)

			assert len(UpperCamelCase							) == 1
			assert (
			    str(warning_info[0].message							)
			    == f"A filesystem protocol was already set for {protocol} and will be overwritten."
			)








"""simple docstring"""





from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader



class      a__				(	__magic_name__   ):


			def __init__(      self      :    List[Any]      ,  UpperCamelCase_      :    pyspark.sql.DataFrame      ,  UpperCamelCase_      :    Optional[NamedSplit] = None      ,  UpperCamelCase_      :    Optional[Features] = None      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    str = None      ,  UpperCamelCase_      :    bool = False      ,  UpperCamelCase_      :    str = None      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    str = "arrow"      ,  **UpperCamelCase_      :    Tuple      ,  ):


						"""simple docstring"""




						super().__init__(
						    split=UpperCamelCase_      ,  features=UpperCamelCase_      ,  cache_dir=UpperCamelCase_      ,  keep_in_memory=UpperCamelCase_      ,  streaming=UpperCamelCase_      ,  **UpperCamelCase_      ,  )
						__UpperCAmelCase	: List[Any]		=					load_from_cache_file
						__UpperCAmelCase	: List[Any]		=					file_format
						__UpperCAmelCase	: str		=					Spark(
						    df=UpperCamelCase_      ,  features=UpperCamelCase_      ,  cache_dir=UpperCamelCase_      ,  working_dir=UpperCamelCase_      ,  **UpperCamelCase_      ,  )


			def       a_     (      self      :    Union[str, Any]):


						"""simple docstring"""




						if self.streaming:
									return self.builder.as_streaming_dataset(split=self.split)
						__UpperCAmelCase	: List[Any]		=					None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
						self.builder.download_and_prepare(
						    download_mode=UpperCamelCase_      ,  file_format=self._file_format      ,  )
						return self.builder.as_dataset(split=self.split)


