Learn AI Series (#99) - Audio Enhancement

avatar

Learn AI Series (#99) - Audio Enhancement

variant-c-10-skyblue.png

What will I learn

  • You will learn spectral subtraction: the classical approach to noise reduction and why it produces "musical noise" artifacts;
  • Wiener filtering: the optimal linear filter for noise reduction using smooth gain masks;
  • deep learning denoising: training U-Net architectures on spectrogram masks to handle non-stationary noise;
  • waveform-domain enhancement: models like DEMUCS and Conv-TasNet that skip the spectrogram entirely;
  • acoustic echo cancellation: adaptive filters that learn room impulse responses in real-time;
  • audio super-resolution: predicting missing high-frequency content from low-bandwidth recordings;
  • real-time constraints: latency budgets, causal architectures, and streaming inference for live audio;
  • evaluation metrics: PESQ, STOI, and SI-SNR for measuring enhancement quality;
  • building a complete multi-stage noise reduction pipeline from scratch.

Requirements

  • A working modern computer running macOS, Windows or Ubuntu;
  • An installed Python 3(.11+) distribution;
  • The ambition to learn AI and machine learning.

Difficulty

  • Beginner

Curriculum (of the Learn AI Series):

Learn AI Series (#99) - Audio Enhancement

Solutions to Episode #98 Exercises

Exercise 1: The intent classifier maps 7 classes (weather, timer, music, lights, news, reminder, search) using keyword overlap scoring. Each intent has a keyword list; classification computes the fraction of keywords matched per intent, normalizes by total score, and rejects below a 0.4 threshold. "What is the weather like" confidently maps to WEATHER. "Tell me about quantum physics" produces low scores across all intents and correctly returns UNKNOWN. Testing 50 random utterances from the combined vocabulary shows clear separation between in-domain (above 0.4) and out-of-domain (below 0.4) queries.

Exercise 2: The slot filler tracks required slots per intent (city for weather, duration for timer, song_or_artist for music). "Set a timer" triggers FILLING state with missing duration -- the system asks "How long?". After "5 minutes", pattern matching extracts the duration, the slot fills, and the state transitions to COMPLETE. Three test dialogues all complete in 2 turns, with the state machine resetting cleanly between conversations.

Exercise 3: The error recovery pipeline simulates ASR noise at three levels. At clean SNR, all words pass through with 0.95 confidence. At medium noise, 30% of words get single-character substitutions (confidence 0.4) -- edit distance correction against a vocabulary recovers most errors. At high noise, words get deletions (20%) and heavy corruption (30%) -- average confidence drops below threshold, triggering "I didn't quite catch that" clarification rather than executing a wrong command. Substitution errors dominate at medium noise; deletions increase at high noise.

On to today's episode

Here we go! We're nearing the end of the audio AI arc and episode 99 brings us to a topic that underpins everything we've built so far. Over the past seven episodes we've covered how machines hear sound (#92), transcribe speech (#93), synthesize it back (#94), classify environmental sounds (#95), generate music (#96), recognize who's speaking (#97), and understand spoken intent (#98). All of those assumed the audio was reasonably clean -- or at least clean enough to work with. Today we deal with the reality that audio in the wild is almost never clean.

Audio enhancement is a family of techniques for taking messy, noisy, degraded recordings and making them sound like they were captured in a proper studio. You're listening to a podcast recorded in someone's kitchen. The fridge hums. Dishes clatter somewhere in the background. Traffic bleeds through the window. The content might be brilliant, but the audio makes you want to reach for the stop button. That is the problem we're solving today.

And here's the thing -- audio enhancement isn't just a nice-to-have cosmetic improvement. It's a critical preprocessing step for every other audio AI system we've built in this series. Speech recognition accuracy drops dramatically in noisy conditions. Speaker verification fails when noise corrupts the voice embeddings we so carefully extracted in episode #97. Intent classification misses commands buried under background chatter. Fix the audio first, and every downstream system works better. The classical methods are elegant and still widely deployed in production. The deep learning methods are dramaticaly better. We'll build both ;-)

Spectral subtraction: the classical approach

The idea behind spectral subtraction is beautifully simple. If you have a noisy signal and can estimate the noise spectrum, just subtract the noise from the signal in the frequency domain. The assumption: noise is additive. The noisy signal Y = S + N, where S is clean speech and N is noise. In the frequency domain: |Y| = |S| + |N| (approximately). So |S| is roughly |Y| - |N|. Let's implement a complete spectral subtractor from scratch using only NumPy:

import numpy as np


class SpectralSubtractor:
    """Classical spectral subtraction
    for noise reduction."""

    def __init__(self, sr=16000,
                 n_fft=1024, hop=256):
        self.sr = sr
        self.n_fft = n_fft
        self.hop = hop

    def stft(self, audio):
        """Short-time Fourier transform."""
        window = np.hanning(self.n_fft)
        n_frames = (
            (len(audio) - self.n_fft)
            // self.hop + 1)
        n_bins = self.n_fft // 2 + 1
        result = np.zeros(
            (n_bins, n_frames),
            dtype=complex)
        for i in range(n_frames):
            start = i * self.hop
            frame = audio[
                start:start + self.n_fft]
            result[:, i] = np.fft.rfft(
                frame * window)
        return result

    def istft(self, stft_matrix, length):
        """Inverse STFT with overlap-add."""
        window = np.hanning(self.n_fft)
        n_frames = stft_matrix.shape[1]
        output = np.zeros(length)
        window_sum = np.zeros(length)
        for i in range(n_frames):
            start = i * self.hop
            frame = np.fft.irfft(
                stft_matrix[:, i])
            end = min(start + self.n_fft,
                      length)
            flen = end - start
            output[start:end] += (
                frame[:flen] * window[:flen])
            window_sum[start:end] += (
                window[:flen] ** 2)
        mask = window_sum > 1e-8
        output[mask] /= window_sum[mask]
        return output

    def denoise(self, noisy, noise_frames=30,
                oversubtract=1.0,
                floor=0.002):
        """Spectral subtraction with
        spectral flooring."""
        spec = self.stft(noisy)
        magnitude = np.abs(spec)
        phase = np.angle(spec)

        # Estimate noise from first
        # N frames (assumed noise-only)
        noise_est = np.mean(
            magnitude[:, :noise_frames],
            axis=1, keepdims=True)

        # Subtract with oversubtraction
        # factor and spectral floor
        clean_mag = np.maximum(
            magnitude
            - oversubtract * noise_est,
            floor * noise_est)

        # Reconstruct with original phase
        clean_spec = clean_mag * np.exp(
            1j * phase)
        return self.istft(
            clean_spec, len(noisy))

    def run(self):
        rng = np.random.RandomState(42)
        dur = 3.0
        n = int(self.sr * dur)
        t = np.arange(n) / self.sr

        # Synthesize "speech": formants
        speech = np.zeros(n)
        for f0, amp in [(150, 0.3),
                        (300, 0.2),
                        (900, 0.1),
                        (2500, 0.05)]:
            speech += amp * np.sin(
                2 * np.pi * f0 * t)
        # Amplitude modulation (syllables)
        speech *= (0.5 + 0.5 * np.sin(
            2 * np.pi * 4 * t))
        # Only active 0.3s-2.5s
        envelope = np.zeros(n)
        envelope[int(0.3*self.sr):
                 int(2.5*self.sr)] = 1.0
        speech *= envelope

        # Add noise
        noise = rng.randn(n) * 0.08
        noisy = speech + noise

        # Denoise
        clean = self.denoise(noisy)

        # Measure SNR improvement
        speech_power = np.mean(
            speech[int(0.5*self.sr):
                   int(2.0*self.sr)] ** 2)
        noise_region = slice(
            0, int(0.2 * self.sr))
        input_noise = np.mean(
            noisy[noise_region] ** 2)
        output_noise = np.mean(
            clean[noise_region] ** 2)

        input_snr = 10 * np.log10(
            speech_power
            / max(input_noise, 1e-10))
        output_snr = 10 * np.log10(
            speech_power
            / max(output_noise, 1e-10))

        print(f"Audio: {dur}s at "
              f"{self.sr} Hz")
        print(f"Input SNR:  "
              f"{input_snr:.1f} dB")
        print(f"Output SNR: "
              f"{output_snr:.1f} dB")
        print(f"Improvement: "
              f"{output_snr - input_snr:.1f}"
              f" dB")


sub = SpectralSubtractor()
sub.run()

It works. But it has a well-known artifact: musical noise. When you subtract a noise estimate from a noisy spectrum, small random fluctuations produce isolated spectral peaks that sound like random tonal blips -- almost like a digital xylophone playing random notes underneath the speech. The spectral_floor parameter helps (instead of setting bins to zero, we set them to a small fraction of the noise estimate), but musical noise is never fully eliminated.

The bigger limitation: spectral subtraction assumes stationary noise -- a constant hum, fan, or hiss. The moment noise changes character (someone coughs, a door slams, traffic varies), the noise estimate becomes wrong and artifacts appear everywhere.

Wiener filtering: the optimal linear approach

The Wiener filter is the mathematically optimal linear filter for noise reduction when you know (or can estimate) the signal and noise power spectra. Instead of hard subtraction, it computes a gain mask -- a value between 0 and 1 for each frequency bin at each time frame. The formula is elegant: gain = SNR / (SNR + 1), where SNR is the estimated signal-to-noise ratio per bin:

class WienerDenoiser:
    """Wiener filter for noise reduction
    with gain smoothing."""

    def __init__(self, sr=16000,
                 n_fft=1024, hop=256):
        self.sr = sr
        self.n_fft = n_fft
        self.hop = hop

    def stft(self, audio):
        window = np.hanning(self.n_fft)
        n_frames = (
            (len(audio) - self.n_fft)
            // self.hop + 1)
        n_bins = self.n_fft // 2 + 1
        result = np.zeros(
            (n_bins, n_frames),
            dtype=complex)
        for i in range(n_frames):
            s = i * self.hop
            result[:, i] = np.fft.rfft(
                audio[s:s+self.n_fft]
                * window)
        return result

    def istft(self, spec, length):
        window = np.hanning(self.n_fft)
        n_frames = spec.shape[1]
        output = np.zeros(length)
        wsum = np.zeros(length)
        for i in range(n_frames):
            s = i * self.hop
            frame = np.fft.irfft(
                spec[:, i])
            e = min(s + self.n_fft, length)
            fl = e - s
            output[s:e] += (
                frame[:fl] * window[:fl])
            wsum[s:e] += window[:fl] ** 2
        mask = wsum > 1e-8
        output[mask] /= wsum[mask]
        return output

    def denoise(self, noisy,
                noise_frames=30,
                smooth_alpha=0.9):
        """Wiener filter with temporal
        gain smoothing."""
        spec = self.stft(noisy)
        power = np.abs(spec) ** 2
        phase = np.angle(spec)

        # Estimate noise power
        noise_power = np.mean(
            power[:, :noise_frames],
            axis=1, keepdims=True)

        # A priori SNR estimate
        signal_power = np.maximum(
            power - noise_power, 0.0)

        # Wiener gain: SNR / (SNR + 1)
        snr = signal_power / (
            noise_power + 1e-10)
        gain = snr / (snr + 1)

        # Temporal smoothing to reduce
        # musical noise
        smoothed = np.zeros_like(gain)
        smoothed[:, 0] = gain[:, 0]
        for i in range(1, gain.shape[1]):
            smoothed[:, i] = (
                smooth_alpha
                * smoothed[:, i-1]
                + (1 - smooth_alpha)
                * gain[:, i])

        clean_spec = (np.abs(spec)
                      * smoothed
                      * np.exp(1j * phase))
        return self.istft(
            clean_spec, len(noisy))

    def compare_methods(self):
        """Compare Wiener vs spectral
        subtraction."""
        rng = np.random.RandomState(42)
        dur = 3.0
        n = int(self.sr * dur)
        t = np.arange(n) / self.sr

        speech = np.zeros(n)
        for f0, a in [(150, 0.3),
                      (450, 0.15),
                      (1200, 0.08)]:
            speech += a * np.sin(
                2*np.pi*f0*t)
        speech *= (0.5 + 0.5 * np.sin(
            2*np.pi*3.5*t))
        env = np.zeros(n)
        env[int(0.3*self.sr):
            int(2.5*self.sr)] = 1.0
        speech *= env

        print(f"{'SNR_in':>8} {'Wiener':>10} "
              f"{'SpSub':>10} {'Winner':>8}")
        print("-" * 40)

        for snr_db in [20, 10, 5, 0, -5]:
            sp_pow = np.mean(speech ** 2)
            noise_pow = sp_pow / (
                10 ** (snr_db / 10))
            noise = rng.randn(n) * np.sqrt(
                noise_pow)
            noisy = speech + noise

            w_clean = self.denoise(noisy)
            w_err = np.mean(
                (speech - w_clean) ** 2)

            sub = SpectralSubtractor(
                self.sr, self.n_fft,
                self.hop)
            s_clean = sub.denoise(noisy)
            s_err = np.mean(
                (speech - s_clean) ** 2)

            winner = ("Wiener"
                      if w_err < s_err
                      else "SpSub")
            print(f"{snr_db:>6}dB "
                  f"{w_err:>10.6f} "
                  f"{s_err:>10.6f} "
                  f"{winner:>8}")


wd = WienerDenoiser()
wd.compare_methods()

The Wiener filter produces smoother output than spectral subtraction because the gain transitions continuously rather than hard-clipping at zero. The temporal smoothing (exponential moving average on the gain) further reduces musical noise by preventing the gain from flickering rapidly between frames. But it shares the same fundamental limitation: it needs a good noise estimate, and it assumes the noise doesn't change dramatically over time.

Notice the comparison: Wiener consistently beats spectral subtraction, especially at lower SNR levels where hard subtraction produces the worst musical noise artifacts. The softer gain curve preserves more of the speech signal while still suppressing noise effectively. At -5dB SNR (where noise is louder than speech!) both methods struggle, but Wiener degrades more gracefully.

Deep learning denoising: spectrogram masks

Neural networks learn to denoise without explicit noise estimation. You train them on pairs of (noisy audio, clean audio) and the network learns a direct mapping. The key insight: the network can handle non-stationary noise because it learns patterns in both speech and noise, adapting its behavior at every time step.

The dominant architecture is the U-Net applied to spectrograms (yes, the same U-Net we built for image segmentation in episode #80). Audio enhancement is, in a way, an image-to-image translation problem -- from a noisy spectrogram image to a clean one:

import torch
import torch.nn as nn
import torch.nn.functional as F


class AudioDenoiseUNet(nn.Module):
    """U-Net for spectrogram mask
    prediction."""

    def __init__(self, in_ch=1,
                 base_ch=32):
        super().__init__()
        self.enc1 = self._block(
            in_ch, base_ch)
        self.enc2 = self._block(
            base_ch, base_ch * 2)
        self.enc3 = self._block(
            base_ch * 2, base_ch * 4)
        self.pool = nn.MaxPool2d(2)

        self.bottleneck = self._block(
            base_ch * 4, base_ch * 8)

        self.up3 = nn.ConvTranspose2d(
            base_ch * 8, base_ch * 4,
            2, stride=2)
        self.dec3 = self._block(
            base_ch * 8, base_ch * 4)
        self.up2 = nn.ConvTranspose2d(
            base_ch * 4, base_ch * 2,
            2, stride=2)
        self.dec2 = self._block(
            base_ch * 4, base_ch * 2)
        self.up1 = nn.ConvTranspose2d(
            base_ch * 2, base_ch,
            2, stride=2)
        self.dec1 = self._block(
            base_ch * 2, base_ch)

        self.out = nn.Sequential(
            nn.Conv2d(base_ch, in_ch, 1),
            nn.Sigmoid())

    def _block(self, in_c, out_c):
        return nn.Sequential(
            nn.Conv2d(in_c, out_c, 3,
                      padding=1),
            nn.BatchNorm2d(out_c),
            nn.ReLU(inplace=True),
            nn.Conv2d(out_c, out_c, 3,
                      padding=1),
            nn.BatchNorm2d(out_c),
            nn.ReLU(inplace=True))

    def forward(self, x):
        """x: (batch, 1, freq, time)
        magnitude spectrogram."""
        e1 = self.enc1(x)
        e2 = self.enc2(self.pool(e1))
        e3 = self.enc3(self.pool(e2))
        b = self.bottleneck(
            self.pool(e3))

        d3 = self.dec3(torch.cat(
            [self.up3(b), e3], dim=1))
        d2 = self.dec2(torch.cat(
            [self.up2(d3), e2], dim=1))
        d1 = self.dec1(torch.cat(
            [self.up1(d2), e1], dim=1))

        mask = self.out(d1)
        return x * mask


model = AudioDenoiseUNet()
total = sum(p.numel()
            for p in model.parameters())
noisy_spec = torch.randn(2, 1, 128, 128)
clean_spec = model(noisy_spec)
print(f"Input: {noisy_spec.shape}")
print(f"Output: {clean_spec.shape}")
print(f"Parameters: {total:,}")
print(f"Mask range: "
      f"[{clean_spec.min():.3f}, "
      f"{clean_spec.max():.3f}]")

The model predicts a mask -- values between 0 and 1 for each time-frequency bin. Bins dominated by speech get mask values near 1 (pass through). Bins dominated by noise get values near 0 (suppress). This is fundamentally more flexible than spectral subtraction because the mask adapts to the content at every single moment, and the network has learned what speech vs noise looks like in the spectrogram domain.

Training data comes from mixing clean speech datasets (like LibriSpeech or VCTK) with noise datasets (MUSAN, AudioSet noise subset, the DNS Challenge dataset) at various SNR levels. You mix clean + noise -> feed the noisy spectrogram -> predict mask -> compare the masked result with the clean spectrogram -> backpropagate. The model learns to handle everything from gentle fan hum at 20dB SNR to construction noise at -5dB SNR.

Waveform-domain models: skipping the spectrogram

Instead of working in the spectrogram domain, some models operate directly on raw waveforms. Conv-TasNet (Time-domain Audio Separation Network) and Meta's DEMUCS architecture (which we mentioned in episode #96 for music source separation) use 1D convolutions on the time-domain signal. This avoids a fundamental problem with spectrogram-based methods: phase reconstruction.

When you predict a magnitude mask and apply it to the noisy magnitude spectrogram, you keep the original (noisy) phase. But phase carries spatial and temporal information that matters for perceptual quality. Waveform-domain models reconstruct everything -- magnitude and phase together:

class SimpleWaveformDenoiser(nn.Module):
    """Simplified waveform-domain
    denoiser inspired by Conv-TasNet."""

    def __init__(self, n_filters=256,
                 kernel_size=20,
                 n_blocks=4,
                 hidden=128):
        super().__init__()
        # Learned encoder (replaces STFT)
        self.encoder = nn.Conv1d(
            1, n_filters, kernel_size,
            stride=kernel_size // 2,
            bias=False)

        # Separator: dilated 1D convs
        layers = []
        for i in range(n_blocks):
            dilation = 2 ** i
            layers.append(nn.Sequential(
                nn.Conv1d(
                    n_filters, hidden, 1),
                nn.PReLU(),
                nn.GroupNorm(1, hidden),
                nn.Conv1d(
                    hidden, hidden, 3,
                    padding=dilation,
                    dilation=dilation,
                    groups=hidden),
                nn.PReLU(),
                nn.GroupNorm(1, hidden),
                nn.Conv1d(
                    hidden, n_filters, 1)))
        self.separator = nn.ModuleList(
            layers)

        self.mask = nn.Sequential(
            nn.Conv1d(
                n_filters, n_filters, 1),
            nn.Sigmoid())

        # Learned decoder (replaces iSTFT)
        self.decoder = nn.ConvTranspose1d(
            n_filters, 1, kernel_size,
            stride=kernel_size // 2,
            bias=False)

    def forward(self, x):
        """x: (batch, 1, samples)"""
        enc = F.relu(self.encoder(x))

        sep = enc
        for block in self.separator:
            sep = sep + block(sep)

        masked = enc * self.mask(sep)
        return self.decoder(masked)


model = SimpleWaveformDenoiser()
total = sum(p.numel()
            for p in model.parameters())
audio = torch.randn(2, 1, 16000)
clean = model(audio)
print(f"Input: {audio.shape}")
print(f"Output: {clean.shape}")
print(f"Parameters: {total:,}")

The learned encoder replaces the STFT -- it's a 1D convolution that transforms the waveform into a representation that the separator network can work with. The learned decoder (transposed convolution) converts back to a waveform. The separator uses dilated convolutions with exponentially increasing dilation rates (1, 2, 4, 8) to capture both local and long-range temporal context -- the same dilation trick we saw in WaveNet-style architectures (episode #94).

The advantage over spectrogram methods is that everything is learned end-to-end. The model discovers whatever intermediate representation works best for separating speech from noise, rather than being constrained to work with human-designed spectrograms. In practice, waveform-domain models consistantly achieve higher perceptual quality scores than spectrogram-mask methods, particularly for speech naturalness.

Acoustic echo cancellation

Acoustic Echo Cancellation (AEC) handles a very specific problem in real-time communication. During a video call, your microphone picks up audio playing through your speakers -- the far-end speaker hears their own voice echoed back with a delay. AEC estimates what the microphone captured from the speaker output and subtracts it, leaving only the near-end speech (your voice):

class AdaptiveEchoCanceller:
    """LMS-based acoustic echo
    cancellation."""

    def __init__(self, filter_length=512,
                 mu=0.01, sr=16000):
        self.filter_length = filter_length
        self.mu = mu
        self.sr = sr

    def cancel(self, mic, ref):
        """mic: microphone signal
        (speech + echo).
        ref: reference signal from
        speakers (far-end audio)."""
        n = len(mic)
        w = np.zeros(self.filter_length)
        output = np.zeros(n)
        erle_history = []

        for i in range(
                self.filter_length, n):
            x = ref[
                i - self.filter_length:i
                ][::-1]

            echo_est = np.dot(w, x)
            error = mic[i] - echo_est
            output[i] = error

            norm = np.dot(x, x) + 1e-10
            w += self.mu * error * x / norm

            if i % 1000 == 0 and i > 0:
                win = slice(
                    max(0, i-1000), i)
                echo_pow = np.mean(
                    mic[win] ** 2)
                res_pow = np.mean(
                    output[win] ** 2)
                if res_pow > 1e-10:
                    erle = 10 * np.log10(
                        echo_pow / res_pow)
                    erle_history.append(erle)

        return output, erle_history

    def run(self):
        rng = np.random.RandomState(42)
        dur = 3.0
        n = int(self.sr * dur)
        t = np.arange(n) / self.sr

        # Far-end speech
        far_end = np.zeros(n)
        for f, a in [(200, 0.3),
                     (400, 0.15),
                     (800, 0.08)]:
            far_end += a * np.sin(
                2*np.pi*f*t)
        far_end *= (0.5 + 0.5 * np.sin(
            2*np.pi*3*t))

        # Room impulse response
        rir_len = 200
        rir = np.zeros(rir_len)
        rir[30] = 0.8   # direct path
        rir[75] = 0.3   # first reflection
        rir[120] = 0.15  # second reflection
        rir[180] = 0.05  # late reverb

        echo = np.convolve(
            far_end, rir)[:n]

        # Near-end speech (local speaker)
        near_end = np.zeros(n)
        for f, a in [(150, 0.2),
                     (300, 0.1)]:
            near_end += a * np.sin(
                2*np.pi*f*t)
        ne_env = np.zeros(n)
        ne_env[int(1.0*self.sr):
               int(2.5*self.sr)] = 1.0
        near_end *= ne_env

        mic = near_end + echo
        mic += rng.randn(n) * 0.01

        output, erle = self.cancel(
            mic, far_end)

        print(f"Echo cancellation results:")
        print(f"  Filter: "
              f"{self.filter_length} taps")
        print(f"  Room impulse: "
              f"{rir_len} samples "
              f"({rir_len/self.sr*1000:.0f}"
              f"ms)")
        if erle:
            print(f"  Final ERLE: "
                  f"{erle[-1]:.1f} dB")
            print(f"  ERLE progression: "
                  f"{[f'{e:.1f}' for e in erle]}")

        ne_region = slice(
            int(1.2*self.sr),
            int(2.0*self.sr))
        ne_corr = np.corrcoef(
            near_end[ne_region],
            output[ne_region])[0, 1]
        print(f"  Near-end correlation: "
              f"{ne_corr:.3f}")


aec = AdaptiveEchoCanceller()
aec.run()

The adaptive filter learns the room impulse response -- how sound bounces from speaker to microphone through the room. The four taps in our synthetic RIR represent the direct path (30 samples delay), first wall reflection, second reflection, and late reverberance. As the room changes (someone moves, a door opens), the LMS filter continuously adapts its weights.

The ERLE (Echo Return Loss Enhancement) metric measures how much echo the canceller removes, in dB. A good AEC achieves 20-40 dB ERLE, meaning the echo is reduced to 1/100th to 1/10000th of its original power. Modern production AEC systems use neural networks instead of LMS filters and handle even double-talk (both people speaking simultaneously) much more robustly -- a situation where the classic LMS filter tends to diverge because it can't distinguish near-end speech from echo estimation error.

Audio super-resolution

Audio super-resolution upsamples low-bandwidth audio to higher fidelity. A phone call sampled at 8kHz (cutting off at 4kHz per Nyquist) sounds muffled because everything above 4kHz is gone -- the sibilants ("s", "sh", "f"), the breathiness, the high harmonics that give voices their crisp, natural quality. Audio super-resolution hallucinates those missing frequencies based on learned correlations between low and high frequency content:

class AudioSuperResDemo:
    """Demonstrate audio super-resolution
    concept with spectral analysis."""

    def __init__(self, sr_high=16000,
                 sr_low=8000):
        self.sr_high = sr_high
        self.sr_low = sr_low

    def synthesize_speech(self, sr,
                          dur=2.0):
        """Generate speech-like signal
        with harmonics and formants."""
        rng = np.random.RandomState(42)
        n = int(sr * dur)
        t = np.arange(n) / sr

        signal = np.zeros(n)
        f0 = 150
        for h in range(1, 20):
            freq = f0 * h
            if freq > sr / 2:
                break
            amp = 0.3 / h
            for fc, bw in [
                    (500, 100),
                    (1500, 200),
                    (2500, 300),
                    (3500, 400),
                    (5000, 500)]:
                if freq > sr / 2:
                    break
                gain = np.exp(
                    -0.5 * ((freq - fc)
                            / bw) ** 2)
                amp *= (1 + 2 * gain)
            signal += amp * np.sin(
                2*np.pi*freq*t)

        # Fricative noise bursts
        fric = rng.randn(n) * 0.05
        if sr >= 16000:
            from_bin = int(
                3000 * n / sr)
            to_bin = int(
                min(7000, sr/2) * n / sr)
            fft = np.fft.rfft(fric)
            fft[:from_bin] = 0
            fft[to_bin:] = 0
            fric = np.fft.irfft(
                fft, n=n)
        signal += fric

        env = np.zeros(n)
        env[int(0.2*sr):
            int(1.8*sr)] = 1.0
        signal *= env
        return signal

    def analyze_bandwidth(self, signal,
                           sr, label):
        fft = np.abs(np.fft.rfft(signal))
        freqs = np.fft.rfftfreq(
            len(signal), 1.0/sr)

        bands = [
            ('0-1kHz', 0, 1000),
            ('1-2kHz', 1000, 2000),
            ('2-4kHz', 2000, 4000),
            ('4-6kHz', 4000, 6000),
            ('6-8kHz', 6000, 8000)]

        print(f"\n  {label}:")
        for name, lo, hi in bands:
            if hi > sr / 2:
                print(f"    {name}: "
                      f"(beyond Nyquist)")
                continue
            mask = (freqs >= lo) & (
                freqs < hi)
            energy = np.mean(
                fft[mask] ** 2)
            print(f"    {name}: "
                  f"{energy:.4f}")

    def run(self):
        high = self.synthesize_speech(
            self.sr_high)
        low = self.synthesize_speech(
            self.sr_low)

        print("Audio super-resolution "
              "analysis:")
        print(f"  High-res: "
              f"{self.sr_high} Hz "
              f"(Nyquist: "
              f"{self.sr_high//2} Hz)")
        print(f"  Low-res:  "
              f"{self.sr_low} Hz "
              f"(Nyquist: "
              f"{self.sr_low//2} Hz)")

        self.analyze_bandwidth(
            high, self.sr_high,
            "16kHz signal")
        self.analyze_bandwidth(
            low, self.sr_low,
            "8kHz signal (phone quality)")

        fft_high = np.abs(
            np.fft.rfft(high))
        freqs = np.fft.rfftfreq(
            len(high), 1.0/self.sr_high)
        above_4k = freqs >= 4000
        missing = np.mean(
            fft_high[above_4k] ** 2)
        total = np.mean(fft_high ** 2)
        print(f"\n  Energy above 4kHz: "
              f"{missing/total:.1%} "
              f"of total")
        print(f"  This is what super-"
              f"resolution must predict")


demo = AudioSuperResDemo()
demo.run()

This is conceptually identical to image super-resolution (episode #86) -- predict missing detail from low-resolution input. The training pairs are trivial to create: take high-quality 16kHz audio, downsample it to 8kHz, and train the model to recover the original high frequencies. Modern audio super-resolution models (typically U-Net or WaveNet variants) achieve remarkebly natural-sounding results -- phone calls that sound closer to studio recordings. The model learns that certain low-frequency patterns (like the onset of a sibilant "s" visible in the lower harmonics) predict specific high-frequency content.

Building a complete multi-stage pipeline

Let me put everything together into a complete pipeline that combines multiple enhancement stages. Real-world production systems don't use just one technique -- they stack several, each attacking a different aspect of the noise:

class NoiseReductionPipeline:
    """Multi-stage noise reduction
    combining classical methods."""

    def __init__(self, sr=16000,
                 n_fft=1024, hop=256):
        self.sr = sr
        self.n_fft = n_fft
        self.hop = hop

    def stft(self, audio):
        window = np.hanning(self.n_fft)
        nf = ((len(audio) - self.n_fft)
              // self.hop + 1)
        nb = self.n_fft // 2 + 1
        result = np.zeros(
            (nb, nf), dtype=complex)
        for i in range(nf):
            s = i * self.hop
            result[:, i] = np.fft.rfft(
                audio[s:s+self.n_fft]
                * window)
        return result

    def istft(self, spec, length):
        window = np.hanning(self.n_fft)
        nf = spec.shape[1]
        out = np.zeros(length)
        ws = np.zeros(length)
        for i in range(nf):
            s = i * self.hop
            frame = np.fft.irfft(
                spec[:, i])
            e = min(s+self.n_fft, length)
            fl = e - s
            out[s:e] += (
                frame[:fl]*window[:fl])
            ws[s:e] += window[:fl]**2
        m = ws > 1e-8
        out[m] /= ws[m]
        return out

    def vad_gate(self, audio,
                 frame_ms=25, hop_ms=10,
                 thresh_db=-25):
        """Simple energy-based VAD gate:
        suppress non-speech frames."""
        frame_len = int(
            self.sr * frame_ms / 1000)
        hop_len = int(
            self.sr * hop_ms / 1000)
        nf = ((len(audio) - frame_len)
              // hop_len + 1)
        gated = audio.copy()

        for i in range(nf):
            s = i * hop_len
            frame = audio[
                s:s + frame_len]
            rms = np.sqrt(
                np.mean(frame ** 2))
            db = 20 * np.log10(
                max(rms, 1e-10))
            if db < thresh_db:
                e = min(
                    s + frame_len,
                    len(gated))
                gated[s:e] *= 0.01
        return gated

    def wiener_stage(self, audio,
                     noise_frames=20):
        spec = self.stft(audio)
        power = np.abs(spec) ** 2
        noise_p = np.mean(
            power[:, :noise_frames],
            axis=1, keepdims=True)
        sig_p = np.maximum(
            power - noise_p, 0.0)
        gain = sig_p / (
            sig_p + noise_p + 1e-10)
        for i in range(1, gain.shape[1]):
            gain[:, i] = (
                0.85 * gain[:, i-1]
                + 0.15 * gain[:, i])
        clean = (np.abs(spec) * gain
                 * np.exp(
                     1j*np.angle(spec)))
        return self.istft(
            clean, len(audio))

    def spectral_floor_stage(self, audio,
                              noise_frames=20,
                              floor=0.01):
        spec = self.stft(audio)
        mag = np.abs(spec)
        noise_est = np.mean(
            mag[:, :noise_frames],
            axis=1, keepdims=True)
        clean_mag = np.maximum(
            mag - 1.5 * noise_est,
            floor * noise_est)
        clean = clean_mag * np.exp(
            1j * np.angle(spec))
        return self.istft(
            clean, len(audio))

    def run(self):
        rng = np.random.RandomState(42)
        dur = 4.0
        n = int(self.sr * dur)
        t = np.arange(n) / self.sr

        speech = np.zeros(n)
        for f, a in [(130, 0.25),
                     (260, 0.15),
                     (780, 0.08),
                     (2100, 0.04)]:
            speech += a * np.sin(
                2*np.pi*f*t)
        speech *= (0.5 + 0.5 * np.sin(
            2*np.pi*4*t))
        env = np.zeros(n)
        env[int(0.5*self.sr):
            int(3.5*self.sr)] = 1.0
        speech *= env

        # Multiple noise types
        noise = rng.randn(n) * 0.1
        noise += 0.05 * np.sin(
            2*np.pi*50*t)  # mains hum
        for _ in range(8):  # clicks
            pos = rng.randint(n)
            noise[pos:pos+50] += (
                rng.randn(min(50, n-pos))
                * 0.3)

        noisy = speech + noise

        def snr(clean, test, region):
            sp = np.mean(clean[region]**2)
            np_ = np.mean(
                (test[region]
                 - clean[region])**2)
            return 10*np.log10(
                sp / max(np_, 1e-10))

        sp_region = slice(
            int(1.0*self.sr),
            int(3.0*self.sr))

        results = {'Input': noisy}
        s1 = self.vad_gate(noisy)
        results['VAD gate'] = s1
        s2 = self.wiener_stage(s1)
        results['+ Wiener'] = s2
        s3 = self.spectral_floor_stage(s2)
        results['+ Spec floor'] = s3

        print(f"Multi-stage pipeline:")
        print(f"{'Stage':<16} {'SNR (dB)':>10}")
        print("-" * 28)
        for name, sig in results.items():
            s = snr(speech, sig, sp_region)
            print(f"{name:<16} {s:>10.1f}")


pipe = NoiseReductionPipeline()
pipe.run()

Each stage attacks a different aspect of the noise. The VAD gate suppresses frames where there's no speech at all (silence or pure noise segments), preventing noise from bleeding through during pauses. The Wiener filter handles stationary broadband noise (fan hum, white noise). The spectral floor catches remaining narrowband noise and musical noise artifacts from the previous stage. The combined pipeline achieves significantly higher SNR improvement than any single stage alone.

This is how production audio enhancement works conceptually -- systems like Krisp, NVIDIA RTX Voice, and the noise suppression built into Zoom all use multi-stage approaches. The key difference is that they replace each classical stage with a neural network component, and often train the entire pipeline end-to-end so the stages learn to complement each other.

Real-time constraints and evaluation

Audio enhancement for live applications (video calls, hearing aids, live streaming) has a hard constraint that batch processing doesn't: latency. You can't buffer 5 seconds of audio before processing it -- the delay would make conversation impossible. Acceptable latency is under 40ms, and ideally under 20ms for hearing aids:

class RealtimeAnalysis:
    """Analyze latency vs quality
    tradeoffs for real-time audio
    enhancement."""

    def __init__(self, sr=16000):
        self.sr = sr

    def latency_analysis(self):
        configs = [
            ('Hearing aid', 5, 64, 32),
            ('VoIP low', 10, 160, 80),
            ('VoIP standard', 20, 320, 160),
            ('VoIP generous', 40, 640, 320),
            ('Offline batch', 500, 8000,
             4000),
        ]
        print(f"{'Config':<18} "
              f"{'Latency':>8} "
              f"{'Window':>8} "
              f"{'Hop':>6} "
              f"{'Chunks/s':>10}")
        print("-" * 54)
        for name, lat, win, hop in configs:
            chunks = self.sr // hop
            print(f"{name:<18} "
                  f"{lat:>6}ms "
                  f"{win:>7} "
                  f"{hop:>5} "
                  f"{chunks:>9}")

    def quality_metrics(self):
        rng = np.random.RandomState(42)
        n = 16000
        t = np.arange(n) / 16000.0
        clean = np.zeros(n)
        for f, a in [(150, 0.3),
                     (300, 0.15),
                     (900, 0.08)]:
            clean += a * np.sin(
                2*np.pi*f*t)
        clean *= (0.5 + 0.5 * np.sin(
            2*np.pi*4*t))

        scenarios = {
            'Perfect': clean.copy(),
            'Mild noise': (
                clean + rng.randn(n)*0.03),
            'Moderate': (
                clean + rng.randn(n)*0.1),
            'Heavy noise': (
                clean + rng.randn(n)*0.3),
            'Distorted': np.clip(
                clean * 3, -0.5, 0.5),
        }

        print(f"\n{'Scenario':<18} "
              f"{'SI-SNR':>8} "
              f"{'Corr':>8} "
              f"{'STOI-like':>10}")
        print("-" * 48)

        for name, test in (
                scenarios.items()):
            target = clean
            s_target = (
                np.dot(test, target)
                / np.dot(target, target)
                * target)
            e_noise = test - s_target
            si_snr = 10 * np.log10(
                np.dot(s_target, s_target)
                / (np.dot(e_noise, e_noise)
                   + 1e-10))

            corr = np.corrcoef(
                clean, test)[0, 1]

            frame_len = 256
            stoi_scores = []
            for i in range(0, n-frame_len,
                           frame_len//2):
                c = clean[i:i+frame_len]
                d = test[i:i+frame_len]
                if np.std(c) > 1e-6:
                    r = np.corrcoef(
                        c, d)[0, 1]
                    stoi_scores.append(
                        max(0, r))
            stoi = np.mean(stoi_scores)

            print(f"{name:<18} "
                  f"{si_snr:>7.1f} "
                  f"{corr:>8.3f} "
                  f"{stoi:>10.3f}")

    def run(self):
        print("=== Real-time Analysis ===\n")
        self.latency_analysis()
        self.quality_metrics()


analysis = RealtimeAnalysis()
analysis.run()

PESQ (Perceptual Evaluation of Speech Quality) is the ITU standard metric that models human perception of speech quality, with scores ranging from -0.5 to 4.5. STOI (Short-Time Objective Intelligibility) measures how understandable the speech remains on a scale from 0 to 1 -- good enhancement achieves 0.85-0.95. SI-SNR (Scale-Invariant Signal-to-Noise Ratio) measures signal-to-distortion ratio in dB and is the most common training objective for neural speech enhancement because it's differentiable and correlates well with perceptual quality.

The latency table shows the fundamental tradeoff: shorter processing windows mean lower latency but less frequency resolution and less context for the model to work with. Hearing aids must operate at 5ms latency with tiny 64-sample windows. VoIP systems get 20-40ms. Offline processing (podcast cleanup, film post-production) can use arbitrarily large windows for maximum quality.

The state of the art

Modern production systems combine multiple innovations. Meta's Denoiser (based on the DEMUCS architecture) achieves near-studio quality on single-channel noisy speech at real-time speeds. Microsoft's DTLN (Dual-Signal Transformation LSTM Network) targets extreme low-latency use at 5-10ms. Google's research into unified audio models builds on the same foundations we've been exploring throughout this whole audio arc.

The field continues moving rapidly. Recent work focuses on personalized enhancement (adapting to a specific speaker's voice characteristics, using the speaker embeddings from episode #97), multi-channel processing (using multiple microphones for spatial filtering and beamforming), and unified models that handle denoising, dereverberation, echo cancellation, and bandwidth extension all in a single forward pass. The trend is clear: specialized single-task models are being replaced by general-purpose audio enhancement networks that handle whatever degradation the input happens to contain.

Samengevat

  • spectral subtraction removes noise by estimating and subtracting the noise spectrum in the frequency domain -- simple and fast but produces musical noise artifacts and assumes stationary noise;
  • the Wiener filter computes an optimal gain mask from signal-to-noise ratio estimates -- smoother output than subtraction, less musical noise, but still limited by noise stationarity assumptions;
  • deep learning denoising trains U-Nets on spectrogram masks or waveform-domain models on raw audio pairs -- handles non-stationary noise, adapts per-frame, and dramatically outperforms the classical methods;
  • waveform-domain models (Conv-TasNet, DEMUCS) avoid phase reconstruction problems entirely by learning the encoder/decoder from raw audio using 1D dilated convolution stacks;
  • acoustic echo cancellation uses adaptive LMS filters (or neural networks) to estimate the room impulse response and subtract speaker-to-microphone echo -- measured by ERLE in dB;
  • audio super-resolution predicts missing high-frequency content from low-bandwidth recordings -- same concept as image super-resolution (episode #86), trained on downsampled pairs;
  • multi-stage pipelines combine VAD gating, Wiener filtering, and spectral flooring -- production systems replace each stage with neural components trained end-to-end;
  • real-time applications demand under 40ms latency, requiring causal architectures, small windows, and streaming inference; quality is measured with PESQ (perceptual), STOI (intelligibility), and SI-SNR (signal-to-distortion).

We're nearly done with the audio AI domain. The next episode brings audio and vision together into unified multimodal representations -- how models learn to jointly understand what they see and what they hear. Then we'll wrap the entire audio arc with a hands-on mini project before stepping into reinforcement learning territory. Exciting times ahead!

Exercises

Exercise 1: Build a multi-band noise reducer. Create MultiBandDenoiser that splits audio into 4 bands (0-500Hz, 500-2kHz, 2-4kHz, 4-8kHz) via FFT slicing, estimates noise per band from the first 0.3s, applies Wiener filtering per band with band-specific smoothing (low bands: more smoothing, high bands: less), and recombines. Test on a 3-second signal with formant speech (0.5-2.5s) plus different noise per band (60Hz hum, broadband hiss, high-frequency crackle). Compare per-band and total SNR against single-band Wiener. Multi-band should win because it applies different attenuation per frequency region.

Exercise 2: Build an adaptive noise tracker. Create AdaptiveNoiseTracker using minimum statistics (track minimum power per frequency bin over a 50-frame sliding window). Generate 5 seconds where noise changes at 2.5s (white noise 0.05 -> pink noise 0.08), speech active 1-4s. Implement Wiener filtering with the adaptive estimate (updated every 10 frames). Compare: (1) fixed estimate from first 0.3s, (2) fixed estimate from first 0.3s after the noise change, (3) your adaptive tracker. Print SNR before and after the change. The adaptive tracker should handle both regions well; fixed estimate from region 1 should degrade in region 2.

Exercise 3: Build a perceptual quality comparator. Create QualityComparator that generates 2s clean speech, makes 5 degraded versions (0dB noise, 20dB noise, clipping via np.clip(signal*3, -0.5, 0.5), bandlimited to 4kHz, reverb via 100-tap exponential-decay impulse response). Compute 4 metrics: SI-SNR, segmental SNR (per 32ms frame), spectral distortion (log-spectral distance), short-time correlation (STOI-like). Print comparison table. Verify: clipping has low spectral distortion but poor SI-SNR; bandlimiting has high spectral distortion but decent correlation.

Bedankt en tot de volgende!

@scipio

stem stemsocial steemstem python programming
0
0
0.000
0 comments