





'''simple docstring'''







import inspect
import re
from hashlib import shaaaa
from typing import Dict, List

from .arrow import arrow
from .audiofolder import audiofolder
from .csv import csv
from .imagefolder import imagefolder
from .json import json
from .pandas import pandas
from .parquet import parquet
from .sql import sql  # noqa F401
from .text import text

def    __UpperCamelCase	(	a    : List[str]     ) ->str:
 snake_case				=			[]
 for line in lines:
  snake_case				=			re.sub(R'''#.*'''					,   ''''''					,   a     )  # remove comments
  if line:
   filtered_lines.append(a     )
 snake_case				=			'''\n'''.join(a     )

 # Make a hash from all this code
 snake_case				=			full_str.encode('''utf-8'''     )
 return shaaaa(a     ).hexdigest()


# get importable module names and hash for caching
_lowercase       						=			{
    'csv': (csv.__name__, _hash_python_lines(inspect.getsource(csv).splitlines())),
    'json': (json.__name__, _hash_python_lines(inspect.getsource(json).splitlines())),
    'pandas': (pandas.__name__, _hash_python_lines(inspect.getsource(pandas).splitlines())),
    'parquet': (parquet.__name__, _hash_python_lines(inspect.getsource(parquet).splitlines())),
    'arrow': (arrow.__name__, _hash_python_lines(inspect.getsource(arrow).splitlines())),
    'text': (text.__name__, _hash_python_lines(inspect.getsource(text).splitlines())),
    'imagefolder': (imagefolder.__name__, _hash_python_lines(inspect.getsource(imagefolder).splitlines())),
    'audiofolder': (audiofolder.__name__, _hash_python_lines(inspect.getsource(audiofolder).splitlines())),
}

# Used to infer the module to use based on the data files extensions
_lowercase       						=			{
    '.csv': ('csv', {}),
    '.tsv': ('csv', {'sep': '\t'}),
    '.json': ('json', {}),
    '.jsonl': ('json', {}),
    '.parquet': ('parquet', {}),
    '.arrow': ('arrow', {}),
    '.txt': ('text', {}),
}
_EXTENSION_TO_MODULE.update({ext: ('imagefolder', {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ('imagefolder', {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext: ('audiofolder', {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ('audiofolder', {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
_lowercase       						=			{'imagefolder', 'audiofolder'}

# Used to filter data files based on extensions given a module name
_lowercase       						=			{}
for _ext, (_module, _) in _EXTENSION_TO_MODULE.items():
   _MODULE_TO_EXTENSIONS.setdefault(_module, []).append(_ext)

_MODULE_TO_EXTENSIONS["imagefolder"].append('.zip')






_MODULE_TO_EXTENSIONS["audiofolder"].append('.zip')








'''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       						=			False, False, False
@dataclass
class 			_lowercase						:
     _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=__a				,		repr=__a					)







     def __call__( self    ) ->	Optional[Any]:
      return self.pa_type







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

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

      snake_case				,		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

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

       with xopen(A__    ,   '''rb'''    ,   use_auth_token=A__    ) as f:
        snake_case				,		snake_case				=			sf.read(A__    )

      else:
       snake_case				,		snake_case				=			sf.read(A__    )

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

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







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







     def     UpperCamelCase			( self    ,   A__    ) ->	pa.StructArray:

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

      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()    ,   )
      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()    ,   )
      snake_case				=			pa.StructArray.from_arrays([bytes_array, path_array]    ,   ['''bytes''', '''path''']    ,   mask=bytes_array.is_null()    )
      return array_cast(A__    ,   self.pa_type    )

