


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


_snake_case      ,       _snake_case      ,       _snake_case             =     False, False, False
@dataclass
class 	UpperCAmelCase_					:

   '''simple docstring'''




   __A :      Optional[int]     		=		None
   __A :      bool     		=		True
   __A :      bool     		=		True
   __A :      Optional[str]     		=		None
   # Automatically constructed
   __A :      ClassVar[str]     		=		"dict"
   __A :      ClassVar[Any]     		=		pa.struct({"bytes": pa.binary(), "path": pa.string()}    )
   __A :      str     		=		field(default="Audio"		, init=UpperCamelCase		, repr=UpperCamelCase    )






   def __call__(		self ):






     """simple docstring"""





     return self.pa_type






   def _snake_case    (		self						,				__A ):






     """simple docstring"""





     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
       lowerCamelCase :					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!)
           lowerCamelCase :					Tuple    =				np.frombuffer(value["bytes"]						,				dtype=np.intaa ).astype(np.floataa ) / 3_2767
         else:
           lowerCamelCase :					Optional[int]    =				np.memmap(value["path"]						,				dtype="h"						,				mode="r" ).astype(np.floataa ) / 3_2767

         lowerCamelCase :					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 _snake_case    (		self						,				__A						,				__A = None ):






     """simple docstring"""





     if not self.decode:
       raise RuntimeError("Decoding is disabled for this feature. Please use Audio(decode=True) instead." )

     lowerCamelCase    ,	lowerCamelCase :					Any    =				(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

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

       with xopen(__A						,				"rb"						,				use_auth_token=__A ) as f:
         lowerCamelCase    ,	lowerCamelCase :					Union[str, Any]    =				sf.read(__A )

     else:
       lowerCamelCase    ,	lowerCamelCase :					Dict    =				sf.read(__A )

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

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






   def _snake_case    (		self ):






     """simple docstring"""





     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 ):






     """simple docstring"""





     if pa.types.is_string(storage.type ):
       lowerCamelCase :					Dict    =				pa.array([None] * len(__A )						,				type=pa.binary() )
       lowerCamelCase :					str    =				pa.StructArray.from_arrays([bytes_array, storage]						,				["bytes", "path"]						,				mask=storage.is_null() )
     elif pa.types.is_binary(storage.type ):
       lowerCamelCase :					Dict    =				pa.array([None] * len(__A )						,				type=pa.string() )
       lowerCamelCase :					Optional[Any]    =				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" ):
       lowerCamelCase :					Dict    =				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:
         lowerCamelCase :					str    =				storage.field("bytes" )
       else:
         lowerCamelCase :					Optional[int]    =				pa.array([None] * len(__A )						,				type=pa.binary() )
       if storage.type.get_field_index("path" ) >= 0:
         lowerCamelCase :					Union[str, Any]    =				storage.field("path" )
       else:
         lowerCamelCase :					Optional[int]    =				pa.array([None] * len(__A )						,				type=pa.string() )
       lowerCamelCase :					Union[str, Any]    =				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 ):






     """simple docstring"""





     @no_op_if_value_is_null
     def path_to_bytes(__A ):
       with xopen(__A						,				"rb" ) as f:
         lowerCamelCase :					Dict    =				f.read()
       return bytes_

     lowerCamelCase :					List[str]    =				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()						,				)
     lowerCamelCase :					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()						,				)
     lowerCamelCase :					Union[str, Any]    =				pa.StructArray.from_arrays([bytes_array, path_array]						,				["bytes", "path"]						,				mask=bytes_array.is_null() )
     return array_cast(__A						,				self.pa_type )







from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


_snake_case             =     {
    '''configuration_vision_encoder_decoder''': ['''VisionEncoderDecoderConfig''', '''VisionEncoderDecoderOnnxConfig''']
}

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _snake_case             =     ['''VisionEncoderDecoderModel''']

try:
   if not is_tf_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _snake_case             =     ['''TFVisionEncoderDecoderModel''']

try:
   if not is_flax_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   _snake_case             =     ['''FlaxVisionEncoderDecoderModel''']

if TYPE_CHECKING:
   from .configuration_vision_encoder_decoder import VisionEncoderDecoderConfig, VisionEncoderDecoderOnnxConfig

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_vision_encoder_decoder import VisionEncoderDecoderModel

   try:
      if not is_tf_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_tf_vision_encoder_decoder import TFVisionEncoderDecoderModel

   try:
      if not is_flax_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_flax_vision_encoder_decoder import FlaxVisionEncoderDecoderModel

else:
   import sys

   _snake_case             =     _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)



