



'''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


A_ ,		A_ ,		A_							: Union[str, Any]						 =	False, False, False



@dataclass
class 							lowercase       :


 """simple docstring"""


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







 def __call__(							self							)		->   str:
        return self.pa_type







 def _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
               _UpperCAmelCase  :						List[str]			     = 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!)
                             _UpperCAmelCase  :						Any			     = np.frombuffer(value["""bytes"""]  ,dtype=np.intaa							).astype(np.floataa							) / 32_767
                      else:
                             _UpperCAmelCase  :						int			     = np.memmap(value["""path"""]  ,dtype="""h"""  ,mode="""r"""							).astype(np.floataa							) / 32_767

                      _UpperCAmelCase  :						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 _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."""							)

        _UpperCAmelCase     ,_UpperCAmelCase  :						List[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

        _UpperCAmelCase  :						Tuple			     = 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:
               _UpperCAmelCase  :						Optional[Any]			     = token_per_repo_id or {}
               _UpperCAmelCase  :						Optional[int]			     = path.split("""::"""							)[-1]
               try:
                      _UpperCAmelCase  :						Optional[Any]			     = string_to_dict(a_  ,config.HUB_DATASETS_URL							)["""repo_id"""]
                      _UpperCAmelCase  :						Tuple			     = token_per_repo_id[repo_id]
               except (ValueError, KeyError):
                      _UpperCAmelCase  :						Optional[Any]			     = None

               with xopen(a_  ,"""rb"""  ,use_auth_token=a_							) as f:
                      _UpperCAmelCase     ,_UpperCAmelCase  :						Dict			     = sf.read(a_							)

        else:
               _UpperCAmelCase     ,_UpperCAmelCase  :						Union[str, Any]			     = sf.read(a_							)

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

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







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







 def _snake_case (							self  ,a_							)		->   pa.StructArray:

        @no_op_if_value_is_null
        def path_to_bytes(a_							):
               with xopen(a_  ,"""rb"""							) as f:
                      _UpperCAmelCase  :						int			     = f.read()
               return bytes_

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








'''simple docstring'''






import unittest

from transformers import load_tool

from .test_tools_common import ToolTesterMixin



class 							lowercase       (  unittest.TestCase					,					_lowerCamelCase				):


 """simple docstring"""


 def _snake_case (							self							)		->   Any:
        _UpperCAmelCase  :						int			     = load_tool("""text-classification"""							)
        self.tool.setup()
        _UpperCAmelCase  :						Tuple			     = load_tool("""text-classification"""  ,remote=a_							)







 def _snake_case (							self							)		->   Union[str, Any]:
        _UpperCAmelCase  :						Tuple			     = self.tool("""That's quite cool"""  ,["""positive""", """negative"""]							)
        self.assertEqual(a_  ,"""positive"""							)







 def _snake_case (							self							)		->   Any:
        _UpperCAmelCase  :						Dict			     = self.remote_tool("""That's quite cool"""  ,["""positive""", """negative"""]							)
        self.assertEqual(a_  ,"""positive"""							)







 def _snake_case (							self							)		->   str:
        _UpperCAmelCase  :						List[Any]			     = self.tool(text="""That's quite cool"""  ,labels=["""positive""", """negative"""]							)
        self.assertEqual(a_  ,"""positive"""							)







 def _snake_case (							self							)		->   Union[str, Any]:
        _UpperCAmelCase  :						Any			     = self.remote_tool(text="""That's quite cool"""  ,labels=["""positive""", """negative"""]							)
        self.assertEqual(a_  ,"""positive"""							)



