Enhancement

kudio.enhance.spectral

Statistical single-channel speech enhancement, as a set of interchangeable parts.

Every method here is the same three decisions:

1. estimate the noise power spectrum, frame by frame;
2. from that, estimate the a priori SNR — how much speech is in this bin;
3. turn that into a gain between 0 and 1 and multiply the spectrum by it.

trad_enhance fuses all three into one function with no parameters, so there is no way to ask what any of them contributed. Splitting them apart is what makes the methods comparable: swap the gain rule and hold the noise estimator still, and the difference you hear is the gain rule.

What separates them in practice:

============ =============================================================== specsub Subtract the noise spectrum. Cheapest, and the reason "musical noise" has a name: isolated surviving bins warble. multiband Spectral subtraction with a per-band over-subtraction factor. Colored noise is not equally loud everywhere, and one global factor over-suppresses where it is quiet. wiener Gain xi / (1 + xi). Smooth, so much less musical noise, at the cost of muffling the speech at low SNR. mmse_stsa Ephraim & Malah (1984): MMSE estimate of the spectral amplitude, assuming Gaussian coefficients. Keeps more speech than Wiener at the same suppression. logmmse Ephraim & Malah (1985): MMSE in the log domain, which is closer to how loudness is perceived. The usual default. omlsa logmmse weighted by a speech-presence probability (Cohen 2001). Suppresses hard where there is probably no speech at all, gently where there might be. spectral_gate A threshold mask rather than a statistical estimator: bins below (noise mean + n sigma) are attenuated. Not principled, very predictable, and good on steady broadband noise. ============ ===============================================================

The decision-directed a priori SNR (Ephraim & Malah) is shared by wiener / mmse_stsa / logmmse / omlsa:

xi(k, l) = alpha * A(k, l-1)^2 / lambda_d + (1 - alpha) * max(gamma - 1, 0)

with alpha = 0.98: it is the smoothing that removes musical noise, and also what makes these methods lag a sudden onset by a frame or two.

Two things to know before reaching for any of them.

They are for noisy audio, and they damage clean audio. Every one trades speech distortion for noise removal. On the test suite's harmonic clip over white noise, the seven over mcra raise the SNR by +7.8 dB on average at 0 dB in and +6.1 at 5 dB; at 10 dB the average is +3.8 and one of them already does harm, and at 20 dB all seven measure worse than doing nothing (−4.1 dB). compare_enhancers(reference=…) reports delta_snr_db precisely so that shows up rather than being assumed away.

They need to hear the noise on its own. Noise estimation works from the quiet parts, so a clip that is wall-to-wall speech gives them almost nothing to measure. At 5 dB SNR, logmmse gained +7.1 dB on the clip with pauses and +0.1 dB on one without. If a recording has no pauses, noise='initial' with a leading second of room tone beats anything adaptive: +9.5 dB against +2.2 for mcra.

>>> clean = kudio.spectral_enhance(y, sr)                    # logmmse
>>> clean = kudio.spectral_enhance(y, sr, method='omlsa', noise='quantile')
>>> for r in kudio.compare_enhancers(y, sr, reference=truth):
...     print(r.method, r.si_sdr_db)

Classes:

Name	Description
Param	One tunable knob, described well enough to build a control from it.
Method	An enhancement method: what it is, and what it lets you turn.
NoiseTracker	A noise estimator as state plus one update, not a whole-file pass.
CustomEnhancer	Anything that denoises, wrapped so it can be ranked with the rest.
EnhanceResult	One run's output, with everything needed to rank it against another.
EnhanceFolderResult	What enhance_folder did.

Functions:

Name	Description
make_noise_tracker	A stateful tracker for method.
estimate_noise_psd	Noise power spectrum for every frame of power (frames, bins).
make_gain_rule	The per-frame gain function for one of RECURSIVE_METHODS.
spectral_enhance	Enhance y with one of the methods in METHODS.
compare_enhancers	Run several methods over the same clip and measure each one.
enhance_folder	Denoise every .wav file under src into dst.

Attributes:

Name	Type	Description
CUSTOM_NOISE		What a custom enhancer's noise column says: it brought its own, and
METHODS	Dict[str, Method]	Every gain rule by name. Each one describes its own parameters -- range,
NOISE_ESTIMATORS	Dict[str, str]	The four noise estimators by name, each with a sentence on what it assumes

CUSTOM_NOISE module-attribute

CUSTOM_NOISE = 'own'

What a custom enhancer's noise column says: it brought its own, and naming one of kudio's estimators beside it would be a lie.

METHODS module-attribute

METHODS: Dict[str, Method] = {
    m.name: m
    for m in (
        Method(
            "specsub",
            "Spectral subtraction",
            "Subtract the noise spectrum. Cheapest, and the origin of musical noise: isolated surviving bins warble between frames.",
            (_OVERSUB, _FLOOR, _GAIN_FLOOR),
            "Boll 1979",
        ),
        Method(
            "multiband",
            "Multi-band spectral subtraction",
            "Spectral subtraction with a per-band over-subtraction factor set from each band's own SNR — colored noise is not equally loud everywhere, and one global factor over-suppresses where it is not.",
            (
                Param(
                    "bands",
                    "bands",
                    6,
                    2,
                    32,
                    1,
                    0,
                    help="Linearly spaced. More bands track colored noise more closely and are noisier to estimate.",
                ),
                _FLOOR,
                _GAIN_FLOOR,
            ),
            "Kamath & Loizou 2002",
        ),
        Method(
            "wiener",
            "Wiener (decision-directed)",
            "Gain xi/(1+xi). Smooth, so far less musical noise than subtraction — it muffles the speech at low SNR instead.",
            (_ALPHA, _GAIN_FLOOR),
        ),
        Method(
            "mmse_stsa",
            "MMSE-STSA",
            "MMSE estimate of the spectral amplitude under a Gaussian model. Keeps more speech than Wiener at the same suppression.",
            (_ALPHA, _GAIN_FLOOR),
            "Ephraim & Malah 1984",
        ),
        Method(
            "logmmse",
            "log-MMSE",
            "MMSE in the log-spectral domain, which is closer to how loudness is perceived. The usual default, and hard to beat without a model.",
            (_ALPHA, _GAIN_FLOOR),
            "Ephraim & Malah 1985",
        ),
        Method(
            "omlsa",
            "OM-LSA",
            "log-MMSE weighted by the probability that there is any speech in the bin at all: suppress hard where there is probably none, gently where there might be.",
            (
                _ALPHA,
                _GAIN_FLOOR,
                Param(
                    "q",
                    "P(no speech)",
                    0.3,
                    0.01,
                    0.95,
                    0.05,
                    2,
                    help="A priori probability that a bin holds no speech. Higher suppresses more and risks clipping onsets.",
                ),
            ),
            "Cohen 2001",
        ),
        Method(
            "spectral_gate",
            "Spectral gate",
            "A threshold mask, not a statistical estimator: attenuate every bin below (noise mean + n sigma), then smooth the mask so the edges do not click. Unprincipled, very predictable.",
            (
                Param(
                    "n_std_thresh",
                    "threshold",
                    1.5,
                    0.0,
                    6.0,
                    0.1,
                    1,
                    " sigma",
                    help="How far above the noise floor a bin has to be to be kept.",
                ),
                Param(
                    "prop_decrease",
                    "strength",
                    1.0,
                    0.0,
                    1.0,
                    0.05,
                    2,
                    help="1.0 applies the mask in full; lower mixes the original back in, which trades noise for artefacts.",
                ),
                Param(
                    "freq_smooth_hz",
                    "smooth (freq)",
                    500.0,
                    0.0,
                    4000.0,
                    50.0,
                    0,
                    " Hz",
                    help="Blur the mask across frequency. Unsmoothed masks are what musical noise sounds like.",
                ),
                Param(
                    "time_smooth_ms",
                    "smooth (time)",
                    50.0,
                    0.0,
                    500.0,
                    10.0,
                    0,
                    " ms",
                    help="Blur the mask across time.",
                ),
            ),
            "Spectral gating, after noisereduce",
        ),
    )
}

Every gain rule by name. Each one describes its own parameters -- range, default, step, unit and a sentence on what turning it costs -- so a GUI or a command line can build its controls from here instead of restating them.

NOISE_ESTIMATORS module-attribute

NOISE_ESTIMATORS: Dict[str, str] = {
    "mcra": "Minima-controlled recursive averaging: tracks the noise while speech is present, by watching each bin's running minimum. The general-purpose choice, and what trad_enhance uses. It starts from the opening frame, so after a quiet opening it can leave noise in some bins for the first second or two.",
    "quantile": "A low quantile of each bin over the whole file, scaled to the mean of steady noise (the 15th percentile of noise power is 0.16 of its mean, 7.9 dB low). Offline only — it needs to see the future — and usually the most accurate of these when the noise is steady and the file is not one continuous shout.",
    "initial": "Average of the first few frames, assumed to be room tone. Exact when that is true and badly wrong when it is not.",
    "minimum": "Running minimum over a sliding window, bias-compensated. Follows non-stationary noise faster than MCRA and is noisier.",
}

The four noise estimators by name, each with a sentence on what it assumes about the recording. Pick one with spectral_enhance(..., noise=name).

RECURSIVE_METHODS module-attribute

RECURSIVE_METHODS = (
    "wiener",
    "mmse_stsa",
    "logmmse",
    "omlsa",
)

Methods whose gain is a function of (xi, gamma) alone, and which therefore run inside DecisionDirected. The other three look at the whole spectrum at once and are handled separately.

Param dataclass

Param(
    name: str,
    label: str,
    default: float,
    low: float,
    high: float,
    step: float = 0.01,
    decimals: int = 2,
    suffix: str = "",
    help: str = "",
)

One tunable knob, described well enough to build a control from it.

Method dataclass

Method(
    name: str,
    label: str,
    summary: str,
    params: Tuple[Param, ...] = (),
    reference: str = "",
)

An enhancement method: what it is, and what it lets you turn.

Attributes:

Name	Type	Description
reference	str	cited source, for the ones that have one

reference class-attribute instance-attribute

reference: str = ''

cited source, for the ones that have one

NoiseTracker

A noise estimator as state plus one update, not a whole-file pass.

Written as a batch function, a tracker cannot be used on a live stream; written twice — once batch, once streaming — the two drift. So the update rule lives here once, and the offline path is a loop over it.

quantile is the exception and is deliberately not a tracker: it needs to see the future.

Methods:

Name	Description
update	One frame in, that frame's noise PSD out.

Attributes:

Name	Type	Description
streamable		can this run on audio that has not finished arriving?

streamable class-attribute instance-attribute

streamable = True

can this run on audio that has not finished arriving?

update

update(power: ndarray) -> ndarray

One frame in, that frame's noise PSD out.

Source code in kudio/enhance/spectral.py

def update(self, power: np.ndarray) -> np.ndarray:
    """One frame in, that frame's noise PSD out."""
    raise NotImplementedError

DecisionDirected

DecisionDirected(gain_fn, alpha: float = 0.98)

The a priori SNR estimator of Ephraim & Malah, as state plus an update.

gain_fn(xi, gamma) -> gain is the only thing that changes between Wiener, MMSE-STSA, log-MMSE and OM-LSA; everything around it — the a posteriori SNR, the smoothing, the feedback of the previous frame's estimate — is identical, and is why they can be compared at all. It is also already recursive, which is what makes those four streamable for free.

Source code in kudio/enhance/spectral.py

def __init__(self, gain_fn, alpha: float = 0.98):
    self.gain_fn, self.alpha = gain_fn, alpha
    self._previous = None             # |A(k, l-1)|^2, the fed-back estimate

CustomEnhancer dataclass

CustomEnhancer(name: str, fn: Any, label: str = '')

Anything that denoises, wrapped so it can be ranked with the rest.

The statistical methods are named by a string; a trained model, a wrapper around someone else's library, or a one-off experiment is a function. Both belong in the same table — "is the model actually better than log-MMSE" is the question people have, and answering it by running two tools and comparing two printouts is how it goes unanswered.

>>> from kudio_enhance.inference import Enhancer
>>> model = Enhancer.load("runs/exp1")
>>> kudio.compare_enhancers(y, sr, ['logmmse', ('model', model.enhance)])

EnhanceResult dataclass

EnhanceResult(
    method: str,
    audio: ndarray,
    seconds: float,
    lufs: float,
    noise_floor_db: float,
    noise: str = "mcra",
    snr_db: Optional[float] = None,
    si_sdr_db: Optional[float] = None,
    delta_snr_db: Optional[float] = None,
    pesq: Optional[float] = None,
    stoi: Optional[float] = None,
    error: Optional[str] = None,
)

One run's output, with everything needed to rank it against another.

A run is a (method, noise estimator) pair, not just a method: the two are separate choices of comparable size, so collapsing them into one name would hide half of what produced the result.

The reference metrics are None when no clean reference was given — the normal case for a real recording, and the reason noise_floor_db and lufs are always filled in.

Attributes:

Name	Type	Description
label	str	method + noise — what actually distinguishes one run.

label property

label: str

method + noise — what actually distinguishes one run.

EnhanceFolderResult dataclass

EnhanceFolderResult(
    written: int,
    total: int,
    outputs: List[Path],
    failed: List[Tuple[Path, str]],
    floors: Tuple[Tuple[Any, float, float], ...] = (),
)

Bases: ConvertResult

What enhance_folder did.

A kudio.ConvertResult — same written / total / failed — plus floors, the before-and-after noise floor of each file when it was asked for. Subclassing rather than inventing a parallel type keeps one shape for "I processed a folder".

Methods:

Name	Description
mean_reduction_db	Average drop in noise floor, or None without a report.

Attributes:

Name	Type	Description
floors	Tuple[Tuple[Any, float, float], ...]	(path, before_dbfs, after_dbfs) per file; empty unless report=True

floors class-attribute instance-attribute

floors: Tuple[Tuple[Any, float, float], ...] = ()

(path, before_dbfs, after_dbfs) per file; empty unless report=True

mean_reduction_db

mean_reduction_db() -> Optional[float]

Average drop in noise floor, or None without a report.

Source code in kudio/enhance/spectral.py

def mean_reduction_db(self) -> Optional[float]:
    """Average drop in noise floor, or ``None`` without a report."""
    usable = [(b, a) for _, b, a in self.floors
              if b is not None and a is not None
              and np.isfinite(b) and np.isfinite(a)]
    if not usable:
        return None
    return float(np.mean([b - a for b, a in usable]))

make_noise_tracker

make_noise_tracker(
    method: str = "mcra", **kwargs
) -> NoiseTracker

A stateful tracker for method.

Raises:

Type	Description
FeatureError	for quantile, which is not a tracker — it is a statistic over the whole file, and there is no honest way to compute it one frame at a time.

Source code in kudio/enhance/spectral.py

def make_noise_tracker(method: str = 'mcra', **kwargs) -> NoiseTracker:
    """A stateful tracker for *method*.

    :raises FeatureError: for ``quantile``, which is not a tracker — it is a
        statistic over the whole file, and there is no honest way to compute it
        one frame at a time.
    """
    if method not in NOISE_ESTIMATORS:
        raise FeatureError(
            f"unknown noise estimator {method!r}. "
            f"Choose from {', '.join(sorted(NOISE_ESTIMATORS))}")
    if method == 'quantile':
        raise FeatureError(
            "'quantile' takes a low quantile over the whole file, so it cannot "
            "run on audio that has not finished arriving. Use 'mcra' (adaptive) "
            "or 'initial' (assumes the opening is room tone) for streaming.")
    cls = {'mcra': _MCRATracker, 'initial': _InitialTracker,
           'minimum': _MinimumTracker}[method]
    _check_noise_kwargs(method, kwargs, cls)
    return cls(**kwargs)

estimate_noise_psd

estimate_noise_psd(
    power: ndarray, method: str = "mcra", **kwargs
) -> ndarray

Noise power spectrum for every frame of power (frames, bins).

See NOISE_ESTIMATORS for what each one assumes. It is as large a lever as the gain rule, because a gain rule can only be as good as the noise it is told about.

Source code in kudio/enhance/spectral.py

def estimate_noise_psd(power: np.ndarray, method: str = 'mcra',
                       **kwargs) -> np.ndarray:
    """Noise power spectrum for every frame of ``power`` ``(frames, bins)``.

    See :data:`NOISE_ESTIMATORS` for what each one assumes. It is as large a
    lever as the gain rule, because a gain rule can only be as good as the
    noise it is told about.
    """
    if method not in NOISE_ESTIMATORS:
        raise FeatureError(
            f"unknown noise estimator {method!r}. "
            f"Choose from {', '.join(sorted(NOISE_ESTIMATORS))}")
    power = np.maximum(np.asarray(power, dtype=np.float64), _EPS)
    if method == 'quantile':
        _check_noise_kwargs(method, kwargs)
        q = float(kwargs.get('quantile', 0.15))
        if not 0.0 < q < 1.0:
            raise FeatureError(f"quantile must be between 0 and 1, got {q}")
        # Steady noise power in a bin is exponentially distributed, so its
        # q-th quantile is -ln(1 - q) times its mean: the 15th percentile sat
        # 7.9 dB under the noise it was meant to describe. Dividing that out
        # took this estimator from the weakest of the four on the test clips
        # (2.8 dB of SI-SDR gained, averaged over the gain rules) to the best
        # (7.8 to 8.4).
        level = np.quantile(power, q, axis=0) / -np.log1p(-q)
        return np.tile(level, (power.shape[0], 1))

    tracker = make_noise_tracker(method, **kwargs)
    return np.stack([tracker.update(frame) for frame in power])

make_gain_rule

make_gain_rule(
    method: str, settings: Optional[dict] = None
) -> Callable[[ndarray, ndarray], ndarray]

The per-frame gain function for one of RECURSIVE_METHODS.

Exposed so the streaming enhancer resolves a method name the same way the offline one does, instead of keeping its own mapping to fall out of step.

Source code in kudio/enhance/spectral.py

def make_gain_rule(method: str, settings: Optional[dict] = None
                   ) -> Callable[[np.ndarray, np.ndarray], np.ndarray]:
    """The per-frame gain function for one of :data:`RECURSIVE_METHODS`.

    Exposed so the streaming enhancer resolves a method name the same way the
    offline one does, instead of keeping its own mapping to fall out of step.
    """
    if method not in RECURSIVE_METHODS:
        raise FeatureError(
            f"{method!r} is not a per-frame gain rule. "
            f"Choose from {', '.join(RECURSIVE_METHODS)}")
    settings = {**METHODS[method].defaults(), **(settings or {})}
    if method == 'omlsa':
        g_min = 10.0 ** (settings.get('gain_floor_db', -25.0) / 20.0)
        return _gain_omlsa_factory(settings['q'], g_min)
    return {'wiener': _gain_wiener,
            'mmse_stsa': _gain_mmse_stsa,
            'logmmse': _gain_logmmse}[method]

spectral_enhance

spectral_enhance(
    y: ndarray,
    sr: int,
    method: str = "logmmse",
    *,
    noise: str = "mcra",
    n_fft: int = 512,
    hop_length: Optional[int] = None,
    window: str = "hann",
    noise_kwargs: Optional[dict] = None,
    **params,
) -> ndarray

Enhance y with one of the methods in METHODS.

>>> clean = kudio.spectral_enhance(y, sr)                       # logmmse
>>> clean = kudio.spectral_enhance(y, sr, method='omlsa', q=0.4)
>>> clean = kudio.spectral_enhance(y, sr, noise='quantile')

The output is the same length as the input, so it lines up sample for sample with what it came from.

Parameters:

Name	Type	Description	Default
method	str	see METHODS; each entry lists its own parameters.	'logmmse'
noise	str	see NOISE_ESTIMATORS. The estimator matters at least as much as the gain rule.	'mcra'
params		method parameters, defaulted from METHODS[method].	{}

Raises:

Type	Description
FeatureError	on an unknown method, non-mono audio, samples that are not finite, or a clip too short to frame.

Source code in kudio/enhance/spectral.py

def spectral_enhance(y: np.ndarray, sr: int, method: str = 'logmmse', *,
                     noise: str = 'mcra',
                     n_fft: int = 512,
                     hop_length: Optional[int] = None,
                     window: str = 'hann',
                     noise_kwargs: Optional[dict] = None,
                     **params) -> np.ndarray:
    """Enhance *y* with one of the methods in :data:`METHODS`.

    >>> clean = kudio.spectral_enhance(y, sr)                       # logmmse
    >>> clean = kudio.spectral_enhance(y, sr, method='omlsa', q=0.4)
    >>> clean = kudio.spectral_enhance(y, sr, noise='quantile')

    The output is the **same length** as the input, so it lines up sample for
    sample with what it came from.

    :param method: see :data:`METHODS`; each entry lists its own parameters.
    :param noise: see :data:`NOISE_ESTIMATORS`. The estimator matters at least
        as much as the gain rule.
    :param params: method parameters, defaulted from ``METHODS[method]``.
    :raises FeatureError: on an unknown method, non-mono audio, samples that
        are not finite, or a clip too short to frame.
    """
    if method not in METHODS:
        raise FeatureError(
            f"unknown method {method!r}. Choose from "
            f"{', '.join(sorted(METHODS))}")
    if sr <= 0:
        raise FeatureError(f"spectral_enhance() needs a positive rate, got {sr}")

    y = np.ascontiguousarray(np.asarray(as_float(y), dtype=np.float32).squeeze())
    if y.ndim != 1:
        raise FeatureError(
            f"spectral_enhance() takes mono audio, got shape {y.shape}")
    if y.size < n_fft:
        raise FeatureError(
            f"clip is shorter than one {n_fft}-sample frame ({y.size} samples); "
            f"pass a smaller n_fft")
    if not np.all(np.isfinite(y)):
        # librosa's own refusal names neither the function nor the fix
        raise FeatureError(
            f"spectral_enhance() got {int(np.count_nonzero(~np.isfinite(y)))} "
            f"sample(s) that are not finite (NaN or inf); repair the clip first")

    hop_length = hop_length or n_fft // 4
    settings = METHODS[method].defaults()
    unknown = set(params) - set(settings)
    if unknown:
        raise FeatureError(
            f"{method!r} does not take {', '.join(sorted(unknown))}. "
            f"It takes: {', '.join(sorted(settings)) or 'nothing'}")
    settings.update(params)
    _check_ranges(method, settings)

    from kudio.core.stft import STFT
    stft = STFT(sr=sr, n_fft=n_fft, hop_length=hop_length, window=window)
    spec = stft.analyse(y)                                  # (frames, bins)
    power = np.maximum(np.abs(spec) ** 2, _EPS)
    noise_psd = estimate_noise_psd(power, noise, **(noise_kwargs or {}))

    g_min = 10.0 ** (settings.get('gain_floor_db', -25.0) / 20.0)

    if method == 'specsub':
        gains = _subtraction_gains(power, noise_psd,
                                   settings['over_subtraction'],
                                   settings['spectral_floor'])
    elif method == 'multiband':
        gains = _multiband_gains(power, noise_psd, settings['bands'],
                                 settings['spectral_floor'])
    elif method == 'spectral_gate':
        gains = _spectral_gate_gains(
            power, noise_psd, n_std_thresh=settings['n_std_thresh'],
            prop_decrease=settings['prop_decrease'],
            freq_smooth_hz=settings['freq_smooth_hz'],
            time_smooth_ms=settings['time_smooth_ms'],
            sr=sr, n_fft=n_fft, hop_length=hop_length)
        g_min = 0.0                     # the gate's own strength is the floor
    else:
        gains = _decision_directed(power, noise_psd, settings['alpha'],
                                   make_gain_rule(method, settings))

    gains = np.clip(gains, g_min, 1.0)
    return stft.synthesise(spec * gains, length=len(y))

compare_enhancers

compare_enhancers(
    y: ndarray,
    sr: int,
    methods: Optional[Sequence[str]] = None,
    *,
    noises: Optional[Sequence[str]] = None,
    reference: Optional[ndarray] = None,
    metrics: bool = True,
    pesq_mode: str = "auto",
    pesq_scale: str = "lqo",
    **common,
) -> List[EnhanceResult]

Run several methods over the same clip and measure each one.

>>> for result in kudio.compare_enhancers(y, sr, reference=clean):
...     print(result)

Both axes can be swept. methods chooses gain rules, noises chooses noise estimators, and giving both compares the full grid:

>>> kudio.compare_enhancers(y, sr, ['logmmse'], noises=list(NOISE_ESTIMATORS))
>>> kudio.compare_enhancers(y, sr, noises=['mcra', 'quantile'])   # 2 x all

That second axis is not decoration: the estimator moves the result as much as the gain rule does, so a comparison that sweeps only methods is looking at half the problem. Which of the two dominates depends on the clip, which is the reason to measure rather than assume.

With a reference you get SNR, SI-SDR and (where installed) PESQ / STOI, plus delta_snr_db: the improvement over the untouched input, which is the number that actually answers "did this help". Without one you get the residual noise floor and the loudness, which is what a real recording allows.

A run that raises is reported with error set rather than taking the whole comparison down — a comparison exists to find out which ones work.

Parameters:

Name	Type	Description	Default
common		passed to every run (n_fft=, alpha=, ...). Anything a given method does not accept is dropped for that method, so one call can hold the framing still across all of them.	{}

Raises:

Type	Description
FeatureError	on an unknown method or estimator, on passing noise= and noises= together, or on a clip with NaN or inf in it.

Source code in kudio/enhance/spectral.py

def compare_enhancers(y: np.ndarray, sr: int,
                      methods: Optional[Sequence[str]] = None, *,
                      noises: Optional[Sequence[str]] = None,
                      reference: Optional[np.ndarray] = None,
                      metrics: bool = True,
                      pesq_mode: str = 'auto',
                      pesq_scale: str = 'lqo',
                      **common) -> List[EnhanceResult]:
    """Run several methods over the same clip and measure each one.

    >>> for result in kudio.compare_enhancers(y, sr, reference=clean):
    ...     print(result)

    **Both axes can be swept.** `methods` chooses gain rules, `noises` chooses
    noise estimators, and giving both compares the full grid:

    >>> kudio.compare_enhancers(y, sr, ['logmmse'], noises=list(NOISE_ESTIMATORS))
    >>> kudio.compare_enhancers(y, sr, noises=['mcra', 'quantile'])   # 2 x all

    That second axis is not decoration: the estimator moves the result as
    much as the gain rule does, so a comparison that sweeps only `methods` is
    looking at half the problem. Which of the two dominates depends on the
    clip, which is the reason to measure rather than assume.

    With a *reference* you get SNR, SI-SDR and (where installed) PESQ / STOI,
    plus **`delta_snr_db`: the improvement over the untouched input**, which is
    the number that actually answers "did this help". Without one you get the
    residual noise floor and the loudness, which is what a real recording
    allows.

    A run that raises is reported with `error` set rather than taking the whole
    comparison down — a comparison exists to find out which ones work.

    :param common: passed to every run (`n_fft=`, `alpha=`, ...). Anything a
        given method does not accept is dropped for that method, so one call
        can hold the framing still across all of them.
    :raises FeatureError: on an unknown method or estimator, on passing
        ``noise=`` and ``noises=`` together, or on a clip with NaN or inf in it.
    """
    entries = [_as_enhancer(m) for m in
               (methods if methods is not None else list(METHODS))]
    names = [e for e in entries if isinstance(e, str)]

    if noises is not None and 'noise' in common:
        raise FeatureError(
            "pass noises=[...] to sweep the estimator, or noise='...' to fix "
            "it — not both")
    estimators = list(noises) if noises is not None \
        else [common.pop('noise', 'mcra')]
    unknown = [n for n in estimators if n not in NOISE_ESTIMATORS]
    if unknown:
        raise FeatureError(f"unknown noise estimator(s): {', '.join(unknown)}")

    taken = {'n_fft', 'hop_length', 'window', 'noise_kwargs'}
    for name in names:
        if name in METHODS:
            taken |= set(METHODS[name].defaults())
    stray = set(common) - taken
    if stray:
        # dropping a key per method is deliberate -- alpha means nothing to
        # specsub -- but one no method takes is a typo, and alpah=0.5 used to
        # run every method on its defaults
        raise FeatureError(f"no method here takes {', '.join(sorted(stray))}")

    y = np.asarray(as_float(y), dtype=np.float32)
    if not np.all(np.isfinite(y)):
        # every run would fail on it, one line each, and the table would
        # still come back as if something had been compared
        raise FeatureError(
            f"compare_enhancers() got {int(np.count_nonzero(~np.isfinite(y)))} "
            f"sample(s) that are not finite (NaN or inf); repair the clip first")
    baseline = (_reference_metrics(reference, y, sr, metrics, pesq_mode,
                                   pesq_scale)
                if reference is not None else None)

    # Warm up before timing anything. librosa's STFT is JIT-compiled and scipy's
    # Bessel/exponential-integral functions import lazily, so whichever run
    # went first was charged ~3 s of one-time cost -- in a table whose entire
    # purpose is comparing runs against each other.
    _warm_up(y, sr, names, estimators[0])

    results: List[EnhanceResult] = []
    for entry in entries:
        if isinstance(entry, CustomEnhancer):
            # it brings its own everything, so the estimator sweep does not
            # apply and saying 'mcra' beside it would be a lie
            results.append(_run_one(entry.name, CUSTOM_NOISE,
                                    lambda: entry(y, sr),
                                    y, sr, reference, baseline, metrics,
                                    pesq_mode, pesq_scale))
            continue

        allowed = set(METHODS[entry].defaults())
        kwargs = {k: v for k, v in common.items()
                  if k in allowed or k in ('n_fft', 'hop_length', 'window',
                                           'noise_kwargs')}
        for estimator in estimators:
            results.append(_run_one(
                entry, estimator,
                lambda name=entry, est=estimator:
                    spectral_enhance(y, sr, name, noise=est, **kwargs),
                y, sr, reference, baseline, metrics,
                pesq_mode, pesq_scale))
    return results

enhance_folder

enhance_folder(
    src: Any,
    dst: PathLike,
    method: Union[
        str, Callable, Tuple[str, Callable]
    ] = "logmmse",
    *,
    noise: str = "mcra",
    n_fft: int = 512,
    hop_length: Optional[int] = None,
    window: str = "hann",
    noise_kwargs: Optional[dict] = None,
    sr: Optional[int] = None,
    subtype: str = "PCM_16",
    overwrite: bool = False,
    progress: Optional[
        Callable[[int, int, Path], None]
    ] = None,
    on_error: str = "collect",
    report: bool = False,
    **params,
) -> EnhanceFolderResult

Denoise every .wav file under src into dst.

The counterpart of kudio.convert_folder, and it mirrors the input's directory structure the same way -- a folder from itself, a list or a .txt list from the deepest folder its files share -- so two files with the same name in different subfolders do not collide.

>>> result = kudio.enhance_folder("noisy/", "clean/", 'logmmse')
>>> print(result)
412/412 written, 0 failed

Parameters:

Name	Type	Description	Default
method	Union[str, Callable, Tuple[str, Callable]]	a name from METHODS, or anything CustomEnhancer accepts — a callable, or (name, callable). A trained model is the obvious second case.	'logmmse'
n_fft	int	as in spectral_enhance, like the three below. A callable method ignores all four.	512
hop_length	Optional[int]	as in spectral_enhance.	None
window	str	as in spectral_enhance.	'hann'
noise_kwargs	Optional[dict]	as in spectral_enhance.	None
report	bool	also measure each file's noise floor before and after, and return them on the result -- worth it when the question is "did this help" rather than "run it".	False
params		method parameters, defaulted from METHODS[method]. A callable takes none — it was configured before it got here.	{}

Returns:

Type	Description
EnhanceFolderResult	an EnhanceFolderResult.

Source code in kudio/enhance/spectral.py

def enhance_folder(src: Any, dst: PathLike,
                   method: Union[str, Callable, Tuple[str, Callable]] = 'logmmse', *,
                   noise: str = 'mcra',
                   n_fft: int = 512,
                   hop_length: Optional[int] = None,
                   window: str = 'hann',
                   noise_kwargs: Optional[dict] = None,
                   sr: Optional[int] = None,
                   subtype: str = 'PCM_16',
                   overwrite: bool = False,
                   progress: Optional[Callable[[int, int, Path], None]] = None,
                   on_error: str = 'collect',
                   report: bool = False,
                   **params) -> EnhanceFolderResult:
    """Denoise every ``.wav`` file under *src* into *dst*.

    The counterpart of :func:`kudio.convert_folder`, and it mirrors the input's
    directory structure the same way -- a folder from itself, a list or a
    ``.txt`` list from the deepest folder its files share -- so two files with
    the same name in different subfolders do not collide.

    >>> result = kudio.enhance_folder("noisy/", "clean/", 'logmmse')
    >>> print(result)
    412/412 written, 0 failed

    :param method: a name from :data:`METHODS`, or anything
        :class:`CustomEnhancer` accepts — a callable, or ``(name, callable)``.
        A trained model is the obvious second case.
    :param n_fft: as in :func:`spectral_enhance`, like the three below. A
        callable *method* ignores all four.
    :param hop_length: as in :func:`spectral_enhance`.
    :param window: as in :func:`spectral_enhance`.
    :param noise_kwargs: as in :func:`spectral_enhance`.
    :param report: also measure each file's noise floor before and after, and
        return them on the result -- worth it when the question is "did this
        help" rather than "run it".
    :param params: method parameters, defaulted from ``METHODS[method]``. A
        callable takes none — it was configured before it got here.

    :returns: an :class:`EnhanceFolderResult`.
    """
    from kudio.core.io import WAVE_SUFFIX, check_file, check_input, \
        file_load, save_wave
    from kudio.core.report import audio_report
    from kudio.exceptions import AudioIOError

    entry = _as_enhancer(method)
    if on_error not in ('collect', 'raise'):
        raise FeatureError(f"on_error must be 'collect' or 'raise', got {on_error!r}")
    if isinstance(entry, CustomEnhancer):
        if params:
            raise FeatureError(
                f"{entry.name!r} is a callable — it takes no method "
                f"parameters, but got {', '.join(sorted(params))}")
        run = entry
    else:
        # Validate the parameters here rather than letting each file trip over
        # them: with on_error='collect' a typo would otherwise come back as 412
        # identical failures instead of one sentence about the typo.
        unknown = set(params) - set(METHODS[entry].defaults())
        if unknown:
            raise FeatureError(
                f"{entry!r} does not take {', '.join(sorted(unknown))}. It "
                f"takes: {', '.join(sorted(METHODS[entry].defaults())) or 'nothing'}")

        settings = METHODS[entry].defaults()
        settings.update(params)
        _check_ranges(entry, settings)

        def run(y, rate):
            return spectral_enhance(y, rate, entry, noise=noise, n_fft=n_fft,
                                    hop_length=hop_length, window=window,
                                    noise_kwargs=noise_kwargs, **params)

    from kudio.core.io import _relative, _source_root
    files, skipped = check_input(src)
    if not files:
        raise AudioIOError(f"no audio files found in {src!r}")

    # the same layout rule as convert_folder: a list used to land flat, so
    # day1/take1 and day2/take1 became take1 and take12
    root = _source_root(src, files)
    out_root = Path(dst)
    out_root.mkdir(parents=True, exist_ok=True)

    outputs, floors = [], []
    failed = [(Path(str(item)), "not an existing .wav file") for item in skipped]
    for n, f in enumerate(files, 1):
        try:
            y, rate = file_load(f, sr=sr, mono=True)
            before = audio_report(y, rate).noise_floor_dbfs if report else None
            out = np.asarray(run(y, rate), dtype=np.float32)
            after = audio_report(out, rate).noise_floor_dbfs if report else None

            target = (out_root / _relative(f, root)).with_suffix(WAVE_SUFFIX)
            target = Path(check_file(target, rename=not overwrite))
            save_wave(target, out, rate, subtype=subtype)
            outputs.append(target)
            if report:
                floors.append((target, before, after))
        except Exception as e:                              # noqa: BLE001
            if on_error == 'raise':
                raise
            failed.append((Path(f), f"{type(e).__name__}: {e}"))
        if progress is not None:
            progress(n, len(files), Path(f))

    return EnhanceFolderResult(written=len(outputs), total=len(files) + len(skipped),
                               outputs=outputs, failed=failed,
                               floors=tuple(floors))

kudio.enhance.streaming

Enhancement for audio that has not finished arriving.

spectral_enhance needs the whole clip. That is fine for a file and useless for a microphone: you cannot monitor a take through a denoiser that only runs once the take is over.

Everything needed was already recursive. The decision-directed a priori SNR feeds the previous frame's estimate into this one, and MCRA updates its noise estimate a frame at a time — both are state plus an update, and StreamEnhancer is the overlap-add machinery around them. It shares those objects with the offline path rather than reimplementing the maths, so the two cannot give different answers to the same question.

>>> enhancer = kudio.StreamEnhancer(sr=16000, method='logmmse')
>>> for block in microphone:
...     play(enhancer.process(block))
>>> tail = enhancer.flush()

What it costs. One analysis frame of latency — latency_seconds, 32 ms at the default 512-point frame and 16 kHz. And it is not free of the offline version's judgement: the noise estimator has only heard what has arrived so far, so the first second is always the worst second.

What cannot stream. quantile is a statistic over the whole file; spectral_gate, specsub and multiband in kudio look at the spectrogram as a block. STREAMABLE_METHODS is the list that does, and asking for anything else raises rather than quietly giving you something different from what the same name does offline.

Classes:

Name	Description
StreamEnhancer	Denoise a stream block by block, with a one-frame delay.

Attributes:

Name	Type	Description
STREAMABLE_METHODS		The methods whose gain depends only on this frame's (xi, gamma).

STREAMABLE_METHODS module-attribute

STREAMABLE_METHODS = RECURSIVE_METHODS

The methods whose gain depends only on this frame's (xi, gamma).

StreamEnhancer

StreamEnhancer(
    sr: int,
    method: str = "logmmse",
    *,
    noise: str = "mcra",
    n_fft: int = 512,
    hop_length: Optional[int] = None,
    window: str = "hann",
    noise_kwargs: Optional[dict] = None,
    **params,
)

Denoise a stream block by block, with a one-frame delay.

Parameters:

Name	Type	Description	Default
sr	int	sample rate.	required
method	str	one of STREAMABLE_METHODS.	'logmmse'
noise	str	'mcra', 'initial' or 'minimum' — see kudio.make_noise_tracker. 'quantile' cannot stream.	'mcra'
n_fft	int	analysis frame; also the latency.	512
hop_length	Optional[int]	defaults to n_fft // 4.	None
params		method parameters, defaulted from METHODS[method].	{}

Methods:

Name	Description
reset	Forget every buffer and estimate, keeping the configuration.
process	Feed a block in; get the enhanced audio that is finished back.
flush	Everything still in flight, once no more audio is coming.

Attributes:

Name	Type	Description
latency_seconds	float	How far behind the output runs: one analysis frame.

Source code in kudio/enhance/streaming.py

def __init__(self, sr: int, method: str = 'logmmse', *,
             noise: str = 'mcra', n_fft: int = 512,
             hop_length: Optional[int] = None,
             window: str = 'hann',
             noise_kwargs: Optional[dict] = None,
             **params):
    if sr <= 0:
        raise FeatureError(f"StreamEnhancer needs a positive rate, got {sr}")
    if method not in STREAMABLE_METHODS:
        offline = "" if method not in METHODS else \
            f" It works offline: kudio.spectral_enhance(y, sr, {method!r})."
        raise FeatureError(
            f"{method!r} does not run on a stream: StreamEnhancer runs the "
            f"decision-directed rules, {', '.join(STREAMABLE_METHODS)}, "
            f"whose state carries from one frame to the next.{offline}")

    settings = METHODS[method].defaults()
    unknown = set(params) - set(settings)
    if unknown:
        raise FeatureError(
            f"{method!r} does not take {', '.join(sorted(unknown))}. "
            f"It takes: {', '.join(sorted(settings))}")
    settings.update(params)
    from kudio.enhance.spectral import _check_ranges
    _check_ranges(method, settings)

    self.sr = int(sr)
    self.method = method
    self.noise = noise
    self.n_fft = int(n_fft)
    self.hop_length = int(hop_length or n_fft // 4)
    self.settings = settings
    self._noise_kwargs = dict(noise_kwargs or {})
    self._g_min = 10.0 ** (settings.get('gain_floor_db', -25.0) / 20.0)

    self._window = _get_window(window, self.n_fft)
    # Analysis and synthesis both apply the window (WOLA), so what has to
    # sum to a constant is w^2, not w. Measuring it beats assuming it: an
    # untested COLA assumption shows up as a periodic amplitude ripple at
    # the hop rate, which is easy to hear and hard to attribute.
    self._wola_norm = _overlap_sum(self._window ** 2, self.hop_length)
    self.reset()

latency_seconds property

latency_seconds: float

How far behind the output runs: one analysis frame.

reset

reset() -> None

Forget every buffer and estimate, keeping the configuration.

Source code in kudio/enhance/streaming.py

def reset(self) -> None:
    """Forget every buffer and estimate, keeping the configuration."""
    self._tracker = make_noise_tracker(self.noise, **self._noise_kwargs)
    self._state = DecisionDirected(
        make_gain_rule(self.method, self.settings),
        self.settings.get('alpha', 0.98))
    self._input = np.zeros(0, dtype=np.float64)
    #: one frame wide; index 0 is always the next sample to hand out, and
    #: the buffer slides by one hop per frame. Every frame therefore lands
    #: at offset 0 *of the slid buffer*, which is what makes successive
    #: frames overlap correctly instead of stacking on top of each other.
    self._acc = np.zeros(self.n_fft, dtype=np.float64)
    self._norm = np.zeros(self.n_fft, dtype=np.float64)
    self._pending: list = []
    self._frames = 0
    self._samples_in = 0
    self._samples_out = 0

process

process(block: ndarray) -> ndarray

Feed a block in; get the enhanced audio that is finished back.

The returned block is not the same length as the input — nothing comes out until a whole frame has arrived, and then it emerges in hop-sized pieces. Concatenate what comes back and call flush at the end; sample i of that concatenation lines up with sample i of the input.

Source code in kudio/enhance/streaming.py

def process(self, block: np.ndarray) -> np.ndarray:
    """Feed a block in; get the enhanced audio that is finished back.

    The returned block is **not** the same length as the input — nothing
    comes out until a whole frame has arrived, and then it emerges in
    hop-sized pieces. Concatenate what comes back and call :meth:`flush`
    at the end; sample *i* of that concatenation lines up with sample *i*
    of the input.
    """
    block = np.asarray(as_float(block), dtype=np.float64)
    if block.ndim > 1 and block.shape[1] == 1:
        block = block[:, 0]
    if block.ndim != 1:
        # reshape(-1) used to read a stereo block as one interleaved
        # signal twice as long
        raise FeatureError(f"StreamEnhancer takes mono blocks, got shape "
                           f"{block.shape}")
    self._input = np.concatenate([self._input, block])
    self._samples_in += len(block)

    while len(self._input) >= self.n_fft:
        self._consume_frame(self._input[:self.n_fft])
        self._input = self._input[self.hop_length:]
    return self._emit()

flush

flush() -> ndarray

Everything still in flight, once no more audio is coming.

Pads the last partial frame with zeros so the tail is processed rather than dropped — a stream that quietly loses its final 30 ms is the kind of bug that only shows up on short takes — and stops where the input stopped, so the whole output is as long as the whole input.

Source code in kudio/enhance/streaming.py

def flush(self) -> np.ndarray:
    """Everything still in flight, once no more audio is coming.

    Pads the last partial frame with zeros so the tail is processed rather
    than dropped — a stream that quietly loses its final 30 ms is the kind
    of bug that only shows up on short takes — and stops where the input
    stopped, so the whole output is as long as the whole input.
    """
    while len(self._input):
        padded = np.zeros(self.n_fft, dtype=np.float64)
        take = min(len(self._input), self.n_fft)
        padded[:take] = self._input[:take]
        self._consume_frame(padded)
        self._input = self._input[self.hop_length:] \
            if len(self._input) > self.hop_length else self._input[:0]

    # drain whatever the accumulator still holds, up to the last sample any
    # frame actually reached
    covered = int(np.count_nonzero(self._norm > 1e-8))
    if covered:
        tail = self._normalise(self._acc[:covered], self._norm[:covered])
        self._pending.append(tail.astype(np.float32))
    self._acc[:] = 0.0
    self._norm[:] = 0.0
    out = self._emit()
    # what lies past the last input sample came out of the zero padding;
    # left on, 16000 samples in came back as 16383
    extra = self._samples_out - self._samples_in
    if extra > 0:
        out = out[:len(out) - extra]
        self._samples_out -= extra
    return out

kudio.enhance.traditional

Traditional (non-DNN) speech enhancement.

trad_enhance implements spectral subtraction / Wiener filtering with MCRA noise estimation (equation numbers reference Loizou, Speech Enhancement: Theory and Practice). wavelet_low_pass_filter is a soft-threshold wavelet denoiser.

Functions:

Name	Description
wavelet_low_pass_filter	Soft-threshold the detail coefficients of a wavelet decomposition.
trad_enhance	Enhance waveform y using spectral subtraction with MCRA noise

INIT

INIT(ns_ps: ndarray, fs: int)

First-frame initialization of the noise-estimation state.

Methods:

Name	Description
weight	Weighted spectral average.
com_min_track	Continuous minimal tracking.
mcra	MCRA algorithm.
mcra2	MCRA-2 algorithm with a frequency-dependent delta [9.60].

Source code in kudio/enhance/traditional.py

def __init__(self, ns_ps: np.ndarray, fs: int):
    self.ns_ps = ns_ps
    self.fs = fs

weight

weight() -> dict

Weighted spectral average.

Source code in kudio/enhance/traditional.py

def weight(self) -> dict:
    """Weighted spectral average."""
    return {'ass': 0.85, 'beta': 1.5, 'noise_ps': self.ns_ps, 'P': self.ns_ps}

com_min_track

com_min_track() -> dict

Continuous minimal tracking.

Source code in kudio/enhance/traditional.py

def com_min_track(self) -> dict:
    """Continuous minimal tracking."""
    return {'n': 2, 'leng': len(self.ns_ps), 'alpha': 0.7, 'beta': 0.96,
            'gamma': 0.998, 'noise_ps': self.ns_ps, 'pxk_old': self.ns_ps,
            'pxk': self.ns_ps, 'pnk_old': self.ns_ps, 'pnk': self.ns_ps}

mcra

mcra() -> dict

MCRA algorithm.

Source code in kudio/enhance/traditional.py

def mcra(self) -> dict:
    """MCRA algorithm."""
    len_val = len(self.ns_ps)
    return {'n': 2, 'ad': 0.95, 'ass': 0.8, 'L': 1000 * 2 // 20, 'delta': 5,
            'ap': 0.2, 'leng': len_val, 'P': self.ns_ps, 'Pmin': self.ns_ps,
            'Ptmp': self.ns_ps, 'pk': np.zeros(len_val), 'noise_ps': self.ns_ps}

mcra2

mcra2() -> dict

MCRA-2 algorithm with a frequency-dependent delta [9.60].

Source code in kudio/enhance/traditional.py

def mcra2(self) -> dict:
    """MCRA-2 algorithm with a frequency-dependent delta [9.60]."""
    len_val = len(self.ns_ps)
    freq_res = self.fs / len_val
    k_1khz = int(1000 // freq_res)
    k_3khz = int(3000 // freq_res)

    low_1 = 2 * np.ones(k_1khz, dtype=int)
    low_2 = 2 * np.ones(k_3khz - k_1khz, dtype=int)
    high = 5 * np.ones(len_val // 2 - k_3khz, dtype=int)
    delta_val = np.concatenate((low_1, low_2, high, high, low_2, low_1))

    return {'n': 2, 'leng': len_val, 'ad': 0.95, 'ass': 0.8, 'ap': 0.2,
            'beta': 0.8, 'beta1': 0.98, 'gamma': 0.998, 'alpha': 0.7,
            'delta': delta_val, 'pk': np.zeros(len_val), 'noise_ps': self.ns_ps,
            'pxk_old': self.ns_ps, 'pxk': self.ns_ps, 'pnk_old': self.ns_ps,
            'pnk': self.ns_ps}

EST

EST(ns_ps: ndarray, para: dict)

Per-frame update of the noise-estimation state.

Methods:

Name	Description
weight	Weighted spectral average [9.30].
com_min_track	Continuous minimal tracking [9.24-9.25].
mcra	MCRA [9.53-9.59].
mcra2	MCRA-2 [9.61, 9.25, 9.53-9.59].

Source code in kudio/enhance/traditional.py

def __init__(self, ns_ps: np.ndarray, para: dict):
    self.ns_ps = ns_ps
    self.para = para

weight

weight() -> dict

Weighted spectral average [9.30].

Source code in kudio/enhance/traditional.py

def weight(self) -> dict:
    """Weighted spectral average [9.30]."""
    ass = self.para['ass']
    beta = self.para['beta']
    noise_ps = self.para['noise_ps']
    P = ass * self.para['P'] + (1 - ass) * self.ns_ps

    smaller = P < beta * noise_ps
    noise_ps[smaller] = ass * noise_ps[smaller] + (1 - ass) * P[smaller]

    self.para['P'] = P
    self.para['noise_ps'] = noise_ps
    return self.para

com_min_track

com_min_track() -> dict

Continuous minimal tracking [9.24-9.25].

Source code in kudio/enhance/traditional.py

def com_min_track(self) -> dict:
    """Continuous minimal tracking [9.24-9.25]."""
    alpha = self.para['alpha']
    beta = self.para['beta']
    gamma = self.para['gamma']
    pxk_old = self.para['pxk_old']
    pnk_old = self.para['pnk_old']
    pnk = self.para['pnk']

    pxk = alpha * pxk_old + (1 - alpha) * self.ns_ps
    for t in range(self.para['leng']):
        if pnk_old[t] <= pxk[t]:
            pnk[t] = (gamma * pnk_old[t]) + \
                (((1 - gamma) / (1 - beta)) * (pxk[t] - beta * pxk_old[t]))
        else:
            pnk[t] = pxk[t]

    self.para['n'] += 1
    self.para['noise_ps'] = pnk
    self.para['pnk'] = pnk
    self.para['pnk_old'] = pnk
    self.para['pxk'] = pxk
    self.para['pxk_old'] = pxk
    return self.para

mcra

mcra() -> dict

MCRA [9.53-9.59].

Source code in kudio/enhance/traditional.py

def mcra(self) -> dict:
    """MCRA [9.53-9.59]."""
    ass = self.para['ass']
    ad = self.para['ad']
    ap = self.para['ap']
    pk = self.para['pk']
    delta = self.para['delta']
    L = self.para['L']
    n = self.para['n']
    noise_ps = self.para['noise_ps']
    Pmin = self.para['Pmin']
    Ptmp = self.para['Ptmp']

    P = ass * self.para['P'] + (1 - ass) * self.ns_ps          # [9.55]
    if n % L == 0:                                             # [9.23]
        Pmin = np.minimum(Ptmp, P)
        Ptmp = P
    else:
        Pmin = np.minimum(Pmin, P)
        Ptmp = np.minimum(Ptmp, P)
    Srk = P / Pmin                                             # [9.58]
    Ikl = (Srk > delta).astype(float)
    pk = ap * pk + (1 - ap) * Ikl                              # [9.59]
    adk = ad + (1 - ad) * pk                                   # [9.54]
    noise_ps = adk * noise_ps + (1 - adk) * self.ns_ps         # [9.53]

    self.para.update(pk=pk, n=n + 1, noise_ps=noise_ps, P=P, Pmin=Pmin, Ptmp=Ptmp)
    return self.para

mcra2

mcra2() -> dict

MCRA-2 [9.61, 9.25, 9.53-9.59].

Source code in kudio/enhance/traditional.py

def mcra2(self) -> dict:
    """MCRA-2 [9.61, 9.25, 9.53-9.59]."""
    ad = self.para['ad']
    ap = self.para['ap']
    beta = self.para['beta']
    gamma = self.para['gamma']
    alpha = self.para['alpha']
    pk = self.para['pk']
    delta = self.para['delta']
    noise_ps = self.para['noise_ps']
    pnk = self.para['pnk']
    pxk_old = self.para['pxk_old']
    pnk_old = self.para['pnk_old']

    pxk = alpha * pxk_old + (1 - alpha) * self.ns_ps           # [9.61]
    for i in range(len(pnk)):                                  # [9.25]
        if pnk_old[i] < pxk[i]:
            pnk[i] = (gamma * pnk_old[i]) + \
                (((1 - gamma) / (1 - beta)) * (pxk[i] - beta * pxk_old[i]))
    Srk = pxk / pnk                                            # [9.57]
    Ikl = (Srk > delta).astype(float)                          # [9.58]
    pk = ap * pk + (1 - ap) * Ikl                              # [9.59]
    adk = ad + (1 - ad) * pk                                   # [9.54]
    noise_ps = adk * noise_ps + (1 - adk) * pxk                # [9.53]

    self.para.update(n=self.para['n'] + 1, pk=pk, noise_ps=noise_ps,
                     pnk=pnk, pnk_old=pnk, pxk=pxk, pxk_old=pxk)
    return self.para

wavelet_low_pass_filter

wavelet_low_pass_filter(
    y: ndarray, thres: float = 0.8, wavelet: str = "db2"
) -> ndarray

Soft-threshold the detail coefficients of a wavelet decomposition.

The threshold is thres times the clip's peak, max |y|. The result has the input's length; a silent clip comes back as it went in.

Source code in kudio/enhance/traditional.py

def wavelet_low_pass_filter(y: np.ndarray, thres: float = 0.8,
                            wavelet: str = 'db2') -> np.ndarray:
    """Soft-threshold the detail coefficients of a wavelet decomposition.

    The threshold is *thres* times the clip's peak, ``max |y|``. The result
    has the input's length; a silent clip comes back as it went in.
    """
    y = as_float(y)                              # float32 stays float32, as pywt keeps it
    if y.ndim != 1:
        raise FeatureError(f"wavelet_low_pass_filter() takes mono audio, got shape {y.shape}")
    # the peak, not the signed maximum: a clip below zero throughout got a
    # negative threshold and came back unfiltered
    level = thres * float(np.nanmax(np.abs(y))) if y.size else 0.0
    if not level > 0:
        # a threshold of zero makes pywt divide by zero: NaN everywhere
        return y.copy()
    coeff = pywt.wavedec(y, wavelet, mode='periodization')
    coeff[1:] = (pywt.threshold(i, value=level, mode='soft')
                 for i in coeff[1:])
    # periodization rounds an odd length up by one sample
    return pywt.waverec(coeff, wavelet, mode='periodization')[:len(y)]

trad_enhance

trad_enhance(y: ndarray, sr: int) -> ndarray

Enhance waveform y using spectral subtraction with MCRA noise estimation and a Wiener-style gain function.

Returns float32 of the same length as y. The last partial frame, under 256 samples, has no frame of its own and comes back silent.

Raises:

Type	Description
FeatureError	for anything but mono audio, or a clip shorter than one 512-sample frame, which would come back as silence.

Source code in kudio/enhance/traditional.py

def trad_enhance(y: np.ndarray, sr: int) -> np.ndarray:
    """Enhance waveform *y* using spectral subtraction with MCRA noise
    estimation and a Wiener-style gain function.

    Returns float32 of the same length as *y*. The last partial frame, under
    256 samples, has no frame of its own and comes back silent.

    :raises FeatureError: for anything but mono audio, or a clip shorter than
        one 512-sample frame, which would come back as silence.
    """
    y = np.asarray(as_float(y), dtype=np.float64)
    if y.ndim != 1:
        raise FeatureError(f"trad_enhance() takes mono audio, got shape {y.shape}")
    if len(y) < 512:
        raise FeatureError(f"trad_enhance() needs at least one 512-sample frame, "
                           f"got {len(y)} samples")
    len_ = 512           # frame size in samples
    PERC = 50            # window overlap as percent of frame
    len1 = len_ * PERC // 100
    len2 = len_ - len1
    Expnt = 2.0          # 1.0: magnitude domain, 2.0: power domain
    beta = 0.002
    win = np.hamming(len_)
    winGain = len2 / sum(win)
    nFFT = 2 * 2 ** 8

    k = 1
    x_old = np.zeros(len1)
    Nframes = len(y) // len2 - 1
    xfinal = np.zeros((Nframes + 1) * len2)

    para = {}
    for n in range(Nframes):
        insign = win * y[k - 1: k + len_ - 1]
        spec = np.fft.fft(insign, nFFT)
        sig = abs(spec)
        # a floor under the power: a frame of digital silence otherwise starts
        # the noise estimate at zero and every gain after it at 0/0
        ns_ps = sig ** 2 + 1e-12  # noisy power spectrum
        theta = np.angle(spec)    # noisy phase

        # noise estimation (MCRA)
        para = INIT(ns_ps, sr).mcra() if n == 0 else EST(ns_ps, para).mcra()
        noise_mu = np.sqrt(para['noise_ps'])

        # posterior SNR; -inf for a frame of digital silence, which the
        # thresholds below read as the lowest SNR, so numpy need not warn
        with np.errstate(divide='ignore'):
            SNRpos = 10 * np.log10(
                np.linalg.norm(sig, 2) ** 2 / np.linalg.norm(noise_mu, 2) ** 2)

        # over-subtraction factor by SNR
        if Expnt == 1.0:
            alpha = 4 if SNRpos < -5.0 else (1 if SNRpos > 20 else 3 - SNRpos * 2 / 20)
        else:
            alpha = 5 if SNRpos < -5.0 else (1 if SNRpos > 20 else 4 - SNRpos * 3 / 20)

        sub_speech = sig ** Expnt - alpha * noise_mu ** Expnt

        # floor negative components at beta * noise
        diffw = sub_speech - beta * noise_mu ** Expnt
        negative = diffw < 0
        sub_speech[negative] = beta * noise_mu[negative] ** Expnt

        # priori SNR
        SNRpri = 10 * np.log10(
            np.linalg.norm(sub_speech ** (1 / Expnt), 2) ** 2
            / np.linalg.norm(noise_mu, 2) ** 2)

        # Wiener gain smoothing parameter
        mel_max = 10
        mel_0 = (1 + 4 * mel_max) / 5
        s = 25 / (mel_max - 1)
        mel = mel_max if SNRpri < -5.0 else (1 if SNRpri > 20 else mel_0 - SNRpri / s)

        g_k = sub_speech / (sub_speech + mel * noise_mu ** Expnt)
        wf_speech = g_k * sig

        # back to time domain with the noisy phase
        x_phase = wf_speech * np.exp(1j * theta)
        xi = np.fft.ifft(x_phase).real

        # overlap-add
        xfinal[k - 1: k + len2 - 1] = x_old + xi[0: len1]
        x_old = xi[len1: len_]
        k = k + len2

    # the last frame's second half has no later frame to overlap with; it
    # used to be dropped, leaving the final 256 samples at zero
    if Nframes > 0:
        xfinal[k - 1: k - 1 + len1] = x_old
    out = np.zeros(len(y), dtype=np.float32)
    n = min(len(y), len(xfinal))
    out[:n] = winGain * xfinal[:n]
    return out