


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


UpperCamelCase__												=      False, False, False

@dataclass
class     UpperCAmelCase__ :



     '''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=_SCREAMING_SNAKE_CASE   ,    repr=_SCREAMING_SNAKE_CASE		)




     def __call__(			self							:							List[str]		):





            """simple docstring"""



            return self.pa_type




     def 	lowerCAmelCase_ (			self							:							Optional[Any]						,   UpperCamelCase							:							Any		):





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

                          _lowercase       :					List[str]     							=      BytesIO(bytes()		)
                          sf.write(__snake_case						,   __snake_case						,   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							:							int						,   UpperCamelCase							:							int						,   UpperCamelCase							:							Tuple = None		):





            """simple docstring"""



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

            _lowercase							,						_lowercase       :					Optional[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

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

                   with xopen(__snake_case						,   '''rb'''						,   use_auth_token=__snake_case		) as f:
                          _lowercase							,						_lowercase       :					str     							=      sf.read(__snake_case		)

            else:
                   _lowercase							,						_lowercase       :					List[Any]     							=      sf.read(__snake_case		)

            _lowercase       :					Optional[Any]     							=      array.T
            if self.mono:
                   _lowercase       :					Dict     							=      librosa.to_mono(__snake_case		)
            if self.sampling_rate and self.sampling_rate != sampling_rate:
                   _lowercase       :					Optional[Any]     							=      librosa.resample(__snake_case						,   orig_sr=__snake_case						,   target_sr=self.sampling_rate		)
                   _lowercase       :					Any     							=      self.sampling_rate

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




     def 	lowerCAmelCase_ (			self							:							Optional[int]		):





            """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							:							List[Any]						,   UpperCamelCase							:							List[Any]		):





            """simple docstring"""



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




     def 	lowerCAmelCase_ (			self							:							Any						,   UpperCamelCase							:							Dict		):





            """simple docstring"""



            @no_op_if_value_is_null
            def path_to_bytes(UpperCamelCase							:							Any		):
                   with xopen(__snake_case						,   '''rb'''		) as f:
                          _lowercase       :					List[Any]     							=      f.read()
                   return bytes_

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






def      UpperCAmelCase_    (    _UpperCAmelCase							:Any  ,						_UpperCAmelCase							:Dict				)	->				Union[str, Any]:
     '''simple docstring'''


     A_          =       [1]
     for i in range(2  ,						_UpperCAmelCase				):
          factorials.append(factorials[-1] * i				)
     assert 0 <= k < factorials[-1] * n, "k out of bounds"

     A_          =       []
     A_          =       list(range(_UpperCAmelCase				)				)

     # Find permutation
     while factorials:
          A_          =       factorials.pop()
          A_   ,    A_          =       divmod(_UpperCAmelCase  ,						_UpperCAmelCase				)
          permutation.append(elements[number]				)
          elements.remove(elements[number]				)
     permutation.append(elements[0]				)

     return permutation


if __name__ == "__main__":
      import doctest

      doctest.testmod()



