
"""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__      :	Union[str, Any]          =    False, False, False







@dataclass
class   _lowercase						:


     '''simple docstring'''





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



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



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase  )->					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__							: Union[str, Any] 							=						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__							: Dict 							=						np.frombuffer(value["bytes"] ,			dtype=np.intaa  ).astype(np.floataa  ) / 3_27_67
                    else:
                         UpperCAmelCase__							: str 							=						np.memmap(value["path"] ,			dtype="h" ,			mode="r"  ).astype(np.floataa  ) / 3_27_67

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

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

               with xopen(A__ ,			"rb" ,			use_auth_token=A__  ) as f:
                    UpperCAmelCase__							,	UpperCAmelCase__							: List[str] 							=						sf.read(A__  )

          else:
               UpperCAmelCase__							,	UpperCAmelCase__							: int 							=						sf.read(A__  )

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

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



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



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase  )->					pa.StructArray:

          @no_op_if_value_is_null
          def path_to_bytes(__UpperCamelCase  ):
               with xopen(A__ ,			"rb"  ) as f:
                    UpperCAmelCase__							: Optional[int] 							=						f.read()
               return bytes_

          UpperCAmelCase__							: 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() ,			)
          UpperCAmelCase__							: List[Any] 							=						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__							: 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  )


"""simple docstring"""




from timeit import timeit



def 							a__ (				lowerCAmelCase	:       int		):






     '''simple docstring'''
     if number < 0:
          raise ValueError("the value of input must not be negative"		)
     UpperCAmelCase__							: Tuple 							=						0
     while number:
          number &= number - 1
          result += 1
     return result



def 							a__ (				lowerCAmelCase	:       int		):






     '''simple docstring'''
     if number < 0:
          raise ValueError("the value of input must not be negative"		)
     UpperCAmelCase__							: Union[str, Any] 							=						0
     while number:
          if number % 2 == 1:
               result += 1
          number >>= 1
     return result



def 							a__ (				):






     '''simple docstring'''

     def do_benchmark(lowerCAmelCase	:       int		) -> None:
          UpperCAmelCase__							: Dict 							=						"import __main__ as z"
          print(F"Benchmark when {number = }:"		)
          print(F"{get_set_bits_count_using_modulo_operator(lowerCAmelCase		) = }"		)
          UpperCAmelCase__							: Tuple 							=						timeit("z.get_set_bits_count_using_modulo_operator(25)"       ,    setup=lowerCAmelCase		)
          print(F"timeit() runs in {timing} seconds"		)
          print(F"{get_set_bits_count_using_brian_kernighans_algorithm(lowerCAmelCase		) = }"		)
          UpperCAmelCase__							: Any 							=						timeit(
              "z.get_set_bits_count_using_brian_kernighans_algorithm(25)"       ,    setup=lowerCAmelCase       ,    )
          print(F"timeit() runs in {timing} seconds"		)

     for number in (25, 37, 58, 0):
          do_benchmark(lowerCAmelCase		)
          print()


if __name__ == "__main__":
    import doctest

    doctest.testmod()
    benchmark()



