
'''simple docstring'''







import textwrap

import pyarrow as pa
import pytest

from datasets import Features, Image
from datasets.packaged_modules.text.text import Text

from ..utils import require_pil
@pytest.fixture
def       snake_case_	(       __SCREAMING_SNAKE_CASE      :				Dict							):



						"""simple docstring"""





						lowercase_			:  Union[str, Any]									= tmp_path / '''text.txt'''
						lowercase_			:  int									= textwrap.dedent(
						    '''\
        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.
        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.
        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.

        Second paragraph:
        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.
        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.
        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.
        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.
        '''							)
						with open(__SCREAMING_SNAKE_CASE   ,   '''w'''   ,   encoding='''utf-8'''							) as f:
												f.write(__SCREAMING_SNAKE_CASE							)
						return str(__SCREAMING_SNAKE_CASE							)
@pytest.fixture
def       snake_case_	(       __SCREAMING_SNAKE_CASE      :				Any   ,   __SCREAMING_SNAKE_CASE      :				Tuple							):



						"""simple docstring"""





						lowercase_			:  Any									= tmp_path / '''text_with_image.txt'''
						with open(__SCREAMING_SNAKE_CASE   ,   '''w'''   ,   encoding='''utf-8'''							) as f:
												f.write(__SCREAMING_SNAKE_CASE							)
						return str(__SCREAMING_SNAKE_CASE							)
@pytest.mark.parametrize('''keep_linebreaks'''   ,   [True, False]							)
def       snake_case_	(       __SCREAMING_SNAKE_CASE      :				List[str]   ,   __SCREAMING_SNAKE_CASE      :				int							):



						"""simple docstring"""





						with open(__SCREAMING_SNAKE_CASE   ,   encoding='''utf-8'''							) as f:
												lowercase_			:  List[Any]									= f.read().splitlines(keepends=__SCREAMING_SNAKE_CASE							)
						lowercase_			:  str									= Text(keep_linebreaks=__SCREAMING_SNAKE_CASE   ,   encoding='''utf-8'''							)
						lowercase_			:  str									= text._generate_tables([[text_file]]							)
						lowercase_			:  Optional[Any]									= pa.concat_tables([table for _, table in generator]							).to_pydict()['''text''']
						assert generated_content == expected_content
@require_pil
def       snake_case_	(       __SCREAMING_SNAKE_CASE      :				Any							):



						"""simple docstring"""





						with open(__SCREAMING_SNAKE_CASE   ,   encoding='''utf-8'''							) as f:
												lowercase_			:  int									= f.read().splitlines()[0]
						lowercase_			:  Optional[Any]									= Text(encoding='''utf-8'''   ,   features=Features({'''image''': Image()}							)							)
						lowercase_			:  Optional[int]									= text._generate_tables([[text_file_with_image]]							)
						lowercase_			:  Optional[Any]									= pa.concat_tables([table for _, table in generator]							)
						assert pa_table.schema.field('''image'''							).type == Image()()
						lowercase_			:  Any									= pa_table.to_pydict()['''image''']
						assert generated_content == [{"path": image_file, "bytes": None}]




@pytest.mark.parametrize('''sample_by'''   ,   ['''line''', '''paragraph''', '''document''']							)
def       snake_case_	(       __SCREAMING_SNAKE_CASE      :				Optional[int]   ,   __SCREAMING_SNAKE_CASE      :				str							):



						"""simple docstring"""





						with open(__SCREAMING_SNAKE_CASE   ,   encoding='''utf-8'''							) as f:
												lowercase_			:  List[Any]									= f.read()
						if sample_by == "line":
												lowercase_			:  List[Any]									= expected_content.splitlines()
						elif sample_by == "paragraph":
												lowercase_			:  List[str]									= expected_content.split('''\n\n'''							)
						elif sample_by == "document":
												lowercase_			:  str									= [expected_content]
						lowercase_			:  str									= Text(sample_by=__SCREAMING_SNAKE_CASE   ,   encoding='''utf-8'''   ,   chunksize=100							)
						lowercase_			:  Any									= text._generate_tables([[text_file]]							)
						lowercase_			:  Optional[Any]									= pa.concat_tables([table for _, table in generator]							).to_pydict()['''text''']
						assert generated_content == expected_content





'''simple docstring'''







import multiprocessing
import os
from typing import BinaryIO, Optional, Union

from .. import Dataset, Features, NamedSplit, config
from ..formatting import query_table
from ..packaged_modules.csv.csv import Csv
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader






class 				lowerCAmelCase__     (						lowerCamelCase_  ):






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = False       ,							__SCREAMING_SNAKE_CASE = False       ,							__SCREAMING_SNAKE_CASE = None       ,							**__SCREAMING_SNAKE_CASE       ,							):


											"""simple docstring"""





											super().__init__(
											    __SCREAMING_SNAKE_CASE       ,							split=__SCREAMING_SNAKE_CASE       ,							features=__SCREAMING_SNAKE_CASE       ,							cache_dir=__SCREAMING_SNAKE_CASE       ,							keep_in_memory=__SCREAMING_SNAKE_CASE       ,							streaming=__SCREAMING_SNAKE_CASE       ,							num_proc=__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE       ,							)
											lowercase_			:  int									= path_or_paths if isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	) else {self.split: path_or_paths}
											lowercase_			:  List[Any]									= Csv(
											    cache_dir=__SCREAMING_SNAKE_CASE       ,							data_files=__SCREAMING_SNAKE_CASE       ,							features=__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE       ,							)






					def 			_snake_case					(     self	):


											"""simple docstring"""





											if self.streaming:
																	lowercase_			:  List[str]									= self.builder.as_streaming_dataset(split=self.split	)
											# Build regular (map-style) dataset
											else:
																	lowercase_			:  Any									= None
																	lowercase_			:  Optional[Any]									= None
																	lowercase_			:  str									= None
																	lowercase_			:  Optional[int]									= None

																	self.builder.download_and_prepare(
																	    download_config=__SCREAMING_SNAKE_CASE       ,							download_mode=__SCREAMING_SNAKE_CASE       ,							verification_mode=__SCREAMING_SNAKE_CASE       ,							base_path=__SCREAMING_SNAKE_CASE       ,							num_proc=self.num_proc       ,							)
																	lowercase_			:  Tuple									= self.builder.as_dataset(
																	    split=self.split       ,							verification_mode=__SCREAMING_SNAKE_CASE       ,							in_memory=self.keep_in_memory	)
											return dataset







class 				lowerCAmelCase__     :






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE = None       ,							__SCREAMING_SNAKE_CASE = None       ,							**__SCREAMING_SNAKE_CASE       ,							):


											"""simple docstring"""





											if num_proc is not None and num_proc <= 0:
																	raise ValueError(F'''num_proc {num_proc} must be an integer > 0.'''	)

											lowercase_			:  List[Any]									= dataset
											lowercase_			:  Dict									= path_or_buf
											lowercase_			:  int									= batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
											lowercase_			:  int									= num_proc
											lowercase_			:  Optional[int]									= '''utf-8'''
											lowercase_			:  Optional[Any]									= to_csv_kwargs






					def 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  List[str]									= self.to_csv_kwargs.pop('''path_or_buf'''       ,							__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Optional[Any]									= self.to_csv_kwargs.pop('''header'''       ,							__SCREAMING_SNAKE_CASE	)
											lowercase_			:  Dict									= self.to_csv_kwargs.pop('''index'''       ,							__SCREAMING_SNAKE_CASE	)

											if isinstance(self.path_or_buf       ,							(str, bytes, os.PathLike)	):
																	with open(self.path_or_buf       ,							'''wb+'''	) as buffer:
																							lowercase_			:  Any									= self._write(file_obj=__SCREAMING_SNAKE_CASE       ,							header=__SCREAMING_SNAKE_CASE       ,							index=__SCREAMING_SNAKE_CASE       ,							**self.to_csv_kwargs	)
											else:
																	lowercase_			:  str									= self._write(file_obj=self.path_or_buf       ,							header=__SCREAMING_SNAKE_CASE       ,							index=__SCREAMING_SNAKE_CASE       ,							**self.to_csv_kwargs	)
											return written






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											lowercase_		,				lowercase_		,				lowercase_		,				lowercase_			:  int									= args

											lowercase_			:  List[str]									= query_table(
											    table=self.dataset.data       ,							key=slice(__SCREAMING_SNAKE_CASE       ,							offset + self.batch_size	)       ,							indices=self.dataset._indices       ,							)
											lowercase_			:  Dict									= batch.to_pandas().to_csv(
											    path_or_buf=__SCREAMING_SNAKE_CASE       ,							header=header if (offset == 0) else False       ,							index=__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)
											return csv_str.encode(self.encoding	)






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											lowercase_			:  Dict									= 0

											if self.num_proc is None or self.num_proc == 1:
																	for offset in logging.tqdm(
																	    range(0       ,							len(self.dataset	)       ,							self.batch_size	)       ,							unit='''ba'''       ,							disable=not logging.is_progress_bar_enabled()       ,							desc='''Creating CSV from Arrow format'''       ,							):
																							lowercase_			:  Any									= self._batch_csv((offset, header, index, to_csv_kwargs)	)
																							written += file_obj.write(__SCREAMING_SNAKE_CASE	)

											else:
																	lowercase_		,				lowercase_			:  Union[str, Any]									= len(self.dataset	), self.batch_size
																	with multiprocessing.Pool(self.num_proc	) as pool:
																							for csv_str in logging.tqdm(
																							    pool.imap(
																							        self._batch_csv       ,							[(offset, header, index, to_csv_kwargs) for offset in range(0       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	)]       ,							)       ,							total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size       ,							unit='''ba'''       ,							disable=not logging.is_progress_bar_enabled()       ,							desc='''Creating CSV from Arrow format'''       ,							):
																													written += file_obj.write(__SCREAMING_SNAKE_CASE	)

											return written



