


'''simple docstring'''






# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


lowercase				=get_logger(__name__)






class    __magic_name__				:
       UpperCAmelCase		 ="dummy_data"
       UpperCAmelCase		 ="datasets"
       UpperCAmelCase		 =False






       def __init__(     self    ,				snake_case    ,				snake_case    ,				snake_case    ,				snake_case = None    ,				snake_case = False    ,				snake_case = True    ,				snake_case = None    ,				)		->    Union[str, Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[int]				       =0
          _UpperCAmelCase  :  int				       =dataset_name
          _UpperCAmelCase  :  Tuple				       =cache_dir
          _UpperCAmelCase  :  Union[str, Any]				       =use_local_dummy_data
          _UpperCAmelCase  :  List[Any]				       =config
          # download_callbacks take a single url as input
          _UpperCAmelCase  :  List[Callable]				       =download_callbacks or []
          # if False, it doesn't load existing files and it returns the paths of the dummy files relative
          # to the dummy_data zip file root
          _UpperCAmelCase  :  int				       =load_existing_dummy_data

          # TODO(PVP, QL) might need to make this more general
          _UpperCAmelCase  :  Union[str, Any]				       =str(snake_case)
          # to be downloaded
          _UpperCAmelCase  :  str				       =None
          _UpperCAmelCase  :  Dict				       =None






       @property
       def      lowerCAmelCase						(     self)		->    int:




          '''simple docstring'''
          if self._dummy_file is None:
             _UpperCAmelCase  :  Optional[Any]				       =self.download_dummy_data()
          return self._dummy_file






       @property
       def      lowerCAmelCase						(     self)		->    int:




          '''simple docstring'''
          if self.config is not None:
             # structure is dummy / config_name / version_name
             return os.path.join('dummy'    ,				self.config.name    ,				self.version_name)
          # structure is dummy / version_name
          return os.path.join('dummy'    ,				self.version_name)






       @property
       def      lowerCAmelCase						(     self)		->    Tuple:




          '''simple docstring'''
          return os.path.join(self.dummy_data_folder    ,				'dummy_data.zip')






       def      lowerCAmelCase						(     self)		->    List[str]:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[Any]				       =(
              self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
          )

          _UpperCAmelCase  :  Dict				       =cached_path(
              snake_case    ,				cache_dir=self.cache_dir    ,				extract_compressed_file=snake_case    ,				force_extract=snake_case)

          return os.path.join(snake_case    ,				self.dummy_file_name)






       @property
       def      lowerCAmelCase						(     self)		->    Any:




          '''simple docstring'''
          return os.path.join(self.datasets_scripts_dir    ,				self.dataset_name    ,				self.dummy_zip_file)






       @property
       def      lowerCAmelCase						(     self)		->    Tuple:




          '''simple docstring'''
          if self._bucket_url is None:
             _UpperCAmelCase  :  Optional[Any]				       =hf_github_url(self.dataset_name    ,				self.dummy_zip_file.replace(os.sep    ,				'/'))
          return self._bucket_url






       @property
       def      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          # return full path if its a dir
          if os.path.isdir(self.dummy_file):
             return self.dummy_file
          # else cut off path to file -> example `xsum`.
          return "/".join(self.dummy_file.replace(os.sep    ,				'/').split('/')[:-1])






       def      lowerCAmelCase						(     self    ,				snake_case    ,				*snake_case)		->    Union[str, Any]:




          '''simple docstring'''
          if self.load_existing_dummy_data:
             # dummy data is downloaded and tested
             _UpperCAmelCase  :  Any				       =self.dummy_file
          else:
             # dummy data cannot be downloaded and only the path to dummy file is returned
             _UpperCAmelCase  :  Optional[Any]				       =self.dummy_file_name

          # special case when data_url is a dict
          if isinstance(snake_case    ,				snake_case):
             return self.create_dummy_data_dict(snake_case    ,				snake_case)
          elif isinstance(snake_case    ,				(list, tuple)):
             return self.create_dummy_data_list(snake_case    ,				snake_case)
          else:
             return self.create_dummy_data_single(snake_case    ,				snake_case)






       def      lowerCAmelCase						(     self    ,				snake_case    ,				*snake_case)		->    List[Any]:




          '''simple docstring'''
          return self.download_and_extract(snake_case)






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case)		->    Dict:




          '''simple docstring'''
          return self.download_and_extract(snake_case)






       def      lowerCAmelCase						(     self    ,				snake_case    ,				*snake_case    ,				**snake_case)		->    Union[str, Any]:




          '''simple docstring'''
          return path






       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          return {}






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  int				       ={}
          for key, single_urls in data_url.items():
             for download_callback in self.download_callbacks:
                if isinstance(snake_case    ,				snake_case):
                   for single_url in single_urls:
                      download_callback(snake_case)
                else:
                   _UpperCAmelCase  :  List[Any]				       =single_urls
                   download_callback(snake_case)
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
             if isinstance(snake_case    ,				snake_case):
                _UpperCAmelCase  :  Optional[int]				       =[os.path.join(snake_case    ,				urllib.parse.quote_plus(Path(snake_case).name)) for x in single_urls]
             else:
                _UpperCAmelCase  :  List[Any]				       =single_urls
                _UpperCAmelCase  :  List[str]				       =os.path.join(snake_case    ,				urllib.parse.quote_plus(Path(snake_case).name))
             _UpperCAmelCase  :  List[Any]				       =value

          # make sure that values are unique
          if all(isinstance(snake_case    ,				snake_case) for i in dummy_data_dict.values()) and len(set(dummy_data_dict.values())) < len(
              dummy_data_dict.values()):
             # append key to value to make its name unique
             _UpperCAmelCase  :  Any				       ={key: value + key for key, value in dummy_data_dict.items()}

          return dummy_data_dict






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case)		->    Dict:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[Any]				       =[]
          # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
          _UpperCAmelCase  :  str				       =all(bool(re.findall('[0-9]{3,}-of-[0-9]{3,}'    ,				snake_case)) for url in data_url)
          _UpperCAmelCase  :  Union[str, Any]				       =all(
              url.startswith('https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed') for url in data_url)
          if data_url and (is_tf_records or is_pubmed_records):
             _UpperCAmelCase  :  Any				       =[data_url[0]] * len(snake_case)
          for single_url in data_url:
             for download_callback in self.download_callbacks:
                download_callback(snake_case)
             # we force the name of each key to be the last file / folder name of the url path
             # if the url has arguments, we need to encode them with urllib.parse.quote_plus
             _UpperCAmelCase  :  List[Any]				       =os.path.join(snake_case    ,				urllib.parse.quote_plus(single_url.split('/')[-1]))
             dummy_data_list.append(snake_case)
          return dummy_data_list






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case)		->    str:




          '''simple docstring'''
          for download_callback in self.download_callbacks:
             download_callback(snake_case)
          # we force the name of each key to be the last file / folder name of the url path
          # if the url has arguments, we need to encode them with urllib.parse.quote_plus
          _UpperCAmelCase  :  Optional[Any]				       =os.path.join(snake_case    ,				urllib.parse.quote_plus(data_url.split('/')[-1]))
          if os.path.exists(snake_case) or not self.load_existing_dummy_data:
             return value
          else:
             # Backward compatibility, maybe deprecate at one point.
             # For many datasets with single url calls to dl_manager.download_and_extract,
             # the dummy_data.zip file is actually the zipped downloaded file
             # while now we expected the dummy_data.zip file to be a directory containing
             # the downloaded file.
             return path_to_dummy_data






       def      lowerCAmelCase						(     self)		->    List[str]:




          '''simple docstring'''
          pass






       def      lowerCAmelCase						(     self)		->    Any:




          '''simple docstring'''
          pass






       def      lowerCAmelCase						(     self    ,				snake_case)		->    Optional[int]:




          '''simple docstring'''
          def _iter_archive_members(snake_case):
             # this preserves the order of the members inside the ZIP archive
             _UpperCAmelCase  :  List[str]				       =Path(self.dummy_file).parent
             _UpperCAmelCase  :  Union[str, Any]				       =path.relative_to(snake_case)
             with ZipFile(self.local_path_to_dummy_data) as zip_file:
                _UpperCAmelCase  :  str				       =zip_file.namelist()
             for member in members:
                if member.startswith(relative_path.as_posix()):
                   yield dummy_parent_path.joinpath(snake_case)

          _UpperCAmelCase  :  List[str]				       =Path(snake_case)
          _UpperCAmelCase  :  List[str]				       =_iter_archive_members(snake_case) if self.use_local_dummy_data else path.rglob('*')
          for file_path in file_paths:
             if file_path.is_file() and not file_path.name.startswith(('.', '__')):
                yield file_path.relative_to(snake_case).as_posix(), file_path.open('rb')






       def      lowerCAmelCase						(     self    ,				snake_case)		->    Optional[int]:




          '''simple docstring'''
          if not isinstance(snake_case    ,				snake_case):
             _UpperCAmelCase  :  int				       =[paths]
          for path in paths:
             if os.path.isfile(snake_case):
                if os.path.basename(snake_case).startswith(('.', '__')):
                   return
                yield path
             else:
                for dirpath, dirnames, filenames in os.walk(snake_case):
                   if os.path.basename(snake_case).startswith(('.', '__')):
                      continue
                   dirnames.sort()
                   for filename in sorted(snake_case):
                      if filename.startswith(('.', '__')):
                         continue
                      yield os.path.join(snake_case    ,				snake_case)





'''simple docstring'''






from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


lowercase				=logging.get_logger(__name__)






class    __magic_name__				(						lowerCAmelCase							):
       UpperCAmelCase		 =["audio_values", "audio_mask"]






       def __init__(     self    ,				snake_case=2_0_4_8    ,				snake_case=1    ,				snake_case=[1_6, 1_6]    ,				snake_case=1_2_8    ,				snake_case=4_4_1_0_0    ,				snake_case=8_6    ,				snake_case=2_0_4_8    ,				snake_case=0.0    ,				**snake_case    ,				)		->    Optional[int]:




          '''simple docstring'''
          super().__init__(
              feature_size=snake_case    ,				sampling_rate=snake_case    ,				padding_value=snake_case    ,				**snake_case    ,				)

          _UpperCAmelCase  :  Dict				       =spectrogram_length
          _UpperCAmelCase  :  Any				       =num_channels
          _UpperCAmelCase  :  Optional[int]				       =patch_size
          _UpperCAmelCase  :  List[str]				       =feature_size // self.patch_size[1]
          _UpperCAmelCase  :  Optional[int]				       =n_fft
          _UpperCAmelCase  :  int				       =sampling_rate // hop_length_to_sampling_rate
          _UpperCAmelCase  :  int				       =sampling_rate
          _UpperCAmelCase  :  List[Any]				       =padding_value
          _UpperCAmelCase  :  Optional[int]				       =mel_filter_bank(
              num_frequency_bins=1 + n_fft // 2    ,				num_mel_filters=snake_case    ,				min_frequency=0.0    ,				max_frequency=2_20_50.0    ,				sampling_rate=snake_case    ,				norm='slaney'    ,				mel_scale='slaney'    ,				).T






       def      lowerCAmelCase						(     self    ,				snake_case)		->    np.ndarray:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       =spectrogram(
              snake_case    ,				window_function(self.n_fft    ,				'hann')    ,				frame_length=self.n_fft    ,				hop_length=self.hop_length    ,				power=2.0    ,				mel_filters=self.mel_filters.T    ,				log_mel='dB'    ,				db_range=80.0    ,				)
          _UpperCAmelCase  :  Any				       =log_spec[:, :-1]
          _UpperCAmelCase  :  List[str]				       =log_spec - 20.0
          _UpperCAmelCase  :  Union[str, Any]				       =np.clip(log_spec / 40.0    ,				-2.0    ,				0.0) + 1.0
          return log_spec






       def __call__(     self    ,				snake_case    ,				snake_case = None    ,				snake_case = True    ,				snake_case = None    ,				snake_case = False    ,				snake_case = False    ,				**snake_case    ,				)		->    BatchFeature:




          '''simple docstring'''

          if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                raise ValueError(
                    'This feature extractor is set to support sampling rate'
                    f" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"
                    f" with {self.sampling_rate} and not {sampling_rate}.")
          else:
             logger.warning(
                 'It is strongly recommended to pass the `sampling_rate` argument to this function. '
                 'Failing to do so can result in silent errors that might be hard to debug.')

          _UpperCAmelCase  :  Tuple				       =isinstance(snake_case    ,				np.ndarray) and len(raw_speech.shape) > 1
          if is_batched_numpy and len(raw_speech.shape) > 2:
             raise ValueError(f"Only mono-channel audio is supported for input to {self}")
          _UpperCAmelCase  :  str				       =is_batched_numpy or (
              isinstance(snake_case    ,				(list, tuple)) and (isinstance(raw_speech[0]    ,				(np.ndarray, tuple, list)))
          )
          if is_batched:
             _UpperCAmelCase  :  List[str]				       =[np.asarray([speech]    ,				dtype=np.floataa).T for speech in raw_speech]
          elif not is_batched and not isinstance(snake_case    ,				np.ndarray):
             _UpperCAmelCase  :  List[Any]				       =np.asarray(snake_case    ,				dtype=np.floataa)
          elif isinstance(snake_case    ,				np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
             _UpperCAmelCase  :  List[Any]				       =raw_speech.astype(np.floataa)
          # always return batch
          if not is_batched:
             _UpperCAmelCase  :  Dict				       =[np.asarray([raw_speech]).T]

          # Convert audio signals to log mel spectrograms, truncate by time axis
          _UpperCAmelCase  :  List[Any]				       =[
              self._np_extract_fbank_features(waveform.squeeze()).T[: self.spectrogram_length] for waveform in raw_speech
          ]
          if isinstance(audio_features[0]    ,				snake_case):
             _UpperCAmelCase  :  Dict				       =[np.asarray(snake_case    ,				dtype=np.floataa) for feature in audio_features]

          # Create audio attention mask
          _UpperCAmelCase  :  Tuple				       =max(
              [ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len for feature in audio_features])  # The maximum number of audio patches in a batch
          if return_attention_mask:
             _UpperCAmelCase  :  Any				       =[
                 (ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len) * [1]
                 + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len) * [0]
                 for feature in audio_features
             ]
             _UpperCAmelCase  :  int				       =np.array(snake_case).astype(np.floataa)

          # convert into correct format for padding
          _UpperCAmelCase  :  List[Any]				       =max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
          _UpperCAmelCase  :  List[str]				       =np.ones([len(snake_case), 1, max_time_len, self.feature_size]).astype(np.floataa)
          _UpperCAmelCase  :  Any				       =padded_audio_features * self.padding_value
          for i in range(len(snake_case)):
             _UpperCAmelCase  :  List[str]				       =audio_features[i]
             _UpperCAmelCase  :  Dict				       =feature

          # return as BatchFeature
          if return_attention_mask:
             _UpperCAmelCase  :  List[str]				       ={'audio_values': padded_audio_features, 'audio_mask': audio_mask}
          else:
             _UpperCAmelCase  :  Optional[int]				       ={'audio_values': padded_audio_features}

          _UpperCAmelCase  :  Any				       =BatchFeature(data=snake_case    ,				tensor_type=snake_case)
          return encoded_inputs

