



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


lowerCamelCase ,	lowerCamelCase ,	lowerCamelCase	 =       False, False, False





@dataclass
class _UpperCamelCase		:





    '''simple docstring'''





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

    def __call__(							self  :  List[Any]):







           '''simple docstring'''

           return self.pa_type

    def 				__lowerCamelCase     (							self  :  List[str]  ,       _lowerCAmelCase  :  Union[str, bytes, dict]):







           '''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(_lowerCAmelCase  ,       _lowerCAmelCase):
                  return {"bytes": None, "path": value}
           elif isinstance(_lowerCAmelCase  ,       _lowerCAmelCase):
                  return {"bytes": value, "path": None}
           elif "array" in value:
                  # convert the audio array to wav bytes
                  __lowercase       	=BytesIO()
                  sf.write(_lowerCAmelCase  ,       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!)
                                __lowercase       	=np.frombuffer(value['bytes']  ,       dtype=np.intaa).astype(np.floataa) / 3_2_7_6_7
                         else:
                                __lowercase       	=np.memmap(value['path']  ,       dtype='h'  ,       mode='r').astype(np.floataa) / 3_2_7_6_7

                         __lowercase       	=BytesIO(bytes())
                         sf.write(_lowerCAmelCase  ,       _lowerCAmelCase  ,       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 				__lowerCamelCase     (							self  :  Union[str, Any]  ,       _lowerCAmelCase  :  dict  ,       _lowerCAmelCase  :  Optional[Dict[str, Union[str, bool, None]]] = None):







           '''simple docstring'''

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

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

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

                  with xopen(_lowerCAmelCase  ,       'rb'  ,       use_auth_token=_lowerCAmelCase) as f:
                         __lowercase	,				__lowercase       	=sf.read(_lowerCAmelCase)

           else:
                  __lowercase	,				__lowercase       	=sf.read(_lowerCAmelCase)

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

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

    def 				__lowerCamelCase     (							self  :  Union[str, Any]):







           '''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 				__lowerCamelCase     (							self  :  Dict  ,       _lowerCAmelCase  :  Union[pa.StringArray, pa.StructArray]):







           '''simple docstring'''

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

    def 				__lowerCamelCase     (							self  :  Optional[int]  ,       _lowerCAmelCase  :  pa.StructArray):







           '''simple docstring'''

           @no_op_if_value_is_null
           def path_to_bytes(_lowerCAmelCase  :  Optional[int]):
                  with xopen(_lowerCAmelCase  ,       'rb') as f:
                         __lowercase       	=f.read()
                  return bytes_

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








'''simple docstring'''



from typing import TYPE_CHECKING

from ...utils import _LazyModule


lowerCamelCase	 =       {"""tokenization_bertweet""": ["""BertweetTokenizer"""]}


if TYPE_CHECKING:
		from .tokenization_bertweet import BertweetTokenizer

else:
		import sys

		lowerCamelCase	 =       _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)



