





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


a__							:							List[str] =get_logger(__name__)

class      snake_case     :




 """simple docstring"""



 SCREAMING_SNAKE_CASE_	:					str					="dummy_data"
 SCREAMING_SNAKE_CASE_	:					List[str]					="datasets"
 SCREAMING_SNAKE_CASE_	:					Dict					=False






 def __init__(       self       :							str							,	__A       :							str							,	__A       :							str							,	__A       :							Union[Version, str]							,	__A       :							Optional[str] = None							,	__A       :							bool = False							,	__A       :							bool = True							,	__A       :							Optional[List[Callable]] = None							,	):
        __UpperCamelCase       =  0
        __UpperCamelCase       =  dataset_name
        __UpperCamelCase       =  cache_dir
        __UpperCamelCase       =  use_local_dummy_data
        __UpperCamelCase       =  config
        # download_callbacks take a single url as input
        __UpperCamelCase       =  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       =  load_existing_dummy_data

        # TODO(PVP, QL) might need to make this more general
        __UpperCamelCase       =  str(__A	)
        # to be downloaded
        __UpperCamelCase       =  None
        __UpperCamelCase       =  None






 @property
 def        _lowerCamelCase					(       self       :							int	):
        if self._dummy_file is None:
               __UpperCamelCase       =  self.download_dummy_data()
        return self._dummy_file






 @property
 def        _lowerCamelCase					(       self       :							Optional[int]	):
        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       :							Union[str, Any]	):
        return os.path.join(self.dummy_data_folder							,	'dummy_data.zip'	)






 def        _lowerCamelCase					(       self       :							Dict	):
        __UpperCamelCase       =  (
            self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
        )

        __UpperCamelCase       =  cached_path(
            __A							,	cache_dir=self.cache_dir							,	extract_compressed_file=__A							,	force_extract=__A	)

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






 @property
 def        _lowerCamelCase					(       self       :							List[str]	):
        return os.path.join(self.datasets_scripts_dir							,	self.dataset_name							,	self.dummy_zip_file	)






 @property
 def        _lowerCamelCase					(       self       :							Tuple	):
        if self._bucket_url is None:
               __UpperCamelCase       =  hf_github_url(self.dataset_name							,	self.dummy_zip_file.replace(os.sep							,	'/'	)	)
        return self._bucket_url






 @property
 def        _lowerCamelCase					(       self       :							Union[str, Any]	):
        # 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       :							Optional[int]							,	__A       :							Optional[int]							,	*__A       :							Optional[int]	):
        if self.load_existing_dummy_data:
               # dummy data is downloaded and tested
               __UpperCamelCase       =  self.dummy_file
        else:
               # dummy data cannot be downloaded and only the path to dummy file is returned
               __UpperCamelCase       =  self.dummy_file_name

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






 def        _lowerCamelCase					(       self       :							Any							,	__A       :							Dict							,	*__A       :							Any	):
        return self.download_and_extract(__A	)






 def        _lowerCamelCase					(       self       :							Optional[Any]							,	__A       :							Tuple							,	__A       :							Dict	):
        return self.download_and_extract(__A	)






 def        _lowerCamelCase					(       self       :							int							,	__A       :							Any							,	*__A       :							List[str]							,	**__A       :							Optional[Any]	):
        return path






 def        _lowerCamelCase					(       self       :							str	):
        return {}






 def        _lowerCamelCase					(       self       :							Any							,	__A       :							int							,	__A       :							Any	):
        __UpperCamelCase       =  {}
        for key, single_urls in data_url.items():
               for download_callback in self.download_callbacks:
                      if isinstance(__A							,	__A	):
                             for single_url in single_urls:
                                    download_callback(__A	)
                      else:
                             __UpperCamelCase       =  single_urls
                             download_callback(__A	)
            # 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(__A							,	__A	):
                      __UpperCamelCase       =  [os.path.join(__A							,	urllib.parse.quote_plus(Path(__A	).name	)	) for x in single_urls]
               else:
                      __UpperCamelCase       =  single_urls
                      __UpperCamelCase       =  os.path.join(__A							,	urllib.parse.quote_plus(Path(__A	).name	)	)
               __UpperCamelCase       =  value

        # make sure that values are unique
        if all(isinstance(__A							,	__A	) 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       =  {key: value + key for key, value in dummy_data_dict.items()}

        return dummy_data_dict






 def        _lowerCamelCase					(       self       :							int							,	__A       :							Dict							,	__A       :							Optional[Any]	):
        __UpperCamelCase       =  []
        # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
        __UpperCamelCase       =  all(bool(re.findall('[0-9]{3,}-of-[0-9]{3,}'							,	__A	)	) for url in data_url	)
        __UpperCamelCase       =  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       =  [data_url[0]] * len(__A	)
        for single_url in data_url:
               for download_callback in self.download_callbacks:
                      download_callback(__A	)
               # 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       =  os.path.join(__A							,	urllib.parse.quote_plus(single_url.split('/'	)[-1]	)	)
               dummy_data_list.append(__A	)
        return dummy_data_list






 def        _lowerCamelCase					(       self       :							Optional[int]							,	__A       :							Any							,	__A       :							List[str]	):
        for download_callback in self.download_callbacks:
               download_callback(__A	)
        # 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       =  os.path.join(__A							,	urllib.parse.quote_plus(data_url.split('/'	)[-1]	)	)
        if os.path.exists(__A	) 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       :							str	):
        pass






 def        _lowerCamelCase					(       self       :							List[Any]	):
        pass






 def        _lowerCamelCase					(       self       :							Tuple							,	__A       :							Dict	):
        def _iter_archive_members(__A       :							Tuple	):
               # this preserves the order of the members inside the ZIP archive
               __UpperCamelCase       =  Path(self.dummy_file	).parent
               __UpperCamelCase       =  path.relative_to(__A	)
               with ZipFile(self.local_path_to_dummy_data	) as zip_file:
                      __UpperCamelCase       =  zip_file.namelist()
               for member in members:
                      if member.startswith(relative_path.as_posix()	):
                             yield dummy_parent_path.joinpath(__A	)

        __UpperCamelCase       =  Path(__A	)
        __UpperCamelCase       =  _iter_archive_members(__A	) 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(__A	).as_posix(), file_path.open('rb'	)






 def        _lowerCamelCase					(       self       :							str							,	__A       :							Optional[Any]	):
        if not isinstance(__A							,	__A	):
               __UpperCamelCase       =  [paths]
        for path in paths:
               if os.path.isfile(__A	):
                      if os.path.basename(__A	).startswith(('.', '__')	):
                             return
                      yield path
               else:
                      for dirpath, dirnames, filenames in os.walk(__A	):
                             if os.path.basename(__A	).startswith(('.', '__')	):
                                    continue
                             dirnames.sort()
                             for filename in sorted(__A	):
                                    if filename.startswith(('.', '__')	):
                                           continue
                                    yield os.path.join(__A							,	__A	)







'''simple docstring'''
from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


a__							:							str =logging.get_logger(__name__)

class      snake_case     (    __lowerCamelCase       ):




 """simple docstring"""



 SCREAMING_SNAKE_CASE_	:					str					=["input_features", "attention_mask"]






 def __init__(       self       :							Union[str, Any]							,	__A       :							Optional[int]=8_0							,	__A       :							Tuple=1_6_0_0_0							,	__A       :							Optional[Any]=8_0							,	__A       :							Any=0.0							,	__A       :							Any=True							,	__A       :							List[str]=True							,	__A       :							str=True							,	**__A       :							List[Any]							,	):
        super().__init__(feature_size=__A							,	sampling_rate=__A							,	padding_value=__A							,	**__A	)
        __UpperCamelCase       =  num_mel_bins
        __UpperCamelCase       =  do_ceptral_normalize
        __UpperCamelCase       =  normalize_means
        __UpperCamelCase       =  normalize_vars
        __UpperCamelCase       =  True






 def        _lowerCamelCase					(       self       :							Union[str, Any]							,	__A       :							np.ndarray							,	):
        __UpperCamelCase       =  waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
        __UpperCamelCase       =  torch.from_numpy(__A	).unsqueeze(0	)
        __UpperCamelCase       =  ta_kaldi.fbank(__A							,	num_mel_bins=self.num_mel_bins							,	sample_frequency=self.sampling_rate	)
        return features.numpy()






 @staticmethod
 def        _lowerCamelCase					(       __A       :							np.ndarray							,	__A       :							int							,	__A       :							Optional[bool] = True							,	__A       :							Optional[bool] = True							,	__A       :							float = 0.0							,	):
        # make sure we normalize float32 arrays
        if normalize_means:
               __UpperCamelCase       =  x[:input_length].mean(axis=0	)
               __UpperCamelCase       =  np.subtract(__A							,	__A	)
        if normalize_vars:
               __UpperCamelCase       =  x[:input_length].std(axis=0	)
               __UpperCamelCase       =  np.divide(__A							,	__A	)

        if input_length < x.shape[0]:
               __UpperCamelCase       =  padding_value

        # make sure array is in float32
        __UpperCamelCase       =  x.astype(np.floataa	)

        return x






 def        _lowerCamelCase					(       self       :							int							,	__A       :							List[np.ndarray]							,	__A       :							Optional[np.ndarray] = None	):
        __UpperCamelCase       =  attention_mask.sum(-1	) if attention_mask is not None else [x.shape[0] for x in input_features]
        return [
            self.utterance_cmvn(__A							,	__A							,	self.normalize_means							,	self.normalize_vars							,	self.padding_value	)
            for x, n in zip(__A							,	__A	)
        ]






 def __call__(       self       :							List[Any]							,	__A       :							Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]							,	__A       :							Union[bool, str, PaddingStrategy] = False							,	__A       :							Optional[int] = None							,	__A       :							bool = False							,	__A       :							Optional[int] = None							,	__A       :							Optional[Union[str, TensorType]] = None							,	__A       :							Optional[int] = None							,	__A       :							Optional[bool] = None							,	**__A       :							Dict							,	):

        if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                      raise ValueError(
                          f'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
                          f''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
                          f''' {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       =  isinstance(__A							,	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       =  is_batched_numpy or (
            isinstance(__A							,	(list, tuple)	) and (isinstance(raw_speech[0]							,	(np.ndarray, tuple, list)	))
        )

        if is_batched:
               __UpperCamelCase       =  [np.asarray(__A							,	dtype=np.floataa	) for speech in raw_speech]
        elif not is_batched and not isinstance(__A							,	np.ndarray	):
               __UpperCamelCase       =  np.asarray(__A							,	dtype=np.floataa	)
        elif isinstance(__A							,	np.ndarray	) and raw_speech.dtype is np.dtype(np.floataa	):
               __UpperCamelCase       =  raw_speech.astype(np.floataa	)

        # always return batch
        if not is_batched:
               __UpperCamelCase       =  [raw_speech]

        # extract fbank features
        __UpperCamelCase       =  [self._extract_fbank_features(__A	) for waveform in raw_speech]

        # convert into correct format for padding
        __UpperCamelCase       =  BatchFeature({'input_features': features}	)

        __UpperCamelCase       =  self.pad(
            __A							,	padding=__A							,	max_length=__A							,	truncation=__A							,	pad_to_multiple_of=__A							,	return_attention_mask=__A							,	**__A							,	)

        # make sure list is in array format
        __UpperCamelCase       =  padded_inputs.get('input_features'	)
        if isinstance(input_features[0]							,	__A	):
               __UpperCamelCase       =  [np.asarray(__A							,	dtype=np.floataa	) for feature in input_features]

        __UpperCamelCase       =  padded_inputs.get('attention_mask'	)
        if attention_mask is not None:
               __UpperCamelCase       =  [np.asarray(__A							,	dtype=np.intaa	) for array in attention_mask]

        # Utterance-level cepstral mean and variance normalization
        if self.do_ceptral_normalize:
               __UpperCamelCase       =  (
                   np.array(__A							,	dtype=np.intaa	)
                   if self._get_padding_strategies(__A							,	max_length=__A	) is not PaddingStrategy.DO_NOT_PAD
                   else None
               )
               __UpperCamelCase       =  self.normalize(
                   padded_inputs['input_features']							,	attention_mask=__A	)

        if return_tensors is not None:
               __UpperCamelCase       =  padded_inputs.convert_to_tensors(__A	)

        return padded_inputs
