

import gc
import unittest

import numpy as np
import torch
import torch.nn.functional as F
from transformers import (
    ClapTextConfig,
    ClapTextModelWithProjection,
    RobertaTokenizer,
    SpeechTaHifiGan,
    SpeechTaHifiGanConfig,
)

from diffusers import (
    AudioLDMPipeline,
    AutoencoderKL,
    DDIMScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_AUDIO_BATCH_PARAMS, TEXT_TO_AUDIO_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 							lowerCAmelCase__  (	__UpperCAmelCase					,							unittest.TestCase						):
					__A     :    Any           =						AudioLDMPipeline
					__A     :    Tuple           =						TEXT_TO_AUDIO_PARAMS
					__A     :    str           =						TEXT_TO_AUDIO_BATCH_PARAMS
					__A     :    Any           =						frozenset(
					    [
					        'num_inference_steps',
					        'num_waveforms_per_prompt',
					        'generator',
					        'latents',
					        'output_type',
					        'return_dict',
					        'callback',
					        'callback_steps',
					    ]						)







					def  _lowercase			(       self      :     Any):
											torch.manual_seed(0)
											A__	:							List[str]						    = UNetaDConditionModel(
											    block_out_channels=(32, 64)			,		layers_per_block=2			,		sample_size=32			,		in_channels=4			,		out_channels=4			,		down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")			,		up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")			,		cross_attention_dim=(32, 64)			,		class_embed_type="simple_projection"			,		projection_class_embeddings_input_dim=32			,		class_embeddings_concat=__SCREAMING_SNAKE_CASE			,		)
											A__	:							List[Any]						    = DDIMScheduler(
											    beta_start=0.0_00_85			,		beta_end=0.0_12			,		beta_schedule="scaled_linear"			,		clip_sample=__SCREAMING_SNAKE_CASE			,		set_alpha_to_one=__SCREAMING_SNAKE_CASE			,		)
											torch.manual_seed(0)
											A__	:							Tuple						    = AutoencoderKL(
											    block_out_channels=[32, 64]			,		in_channels=1			,		out_channels=1			,		down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]			,		up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]			,		latent_channels=4			,		)
											torch.manual_seed(0)
											A__	:							Union[str, Any]						    = ClapTextConfig(
											    bos_token_id=0			,		eos_token_id=2			,		hidden_size=32			,		intermediate_size=37			,		layer_norm_eps=1e-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		pad_token_id=1			,		vocab_size=1000			,		projection_dim=32			,		)
											A__	:							Any						    = ClapTextModelWithProjection(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = RobertaTokenizer.from_pretrained("hf-internal-testing/tiny-random-roberta"			,		model_max_length=77)

											A__	:							str						    = SpeechTaHifiGanConfig(
											    model_in_dim=8			,		sampling_rate=1_6000			,		upsample_initial_channel=16			,		upsample_rates=[2, 2]			,		upsample_kernel_sizes=[4, 4]			,		resblock_kernel_sizes=[3, 7]			,		resblock_dilation_sizes=[[1, 3, 5], [1, 3, 5]]			,		normalize_before=__SCREAMING_SNAKE_CASE			,		)

											A__	:							Union[str, Any]						    = SpeechTaHifiGan(__SCREAMING_SNAKE_CASE)

											A__	:							List[Any]						    = {
											    "unet": unet,
											    "scheduler": scheduler,
											    "vae": vae,
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											    "vocoder": vocoder,
											}
											return components







					def  _lowercase			(       self      :     Tuple			,		_A      :     Optional[Any]			,		_A      :     str=0):

											if str(__SCREAMING_SNAKE_CASE).startswith("mps"):
																	A__	:							Dict						    = torch.manual_seed(__SCREAMING_SNAKE_CASE)
											else:
																	A__	:							int						    = torch.Generator(device=__SCREAMING_SNAKE_CASE).manual_seed(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = {
											    "prompt": "A hammer hitting a wooden surface",
											    "generator": generator,
											    "num_inference_steps": 2,
											    "guidance_scale": 6.0,
											}
											return inputs







					def  _lowercase			(       self      :     Optional[int]):
											A__	:							Optional[int]						    = "cpu"  # ensure determinism for the device-dependent torch.Generator

											A__	:							Union[str, Any]						    = self.get_dummy_components()
											A__	:							Any						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							Union[str, Any]						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							List[Any]						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							int						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE)
											A__	:							Tuple						    = output.audios[0]

											assert audio.ndim == 1
											assert len(__SCREAMING_SNAKE_CASE) == 256

											A__	:							List[str]						    = audio[:10]
											A__	:							Tuple						    = np.array(
											    [-0.00_50, 0.00_50, -0.00_60, 0.00_33, -0.00_26, 0.00_33, -0.00_27, 0.00_33, -0.00_28, 0.00_33])

											assert np.abs(audio_slice - expected_slice).max() < 1e-2







					def  _lowercase			(       self      :     Union[str, Any]):
											A__	:							List[Any]						    = self.get_dummy_components()
											A__	:							List[str]						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							List[str]						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											A__	:							Union[str, Any]						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							Any						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = 3 * [inputs["prompt"]]

											# forward
											A__	:							Tuple						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE)
											A__	:							Tuple						    = output.audios[0]

											A__	:							int						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[int]						    = 3 * [inputs.pop("prompt")]

											A__	:							Union[str, Any]						    = audioldm_pipe.tokenizer(
											    __SCREAMING_SNAKE_CASE			,		padding="max_length"			,		max_length=audioldm_pipe.tokenizer.model_max_length			,		truncation=__SCREAMING_SNAKE_CASE			,		return_tensors="pt"			,		)
											A__	:							int						    = text_inputs["input_ids"].to(__SCREAMING_SNAKE_CASE)

											A__	:							Optional[Any]						    = audioldm_pipe.text_encoder(
											    __SCREAMING_SNAKE_CASE			,		)
											A__	:							Dict						    = prompt_embeds.text_embeds
											# additional L_2 normalization over each hidden-state
											A__	:							Any						    = F.normalize(__SCREAMING_SNAKE_CASE			,		dim=-1)

											A__	:							str						    = prompt_embeds

											# forward
											A__	:							int						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE)
											A__	:							str						    = output.audios[0]

											assert np.abs(audio_a - audio_a).max() < 1e-2







					def  _lowercase			(       self      :     Optional[int]):
											A__	:							Optional[Any]						    = self.get_dummy_components()
											A__	:							List[str]						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											A__	:							Any						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							str						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = 3 * ["this is a negative prompt"]
											A__	:							Optional[int]						    = negative_prompt
											A__	:							str						    = 3 * [inputs["prompt"]]

											# forward
											A__	:							Optional[int]						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE)
											A__	:							Dict						    = output.audios[0]

											A__	:							List[Any]						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[int]						    = 3 * [inputs.pop("prompt")]

											A__	:							Dict						    = []
											for p in [prompt, negative_prompt]:
																	A__	:							Dict						    = audioldm_pipe.tokenizer(
																	    __SCREAMING_SNAKE_CASE			,		padding="max_length"			,		max_length=audioldm_pipe.tokenizer.model_max_length			,		truncation=__SCREAMING_SNAKE_CASE			,		return_tensors="pt"			,		)
																	A__	:							int						    = text_inputs["input_ids"].to(__SCREAMING_SNAKE_CASE)

																	A__	:							List[str]						    = audioldm_pipe.text_encoder(
																	    __SCREAMING_SNAKE_CASE			,		)
																	A__	:							Optional[Any]						    = text_embeds.text_embeds
																	# additional L_2 normalization over each hidden-state
																	A__	:							Union[str, Any]						    = F.normalize(__SCREAMING_SNAKE_CASE			,		dim=-1)

																	embeds.append(__SCREAMING_SNAKE_CASE)

											A__    ,       A__	:							Optional[Any]						    = embeds

											# forward
											A__	:							Any						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE)
											A__	:							int						    = output.audios[0]

											assert np.abs(audio_a - audio_a).max() < 1e-2







					def  _lowercase			(       self      :     List[str]):
											A__	:							List[str]						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
											A__	:							Optional[int]						    = self.get_dummy_components()
											A__	:							List[Any]						    = PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							Dict						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							List[Any]						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Tuple						    = "egg cracking"
											A__	:							Any						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE			,		negative_prompt=__SCREAMING_SNAKE_CASE)
											A__	:							List[Any]						    = output.audios[0]

											assert audio.ndim == 1
											assert len(__SCREAMING_SNAKE_CASE) == 256

											A__	:							Dict						    = audio[:10]
											A__	:							List[str]						    = np.array(
											    [-0.00_51, 0.00_50, -0.00_60, 0.00_34, -0.00_26, 0.00_33, -0.00_27, 0.00_33, -0.00_28, 0.00_32])

											assert np.abs(audio_slice - expected_slice).max() < 1e-2







					def  _lowercase			(       self      :     List[str]):
											A__	:							Tuple						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
											A__	:							List[str]						    = self.get_dummy_components()
											A__	:							Optional[int]						    = PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE)
											A__	:							Optional[int]						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							Any						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							List[str]						    = "A hammer hitting a wooden surface"

											# test num_waveforms_per_prompt=1 (default)
											A__	:							Union[str, Any]						    = audioldm_pipe(__SCREAMING_SNAKE_CASE			,		num_inference_steps=2).audios

											assert audios.shape == (1, 256)

											# test num_waveforms_per_prompt=1 (default) for batch of prompts
											A__	:							Union[str, Any]						    = 2
											A__	:							int						    = audioldm_pipe([prompt] * batch_size			,		num_inference_steps=2).audios

											assert audios.shape == (batch_size, 256)

											# test num_waveforms_per_prompt for single prompt
											A__	:							Union[str, Any]						    = 2
											A__	:							int						    = audioldm_pipe(__SCREAMING_SNAKE_CASE			,		num_inference_steps=2			,		num_waveforms_per_prompt=__SCREAMING_SNAKE_CASE).audios

											assert audios.shape == (num_waveforms_per_prompt, 256)

											# test num_waveforms_per_prompt for batch of prompts
											A__	:							Union[str, Any]						    = 2
											A__	:							int						    = audioldm_pipe(
											    [prompt] * batch_size			,		num_inference_steps=2			,		num_waveforms_per_prompt=__SCREAMING_SNAKE_CASE).audios

											assert audios.shape == (batch_size * num_waveforms_per_prompt, 256)







					def  _lowercase			(       self      :     Tuple):
											A__	:							Union[str, Any]						    = "cpu"  # ensure determinism for the device-dependent torch.Generator
											A__	:							Dict						    = self.get_dummy_components()
											A__	:							Dict						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							Any						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)
											A__	:							Optional[Any]						    = audioldm_pipe.vocoder.config.sampling_rate

											A__	:							Optional[int]						    = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Optional[int]						    = audioldm_pipe(audio_length_in_s=0.0_16			,		**__SCREAMING_SNAKE_CASE)
											A__	:							Any						    = output.audios[0]

											assert audio.ndim == 1
											assert len(__SCREAMING_SNAKE_CASE) / vocoder_sampling_rate == 0.0_16

											A__	:							str						    = audioldm_pipe(audio_length_in_s=0.0_32			,		**__SCREAMING_SNAKE_CASE)
											A__	:							Dict						    = output.audios[0]

											assert audio.ndim == 1
											assert len(__SCREAMING_SNAKE_CASE) / vocoder_sampling_rate == 0.0_32







					def  _lowercase			(       self      :     Union[str, Any]):
											A__	:							Any						    = self.get_dummy_components()
											A__	:							str						    = AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
											A__	:							List[Any]						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							int						    = ["hey"]

											A__	:							str						    = audioldm_pipe(__SCREAMING_SNAKE_CASE			,		num_inference_steps=1)
											A__	:							List[Any]						    = output.audios.shape
											assert audio_shape == (1, 256)

											A__	:							Optional[int]						    = audioldm_pipe.vocoder.config
											config.model_in_dim *= 2
											A__	:							int						    = SpeechTaHifiGan(__SCREAMING_SNAKE_CASE).to(__SCREAMING_SNAKE_CASE)
											A__	:							List[str]						    = audioldm_pipe(__SCREAMING_SNAKE_CASE			,		num_inference_steps=1)
											A__	:							str						    = output.audios.shape
											# waveform shape is unchanged, we just have 2x the number of mel channels in the spectrogram
											assert audio_shape == (1, 256)







					def  _lowercase			(       self      :     Optional[Any]):
											self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE)







					def  _lowercase			(       self      :     str):
											self._test_inference_batch_single_identical(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE)







					@unittest.skipIf(
					    torch_device != "cuda" or not is_xformers_available()			,		reason="XFormers attention is only available with CUDA and `xformers` installed"			,		)
					def  _lowercase			(       self      :     Any):
											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE)




@slow
class 							lowerCAmelCase__  (	unittest.TestCase						):







					def  _lowercase			(       self      :     List[str]):
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()







					def  _lowercase			(       self      :     List[Any]			,		_A      :     Optional[int]			,		_A      :     Optional[int]="cpu"			,		_A      :     List[Any]=torch.floataa			,		_A      :     Optional[Any]=0):
											A__	:							Tuple						    = torch.Generator(device=__SCREAMING_SNAKE_CASE).manual_seed(__SCREAMING_SNAKE_CASE)
											A__	:							Tuple						    = np.random.RandomState(__SCREAMING_SNAKE_CASE).standard_normal((1, 8, 128, 16))
											A__	:							Any						    = torch.from_numpy(__SCREAMING_SNAKE_CASE).to(device=__SCREAMING_SNAKE_CASE			,		dtype=__SCREAMING_SNAKE_CASE)
											A__	:							Tuple						    = {
											    "prompt": "A hammer hitting a wooden surface",
											    "latents": latents,
											    "generator": generator,
											    "num_inference_steps": 3,
											    "guidance_scale": 2.5,
											}
											return inputs







					def  _lowercase			(       self      :     Optional[Any]):
											A__	:							Optional[int]						    = AudioLDMPipeline.from_pretrained("cvssp/audioldm")
											A__	:							Tuple						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							List[str]						    = self.get_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							List[str]						    = 25
											A__	:							Any						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE).audios[0]

											assert audio.ndim == 1
											assert len(__SCREAMING_SNAKE_CASE) == 8_1920

											A__	:							List[str]						    = audio[7_7230:7_7240]
											A__	:							Any						    = np.array(
											    [-0.48_84, -0.46_07, 0.00_23, 0.50_07, 0.58_96, 0.51_51, 0.38_13, -0.02_08, -0.36_87, -0.43_15])
											A__	:							List[Any]						    = np.abs(expected_slice - audio_slice).max()
											assert max_diff < 1e-2







					def  _lowercase			(       self      :     Union[str, Any]):
											A__	:							int						    = AudioLDMPipeline.from_pretrained("cvssp/audioldm")
											A__	:							Optional[int]						    = LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config)
											A__	:							List[str]						    = audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

											A__	:							Dict						    = self.get_inputs(__SCREAMING_SNAKE_CASE)
											A__	:							Tuple						    = audioldm_pipe(**__SCREAMING_SNAKE_CASE).audios[0]

											assert audio.ndim == 1
											assert len(__SCREAMING_SNAKE_CASE) == 8_1920

											A__	:							Optional[int]						    = audio[2_7780:2_7790]
											A__	:							str						    = np.array([-0.21_31, -0.08_73, -0.01_24, -0.01_89, 0.05_69, 0.13_73, 0.18_83, 0.28_86, 0.32_97, 0.22_12])
											A__	:							Union[str, Any]						    = np.abs(expected_slice - audio_slice).max()
											assert max_diff < 3e-2




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

import fsspec

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




class 							lowerCAmelCase__  (	UpperCamelCase						):







					def __init__(       self      :     Dict			,		_A      :     NestedDataStructureLike[PathLike]			,		_A      :     Optional[NamedSplit] = None			,		_A      :     Optional[Features] = None			,		_A      :     str = None			,		_A      :     bool = False			,		_A      :     bool = False			,		_A      :     Optional[str] = None			,		_A      :     Optional[int] = None			,		**_A      :     Dict			,		):
											super().__init__(
											    _A			,		split=_A			,		features=_A			,		cache_dir=_A			,		keep_in_memory=_A			,		streaming=_A			,		num_proc=_A			,		**_A			,		)
											A__	:							Optional[int]						    = field
											A__	:							Dict						    = path_or_paths if isinstance(_A			,		_A) else {self.split: path_or_paths}
											A__	:							int						    = Json(
											    cache_dir=_A			,		data_files=_A			,		features=_A			,		field=_A			,		**_A			,		)







					def  _lowercase			(       self      :     Dict):
											# Build iterable dataset
											if self.streaming:
																	A__	:							Union[str, Any]						    = self.builder.as_streaming_dataset(split=self.split)
											# Build regular (map-style) dataset
											else:
																	A__	:							Optional[Any]						    = None
																	A__	:							Any						    = None
																	A__	:							Tuple						    = None
																	A__	:							Tuple						    = None

																	self.builder.download_and_prepare(
																	    download_config=_A			,		download_mode=_A			,		verification_mode=_A			,		base_path=_A			,		num_proc=self.num_proc			,		)
																	A__	:							Any						    = self.builder.as_dataset(
																	    split=self.split			,		verification_mode=_A			,		in_memory=self.keep_in_memory)
											return dataset




class 							lowerCAmelCase__  :







					def __init__(       self      :     Union[str, Any]			,		_A      :     Dataset			,		_A      :     Union[PathLike, BinaryIO]			,		_A      :     Optional[int] = None			,		_A      :     Optional[int] = None			,		**_A      :     Any			,		):
											if num_proc is not None and num_proc <= 0:
																	raise ValueError(F'num_proc {num_proc} must be an integer > 0.')

											A__	:							Tuple						    = dataset
											A__	:							Tuple						    = path_or_buf
											A__	:							int						    = batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
											A__	:							Tuple						    = num_proc
											A__	:							Any						    = "utf-8"
											A__	:							Optional[int]						    = to_json_kwargs







					def  _lowercase			(       self      :     Union[str, Any]):
											A__	:							Any						    = self.to_json_kwargs.pop("path_or_buf"			,		_A)
											A__	:							Union[str, Any]						    = self.to_json_kwargs.pop("orient"			,		"records")
											A__	:							List[Any]						    = self.to_json_kwargs.pop("lines"			,		True if orient == "records" else False)
											A__	:							List[Any]						    = self.to_json_kwargs.pop("index"			,		False if orient in ["split", "table"] else True)
											A__	:							int						    = self.to_json_kwargs.pop("compression"			,		_A)

											if compression not in [None, "infer", "gzip", "bz2", "xz"]:
																	raise NotImplementedError(F'`datasets` currently does not support {compression} compression')

											if isinstance(self.path_or_buf			,		(str, bytes, os.PathLike)):
																	with fsspec.open(self.path_or_buf			,		"wb"			,		compression=_A) as buffer:
																							A__	:							int						    = self._write(file_obj=_A			,		orient=_A			,		lines=_A			,		index=_A			,		**self.to_json_kwargs)
											else:
																	if compression:
																							raise NotImplementedError(
																							    F'The compression parameter is not supported when writing to a buffer, but compression={compression}'
																							    " was passed. Please provide a local path instead.")
																	A__	:							Optional[Any]						    = self._write(
																	    file_obj=self.path_or_buf			,		orient=_A			,		lines=_A			,		index=_A			,		**self.to_json_kwargs)
											return written







					def  _lowercase			(       self      :     List[Any]			,		_A      :     int):
											A__    ,       A__    ,       A__    ,       A__    ,       A__	:							Union[str, Any]						    = args

											A__	:							Dict						    = query_table(
											    table=self.dataset.data			,		key=slice(_A			,		offset + self.batch_size)			,		indices=self.dataset._indices			,		)
											A__	:							Dict						    = batch.to_pandas().to_json(
											    path_or_buf=_A			,		orient=_A			,		lines=_A			,		index=_A			,		**_A)
											if not json_str.endswith("\n"):
																	json_str += "\n"
											return json_str.encode(self.encoding)







					def  _lowercase			(       self      :     List[str]			,		_A      :     BinaryIO			,		_A      :     str			,		_A      :     List[Any]			,		_A      :     Optional[Any]			,		**_A      :     Optional[Any]			,		):
											A__	:							Any						    = 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 json from Arrow format"			,		):
																							A__	:							List[str]						    = self._batch_json((offset, orient, lines, index, to_json_kwargs))
																							written += file_obj.write(_A)
											else:
																	A__    ,       A__	:							Union[str, Any]						    = len(self.dataset), self.batch_size
																	with multiprocessing.Pool(self.num_proc) as pool:
																							for json_str in logging.tqdm(
																							    pool.imap(
																							        self._batch_json			,		[(offset, orient, lines, index, to_json_kwargs) for offset in range(0			,		_A			,		_A)]			,		)			,		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 json from Arrow format"			,		):
																													written += file_obj.write(_A)

											return written