






import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
	from .features import FeatureType


__lowerCamelCase						,   __lowerCamelCase						,   __lowerCamelCase :      str										=				False, False, False



@dataclass
class 							SCREAMING_SNAKE_CASE__						:





					"""simple docstring"""






					a_     =		None
					a_     =		True
					a_     =		True
					a_     =		None
					# Automatically constructed
					a_     =		"dict"
					a_     =		pa.struct({"bytes": pa.binary(), "path": pa.string()}						)
					a_     =		field(default="Audio"			,       init=UpperCamelCase_			,       repr=UpperCamelCase_						)



					def __call__( self					:  Optional[int]	):
						return self.pa_type



					def 	_lowercase					( self					:  int							,		__A					:  Union[str, bytes, dict]	):
						try:
							import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
						except ImportError as err:
							raise ImportError("To support encoding audio data, please install 'soundfile'."	) from err
						if isinstance(__A							,		__A	):
							return {"bytes": None, "path": value}
						elif isinstance(__A							,		__A	):
							return {"bytes": value, "path": None}
						elif "array" in value:
							# convert the audio array to wav bytes
							snake_case__			:				List[Any]       =  BytesIO()
							sf.write(__A							,		value["array"]							,		value["sampling_rate"]							,		format="wav"	)
							return {"bytes": buffer.getvalue(), "path": None}
						elif value.get("path"	) is not None and os.path.isfile(value["path"]	):
							# we set "bytes": None to not duplicate the data if they're already available locally
							if value["path"].endswith("pcm"	):
								# "PCM" only has raw audio bytes
								if value.get("sampling_rate"	) is None:
									# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
									raise KeyError("To use PCM files, please specify a 'sampling_rate' in Audio object"	)
								if value.get("bytes"	):
									# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
									snake_case__			:				Tuple       =  np.frombuffer(value["bytes"]							,		dtype=np.intaa	).astype(np.floataa	) / 3_2_7_6_7
								else:
									snake_case__			:				Union[str, Any]       =  np.memmap(value["path"]							,		dtype="h"							,		mode="r"	).astype(np.floataa	) / 3_2_7_6_7

								snake_case__			:				Optional[Any]       =  BytesIO(bytes()	)
								sf.write(__A							,		__A							,		value["sampling_rate"]							,		format="wav"	)
								return {"bytes": buffer.getvalue(), "path": None}
							else:
								return {"bytes": None, "path": value.get("path"	)}
						elif value.get("bytes"	) is not None or value.get("path"	) is not None:
							# store the audio bytes, and path is used to infer the audio format using the file extension
							return {"bytes": value.get("bytes"	), "path": value.get("path"	)}
						else:
							raise ValueError(
							    f'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''	)



					def 	_lowercase					( self					:  int							,		__A					:  dict							,		__A					:  Optional[Dict[str, Union[str, bool, None]]] = None	):
						if not self.decode:
							raise RuntimeError("Decoding is disabled for this feature. Please use Audio(decode=True) instead."	)

						snake_case__,						snake_case__			:				str       =  (value["path"], BytesIO(value["bytes"]	)) if value["bytes"] is not None else (value["path"], None)
						if path is None and file is None:
							raise ValueError(f'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'''	)

						try:
							import librosa
							import soundfile as sf
						except ImportError as err:
							raise ImportError("To support decoding audio files, please install 'librosa' and 'soundfile'."	) from err

						snake_case__			:				int       =  xsplitext(__A	)[1][1:].lower() if path is not None else None
						if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
							raise RuntimeError(
							    "Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, "
							    "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "	)
						elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
							raise RuntimeError(
							    "Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, "
							    "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "	)

						if file is None:
							snake_case__			:				Dict       =  token_per_repo_id or {}
							snake_case__			:				Tuple       =  path.split("::"	)[-1]
							try:
								snake_case__			:				str       =  string_to_dict(__A							,		config.HUB_DATASETS_URL	)["repo_id"]
								snake_case__			:				List[Any]       =  token_per_repo_id[repo_id]
							except (ValueError, KeyError):
								snake_case__			:				Any       =  None

							with xopen(__A							,		"rb"							,		use_auth_token=__A	) as f:
								snake_case__,						snake_case__			:				Any       =  sf.read(__A	)

						else:
							snake_case__,						snake_case__			:				Tuple       =  sf.read(__A	)

						snake_case__			:				int       =  array.T
						if self.mono:
							snake_case__			:				Optional[int]       =  librosa.to_mono(__A	)
						if self.sampling_rate and self.sampling_rate != sampling_rate:
							snake_case__			:				Any       =  librosa.resample(__A							,		orig_sr=__A							,		target_sr=self.sampling_rate	)
							snake_case__			:				str       =  self.sampling_rate

						return {"path": path, "array": array, "sampling_rate": sampling_rate}



					def 	_lowercase					( self					:  Union[str, Any]	):
						from .features import Value

						if self.decode:
							raise ValueError("Cannot flatten a decoded Audio feature."	)
						return {
						    "bytes": Value("binary"	),
						    "path": Value("string"	),
						}



					def 	_lowercase					( self					:  Any							,		__A					:  Union[pa.StringArray, pa.StructArray]	):
						if pa.types.is_string(storage.type	):
							snake_case__			:				Optional[int]       =  pa.array([None] * len(__A	)							,		type=pa.binary()	)
							snake_case__			:				Dict       =  pa.StructArray.from_arrays([bytes_array, storage]							,		["bytes", "path"]							,		mask=storage.is_null()	)
						elif pa.types.is_binary(storage.type	):
							snake_case__			:				Any       =  pa.array([None] * len(__A	)							,		type=pa.string()	)
							snake_case__			:				Tuple       =  pa.StructArray.from_arrays([storage, path_array]							,		["bytes", "path"]							,		mask=storage.is_null()	)
						elif pa.types.is_struct(storage.type	) and storage.type.get_all_field_indices("array"	):
							snake_case__			:				Tuple       =  pa.array([Audio().encode_example(__A	) if x is not None else None for x in storage.to_pylist()]	)
						elif pa.types.is_struct(storage.type	):
							if storage.type.get_field_index("bytes"	) >= 0:
								snake_case__			:				List[Any]       =  storage.field("bytes"	)
							else:
								snake_case__			:				Union[str, Any]       =  pa.array([None] * len(__A	)							,		type=pa.binary()	)
							if storage.type.get_field_index("path"	) >= 0:
								snake_case__			:				Tuple       =  storage.field("path"	)
							else:
								snake_case__			:				str       =  pa.array([None] * len(__A	)							,		type=pa.string()	)
							snake_case__			:				Tuple       =  pa.StructArray.from_arrays([bytes_array, path_array]							,		["bytes", "path"]							,		mask=storage.is_null()	)
						return array_cast(__A							,		self.pa_type	)






					def 	_lowercase					( self					:  List[str]							,		__A					:  pa.StructArray	):

						@no_op_if_value_is_null
						def path_to_bytes(__A					:  Union[str, Any]	):
							with xopen(__A							,		"rb"	) as f:
								snake_case__			:				List[str]       =  f.read()
							return bytes_

						snake_case__			:				Optional[Any]       =  pa.array(
						    [
						        (path_to_bytes(x["path"]	) if x["bytes"] is None else x["bytes"]) if x is not None else None
						        for x in storage.to_pylist()
						    ]							,		type=pa.binary()							,		)
						snake_case__			:				Dict       =  pa.array(
						    [os.path.basename(__A	) if path is not None else None for path in storage.field("path"	).to_pylist()]							,		type=pa.string()							,		)
						snake_case__			:				Optional[int]       =  pa.StructArray.from_arrays([bytes_array, path_array]							,		["bytes", "path"]							,		mask=bytes_array.is_null()	)
						return array_cast(__A							,		self.pa_type	)











from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch



@require_torch
@require_tf
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				Optional[Any]       =  SMALL_MODEL_IDENTIFIER
						snake_case__			:				Any       =  "pt"
						snake_case__			:				Any       =  "tf"



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  List[Any]	):
						snake_case__			:				int       =  AutoModel.from_pretrained(self.test_model	)
						model_pt.save_pretrained(__A	)



					def 	_lowercase					( self					:  Optional[int]							,		__A					:  Tuple	):
						snake_case__			:				List[Any]       =  TFAutoModel.from_pretrained(self.test_model							,		from_pt=__A	)
						model_tf.save_pretrained(__A	)



					def 	_lowercase					( self					:  str	):
						snake_case__			:				Optional[Any]       =  "mock_framework"

						# Framework provided - return whatever the user provides
						snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model							,		__A	)
						self.assertEqual(__A							,		__A	)

						# Local checkpoint and framework provided - return provided framework
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				Optional[int]       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				int       =  FeaturesManager.determine_framework(__A							,		__A	)
							self.assertEqual(__A							,		__A	)



					def 	_lowercase					( self					:  Dict	):
						# PyTorch checkpoint
						with TemporaryDirectory() as local_pt_ckpt:
							self._setup_pt_ckpt(__A	)
							snake_case__			:				List[str]       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_pt	)

						# TensorFlow checkpoint
						with TemporaryDirectory() as local_tf_ckpt:
							self._setup_tf_ckpt(__A	)
							snake_case__			:				Tuple       =  FeaturesManager.determine_framework(__A	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Invalid local checkpoint
						with TemporaryDirectory() as local_invalid_ckpt:
							with self.assertRaises(__A	):
								snake_case__			:				int       =  FeaturesManager.determine_framework(__A	)






					def 	_lowercase					( self					:  Dict	):
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	):
							snake_case__			:				List[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# PyTorch not in environment -> use TensorFlow
						snake_case__			:				Tuple       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				int       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_tf	)

						# Both in environment -> use PyTorch
						snake_case__			:				Dict       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[int]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)
							self.assertEqual(__A							,		self.framework_pt	)

						# Both not in environment -> raise error
						snake_case__			:				List[str]       =  MagicMock(return_value=__A	)
						snake_case__			:				Optional[Any]       =  MagicMock(return_value=__A	)
						with patch("transformers.onnx.features.is_tf_available"							,		__A	), patch(
						    "transformers.onnx.features.is_torch_available"							,		__A	):
							with self.assertRaises(__A	):
								snake_case__			:				Optional[Any]       =  FeaturesManager.determine_framework(self.test_model	)



