Datasets

kudio.core.synth

Noisy-speech synthesis: the whole chain between a talker and a file.

A recording is degraded three ways and this class puts all three in. Noise is additive, a room is convolutive, and the channel is neither, so they are applied in the order the signal actually meets them:

talker -> room (rt60=) -> + noise (snr_ratio=) -> link (channel=) -> file

Passing rt60= reverberates the clean signal before the noise is mixed in, which is the order a microphone meets them: the room acts on the talker, and the noise arrives at the microphone alongside the result. The SNR is therefore measured against the reverberant speech, because that is what is actually competing with the noise. Passing channel= runs the finished mixture through a link -- the wire carries whatever the microphone produced, noise included, so it comes last.

The channel is not a fourth axis. snr_ratio and rt60 are swept; channel is not, and it does not appear in the output names. A corpus is recorded over a phone line or it is not -- that is a property of the corpus, not a dimension to cross with the others, and four axes already multiply enough. To compare two links, run the Synthesizer twice into two directories.

The clean file stays the training target, so a model trained on this data is asked to undo all of it. write_targets=True also writes the reverberant-but-clean signal -- the room and nothing else -- so a model can be asked to remove only the noise and leave the room alone, which is a different and easier problem. manifest records which is which.

Classes:

Name	Description
Synthesizer	Mix clean audio with noise at the requested SNRs.

Synthesizer

Synthesizer(
    clean: Any,
    noise: Any,
    out_path: str = "",
    snr_ratio: Union[float, Sequence[float]] = (-5, 0, 5),
    rt60: Optional[Union[float, Sequence[float]]] = None,
    drr_db: float = 0.0,
    channel: Optional[Union[str, dict]] = None,
    write_targets: bool = False,
    root: Optional[PathLike] = None,
    subtype: str = "PCM_16",
)

Mix clean audio with noise at the requested SNRs.

>>> syx = Synthesizer('data/train/clean', 'data/train/noise',
...                   out_path='data/train/mixed', snr_ratio=(-5, 0, 5))
>>> syn_list = syx.syn(mode='inc')     # every clean x noise x SNR
>>> syn_list = syx.syn(mode='reg')     # one output per clean file
>>> syn_list = syx.syn_extra_mode('data/backgroundnoise', background_snr=(5,))

syn returns [(out_path, clean, noise, snr), ...].

Parameters:

Name	Type	Description	Default
clean	Any	any input kudio.check_input accepts -- a directory, a .wav file, a .txt list with one path per line, or a list of those. A list is how a subset is synthesised without copying it out first.	required
noise	Any	the same, for the noise.	required
out_path	str	output directory for the mixed files (default mixed). With overwrite=True syn empties it first, and refuses to when it is the root of a filesystem or holds an input file, the working directory or the home directory.	''
snr_ratio	Union[float, Sequence[float]]	one or more SNR values in dB.	(-5, 0, 5)
rt60	Optional[Union[float, Sequence[float]]]	one or more reverberation times in seconds. None (the default) puts no room on the speech and leaves it out of the output names. Several values are another axis: every output the mode would write is written once per room, in reg mode as much as in inc.	None
drr_db	float	direct-to-reverberant ratio for the rooms, in dB. Distance in a readable unit -- +10 and above is a close microphone, 0 is a metre or two, negative is the far end of a hall.	0.0
channel	Optional[Union[str, dict]]	a link for the finished mixture to travel down: a preset name ('telephone', 'voip', 'mu_law', ...), a dict of kudio.CHANNEL_DEFAULTS, or None for none. Constant across the run -- see the module docstring on why it is not an axis. Packet loss is seeded per output file, so a rerun drops the same packets.	None
write_targets	bool	also write the reverberant-but-clean signal to a TARGETS folder inside out_path, and record it in manifest. Only meaningful with rt60=: with no room the clean file already is that signal. Anything that later collects .wav files from out_path by folder collects these too.	False
root	Optional[PathLike]	the folder the clean files' layout is measured from; the output mirrors that layout. Defaults to clean when it is a directory and otherwise to the deepest folder the clean files share, so two talkers' sx133.wav stay two files.	None
subtype	str	how the files are written: 'PCM_16' (the default), 'PCM_24', 'FLOAT', ... A mixture is a sum and can land past full scale. A fixed-point subtype clips it there (every PCM_* width rounded to its nearest step), and syn warns with the count -- the files are in clipped. 'FLOAT' keeps the samples as they are.	'PCM_16'

Methods:

Name	Description
waveform	Load the clean and noise file lists into waveform data.
syn	Generate the noisy dataset.
manifest	The last run as kudio.Pair records, rooms included.
syn_extra_mode	Two-stage mixing: noise x background first, then everything x clean.
syn_waves	Mix one (save_dir, clean_file, noise_file, snr) entry to disk.

Attributes:

Name	Type	Description
rooms	Dict[Path, tuple]	room per output file, keyed by the output path rather than held in a
channel	dict	the link, normalised once. Empty dict means no channel, and is
channels	Dict[Path, tuple]	(spec, seed) per output file, for the same reason rooms is a
targets	Dict[Path, Path]	output file -> the reverberant-clean target written beside it
clipped	Dict[Path, float]	output file -> its peak, for every file of the last run that went
seed	Optional[int]	the base every per-file draw of the last run came from. Passed back
clean_root	Optional[Path]	where the output layout is measured from; None puts every
target_sr	Optional[int]	target output rate; None keeps each clean file's own rate

Source code in kudio/core/synth.py

def __init__(self, clean: Any, noise: Any, out_path: str = '',
             snr_ratio: Union[float, Sequence[float]] = (-5, 0, 5),
             rt60: Optional[Union[float, Sequence[float]]] = None,
             drr_db: float = 0.0,
             channel: Optional[Union[str, dict]] = None,
             write_targets: bool = False,
             root: Optional[PathLike] = None,
             subtype: str = 'PCM_16'):
    for what, given in (('clean', clean), ('noise', noise)):
        if isinstance(given, (str, PurePath)) and not Path(given).exists():
            raise FileNotFoundError(f"{what} input not found: {str(given)!r}")
    self.cleanWaves, bad_clean = check_input(clean)
    self.noiseWaves, bad_noise = check_input(noise)
    if not self.cleanWaves:
        raise ValueError(f"No wave files found in clean path: {clean}")
    if not self.noiseWaves:
        raise ValueError(f"No wave files found in noise path: {noise}")
    skipped = bad_clean + bad_noise
    if skipped:
        # a list or a manifest names files one by one, so one that is not
        # there is a mistake worth hearing about, not a file to step over
        warnings.warn(
            f"Synthesizer skipped {len(skipped)} input(s) that are not "
            f"existing .wav files, e.g. {str(skipped[0])!r}",
            UserWarning, stacklevel=2)

    self.noisyDir = Path('mixed' if out_path == '' else out_path)

    if isinstance(snr_ratio, str):
        raise TypeError("snr_ratio can't be str")
    self.snrRatio = list(snr_ratio) if hasattr(snr_ratio, '__iter__') else [snr_ratio]
    if not self.snrRatio:
        raise ValueError("snr_ratio is empty")
    if not all(np.isfinite(float(v)) for v in self.snrRatio):
        raise ValueError(f"snr_ratio values must be finite, got {self.snrRatio}")

    if isinstance(rt60, str):
        raise TypeError("rt60 can't be str")
    if rt60 is None:
        self.rt60Ratio: List[float] = []
    elif hasattr(rt60, '__iter__'):
        self.rt60Ratio = [float(v) for v in rt60]
    else:
        self.rt60Ratio = [float(rt60)]
    if any(v <= 0 for v in self.rt60Ratio):
        raise ValueError(f"rt60 values must be positive, got {self.rt60Ratio}")
    self.drr_db = float(drr_db)
    #: room per output file, keyed by the output path rather than held in a
    #: list beside `synWavesList`. Two lists that have to stay in step is
    #: the failure this codebase keeps meeting; a mapping cannot drift.
    self.rooms: Dict[Path, tuple] = {}

    from kudio.effects.channel import channel_spec, channel_tag
    #: the link, normalised once. Empty dict means no channel, and is
    #: falsy, so every downstream check is `if self.channel:`.
    self.channel: dict = channel_spec(channel)
    self.channel_tag: str = channel_tag(self.channel)
    #: (spec, seed) per output file, for the same reason `rooms` is a
    #: mapping: the seed has to reach the worker that writes that file.
    self.channels: Dict[Path, tuple] = {}

    self.write_targets = bool(write_targets)
    if self.write_targets and not self.rt60Ratio:
        warnings.warn(
            "Synthesizer(write_targets=True) without rt60= writes nothing: "
            "with no room the clean file already is the reverberant-clean "
            "signal. Pass rt60=, or drop write_targets.",
            UserWarning, stacklevel=2)
    #: output file -> the reverberant-clean target written beside it
    self.targets: Dict[Path, Path] = {}

    import soundfile as sf
    if not sf.check_format('WAV', str(subtype).upper()):
        raise ValueError(f"subtype {subtype!r} cannot be written to a .wav "
                         f"file; see soundfile.available_subtypes('WAV')")
    self.subtype = str(subtype).upper()
    #: output file -> its peak, for every file of the last run that went
    #: past full scale and was clipped there when written
    self.clipped: Dict[Path, float] = {}
    #: the base every per-file draw of the last run came from. Passed back
    #: as ``syn(seed=...)`` it rebuilds that run, even one given no seed.
    self.seed: Optional[int] = None

    self.synWavesList: Optional[list] = None
    self.clean_dir_ = clean
    #: where the output layout is measured from; ``None`` puts every
    #: output directly in `noisyDir`
    self.clean_root: Optional[Path] = _source_root(clean, self.cleanWaves, root)
    #: target output rate; ``None`` keeps each clean file's own rate
    self.target_sr: Optional[int] = None

rooms instance-attribute

rooms: Dict[Path, tuple] = {}

room per output file, keyed by the output path rather than held in a list beside synWavesList. Two lists that have to stay in step is the failure this codebase keeps meeting; a mapping cannot drift.

channel instance-attribute

channel: dict = channel_spec(channel)

the link, normalised once. Empty dict means no channel, and is falsy, so every downstream check is if self.channel:.

channels instance-attribute

channels: Dict[Path, tuple] = {}

(spec, seed) per output file, for the same reason rooms is a mapping: the seed has to reach the worker that writes that file.

targets instance-attribute

targets: Dict[Path, Path] = {}

output file -> the reverberant-clean target written beside it

clipped instance-attribute

clipped: Dict[Path, float] = {}

output file -> its peak, for every file of the last run that went past full scale and was clipped there when written

seed instance-attribute

seed: Optional[int] = None

the base every per-file draw of the last run came from. Passed back as syn(seed=...) it rebuilds that run, even one given no seed.

clean_root instance-attribute

clean_root: Optional[Path] = _source_root(
    clean, self.cleanWaves, root
)

where the output layout is measured from; None puts every output directly in noisyDir

target_sr instance-attribute

target_sr: Optional[int] = None

target output rate; None keeps each clean file's own rate

waveform

waveform(
    is_pool: bool = False,
) -> Tuple[
    List[Tuple[ndarray, int]], List[Tuple[ndarray, int]]
]

Load the clean and noise file lists into waveform data.

Source code in kudio/core/synth.py

def waveform(self, is_pool: bool = False
             ) -> Tuple[List[Tuple[np.ndarray, int]], List[Tuple[np.ndarray, int]]]:
    """Load the clean and noise file lists into waveform data."""
    from kudio.core.io import load_waves
    return (load_waves(self.cleanWaves, use_pool=is_pool),
            load_waves(self.noiseWaves, use_pool=is_pool))

syn

syn(
    mode: str = "regular",
    mkdir_parents: bool = True,
    is_add_clean: bool = False,
    rdn_choice_num: int = 0,
    overwrite: bool = True,
    is_pool: bool = False,
    is_silence: bool = False,
    p_silence: float = 0.02,
    target_sr: Optional[int] = None,
    desired_sample: Optional[int] = None,
    seed: Optional[int] = None,
) -> Optional[list]

Generate the noisy dataset.

mode 'reg'/'regular': one output per clean file, the noises and SNRs taken in turn -- and every noise x SNR combination comes up once in each len(noise) * len(snr) outputs, whatever the two lengths are. mode 'inc'/'increment': full cartesian product clean x noise x SNR.

Parameters:

Name	Type	Description	Default
rdn_choice_num	int	synthesise this many clean files, drawn at random without replacement, instead of all of them.	0
is_silence	bool	also write a p_silence share of the outputs again as noise alone, named ..._n00.wav. Only that share: the mixtures keep their speech.	False
target_sr	Optional[int]	resample the mixture to this rate. None (the default) keeps each clean file's own rate; the noise is always resampled to match its clean partner.	None
desired_sample	Optional[int]	deprecated alias for target_sr.	None
seed	Optional[int]	the base of everything random in the run. Each output file draws its noise crop, its room and its packet loss from the base and its own path, so a rerun rebuilds the same files whether or not they went through a process pool. None draws a base from numpy's global RNG -- so seeding that reproduces the run too -- and records it in seed. The global RNG is never reseeded.	None

Source code in kudio/core/synth.py

def syn(self, mode: str = 'regular',
        mkdir_parents: bool = True,
        is_add_clean: bool = False,
        rdn_choice_num: int = 0,
        overwrite: bool = True,
        is_pool: bool = False,
        is_silence: bool = False,
        p_silence: float = 0.02,
        target_sr: Optional[int] = None,
        desired_sample: Optional[int] = None,
        seed: Optional[int] = None) -> Optional[list]:
    """Generate the noisy dataset.

    mode ``'reg'``/``'regular'``: one output per clean file, the noises and
    SNRs taken in turn -- and every noise x SNR combination comes up once
    in each ``len(noise) * len(snr)`` outputs, whatever the two lengths
    are. mode ``'inc'``/``'increment'``: full cartesian product
    clean x noise x SNR.

    :param rdn_choice_num: synthesise this many clean files, drawn at
        random without replacement, instead of all of them.
    :param is_silence: also write a *p_silence* share of the outputs again
        as noise alone, named ``..._n00.wav``. Only that share: the
        mixtures keep their speech.
    :param target_sr: resample the mixture to this rate. ``None`` (the
        default) keeps each clean file's own rate; the noise is always
        resampled to match its clean partner.
    :param desired_sample: deprecated alias for *target_sr*.
    :param seed: the base of everything random in the run. Each output
        file draws its noise crop, its room and its packet loss from the
        base and its own path, so a rerun rebuilds the same files whether
        or not they went through a process pool. ``None`` draws a base
        from numpy's global RNG -- so seeding that reproduces the run too
        -- and records it in :attr:`seed`. The global RNG is never
        reseeded.
    """
    if mode in ('max', 'extra'):
        raise ValueError("Use 'syn_extra_mode' to run extra mode.")
    if mode not in ('reg', 'regular', 'inc', 'increment'):
        raise ValueError(f"Unknown mode: {mode!r}")
    if desired_sample is not None:
        warnings.warn(
            "Synthesizer.syn(desired_sample=...) is deprecated and will be "
            "removed in a future release; use target_sr=... instead.",
            DeprecationWarning, stacklevel=2)
        if target_sr is None:
            target_sr = desired_sample
    if target_sr is not None and not target_sr > 0:
        # checked before the output folder is emptied, not on the first file
        raise ValueError(f"target_sr must be a positive rate or None, got {target_sr}")
    log.info("==== Audio synthesizing ====")
    if self.noisyDir.is_dir() and any(self.noisyDir.iterdir()):
        if overwrite:
            self._check_clearable()
            log.warning("Output path exists, overwriting -> %s", self.noisyDir)
            shutil.rmtree(self.noisyDir)
        else:
            # a warning rather than a log line: returning None from a
            # call that looked like it ran is easy to miss
            warnings.warn(f"{self.noisyDir} is not empty, so nothing was "
                          f"synthesised. Pass overwrite=True to replace it, "
                          f"or point out_path= somewhere else.",
                          UserWarning, stacklevel=2)
            return None
    self.noisyDir.mkdir(parents=True, exist_ok=True)

    # Drawn once the run is certain to happen. `_base_seed` writes
    # `self.seed`, and that number is the only way back to the corpus
    # already in the output folder, so a call that aborts above has to
    # leave it -- and `target_sr` with it -- naming what is really there.
    # Nothing between here and the check wants either: the first draw off
    # `rng` is the random subset just below.
    base = self._base_seed(seed)
    rng = np.random.default_rng(base)
    self.target_sr = target_sr

    # optionally sample N clean files at random
    if rdn_choice_num > len(self.cleanWaves):
        warnings.warn(f"rdn_choice_num={rdn_choice_num} asks for more clean "
                      f"files than there are ({len(self.cleanWaves)}), so "
                      f"all of them are used.", UserWarning, stacklevel=2)
    if len(self.cleanWaves) >= rdn_choice_num > 0:
        picks = rng.choice(len(self.cleanWaves), size=rdn_choice_num,
                           replace=False)
        clnWaves = [self.cleanWaves[k] for k in picks]
    else:
        clnWaves = list(self.cleanWaves)

    # optionally copy the raw clean files alongside the output
    if is_add_clean:
        dir_ = self.noisyDir / "RAW_DATA"
        for cln in clnWaves:
            # mirrored like the mixtures, or two talkers' sa1.wav would
            # be one copy
            where = dir_ / (_relative(cln, self.clean_root).parent
                            if mkdir_parents else Path())
            where.mkdir(parents=True, exist_ok=True)
            shutil.copy(cln, where / f"{cln.stem}_raw.wav")

    clnLen, nosLen, snrLen = len(clnWaves), len(self.noiseWaves), len(self.snrRatio)
    log.info("Clean: %d, Noise: %d, SNR: %s", clnLen, nosLen, self.snrRatio)

    if mode in ('reg', 'regular'):
        log.info("%d SNRs x %d noises -> %d noisy files", snrLen, nosLen, clnLen)
        noise_idx, snr_idx = _reg_pairing(clnLen, nosLen, snrLen)
        nosWaves = [self.noiseWaves[k] for k in noise_idx]
        snrSeqs = [self.snrRatio[k] for k in snr_idx]
        if clnLen < nosLen * snrLen:
            log.info("%d outputs hold %d of the %d noise x SNR combinations",
                     clnLen, clnLen, nosLen * snrLen)
    else:
        log.info("%d noisy files from %d SNRs, %d noises and %d room(s)",
                 clnLen * nosLen * snrLen * max(1, len(self.rt60Ratio)),
                 snrLen, nosLen, max(1, len(self.rt60Ratio)))
        clnWaves = [c for c in clnWaves for _ in range(nosLen * snrLen)]
        nosWaves = [n for n in self.noiseWaves for _ in range(snrLen)] * clnLen
        snrSeqs = list(self.snrRatio) * (clnLen * nosLen)

    # The room is a fourth axis. Repeating the three existing sequences
    # rather than regenerating them keeps `reg` and `inc` meaning exactly
    # what they meant before, with or without rooms.
    rooms = self.rt60Ratio or [None]
    if len(rooms) > 1:
        repeat = len(rooms)
        clnWaves = [c for c in clnWaves for _ in range(repeat)]
        nosWaves = [n for n in nosWaves for _ in range(repeat)]
        snrSeqs = [s for s in snrSeqs for _ in range(repeat)]
        rt60Seqs = list(rooms) * (len(clnWaves) // repeat)
    else:
        rt60Seqs = [rooms[0]] * len(clnWaves)

    layout: Dict[Path, Path] = {}
    outWaves = []
    for cln, nos, snr, rt in zip(clnWaves, nosWaves, snrSeqs, rt60Seqs):
        snr_tag = f"n{_snr_text(snr)}" if float(snr) < 0 else f"{_snr_text(snr)}dB"
        parts = [cln.stem, nos.stem, snr_tag]
        if rt is not None:
            parts.append(f"rt{int(round(rt * 1000))}ms")
        if mkdir_parents:
            if cln not in layout:
                layout[cln] = _relative(cln, self.clean_root).parent
            outDir = self.noisyDir / layout[cln]
        else:
            outDir = self.noisyDir
        # joined rather than `with_suffix`, which would take '.b_noise_5dB'
        # off 'a.b_noise_5dB' and leave every output of 'a.b.wav' as 'a.wav'
        outWaves.append(outDir / ('_'.join(parts) + '.wav'))

    # Normalised here as well as in the manifest: an SNR list given as
    # np.arange carries np.int64, which `json` cannot serialise, so a
    # caller writing `syn()`'s own tuples out hit what the manifest path
    # was already protected from.
    entries = list(zip(outWaves, clnWaves, nosWaves,
                       [_snr_value(s) for s in snrSeqs]))
    rt_of = dict(zip(outWaves, rt60Seqs))
    entries = self._one_clean_per_name(entries, mkdir_parents)
    self._warn_collisions([out for out, *_ in entries], mkdir_parents)
    silent: Set[Path] = set()
    if is_silence and 0 < p_silence < 1:
        extra = self._silence_share(entries, rng, p_silence)
        silent = {out for out, *_ in extra}
        entries += extra

    # Each file is seeded from the base seed and its own path, so the same
    # request rebuilds the same dataset and two files that share an RT60
    # still get different rooms.
    self.rooms, self.channels, self.targets = {}, {}, {}
    seeds = self._seeds(entries, base, self.noisyDir)
    targets_dir = self.noisyDir / "TARGETS"
    for out, file_seed in seeds.items():
        rt = rt_of.get(out)
        if rt is not None:
            self.rooms[out] = (float(rt), self.drr_db, file_seed)
            if self.write_targets:
                # mirror the mixture's own layout, so the two names line up
                self.targets[out] = targets_dir / out.relative_to(self.noisyDir)
        if self.channel:
            # a different derivation from the room's, so a file does not
            # get the same number twice and lose the packets where the
            # room happens to be loudest
            self.channels[out] = (self.channel,
                                  (file_seed * 2654435761) & 0x7FFFFFFF)

    self.synWavesList = entries
    self._noise_only = silent
    peaks = self.__syn_pool(entries, is_pool=is_pool, subtype=self.subtype,
                            seeds=seeds, silent=silent, rooms=self.rooms,
                            channels=self.channels, targets=self.targets)
    message = self._clipping(entries, peaks)
    if message:
        warnings.warn(message, UserWarning, stacklevel=2)
    return self.synWavesList

manifest

manifest() -> List[Pair]

The last run as kudio.Pair records, rooms included.

syn returns four-tuples and always has; a fifth element that appeared only when rt60= was passed would be a return shape that depends on a keyword, which is the bug this codebase has met most often. The room is carried here instead, where it travels with the output path it belongs to.

Each row is a file this run wrote and the clean signal it was built from. A clean file that lost its output name to another one is not here -- there is no mixture of it to point at -- and syn warns when that happens. Nor are the noise-only files is_silence=True adds: there is no speech in them, so the clean file is not their reference, and they are in what syn returns instead.

>>> pairs = syx.manifest()
>>> kudio.save_manifest("runs/exp/manifest.json", pairs)

Source code in kudio/core/synth.py

def manifest(self) -> List[Pair]:
    """The last run as :class:`kudio.Pair` records, rooms included.

    :meth:`syn` returns four-tuples and always has; a fifth element that
    appeared only when ``rt60=`` was passed would be a return shape that
    depends on a keyword, which is the bug this codebase has met most
    often. The room is carried here instead, where it travels with the
    output path it belongs to.

    Each row is a file this run wrote and the clean signal it was built
    from. A clean file that lost its output name to another one is not
    here -- there is no mixture of it to point at -- and :meth:`syn` warns
    when that happens. Nor are the noise-only files ``is_silence=True``
    adds: there is no speech in them, so the clean file is not their
    reference, and they are in what :meth:`syn` returns instead.

    >>> pairs = syx.manifest()                           # doctest: +SKIP
    >>> kudio.save_manifest("runs/exp/manifest.json", pairs)  # doctest: +SKIP
    """
    if not self.synWavesList:
        return []
    tag = self.channel_tag or None
    # forward slashes, as save_manifest writes them, so a manifest read
    # back compares equal to the one that was saved
    return [Pair(noisy=Path(out).as_posix(), clean=Path(cln).as_posix(),
                 noise=Path(nos).stem,
                 snr_db=_snr_value(snr),
                 rt60=self.rooms.get(Path(out), (None,))[0],
                 channel=tag if Path(out) in self.channels else None,
                 target=(self.targets[Path(out)].as_posix()
                         if Path(out) in self.targets else None))
            for out, cln, nos, snr in self.synWavesList
            if Path(out) not in getattr(self, '_noise_only', ())]

syn_extra_mode

syn_extra_mode(
    background_path: Any = "../data/backgroundnoise",
    background_snr: Union[float, Sequence[float]] = (5,),
    seed: Optional[int] = None,
    is_pool: bool = True,
) -> list

Two-stage mixing: noise x background first, then everything x clean.

Stage one mixes every noise with every background at background_snr in a temporary folder. Stage two mixes every clean file with each of those mixtures, noises and backgrounds at every SNR, and writes a 2% share again as noise alone. The output folder is emptied first, under the same guard as syn; seed is as there. Rooms, the link and resampling are not applied in this mode, and the output is written flat, one folder for everything.

Names carry dB on both signs, n5dB and 5dB, where syn writes n5. Corpora on disk already use both spellings, so neither changes; match on the manifest, which holds each SNR as a number, rather than on the name.

Parameters:

Name	Type	Description	Default
is_pool	bool	write stage two through a process pool once it holds more than 200 files. A pool needs the calling script's top level behind if __name__ == '__main__': on Windows and macOS.	True

Source code in kudio/core/synth.py

def syn_extra_mode(self, background_path: Any = '../data/backgroundnoise',
                   background_snr: Union[float, Sequence[float]] = (5,),
                   seed: Optional[int] = None,
                   is_pool: bool = True) -> list:
    """Two-stage mixing: noise x background first, then everything x clean.

    Stage one mixes every noise with every background at *background_snr*
    in a temporary folder. Stage two mixes every clean file with each of
    those mixtures, noises and backgrounds at every SNR, and writes a 2%
    share again as noise alone. The output folder is emptied first, under
    the same guard as :meth:`syn`; *seed* is as there. Rooms, the link
    and resampling are not applied in this mode, and the output is written
    flat, one folder for everything.

    Names carry ``dB`` on both signs, ``n5dB`` and ``5dB``, where
    :meth:`syn` writes ``n5``. Corpora on disk already use both spellings,
    so neither changes; match on the manifest, which holds each SNR as a
    number, rather than on the name.

    :param is_pool: write stage two through a process pool once it holds
        more than 200 files. A pool needs the calling script's top level
        behind ``if __name__ == '__main__':`` on Windows and macOS.
    """
    if isinstance(background_snr, str):
        raise TypeError("background_snr can't be str")
    bgSNR = list(background_snr) if hasattr(background_snr, '__iter__') else [background_snr]
    bgNoiseWaves, _ = check_input(background_path)
    if not bgNoiseWaves:
        raise ValueError(f"No wave files found in: {background_path}")
    if self.noisyDir.is_dir() and any(self.noisyDir.iterdir()):
        self._check_clearable(bgNoiseWaves)
        shutil.rmtree(self.noisyDir)
    self.noisyDir.mkdir(parents=True, exist_ok=True)
    # after the guard, like `syn`: `_base_seed` writes `self.seed`, and a
    # run that refuses to clear the folder must not take the seed that
    # names the corpus already sitting in it
    base = self._base_seed(seed)
    rng = np.random.default_rng(base)
    self.rooms, self.channels, self.targets = {}, {}, {}
    # no resampling in this mode, whatever an earlier syn() asked for
    self.target_sr = None
    cleanWaves = self.cleanWaves
    for cln in cleanWaves:
        shutil.copy(cln, self.noisyDir / f"{cln.stem}_n0{cln.suffix}")

    def name(cln, nos, snr) -> str:
        tag = f"n{_snr_text(snr)}" if float(snr) < 0 else _snr_text(snr)
        return f"{cln.stem}_{nos.stem}_{tag}dB.wav"

    with tempfile.TemporaryDirectory(prefix="kudio-extra-") as tmp:
        # stage 1: noise x background noise into a temporary directory,
        # written as float -- an intermediate has no business clipping
        temporaryPath = Path(tmp)
        fullCleanWaves = [n for n in self.noiseWaves
                          for _ in range(len(bgNoiseWaves) * len(bgSNR))]
        fullNoiseWaves = [b for b in bgNoiseWaves
                          for _ in range(len(bgSNR))] * len(self.noiseWaves)
        fullSnrSequence = bgSNR * (len(self.noiseWaves) * len(bgNoiseWaves))
        fullOutPaths = [temporaryPath / name(cln, nos, snr) for cln, nos, snr
                        in zip(fullCleanWaves, fullNoiseWaves, fullSnrSequence)]
        stage1 = list(zip(fullOutPaths, fullCleanWaves, fullNoiseWaves,
                          fullSnrSequence))
        self.__syn_pool(stage1, is_pool=False, subtype='FLOAT',
                        seeds=self._seeds(stage1, base, temporaryPath))

        # stage 2: clean x (stage-1 output + noise + background noise)
        noiseWaves = fullOutPaths + list(self.noiseWaves) + list(bgNoiseWaves)
        fullCleanWaves = [c for c in cleanWaves
                          for _ in range(len(noiseWaves) * len(self.snrRatio))]
        fullNoiseWaves = [n for n in noiseWaves
                          for _ in range(len(self.snrRatio))] * len(cleanWaves)
        fullSnrSequence = list(self.snrRatio) * (len(cleanWaves) * len(noiseWaves))
        fullOutPaths = [self.noisyDir / name(cln, nos, snr) for cln, nos, snr
                        in zip(fullCleanWaves, fullNoiseWaves, fullSnrSequence)]
        entries = list(zip(fullOutPaths, fullCleanWaves, fullNoiseWaves,
                           fullSnrSequence))
        self._warn_collisions([out for out, *_ in entries], False)
        extra = self._silence_share(entries, rng, 0.02, flat=True)
        entries += extra
        self.synWavesList = entries
        self._noise_only = {out for out, *_ in extra}
        peaks = self.__syn_pool(entries, is_pool=is_pool, subtype=self.subtype,
                                seeds=self._seeds(entries, base, self.noisyDir),
                                silent={out for out, *_ in extra})
    message = self._clipping(entries, peaks)
    if message:
        warnings.warn(message, UserWarning, stacklevel=2)
    return self.synWavesList

syn_waves staticmethod

syn_waves(
    synWavesList: Union[tuple, Sequence[tuple]],
    isSilence: bool,
    target_sr: Optional[int],
    num: Optional[int] = None,
    rooms: Optional[Dict[Path, tuple]] = None,
    channels: Optional[Dict[Path, tuple]] = None,
    targets: Optional[Dict[Path, Path]] = None,
    seeds: Optional[Dict[Path, int]] = None,
    silent: Optional[Set[Path]] = None,
    subtype: str = "PCM_16",
) -> float

Mix one (save_dir, clean_file, noise_file, snr) entry to disk.

With num given, synWavesList is the full list and num its index (multiprocessing form); otherwise it is a single entry.

rooms maps an output path to (rt60, drr_db, seed). The room is applied to the clean signal before the noise, and the SNR is then measured against the reverberant result -- the order a microphone meets them, and the only one in which the requested SNR is the SNR you get.

channels maps an output path to (spec, seed) and is applied to the finished mixture, since the wire carries the noise too. targets maps an output path to where the reverberant-clean signal goes; it is written before the noise arrives and before the channel, so it is the speech as the room left it and nothing else.

seeds maps an output path to the seed its noise crop is drawn from, so the crop does not depend on which worker wrote the file; a path missing from it draws from numpy's global RNG. Outputs in silent are written as the noise alone; isSilence=True does that to all.

Returns the largest absolute sample written, mixture or target. Past 1.0, a fixed-point subtype clipped it.

Source code in kudio/core/synth.py

@staticmethod
def syn_waves(synWavesList: Union[tuple, Sequence[tuple]], isSilence: bool,
              target_sr: Optional[int],
              num: Optional[int] = None,
              rooms: Optional[Dict[Path, tuple]] = None,
              channels: Optional[Dict[Path, tuple]] = None,
              targets: Optional[Dict[Path, Path]] = None,
              seeds: Optional[Dict[Path, int]] = None,
              silent: Optional[Set[Path]] = None,
              subtype: str = 'PCM_16') -> float:
    """Mix one ``(save_dir, clean_file, noise_file, snr)`` entry to disk.

    With ``num`` given, ``synWavesList`` is the full list and ``num`` its
    index (multiprocessing form); otherwise it is a single entry.

    *rooms* maps an output path to ``(rt60, drr_db, seed)``. The room is
    applied to the clean signal **before** the noise, and the SNR is then
    measured against the reverberant result -- the order a microphone meets
    them, and the only one in which the requested SNR is the SNR you get.

    *channels* maps an output path to ``(spec, seed)`` and is applied to
    the finished mixture, since the wire carries the noise too. *targets*
    maps an output path to where the reverberant-clean signal goes; it is
    written before the noise arrives and before the channel, so it is the
    speech as the room left it and nothing else.

    *seeds* maps an output path to the seed its noise crop is drawn from,
    so the crop does not depend on which worker wrote the file; a path
    missing from it draws from numpy's global RNG. Outputs in *silent* are
    written as the noise alone; ``isSilence=True`` does that to all.

    Returns the largest absolute sample written, mixture or target. Past
    1.0, a fixed-point *subtype* clipped it.
    """
    save_dir, clean_file, noise_file, snr = \
        synWavesList if num is None else synWavesList[num]
    out = Path(save_dir)
    # target_sr=None keeps the clean file's native rate; the noise is then
    # resampled to whatever the clean file turned out to be, so the two are
    # never mixed at different rates
    y_clean, sr_clean = _load(clean_file, target_sr, "clean")
    y_noise, _ = _load(noise_file, sr_clean, "noise")
    if not len(y_noise):
        raise SynthesisError(f"noise file is empty: {noise_file}. Drop it "
                             f"from the noise set.")

    room = (rooms or {}).get(out)
    if room is not None:
        from kudio.core.room import apply_rir, rir
        rt60_s, drr_db, room_seed = room
        # trim=True keeps the length, so the clean file stays a
        # sample-aligned target; the direct path sits at sample 0 of the
        # impulse response, so nothing is delayed either
        y_clean = apply_rir(
            y_clean, rir(int(sr_clean), rt60=rt60_s, drr_db=drr_db,
                         seed=room_seed))

    peak = 0.0
    target_path = (targets or {}).get(out)
    if target_path is not None:
        peak = float(np.max(np.abs(y_clean))) if y_clean.size else 0.0
        save_wave(target_path, y_clean, int(sr_clean), subtype=subtype)

    clean_pwr = np.sum(np.abs(y_clean) ** 2) / len(y_clean) if len(y_clean) else 0.0
    if not clean_pwr > 0.0:
        # the noise is scaled against the speech, so silent speech made a
        # silent mixture: a file of zeros labelled with an SNR
        raise SynthesisError(
            f"clean file has no signal to set an SNR against: {clean_file}. "
            f"It is empty or digital silence. Drop it from the clean set.")
    if len(y_noise) < len(y_clean):
        repeats = len(y_clean) // len(y_noise) + 1
        y_noise = np.tile(y_noise, repeats)[:len(y_clean)]
    else:  # random crop (no-op when both are the same length)
        span = len(y_noise) - len(y_clean) + 1
        crop_seed = (seeds or {}).get(out)
        if crop_seed is None:
            ind = np.random.randint(0, span)
        else:
            ind = int(np.random.default_rng([crop_seed, _CROP_STREAM])
                      .integers(0, span))
        y_noise = y_noise[ind:ind + len(y_clean)]

    y_noise = y_noise - np.mean(y_noise)
    spread = float(np.std(y_noise))
    if not spread > 0.0:
        # Digital silence, or a constant: there is no scaling that puts
        # this at the requested SNR, and dividing by it gave every sample
        # NaN, which int16 wrote as zero -- a mixture with no speech in it
        # and nothing said. One truncated download in a noise corpus was
        # enough.
        raise SynthesisError(
            f"noise file has no variation and cannot be scaled to "
            f"{_snr_value(snr)} dB: {noise_file}. It is digital silence or "
            f"a constant level (std 0 after removing the mean), so there "
            f"is no gain that makes it a noise floor. Drop the file from "
            f"the noise set, or trim it to the part that has signal in it.")
    noise_variance = clean_pwr / (10 ** (snr / 10))
    noise = np.sqrt(noise_variance) * y_noise / spread
    if not np.all(np.isfinite(noise)):
        # belt and braces: a near-silent file scales by a vast factor, and
        # a mixture of inf or NaN must never reach the disk as zeros
        raise SynthesisError(
            f"scaling {noise_file} to {_snr_value(snr)} dB against "
            f"{clean_file} produced values that are not finite. The noise "
            f"file is almost certainly degenerate -- check its level.")

    y_noisy = noise if isSilence or out in (silent or ()) else y_clean + noise

    link = (channels or {}).get(out)
    if link is not None:
        from kudio.effects.channel import apply_channel
        spec, link_seed = link
        # the mixture, not the speech: a wire carries whatever reached the
        # microphone, and the noise reached it too
        y_noisy = apply_channel(y_noisy, int(sr_clean), spec,
                                seed=link_seed)

    if y_noisy.size:
        peak = max(peak, float(np.max(np.abs(y_noisy))))
    save_wave(out, y_noisy, int(sr_clean), subtype=subtype)
    return peak

kudio.core.dataset

Dataset manifests: record which mixture came from which clean file.

Synthesizer already returns the pairing it wrote. Writing that down beside the audio is what stops the pairing from being re-derived later by pattern matching on filenames — the point at which one renamed directory silently trains a model against the wrong targets.

The manifest is a plain JSON list of objects, so it stays readable and diffable, and any other tool can consume it without importing kudio.

Classes:

Name	Description
Pair	One training example: a degraded file and the clean signal behind it.
Label	A named span of a recording, in seconds. end == start is a point.

Functions:

Name	Description
save_manifest	Write pairs to path as JSON. Parent directories are created.
load_manifest	Read a manifest written by save_manifest.
save_labels	Write labels as JSON, or as an Audacity label track.
load_labels	Read labels written by save_labels, or by Audacity.
split_pairs	Shuffle once, then cut into (train, validation, test).

Pair dataclass

Pair(
    noisy: str,
    clean: str,
    noise: str = "",
    snr_db: Optional[float] = None,
    rt60: Optional[float] = None,
    channel: Optional[str] = None,
    target: Optional[str] = None,
)

One training example: a degraded file and the clean signal behind it.

Frozen, therefore hashable — splits are checked for overlap by putting the pairs in a set, which silently fails to catch anything if they are not.

Methods:

Name	Description
exists	True when both sides are actually on disk.

Attributes:

Name	Type	Description
snr_db	Optional[float]	an int when the SNR is a whole number of dB, as it always was; a float
rt60	Optional[float]	reverberation time of the room the clean signal was put through, in
channel	Optional[str]	the link the mixture travelled down, as kudio.channel_tag
target	Optional[str]	the reverberant-but-clean signal, when one was written. clean is

snr_db class-attribute instance-attribute

snr_db: Optional[float] = None

an int when the SNR is a whole number of dB, as it always was; a float when it is not, rather than that float cut down to an int

rt60 class-attribute instance-attribute

rt60: Optional[float] = None

reverberation time of the room the clean signal was put through, in seconds, or None when it was not. Optional and last, so a manifest written before rooms existed still loads and a manifest written with them still loads in an older kudio -- see load_manifest.

channel class-attribute instance-attribute

channel: Optional[str] = None

the link the mixture travelled down, as kudio.channel_tag writes it ('telephone', 'mu_law+loss5%x4', ...), or None. A string rather than the spec dict it came from: this dataclass is frozen so that a split can be checked for overlap with a set, and a dict field would quietly make it unhashable again.

target class-attribute instance-attribute

target: Optional[str] = None

the reverberant-but-clean signal, when one was written. clean is still the dry target; this is the same speech with the room left on, for training that should remove the noise and leave the room.

exists

exists() -> bool

True when both sides are actually on disk.

Source code in kudio/core/dataset.py

def exists(self) -> bool:
    """True when both sides are actually on disk."""
    return Path(self.noisy).is_file() and Path(self.clean).is_file()

Label dataclass

Label(start: float, end: float, text: str = '')

A named span of a recording, in seconds. end == start is a point.

What kudio.vad finds, what an annotator marks, what a dataset builder cuts on — all the same shape, so they get one.

save_manifest

save_manifest(
    path: PathLike, pairs: Sequence[Pair]
) -> Path

Write pairs to path as JSON. Parent directories are created.

Paths are written with forward slashes, so a manifest made on Windows loads on Linux and macOS too.

Source code in kudio/core/dataset.py

def save_manifest(path: PathLike, pairs: Sequence[Pair]) -> Path:
    """Write *pairs* to *path* as JSON. Parent directories are created.

    Paths are written with forward slashes, so a manifest made on Windows
    loads on Linux and macOS too.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump([_portable(asdict(p)) for p in pairs], fh, indent=2,
                  ensure_ascii=False)
    return path

load_manifest

load_manifest(path: PathLike) -> List[Pair]

Read a manifest written by save_manifest.

Extra keys are dropped with a UserWarning rather than raising — a manifest written by a later version should still load — but a missing noisy or clean is an error, since there is no example without them.

Source code in kudio/core/dataset.py

def load_manifest(path: PathLike) -> List[Pair]:
    """Read a manifest written by :func:`save_manifest`.

    Extra keys are dropped with a ``UserWarning`` rather than raising — a
    manifest written by a later version should still load — but a missing
    ``noisy`` or ``clean`` is an error, since there is no example without them.
    """
    with open(path, "r", encoding="utf-8") as fh:
        rows = json.load(fh)
    if not isinstance(rows, list):
        raise ValueError(f"manifest is not a JSON list: {path}")

    known = {f.name for f in fields(Pair)}
    out: List[Pair] = []
    dropped: Dict[str, int] = {}
    for n, row in enumerate(rows, 1):
        if not isinstance(row, dict):
            raise ValueError(f"manifest {path}: row {n} is not an object: {row!r}")
        missing = [k for k in ('noisy', 'clean') if not row.get(k)]
        if missing:
            raise ValueError(f"manifest {path}: row {n} has no "
                             f"{' or '.join(missing)}; every row needs both "
                             f"'noisy' and 'clean'")
        unknown = set(row) - known
        for key in unknown:
            dropped[key] = dropped.get(key, 0) + 1
        out.append(Pair(**{k: v for k, v in row.items() if k in known}))
    if dropped:
        warnings.warn(f"manifest {path}: ignored unknown key(s) "
                      f"{', '.join(sorted(dropped))}", UserWarning, stacklevel=2)
    return out

save_labels

save_labels(
    path: PathLike,
    labels: Sequence[Label],
    fmt: Optional[str] = None,
) -> Path

Write labels as JSON, or as an Audacity label track.

>>> kudio.save_labels("clip.labels.json", labels)
>>> kudio.save_labels("clip.txt", labels)          # Audacity, ELAN, ...

The format follows the suffix unless you say otherwise: .txt / .tsv means Audacity's tab-separated start end text, anything else means JSON. Audacity's format is three columns of plain text that half the audio world can already read, which is worth more than a private format that only kudio understands.

Source code in kudio/core/dataset.py

def save_labels(path: PathLike, labels: Sequence[Label],
                fmt: Optional[str] = None) -> Path:
    """Write *labels* as JSON, or as an **Audacity label track**.

    >>> kudio.save_labels("clip.labels.json", labels)
    >>> kudio.save_labels("clip.txt", labels)          # Audacity, ELAN, ...

    The format follows the suffix unless you say otherwise: ``.txt`` / ``.tsv``
    means Audacity's tab-separated ``start end text``, anything else means
    JSON. Audacity's format is three columns of plain text that half the audio
    world can already read, which is worth more than a private format that only
    kudio understands.
    """
    path, fmt = _label_path(path, fmt)
    path.parent.mkdir(parents=True, exist_ok=True)
    if fmt == 'json':
        with open(path, "w", encoding="utf-8") as fh:
            json.dump([asdict(x) for x in labels], fh, indent=2,
                      ensure_ascii=False)
        return path

    with open(path, "w", encoding="utf-8", newline="\n") as fh:
        for label in labels:
            # tabs are the separator, so a tab inside a label would split the
            # row -- Audacity has the same problem and also just strips them
            text = label.text.replace("\t", " ").replace("\n", " ")
            # microseconds: 60x finer than a sample at 16 kHz and still 5x
            # finer at 192 kHz, so the rounding cannot move a boundary
            fh.write(f"{label.start:.6f}\t{label.end:.6f}\t{text}\n")
    return path

load_labels

load_labels(
    path: PathLike, fmt: Optional[str] = None
) -> List[Label]

Read labels written by save_labels, or by Audacity.

Audacity follows a label that has a frequency range with a second row, \<TAB>low<TAB>high. Those rows are skipped: a Label holds times, not frequencies.

Raises:

Type	Description
ValueError	on a malformed row, naming the line — a label file is usually hand-edited at some point, and "line 12 has 2 columns" is worth more than an IndexError.

Source code in kudio/core/dataset.py

def load_labels(path: PathLike, fmt: Optional[str] = None) -> List[Label]:
    """Read labels written by :func:`save_labels`, or by Audacity.

    Audacity follows a label that has a frequency range with a second row,
    ``\\<TAB>low<TAB>high``. Those rows are skipped: a :class:`Label` holds
    times, not frequencies.

    :raises ValueError: on a malformed row, naming the line — a label file is
        usually hand-edited at some point, and "line 12 has 2 columns" is worth
        more than an IndexError.
    """
    path, fmt = _label_path(path, fmt)
    if fmt == 'json':
        with open(path, "r", encoding="utf-8") as fh:
            rows = json.load(fh)
        if not isinstance(rows, list):
            raise ValueError(f"label file is not a JSON list: {path}")
        known = {f.name for f in fields(Label)}
        return [Label(**{k: v for k, v in row.items() if k in known})
                for row in rows]

    labels = []
    with open(path, "r", encoding="utf-8") as fh:
        for n, line in enumerate(fh, 1):
            if not line.strip() or line.startswith("\\"):
                continue
            parts = line.rstrip("\n").split("\t")
            if len(parts) < 2:
                raise ValueError(
                    f"{path} line {n}: expected 'start<TAB>end<TAB>text', "
                    f"found {len(parts)} column(s)")
            try:
                start, end = float(parts[0]), float(parts[1])
            except ValueError:
                raise ValueError(
                    f"{path} line {n}: {parts[0]!r} and {parts[1]!r} are not "
                    f"times in seconds") from None
            labels.append(Label(start=start, end=end,
                                text="\t".join(parts[2:]).strip()))
    return labels

split_pairs

split_pairs(
    pairs: Sequence[Pair],
    val_split: float = 0.1,
    test_split: float = 0.1,
    seed: Optional[int] = None,
) -> Tuple[List[Pair], List[Pair], List[Pair]]

Shuffle once, then cut into (train, validation, test).

An example listed more than once is cut down to one first, with a warning, since dealing the copies out would put the same file in two splits and then score it as if it had been held out. Rows that disagree about one mixture file raise instead — see _distinct_examples.

Splitting a tiny set by percentage can leave nothing to train on, so when the shares would leave training empty, the validation and test sets are given up instead: an empty validation set is visible, and a run with no training data at all is not a run. Anything above that is split as asked, even when training ends up with a single example.

Source code in kudio/core/dataset.py

def split_pairs(pairs: Sequence[Pair], val_split: float = 0.1,
                test_split: float = 0.1, seed: Optional[int] = None
                ) -> Tuple[List[Pair], List[Pair], List[Pair]]:
    """Shuffle once, then cut into ``(train, validation, test)``.

    An example listed more than once is cut down to one first, with a
    warning, since dealing the copies out would put the same file in two
    splits and then score it as if it had been held out. Rows that disagree
    about one mixture file raise instead — see :func:`_distinct_examples`.

    Splitting a tiny set by percentage can leave nothing to train on, so when
    the shares would leave training empty, the validation and test sets are
    given up instead: an empty validation set is visible, and a run with no
    training data at all is not a run. Anything above that is split as asked,
    even when training ends up with a single example.
    """
    if not 0.0 <= val_split < 1.0 or not 0.0 <= test_split < 1.0:
        raise ValueError("splits must be in [0, 1)")
    if val_split + test_split >= 1.0:
        raise ValueError(
            f"val_split + test_split must leave something to train on, "
            f"got {val_split} + {test_split}")

    items = _distinct_examples(pairs)
    rng = np.random.default_rng(seed)
    rng.shuffle(items)

    n = len(items)
    n_test = int(round(n * test_split))
    n_val = int(round(n * val_split))
    if n - n_val - n_test < 1:
        n_val = n_test = 0
    return (items[n_val + n_test:], items[:n_val], items[n_val:n_val + n_test])