






"""simple docstring"""

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


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




@dataclass
class  _UpperCAmelCase :
 _lowerCAmelCase     :    List[str]	      =	None
 _lowerCAmelCase     :    Any	      =	True
 _lowerCAmelCase     :    List[Any]	      =	True
 _lowerCAmelCase     :    int	      =	None
 # Automatically constructed
 _lowerCAmelCase     :    Optional[Any]	      =	"""dict"""
 _lowerCAmelCase     :    Tuple	      =	pa.struct({"""bytes""": pa.binary(), """path""": pa.string()})
 _lowerCAmelCase     :    List[str]	      =	field(default="""Audio"""					,  init=lowercase__					,  repr=lowercase__)

 def __call__(  self	: Tuple       ):
     return self.pa_type

 def 		_snake_case      (  self	: Tuple						,  lowercase_	: 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(UpperCAmelCase__						,  UpperCAmelCase__       ):
         return {"bytes": None, "path": value}
     elif isinstance(UpperCAmelCase__						,  UpperCAmelCase__       ):
         return {"bytes": value, "path": None}
     elif "array" in value:
         # convert the audio array to wav bytes
         snake_case_				:  Optional[int]       =			BytesIO()
         sf.write(UpperCAmelCase__						,  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_				:  Optional[int]       =			np.frombuffer(value['''bytes''']						,  dtype=np.intaa       ).astype(np.floataa       ) / 32767
             else:
                 snake_case_				:  Union[str, Any]       =			np.memmap(value['''path''']						,  dtype='''h'''						,  mode='''r'''       ).astype(np.floataa       ) / 32767

             snake_case_				:  Tuple       =			BytesIO(bytes()       )
             sf.write(UpperCAmelCase__						,  UpperCAmelCase__						,  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 		_snake_case      (  self	: int						,  lowercase_	: dict						,  lowercase_	: 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_				:  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_				:  Union[str, Any]       =			xsplitext(UpperCAmelCase__       )[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_				:  List[str]       =			token_per_repo_id or {}
         snake_case_				:  Dict       =			path.split('''::'''       )[-1]
         try:
             snake_case_				:  Optional[Any]       =			string_to_dict(UpperCAmelCase__						,  config.HUB_DATASETS_URL       )['''repo_id''']
             snake_case_				:  Optional[int]       =			token_per_repo_id[repo_id]
         except (ValueError, KeyError):
             snake_case_				:  List[Any]       =			None

         with xopen(UpperCAmelCase__						,  '''rb'''						,  use_auth_token=UpperCAmelCase__       ) as f:
             snake_case_				:  Any       =			sf.read(UpperCAmelCase__       )

     else:
         snake_case_				:  Optional[Any]       =			sf.read(UpperCAmelCase__       )

     snake_case_				:  int       =			array.T
     if self.mono:
         snake_case_				:  Tuple       =			librosa.to_mono(UpperCAmelCase__       )
     if self.sampling_rate and self.sampling_rate != sampling_rate:
         snake_case_				:  Union[str, Any]       =			librosa.resample(UpperCAmelCase__						,  orig_sr=UpperCAmelCase__						,  target_sr=self.sampling_rate       )
         snake_case_				:  str       =			self.sampling_rate

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

 def 		_snake_case      (  self	: str       ):
     from .features import Value

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

 def 		_snake_case      (  self	: Any						,  lowercase_	: Union[pa.StringArray, pa.StructArray]       ):
     if pa.types.is_string(storage.type       ):
         snake_case_				:  List[Any]       =			pa.array([None] * len(UpperCAmelCase__       )						,  type=pa.binary()       )
         snake_case_				:  Tuple       =			pa.StructArray.from_arrays([bytes_array, storage]						,  ['''bytes''', '''path''']						,  mask=storage.is_null()       )
     elif pa.types.is_binary(storage.type       ):
         snake_case_				:  List[Any]       =			pa.array([None] * len(UpperCAmelCase__       )						,  type=pa.string()       )
         snake_case_				:  List[str]       =			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_				:  int       =			pa.array([Audio().encode_example(UpperCAmelCase__       ) 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_				:  Any       =			storage.field('''bytes'''       )
         else:
             snake_case_				:  Optional[int]       =			pa.array([None] * len(UpperCAmelCase__       )						,  type=pa.binary()       )
         if storage.type.get_field_index('''path'''       ) >= 0:
             snake_case_				:  Optional[Any]       =			storage.field('''path'''       )
         else:
             snake_case_				:  Dict       =			pa.array([None] * len(UpperCAmelCase__       )						,  type=pa.string()       )
         snake_case_				:  Optional[Any]       =			pa.StructArray.from_arrays([bytes_array, path_array]						,  ['''bytes''', '''path''']						,  mask=storage.is_null()       )
     return array_cast(UpperCAmelCase__						,  self.pa_type       )






 def 		_snake_case      (  self	: Tuple						,  lowercase_	: pa.StructArray       ):

     @no_op_if_value_is_null
     def path_to_bytes(lowercase_	: Tuple       ):
         with xopen(UpperCAmelCase__						,  '''rb'''       ) as f:
             snake_case_				:  str       =			f.read()
         return bytes_

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










"""simple docstring"""

import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / '''utils'''))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowercase__					:						Any												=			get_tests_dir('''fixtures''')




class  _UpperCAmelCase (    unittest.TestCase):

 def 		_snake_case      (  self	: Tuple       ):
     # A mock response for an HTTP head request to emulate server down
     snake_case_				:  str       =			mock.Mock()
     snake_case_				:  Optional[Any]       =			500
     snake_case_				:  str       =			{}
     snake_case_				:  Optional[int]       =			HTTPError
     snake_case_				:  Tuple       =			{}

     # Download this model to make sure it's in the cache.
     snake_case_				:  List[str]       =			WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''       )
     # Under the mock environment we get a 500 error when trying to reach the model.
     with mock.patch('''requests.Session.request'''						,  return_value=lowercase_       ) as mock_head:
         snake_case_				:  int       =			WavaVecaFeatureExtractor.from_pretrained('''hf-internal-testing/tiny-random-wav2vec2'''       )
         # This check we did call the fake head request
         mock_head.assert_called()






 def 		_snake_case      (  self	: Union[str, Any]       ):
     # This test is for deprecated behavior and can be removed in v5
     snake_case_				:  str       =			WavaVecaFeatureExtractor.from_pretrained(
         '''https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'''       )





@is_staging_test
class  _UpperCAmelCase (    unittest.TestCase):

 @classmethod
 def 		_snake_case      (  cls	: int       ):
     snake_case_				:  Dict       =			TOKEN
     HfFolder.save_token(lowercase_       )

 @classmethod
 def 		_snake_case      (  cls	: str       ):
     try:
         delete_repo(token=cls._token						,  repo_id='''test-feature-extractor'''       )
     except HTTPError:
         pass

     try:
         delete_repo(token=cls._token						,  repo_id='''valid_org/test-feature-extractor-org'''       )
     except HTTPError:
         pass

     try:
         delete_repo(token=cls._token						,  repo_id='''test-dynamic-feature-extractor'''       )
     except HTTPError:
         pass

 def 		_snake_case      (  self	: Any       ):
     snake_case_				:  int       =			WavaVecaFeatureExtractor.from_pretrained(lowercase_       )
     feature_extractor.push_to_hub('''test-feature-extractor'''						,  use_auth_token=self._token       )

     snake_case_				:  Optional[Any]       =			WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )

     # Reset repo
     delete_repo(token=self._token						,  repo_id='''test-feature-extractor'''       )

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
         feature_extractor.save_pretrained(
             lowercase_						,  repo_id='''test-feature-extractor'''						,  push_to_hub=lowercase_						,  use_auth_token=self._token       )

     snake_case_				:  Optional[int]       =			WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )

 def 		_snake_case      (  self	: Dict       ):
     snake_case_				:  Union[str, Any]       =			WavaVecaFeatureExtractor.from_pretrained(lowercase_       )
     feature_extractor.push_to_hub('''valid_org/test-feature-extractor'''						,  use_auth_token=self._token       )

     snake_case_				:  Optional[int]       =			WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor'''       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )

     # Reset repo
     delete_repo(token=self._token						,  repo_id='''valid_org/test-feature-extractor'''       )

     # Push to hub via save_pretrained
     with tempfile.TemporaryDirectory() as tmp_dir:
         feature_extractor.save_pretrained(
             lowercase_						,  repo_id='''valid_org/test-feature-extractor-org'''						,  push_to_hub=lowercase_						,  use_auth_token=self._token       )

     snake_case_				:  int       =			WavaVecaFeatureExtractor.from_pretrained('''valid_org/test-feature-extractor-org'''       )
     for k, v in feature_extractor.__dict__.items():
         self.assertEqual(lowercase_						,  getattr(lowercase_						,  lowercase_       )       )






 def 		_snake_case      (  self	: str       ):
     CustomFeatureExtractor.register_for_auto_class()
     snake_case_				:  Optional[Any]       =			CustomFeatureExtractor.from_pretrained(lowercase_       )

     feature_extractor.push_to_hub('''test-dynamic-feature-extractor'''						,  use_auth_token=self._token       )

     # This has added the proper auto_map field to the config
     self.assertDictEqual(
         feature_extractor.auto_map						,  {'''AutoFeatureExtractor''': '''custom_feature_extraction.CustomFeatureExtractor'''}						,  )

     snake_case_				:  Any       =			AutoFeatureExtractor.from_pretrained(
         f"{USER}/test-dynamic-feature-extractor"						,  trust_remote_code=lowercase_       )
     # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
     self.assertEqual(new_feature_extractor.__class__.__name__						,  '''CustomFeatureExtractor'''       )





