






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


a_       ,					a_       ,					a_		:						Optional[Any]             =    False, False, False

@dataclass
class 		_snake_case  :
			_lowercase		:    Optional[int]			     =							None
			_lowercase		:    bool			     =							True
			_lowercase		:    bool			     =							True
			_lowercase		:    Optional[str]			     =							None
			# Automatically constructed
			_lowercase		:    ClassVar[str]			     =							"dict"
			_lowercase		:    ClassVar[Any]			     =							pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}					)
			_lowercase		:    str			     =							field(default='''Audio'''   ,    init=A__   ,    repr=A__					)






			def __call__(       self)      ->       Union[str, Any]:
						return self.pa_type






			def    SCREAMING_SNAKE_CASE__    (       self		,					a)      ->       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
									SCREAMING_SNAKE_CASE						= 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!)
															SCREAMING_SNAKE_CASE						= np.frombuffer(value['bytes']		,					dtype=np.intaa).astype(np.floataa) / 3_2767
												else:
															SCREAMING_SNAKE_CASE						= np.memmap(value['path']		,					dtype='h'		,					mode='r').astype(np.floataa) / 3_2767

												SCREAMING_SNAKE_CASE						= 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    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a = None)      ->       dict:
						if not self.decode:
									raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.')

						SCREAMING_SNAKE_CASE				,						SCREAMING_SNAKE_CASE						= (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

						SCREAMING_SNAKE_CASE						= 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:
									SCREAMING_SNAKE_CASE						= token_per_repo_id or {}
									SCREAMING_SNAKE_CASE						= path.split('::')[-1]
									try:
												SCREAMING_SNAKE_CASE						= string_to_dict(a		,					config.HUB_DATASETS_URL)['repo_id']
												SCREAMING_SNAKE_CASE						= token_per_repo_id[repo_id]
									except (ValueError, KeyError):
												SCREAMING_SNAKE_CASE						= None

									with xopen(a		,					'rb'		,					use_auth_token=a) as f:
												SCREAMING_SNAKE_CASE				,						SCREAMING_SNAKE_CASE						= sf.read(a)

						else:
									SCREAMING_SNAKE_CASE				,						SCREAMING_SNAKE_CASE						= sf.read(a)

						SCREAMING_SNAKE_CASE						= array.T
						if self.mono:
									SCREAMING_SNAKE_CASE						= librosa.to_mono(a)
						if self.sampling_rate and self.sampling_rate != sampling_rate:
									SCREAMING_SNAKE_CASE						= librosa.resample(a		,					orig_sr=a		,					target_sr=self.sampling_rate)
									SCREAMING_SNAKE_CASE						= self.sampling_rate

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






			def    SCREAMING_SNAKE_CASE__    (       self)      ->       Union["FeatureType", Dict[str, "FeatureType"]]:
						from .features import Value

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






			def    SCREAMING_SNAKE_CASE__    (       self		,					a)      ->       pa.StructArray:
						if pa.types.is_string(storage.type):
									SCREAMING_SNAKE_CASE						= pa.array([None] * len(a)		,					type=pa.binary())
									SCREAMING_SNAKE_CASE						= pa.StructArray.from_arrays([bytes_array, storage]		,					['bytes', 'path']		,					mask=storage.is_null())
						elif pa.types.is_binary(storage.type):
									SCREAMING_SNAKE_CASE						= pa.array([None] * len(a)		,					type=pa.string())
									SCREAMING_SNAKE_CASE						= 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'):
									SCREAMING_SNAKE_CASE						= 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:
												SCREAMING_SNAKE_CASE						= storage.field('bytes')
									else:
												SCREAMING_SNAKE_CASE						= pa.array([None] * len(a)		,					type=pa.binary())
									if storage.type.get_field_index('path') >= 0:
												SCREAMING_SNAKE_CASE						= storage.field('path')
									else:
												SCREAMING_SNAKE_CASE						= pa.array([None] * len(a)		,					type=pa.string())
									SCREAMING_SNAKE_CASE						= pa.StructArray.from_arrays([bytes_array, path_array]		,					['bytes', 'path']		,					mask=storage.is_null())
						return array_cast(a		,					self.pa_type)







			def    SCREAMING_SNAKE_CASE__    (       self		,					a)      ->       pa.StructArray:

						@no_op_if_value_is_null
						def path_to_bytes(a):
									with xopen(a		,					'rb') as f:
												SCREAMING_SNAKE_CASE						= f.read()
									return bytes_

						SCREAMING_SNAKE_CASE						= 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()		,					)
						SCREAMING_SNAKE_CASE						= pa.array(
						    [os.path.basename(a) if path is not None else None for path in storage.field('path').to_pylist()]		,					type=pa.string()		,					)
						SCREAMING_SNAKE_CASE						= pa.StructArray.from_arrays([bytes_array, path_array]		,					['bytes', 'path']		,					mask=bytes_array.is_null())
						return array_cast(a		,					self.pa_type)












# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import doctest
import sys
import warnings
from os.path import abspath, dirname, join

import _pytest

from transformers.testing_utils import HfDoctestModule, HfDocTestParser


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
a_		:						List[str]             =    abspath(join(dirname(__file__), 'src'))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action='ignore', category=FutureWarning)






def 				lowerCamelCase__				(_UpperCAmelCase):
			config.addinivalue_line(
			    'markers'							,       'is_pt_tf_cross_test: mark test to run only when PT and TF interactions are tested')
			config.addinivalue_line(
			    'markers'							,       'is_pt_flax_cross_test: mark test to run only when PT and FLAX interactions are tested')
			config.addinivalue_line('markers'							,       'is_pipeline_test: mark test to run only when pipelines are tested')
			config.addinivalue_line('markers'							,       'is_staging_test: mark test to run only in the staging environment')
			config.addinivalue_line('markers'							,       'accelerate_tests: mark test that require accelerate')
			config.addinivalue_line('markers'							,       'tool_tests: mark the tool tests that are run on their specific schedule')






def 				lowerCamelCase__				(_UpperCAmelCase):
			from transformers.testing_utils import pytest_addoption_shared

			pytest_addoption_shared(_UpperCAmelCase)






def 				lowerCamelCase__				(_UpperCAmelCase):
			from transformers.testing_utils import pytest_terminal_summary_main

			SCREAMING_SNAKE_CASE						= terminalreporter.config.getoption('--make-reports')
			if make_reports:
						pytest_terminal_summary_main(_UpperCAmelCase							,       id=_UpperCAmelCase)






def 				lowerCamelCase__				(_UpperCAmelCase							,       _UpperCAmelCase):
			# If no tests are collected, pytest exists with code 5, which makes the CI fail.
			if exitstatus == 5:
						SCREAMING_SNAKE_CASE						= 0


# Doctest custom flag to ignore output.
a_		:						List[Any]             =    doctest.register_optionflag('IGNORE_RESULT')

a_		:						Any             =    doctest.OutputChecker

class 		_snake_case  (  A__					):






			def    SCREAMING_SNAKE_CASE__    (       self		,					a		,					a		,					a)      ->       List[Any]:
						if IGNORE_RESULT & optionflags:
									return True
						return OutputChecker.check_output(self		,					a		,					a		,					a)



a_		:						int             =    CustomOutputChecker
a_		:						List[str]             =    HfDoctestModule







a_		:						Optional[int]             =    HfDocTestParser




