



'''simple docstring'''



import io
import json

import fsspec
import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.json import JsonDatasetReader, JsonDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases



def   __lowerCAmelCase					(							lowerCamelCase   :						str				,		lowerCamelCase   :						Union[str, Any]					):






	'''simple docstring'''




	assert isinstance(lowerCamelCase				,		lowerCamelCase					)
	assert dataset.num_rows == 4
	assert dataset.num_columns == 3
	assert dataset.column_names == ["col_1", "col_2", "col_3"]
	for feature, expected_dtype in expected_features.items():
		assert dataset.features[feature].dtype == expected_dtype



@pytest.mark.parametrize("keep_in_memory"				,		[False, True]					)
def   __lowerCAmelCase					(							lowerCamelCase   :						Tuple				,		lowerCamelCase   :						int				,		lowerCamelCase   :						Optional[int]					):






	'''simple docstring'''




	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
		__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		cache_dir=lowerCamelCase				,		keep_in_memory=lowerCamelCase					).read()
	_check_json_dataset(lowerCamelCase				,		lowerCamelCase					)



@pytest.mark.parametrize(
    "features"				,		[
        None,
        {"col_1": "string", "col_2": "int64", "col_3": "float64"},
        {"col_1": "string", "col_2": "string", "col_3": "string"},
        {"col_1": "int32", "col_2": "int32", "col_3": "int32"},
        {"col_1": "float32", "col_2": "float32", "col_3": "float32"},
    ]				,		)
def   __lowerCAmelCase					(							lowerCamelCase   :						List[Any]				,		lowerCamelCase   :						int				,		lowerCamelCase   :						Dict					):






	'''simple docstring'''




	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	__lowerCAmelCase												=  features.copy() if features else default_expected_features
	__lowerCAmelCase												=  (
	    Features({feature: Value(lowerCamelCase					) for feature, dtype in features.items()}					) if features is not None else None
	)
	__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		features=lowerCamelCase				,		cache_dir=lowerCamelCase					).read()
	_check_json_dataset(lowerCamelCase				,		lowerCamelCase					)



@pytest.mark.parametrize(
    "features"				,		[
        None,
        {"col_3": "float64", "col_1": "string", "col_2": "int64"},
    ]				,		)
def   __lowerCAmelCase					(							lowerCamelCase   :						List[Any]				,		lowerCamelCase   :						Optional[int]				,		lowerCamelCase   :						Any					):






	'''simple docstring'''




	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_3": "float64", "col_1": "string", "col_2": "int64"}
	__lowerCAmelCase												=  features.copy() if features else default_expected_features
	__lowerCAmelCase												=  (
	    Features({feature: Value(lowerCamelCase					) for feature, dtype in features.items()}					) if features is not None else None
	)
	__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		features=lowerCamelCase				,		cache_dir=lowerCamelCase					).read()
	assert isinstance(lowerCamelCase				,		lowerCamelCase					)
	assert dataset.num_rows == 2
	assert dataset.num_columns == 3
	assert dataset.column_names == ["col_3", "col_1", "col_2"]
	for feature, expected_dtype in expected_features.items():
		assert dataset.features[feature].dtype == expected_dtype



def   __lowerCAmelCase					(							lowerCamelCase   :						Any				,		lowerCamelCase   :						Optional[int]					):






	'''simple docstring'''




	__lowerCAmelCase												=  {"col_2": "int64", "col_3": "float64", "col_1": "string"}
	__lowerCAmelCase												=  features.copy()
	__lowerCAmelCase												=  (
	    Features({feature: Value(lowerCamelCase					) for feature, dtype in features.items()}					) if features is not None else None
	)
	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		features=lowerCamelCase				,		cache_dir=lowerCamelCase					).read()
	assert isinstance(lowerCamelCase				,		lowerCamelCase					)
	assert dataset.num_rows == 2
	assert dataset.num_columns == 3
	assert dataset.column_names == ["col_2", "col_3", "col_1"]
	for feature, expected_dtype in expected_features.items():
		assert dataset.features[feature].dtype == expected_dtype



@pytest.mark.parametrize("split"				,		[None, NamedSplit("train"					), "train", "test"]					)
def   __lowerCAmelCase					(							lowerCamelCase   :						List[Any]				,		lowerCamelCase   :						Optional[Any]				,		lowerCamelCase   :						Union[str, Any]					):






	'''simple docstring'''




	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		cache_dir=lowerCamelCase				,		split=lowerCamelCase					).read()
	_check_json_dataset(lowerCamelCase				,		lowerCamelCase					)
	assert dataset.split == split if split else "train"



@pytest.mark.parametrize("path_type"				,		[str, list]					)
def   __lowerCAmelCase					(							lowerCamelCase   :						Any				,		lowerCamelCase   :						List[str]				,		lowerCamelCase   :						Optional[int]					):






	'''simple docstring'''




	if issubclass(lowerCamelCase				,		lowerCamelCase					):
		__lowerCAmelCase												=  jsonl_path
	elif issubclass(lowerCamelCase				,		lowerCamelCase					):
		__lowerCAmelCase												=  [jsonl_path]
	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		cache_dir=lowerCamelCase					).read()
	_check_json_dataset(lowerCamelCase				,		lowerCamelCase					)



def   __lowerCAmelCase					(							lowerCamelCase   :						Optional[int]				,		lowerCamelCase   :						int				,		lowerCamelCase   :						str=("train",)					):






	'''simple docstring'''




	assert isinstance(lowerCamelCase				,		lowerCamelCase					)
	for split in splits:
		__lowerCAmelCase												=  dataset_dict[split]
		assert dataset.num_rows == 4
		assert dataset.num_columns == 3
		assert dataset.column_names == ["col_1", "col_2", "col_3"]
		for feature, expected_dtype in expected_features.items():
			assert dataset.features[feature].dtype == expected_dtype



@pytest.mark.parametrize("keep_in_memory"				,		[False, True]					)
def   __lowerCAmelCase					(							lowerCamelCase   :						Dict				,		lowerCamelCase   :						int				,		lowerCamelCase   :						List[str]					):






	'''simple docstring'''




	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
		__lowerCAmelCase												=  JsonDatasetReader({"train": jsonl_path}				,		cache_dir=lowerCamelCase				,		keep_in_memory=lowerCamelCase					).read()
	_check_json_datasetdict(lowerCamelCase				,		lowerCamelCase					)



@pytest.mark.parametrize(
    "features"				,		[
        None,
        {"col_1": "string", "col_2": "int64", "col_3": "float64"},
        {"col_1": "string", "col_2": "string", "col_3": "string"},
        {"col_1": "int32", "col_2": "int32", "col_3": "int32"},
        {"col_1": "float32", "col_2": "float32", "col_3": "float32"},
    ]				,		)
def   __lowerCAmelCase					(							lowerCamelCase   :						int				,		lowerCamelCase   :						Any				,		lowerCamelCase   :						Dict					):






	'''simple docstring'''




	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	__lowerCAmelCase												=  features.copy() if features else default_expected_features
	__lowerCAmelCase												=  (
	    Features({feature: Value(lowerCamelCase					) for feature, dtype in features.items()}					) if features is not None else None
	)
	__lowerCAmelCase												=  JsonDatasetReader({"train": jsonl_path}				,		features=lowerCamelCase				,		cache_dir=lowerCamelCase					).read()
	_check_json_datasetdict(lowerCamelCase				,		lowerCamelCase					)



@pytest.mark.parametrize("split"				,		[None, NamedSplit("train"					), "train", "test"]					)
def   __lowerCAmelCase					(							lowerCamelCase   :						int				,		lowerCamelCase   :						Optional[int]				,		lowerCamelCase   :						int					):






	'''simple docstring'''




	if split:
		__lowerCAmelCase												=  {split: jsonl_path}
	else:
		__lowerCAmelCase												=  "train"
		__lowerCAmelCase												=  {"train": jsonl_path, "test": jsonl_path}
	__lowerCAmelCase												=  tmp_path / "cache"
	__lowerCAmelCase												=  {"col_1": "string", "col_2": "int64", "col_3": "float64"}
	__lowerCAmelCase												=  JsonDatasetReader(lowerCamelCase				,		cache_dir=lowerCamelCase					).read()
	_check_json_datasetdict(lowerCamelCase				,		lowerCamelCase				,		splits=list(path.keys()					)					)
	assert all(dataset[split].split == split for split in path.keys()					)



def   __lowerCAmelCase					(							lowerCamelCase   :						Optional[Any]					):






	'''simple docstring'''




	return json.load(lowerCamelCase					)



def   __lowerCAmelCase					(							lowerCamelCase   :						Tuple					):






	'''simple docstring'''




	return [json.loads(lowerCamelCase					) for line in buffer]






class 					UpperCAmelCase__	:



					@pytest.mark.parametrize("lines, load_json_function"			,				[(True, load_json_lines), (False, load_json)]     )
					def        UpperCAmelCase_ (    self			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase     )     ->			int:
						with io.BytesIO() as buffer:
							JsonDatasetWriter(UpperCamelCase			,				UpperCamelCase			,				lines=UpperCamelCase     ).write()
							buffer.seek(0     )
							__lowerCAmelCase												=  load_json_function(UpperCamelCase     )
						assert isinstance(UpperCamelCase			,				UpperCamelCase     )
						assert isinstance(exported_content[0]			,				UpperCamelCase     )
						assert len(UpperCamelCase     ) == 10



					@pytest.mark.parametrize(
					    "orient, container, keys, len_at"			,				[
					        ("records", list, {"tokens", "labels", "answers", "id"}, None),
					        ("split", dict, {"columns", "data"}, "data"),
					        ("index", dict, set("0123456789"     ), None),
					        ("columns", dict, {"tokens", "labels", "answers", "id"}, "tokens"),
					        ("values", list, None, None),
					        ("table", dict, {"schema", "data"}, "data"),
					    ]			,				)
					def        UpperCAmelCase_ (    self			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase     )     ->			int:
						with io.BytesIO() as buffer:
							JsonDatasetWriter(UpperCamelCase			,				UpperCamelCase			,				lines=UpperCamelCase			,				orient=UpperCamelCase     ).write()
							buffer.seek(0     )
							__lowerCAmelCase												=  load_json(UpperCamelCase     )
						assert isinstance(UpperCamelCase			,				UpperCamelCase     )
						if keys:
							if container is dict:
								assert exported_content.keys() == keys
							else:
								assert exported_content[0].keys() == keys
						else:
							assert not hasattr(UpperCamelCase			,				"keys"     ) and not hasattr(exported_content[0]			,				"keys"     )
						if len_at:
							assert len(exported_content[len_at]     ) == 10
						else:
							assert len(UpperCamelCase     ) == 10



					@pytest.mark.parametrize("lines, load_json_function"			,				[(True, load_json_lines), (False, load_json)]     )
					def        UpperCAmelCase_ (    self			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase     )     ->			str:
						with io.BytesIO() as buffer:
							JsonDatasetWriter(UpperCamelCase			,				UpperCamelCase			,				lines=UpperCamelCase			,				num_proc=2     ).write()
							buffer.seek(0     )
							__lowerCAmelCase												=  load_json_function(UpperCamelCase     )
						assert isinstance(UpperCamelCase			,				UpperCamelCase     )
						assert isinstance(exported_content[0]			,				UpperCamelCase     )
						assert len(UpperCamelCase     ) == 10



					@pytest.mark.parametrize(
					    "orient, container, keys, len_at"			,				[
					        ("records", list, {"tokens", "labels", "answers", "id"}, None),
					        ("split", dict, {"columns", "data"}, "data"),
					        ("index", dict, set("0123456789"     ), None),
					        ("columns", dict, {"tokens", "labels", "answers", "id"}, "tokens"),
					        ("values", list, None, None),
					        ("table", dict, {"schema", "data"}, "data"),
					    ]			,				)
					def        UpperCAmelCase_ (    self			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase     )     ->			Union[str, Any]:
						with io.BytesIO() as buffer:
							JsonDatasetWriter(UpperCamelCase			,				UpperCamelCase			,				lines=UpperCamelCase			,				orient=UpperCamelCase			,				num_proc=2     ).write()
							buffer.seek(0     )
							__lowerCAmelCase												=  load_json(UpperCamelCase     )
						assert isinstance(UpperCamelCase			,				UpperCamelCase     )
						if keys:
							if container is dict:
								assert exported_content.keys() == keys
							else:
								assert exported_content[0].keys() == keys
						else:
							assert not hasattr(UpperCamelCase			,				"keys"     ) and not hasattr(exported_content[0]			,				"keys"     )
						if len_at:
							assert len(exported_content[len_at]     ) == 10
						else:
							assert len(UpperCamelCase     ) == 10



					def        UpperCAmelCase_ (    self			,				UpperCamelCase     )     ->			Any:
						with pytest.raises(UpperCamelCase     ):
							with io.BytesIO() as buffer:
								JsonDatasetWriter(UpperCamelCase			,				UpperCamelCase			,				num_proc=0     )



					@pytest.mark.parametrize("compression, extension"			,				[("gzip", "gz"), ("bz2", "bz2"), ("xz", "xz")]     )
					def        UpperCAmelCase_ (    self			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase			,				UpperCamelCase     )     ->			Tuple:
						__lowerCAmelCase												=  tmp_path_factory.mktemp("data"     ) / F'''test.json.{extension}'''
						__lowerCAmelCase												=  str(shared_datadir / F'''test_file.json.{extension}'''     )
						JsonDatasetWriter(UpperCamelCase			,				UpperCamelCase			,				compression=UpperCamelCase     ).write()

						with fsspec.open(UpperCamelCase			,				"rb"			,				compression="infer"     ) as f:
							__lowerCAmelCase												=  f.read()
						with fsspec.open(UpperCamelCase			,				"rb"			,				compression="infer"     ) as f:
							__lowerCAmelCase												=  f.read()
						assert exported_content == original_content




'''simple docstring'''



import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np



def   __lowerCAmelCase					(							lowerCamelCase   :						bytes				,		lowerCamelCase   :						int					):






	'''simple docstring'''




	__lowerCAmelCase												=  f'''{sampling_rate}'''
	__lowerCAmelCase												=  "1"
	__lowerCAmelCase												=  "f32le"
	__lowerCAmelCase												=  [
	    "ffmpeg",
	    "-i",
	    "pipe:0",
	    "-ac",
	    ac,
	    "-ar",
	    ar,
	    "-f",
	    format_for_conversion,
	    "-hide_banner",
	    "-loglevel",
	    "quiet",
	    "pipe:1",
	]

	try:
		with subprocess.Popen(lowerCamelCase				,		stdin=subprocess.PIPE				,		stdout=subprocess.PIPE					) as ffmpeg_process:
			__lowerCAmelCase												=  ffmpeg_process.communicate(lowerCamelCase					)
	except FileNotFoundError as error:
		raise ValueError("ffmpeg was not found but is required to load audio files from filename"					) from error
	__lowerCAmelCase												=  output_stream[0]
	__lowerCAmelCase												=  np.frombuffer(lowerCamelCase				,		np.floataa					)
	if audio.shape[0] == 0:
		raise ValueError("Malformed soundfile"					)
	return audio



def   __lowerCAmelCase					(							lowerCamelCase   :						int				,		lowerCamelCase   :						float				,		lowerCamelCase   :						str = "f32le"				,		):






	'''simple docstring'''




	__lowerCAmelCase												=  f'''{sampling_rate}'''
	__lowerCAmelCase												=  "1"
	if format_for_conversion == "s16le":
		__lowerCAmelCase												=  2
	elif format_for_conversion == "f32le":
		__lowerCAmelCase												=  4
	else:
		raise ValueError(f'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''					)

	__lowerCAmelCase												=  platform.system()
	if system == "Linux":
		__lowerCAmelCase												=  "alsa"
		__lowerCAmelCase												=  "default"
	elif system == "Darwin":
		__lowerCAmelCase												=  "avfoundation"
		__lowerCAmelCase												=  ":0"
	elif system == "Windows":
		__lowerCAmelCase												=  "dshow"
		__lowerCAmelCase												=  "default"

	__lowerCAmelCase												=  [
	    "ffmpeg",
	    "-f",
	    format_,
	    "-i",
	    input_,
	    "-ac",
	    ac,
	    "-ar",
	    ar,
	    "-f",
	    format_for_conversion,
	    "-fflags",
	    "nobuffer",
	    "-hide_banner",
	    "-loglevel",
	    "quiet",
	    "pipe:1",
	]
	__lowerCAmelCase												=  int(round(sampling_rate * chunk_length_s					)					) * size_of_sample
	__lowerCAmelCase												=  _ffmpeg_stream(lowerCamelCase				,		lowerCamelCase					)
	for item in iterator:
		yield item



def   __lowerCAmelCase					(							lowerCamelCase   :						int				,		lowerCamelCase   :						float				,		lowerCamelCase   :						Optional[int] = None				,		lowerCamelCase   :						Optional[Union[Tuple[float, float], float]] = None				,		lowerCamelCase   :						str = "f32le"				,		):






	'''simple docstring'''




	if stream_chunk_s is not None:
		__lowerCAmelCase												=  stream_chunk_s
	else:
		__lowerCAmelCase												=  chunk_length_s

	__lowerCAmelCase												=  ffmpeg_microphone(lowerCamelCase				,		lowerCamelCase				,		format_for_conversion=lowerCamelCase					)
	if format_for_conversion == "s16le":
		__lowerCAmelCase												=  np.intaa
		__lowerCAmelCase												=  2
	elif format_for_conversion == "f32le":
		__lowerCAmelCase												=  np.floataa
		__lowerCAmelCase												=  4
	else:
		raise ValueError(f'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''					)

	if stride_length_s is None:
		__lowerCAmelCase												=  chunk_length_s / 6
	__lowerCAmelCase												=  int(round(sampling_rate * chunk_length_s					)					) * size_of_sample
	if isinstance(lowerCamelCase				,		(int, float)					):
		__lowerCAmelCase												=  [stride_length_s, stride_length_s]

	__lowerCAmelCase												=  int(round(sampling_rate * stride_length_s[0]					)					) * size_of_sample
	__lowerCAmelCase												=  int(round(sampling_rate * stride_length_s[1]					)					) * size_of_sample
	__lowerCAmelCase												=  datetime.datetime.now()
	__lowerCAmelCase												=  datetime.timedelta(seconds=lowerCamelCase					)
	for item in chunk_bytes_iter(lowerCamelCase				,		lowerCamelCase				,		stride=(stride_left, stride_right)				,		stream=lowerCamelCase					):
		# Put everything back in numpy scale
		__lowerCAmelCase												=  np.frombuffer(item["raw"]				,		dtype=lowerCamelCase					)
		__lowerCAmelCase												=  (
		    item["stride"][0] // size_of_sample,
		    item["stride"][1] // size_of_sample,
		)
		__lowerCAmelCase												=  sampling_rate
		audio_time += delta
		if datetime.datetime.now() > audio_time + 10 * delta:
			# We're late !! SKIP
			continue
		yield item



def   __lowerCAmelCase					(							lowerCamelCase   :						Optional[Any]				,		lowerCamelCase   :						int				,		lowerCamelCase   :						Tuple[int, int]				,		lowerCamelCase   :						bool = False					):






	'''simple docstring'''




	__lowerCAmelCase												=  B""
	__lowerCAmelCase							,       __lowerCAmelCase												=  stride
	if stride_left + stride_right >= chunk_len:
		raise ValueError(
		    f'''Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}'''					)
	__lowerCAmelCase												=  0
	for raw in iterator:
		acc += raw
		if stream and len(lowerCamelCase					) < chunk_len:
			__lowerCAmelCase												=  (_stride_left, 0)
			yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
		else:
			while len(lowerCamelCase					) >= chunk_len:
				# We are flushing the accumulator
				__lowerCAmelCase												=  (_stride_left, stride_right)
				__lowerCAmelCase												=  {"raw": acc[:chunk_len], "stride": stride}
				if stream:
					__lowerCAmelCase												=  False
				yield item
				__lowerCAmelCase												=  stride_left
				__lowerCAmelCase												=  acc[chunk_len - stride_left - stride_right :]
    # Last chunk
	if len(lowerCamelCase					) > stride_left:
		__lowerCAmelCase												=  {"raw": acc, "stride": (_stride_left, 0)}
		if stream:
			__lowerCAmelCase												=  False
		yield item




def   __lowerCAmelCase					(							lowerCamelCase   :						Optional[int]				,		lowerCamelCase   :						int					):






	'''simple docstring'''




	__lowerCAmelCase												=  2**24  # 16Mo
	try:
		with subprocess.Popen(lowerCamelCase				,		stdout=subprocess.PIPE				,		bufsize=lowerCamelCase					) as ffmpeg_process:
			while True:
				__lowerCAmelCase												=  ffmpeg_process.stdout.read(lowerCamelCase					)
				if raw == b"":
					break
				yield raw
	except FileNotFoundError as error:
		raise ValueError("ffmpeg was not found but is required to stream audio files from filename"					) from error