Source code for speechbrain.dataio.legacy

"""SpeechBrain Extended CSV Compatibility."""
from speechbrain.dataio.dataset import DynamicItemDataset
import collections
import csv
import pickle
import logging
import torch
import torchaudio
import re

logger = logging.getLogger(__name__)

TORCHAUDIO_FORMATS = ["wav", "flac", "aac", "ogg", "flac", "mp3"]
ITEM_POSTFIX = "_data"

CSVItem = collections.namedtuple("CSVItem", ["data", "format", "opts"])
CSVItem.__doc__ = """The Legacy Extended CSV Data item triplet"""

[docs] class ExtendedCSVDataset(DynamicItemDataset): """Extended CSV compatibility for DynamicItemDataset. Uses the SpeechBrain Extended CSV data format, where the CSV must have an 'ID' and 'duration' fields. The rest of the fields come in triplets: ``<name>, <name>_format, <name>_opts`` These add a <name>_sb_data item in the dict. Additionally, a basic DynamicItem (see DynamicItemDataset) is created, which loads the _sb_data item. Bash-like string replacements with $to_replace are supported. NOTE ---- Mapping from legacy interface: - csv_file -> csvpath - sentence_sorting -> sorting, and "random" is not supported, use e.g. ``make_dataloader(..., shuffle = (sorting=="random"))`` - avoid_if_shorter_than -> min_duration - avoid_if_longer_than -> max_duration - csv_read -> output_keys, and if you want IDs add "id" as key Arguments --------- csvpath : str, path Path to extended CSV. replacements : dict Used for Bash-like $-prefixed substitution, e.g. ``{"data_folder": "/home/speechbrain/data"}``, which would transform `$data_folder/utt1.wav` into `/home/speechbain/data/utt1.wav` sorting : {"original", "ascending", "descending"} Keep CSV order, or sort ascending or descending by duration. min_duration : float, int Minimum duration in seconds. Discards other entries. max_duration : float, int Maximum duration in seconds. Discards other entries. dynamic_items : list Configuration for extra dynamic items produced when fetching an example. List of DynamicItems or dicts with keys:: func: <callable> # To be called takes: <list> # key or list of keys of args this takes provides: key # key or list of keys that this provides NOTE: A dynamic item is automatically added for each CSV data-triplet output_keys : list, None The list of output keys to produce. You can refer to the names of the CSV data-triplets. E.G. if the CSV has: wav,wav_format,wav_opts, then the Dataset has a dynamic item output available with key ``"wav"`` NOTE: If None, read all existing. """ def __init__( self, csvpath, replacements={}, sorting="original", min_duration=0, max_duration=36000, dynamic_items=[], output_keys=[], ): if sorting not in ["original", "ascending", "descending"]: clsname = self.__class__.__name__ raise ValueError(f"{clsname} doesn't support {sorting} sorting") # Load the CSV, init class data, di_to_add, data_names = load_sb_extended_csv( csvpath, replacements ) super().__init__(data, dynamic_items, output_keys) self.pipeline.add_dynamic_items(di_to_add) # Handle filtering, sorting: reverse = False sort_key = None if sorting == "ascending" or "descending": sort_key = "duration" if sorting == "descending": reverse = True filtered_sorted_ids = self._filtered_sorted_ids( key_min_value={"duration": min_duration}, key_max_value={"duration": max_duration}, sort_key=sort_key, reverse=reverse, ) self.data_ids = filtered_sorted_ids # Handle None output_keys (differently than Base) if not output_keys: self.set_output_keys(data_names)
[docs] def load_sb_extended_csv(csv_path, replacements={}): """Loads SB Extended CSV and formats string values. Uses the SpeechBrain Extended CSV data format, where the CSV must have an 'ID' and 'duration' fields. The rest of the fields come in triplets: ``<name>, <name>_format, <name>_opts``. These add a <name>_sb_data item in the dict. Additionally, a basic DynamicItem (see DynamicItemDataset) is created, which loads the _sb_data item. Bash-like string replacements with $to_replace are supported. This format has its restriction, but they allow some tasks to have loading specified by the CSV. Arguments ---------- csv_path : str Path to the CSV file. replacements : dict Optional dict: e.g. ``{"data_folder": "/home/speechbrain/data"}`` This is used to recursively format all string values in the data. Returns ------- dict CSV data with replacements applied. list List of DynamicItems to add in DynamicItemDataset. """ with open(csv_path, newline="") as csvfile: result = {} reader = csv.DictReader(csvfile, skipinitialspace=True) variable_finder = re.compile(r"\$([\w.]+)") if not reader.fieldnames[0] == "ID": raise KeyError( "CSV has to have an 'ID' field, with unique ids" " for all data points" ) if not reader.fieldnames[1] == "duration": raise KeyError( "CSV has to have an 'duration' field, " "with the length of the data point in seconds." ) if not len(reader.fieldnames[2:]) % 3 == 0: raise ValueError( "All named fields must have 3 entries: " "<name>, <name>_format, <name>_opts" ) names = reader.fieldnames[2::3] for row in reader: # Make a triplet for each name data_point = {} # ID: data_id = row["ID"] del row["ID"] # This is used as a key in result, instead. # Duration: data_point["duration"] = float(row["duration"]) del row["duration"] # This is handled specially. if data_id in result: raise ValueError(f"Duplicate id: {data_id}") # Replacements: # Only need to run these in the actual data, # not in _opts, _format for key, value in list(row.items())[::3]: try: row[key] = variable_finder.sub( lambda match: replacements[match[1]], value ) except KeyError: raise KeyError( f"The item {value} requires replacements " "which were not supplied." ) for i, name in enumerate(names): triplet = CSVItem(*list(row.values())[i * 3 : i * 3 + 3]) data_point[name + ITEM_POSTFIX] = triplet result[data_id] = data_point # Make a DynamicItem for each CSV entry # _read_csv_item delegates reading to further dynamic_items_to_add = [] for name in names: di = { "func": _read_csv_item, "takes": name + ITEM_POSTFIX, "provides": name, } dynamic_items_to_add.append(di) return result, dynamic_items_to_add, names
def _read_csv_item(item): """Reads the different formats supported in SB Extended CSV. Delegates to the relevant functions. """ opts = _parse_csv_item_opts(item.opts) if item.format in TORCHAUDIO_FORMATS: audio, _ = torchaudio.load( return audio.squeeze(0) elif item.format == "pkl": return read_pkl(, opts) elif item.format == "string": # Just implement string reading here. # NOTE: No longer supporting # lab2ind mapping like before. # Try decoding string string = try: string = string.decode("utf-8") except AttributeError: pass # Splitting elements with ' ' string = string.split(" ") return string else: raise TypeError(f"Don't know how to read {item.format}") def _parse_csv_item_opts(entry): """Parse the _opts field in a SB Extended CSV item.""" # Accepting even slightly weirdly formatted entries: entry = entry.strip() if len(entry) == 0: return {} opts = {} for opt in entry.split(" "): opt_name, opt_val = opt.split(":") opts[opt_name] = opt_val return opts
[docs] def read_pkl(file, data_options={}, lab2ind=None): """This function reads tensors store in pkl format. Arguments --------- file : str The path to file to read. data_options : dict, optional A dictionary containing options for the reader. lab2ind : dict, optional Mapping from label to integer indices. Returns ------- numpy.array The array containing the read signal. """ # Trying to read data try: with open(file, "rb") as f: pkl_element = pickle.load(f) except pickle.UnpicklingError: err_msg = "cannot read the pkl file %s" % (file) raise ValueError(err_msg) type_ok = False if isinstance(pkl_element, list): if isinstance(pkl_element[0], float): tensor = torch.FloatTensor(pkl_element) type_ok = True if isinstance(pkl_element[0], int): tensor = torch.LongTensor(pkl_element) type_ok = True if isinstance(pkl_element[0], str): # convert string to integer as specified in self.label_dict if lab2ind is not None: for index, val in enumerate(pkl_element): pkl_element[index] = lab2ind[val] tensor = torch.LongTensor(pkl_element) type_ok = True if not (type_ok): err_msg = ( "The pkl file %s can only contain list of integers, " "floats, or strings. Got %s" ) % (file, type(pkl_element[0])) raise ValueError(err_msg) else: tensor = pkl_element tensor_type = tensor.dtype # Conversion to 32 bit (if needed) if tensor_type == "float64": tensor = tensor.astype("float32") if tensor_type == "int64": tensor = tensor.astype("int32") return tensor