| | |
|---|
| | | from typing import Mapping |
|---|
| | | from typing import Tuple |
|---|
| | | from typing import Union |
|---|
| | | |
|---|
| | | import h5py |
|---|
| | | try: |
|---|
| | | import h5py |
|---|
| | | except: |
|---|
| | | print("If you want use h5py dataset, please pip install h5py, and try it again") |
|---|
| | | import humanfriendly |
|---|
| | | import kaldiio |
|---|
| | | import numpy as np |
|---|
| | | import torch |
|---|
| | | from torch.utils.data.dataset import Dataset |
|---|
| | | from typeguard import check_argument_types |
|---|
| | | from typeguard import check_return_type |
|---|
| | | |
|---|
| | | from funasr.fileio.npy_scp import NpyScpReader |
|---|
| | | from funasr.fileio.rand_gen_dataset import FloatRandomGenerateDataset |
|---|
| | |
|---|
| | | |
|---|
| | | class AdapterForSoundScpReader(collections.abc.Mapping): |
|---|
| | | def __init__(self, loader, dtype=None): |
|---|
| | | assert check_argument_types() |
|---|
| | | self.loader = loader |
|---|
| | | self.dtype = dtype |
|---|
| | | self.rate = None |
|---|
| | |
|---|
| | | # NOTE(kamo): SoundScpReader doesn't support pipe-fashion |
|---|
| | | # like Kaldi e.g. "cat a.wav |". |
|---|
| | | # NOTE(kamo): The audio signal is normalized to [-1,1] range. |
|---|
| | | loader = SoundScpReader(path, dest_sample_rate, normalize=True, always_2d=False) |
|---|
| | | loader = SoundScpReader(path, normalize=True, always_2d=False, dest_sample_rate = dest_sample_rate) |
|---|
| | | |
|---|
| | | # SoundScpReader.__getitem__() returns Tuple[int, ndarray], |
|---|
| | | # but ndarray is desired, so Adapter class is inserted here |
|---|
| | |
|---|
| | | max_cache_fd: int = 0, |
|---|
| | | dest_sample_rate: int = 16000, |
|---|
| | | ): |
|---|
| | | assert check_argument_types() |
|---|
| | | if len(path_name_type_list) == 0: |
|---|
| | | raise ValueError( |
|---|
| | | '1 or more elements are required for "path_name_type_list"' |
|---|
| | |
|---|
| | | return _mes |
|---|
| | | |
|---|
| | | def __getitem__(self, uid: Union[str, int]) -> Tuple[str, Dict[str, np.ndarray]]: |
|---|
| | | assert check_argument_types() |
|---|
| | | |
|---|
| | | # Change integer-id to string-id |
|---|
| | | if isinstance(uid, int): |
|---|
| | |
|---|
| | | self.cache[uid] = data |
|---|
| | | |
|---|
| | | retval = uid, data |
|---|
| | | assert check_return_type(retval) |
|---|
| | | return retval |
|---|
