kirad.augmentations

kirad.augmentations

Data stage (optional): RawBoost waveform augmentation, applied in KintsugiAudioDataset.

Enabled per config via augmentations: [{class_name, params, aug_prob}]; runs on CPU numpy.

RawBoost

RawBoost waveform augmentation (LnL/ISD/SSI noise); default algo 4 = all three.

Plug-and-play RawBoost augmentation (official defaults) The following are plug-and-play algorithm presets from the official paper: https://arxiv.org/abs/2111.04433 Preset #4 seems to be the top-performing one.

See the paper for the meaning of all other parameters algo: 0 - None 1 - LnL 2 - ISD 3 - SSI 4 - LnL → ISD → SSI 5 - LnL → ISD 6 - LnL → SSI 7 - ISD → SSI 8 - LnL + ISD (parallel summed)

Source code in kirad/augmentations.py

class RawBoost:
    """RawBoost waveform augmentation (LnL/ISD/SSI noise); default algo 4 = all three.

    Plug-and-play RawBoost augmentation (official defaults)
    The following are plug-and-play algorithm presets from the official paper:
    https://arxiv.org/abs/2111.04433
    Preset #4 seems to be the top-performing one.

    See the paper for the meaning of all other parameters
    algo:
      0 - None
      1 - LnL
      2 - ISD
      3 - SSI
      4 - LnL → ISD → SSI
      5 - LnL → ISD
      6 - LnL → SSI
      7 - ISD → SSI
      8 - LnL + ISD (parallel summed)
    """

    def __init__(
        self,
        algo=4,
        fs=16000,  # HARDCODED: assumes 16 kHz audio; maxF=8000 is its Nyquist
        nBands=5,
        minF=20,
        maxF=8000,
        minBW=100,
        maxBW=1000,
        minCoeff=10,
        maxCoeff=100,
        minG=0,
        maxG=0,
        minBiasLinNonLin=5,
        maxBiasLinNonLin=20,
        N_f=5,
        P=10,
        g_sd=2,
        SNRmin=10,
        SNRmax=40,
    ):
        """Store RawBoost parameters (paper defaults); raises if algo > 8."""

        if algo > 8:
            raise ValueError(f"Invalid algo: {algo}")

        self.algo = algo
        self.fs = fs

        # Notch-filter / LnL parameters.
        self.nBands = nBands
        self.minF = minF
        self.maxF = maxF
        self.minBW = minBW
        self.maxBW = maxBW
        self.minCoeff = minCoeff
        self.maxCoeff = maxCoeff
        self.minG = minG
        self.maxG = maxG
        self.minBias = minBiasLinNonLin
        self.maxBias = maxBiasLinNonLin
        self.N_f = N_f

        # ISD parameters.
        self.P = P
        self.g_sd = g_sd

        # SSI SNR range (dB).
        self.SNRmin = SNRmin
        self.SNRmax = SNRmax

    def __call__(self, x: torch.tensor):
        """Augment one waveform [T]; returns a tensor with the input dtype and device.

        Note: calls x.numpy(), so x must be a CPU tensor.
        """

        # Work in numpy on CPU; remember dtype/device to restore them at the end.
        orig_dtype, orig_device = x.dtype, x.device
        x = x.numpy()

        # convenience function aliases
        LnL = lambda sig: LnL_convolutive_noise(
            sig,
            self.N_f,
            self.nBands,
            self.minF,
            self.maxF,
            self.minBW,
            self.maxBW,
            self.minCoeff,
            self.maxCoeff,
            self.minG,
            self.maxG,
            self.minBias,
            self.maxBias,
            self.fs,
        )

        ISD = lambda sig: ISD_additive_noise(sig, self.P, self.g_sd)

        SSI = lambda sig: SSI_additive_noise(
            sig,
            self.SNRmin,
            self.SNRmax,
            self.nBands,
            self.minF,
            self.maxF,
            self.minBW,
            self.maxBW,
            self.minCoeff,
            self.maxCoeff,
            self.minG,
            self.maxG,
            self.fs,
        )

        # ----- Algorithms -----
        if self.algo == 1:  # LnL
            x = LnL(x)

        if self.algo == 2:  # ISD
            x = ISD(x)

        if self.algo == 3:  # SSI
            x = SSI(x)

        if self.algo == 4:  # LnL → ISD → SSI
            x = SSI(ISD(LnL(x)))

        if self.algo == 5:  # LnL → ISD
            x = ISD(LnL(x))

        if self.algo == 6:  # LnL → SSI
            x = SSI(LnL(x))

        if self.algo == 7:  # ISD → SSI
            x = SSI(ISD(x))

        if self.algo == 8:  # parallel LnL + ISD
            x = LnL(x) + ISD(x)

        return torch.tensor(x, dtype=orig_dtype, device=orig_device)

algo = algo instance-attribute

fs = fs instance-attribute

nBands = nBands instance-attribute

minF = minF instance-attribute

maxF = maxF instance-attribute

minBW = minBW instance-attribute

maxBW = maxBW instance-attribute

minCoeff = minCoeff instance-attribute

maxCoeff = maxCoeff instance-attribute

minG = minG instance-attribute

maxG = maxG instance-attribute

minBias = minBiasLinNonLin instance-attribute

maxBias = maxBiasLinNonLin instance-attribute

N_f = N_f instance-attribute

P = P instance-attribute

g_sd = g_sd instance-attribute

SNRmin = SNRmin instance-attribute

SNRmax = SNRmax instance-attribute

__init__(algo=4, fs=16000, nBands=5, minF=20, maxF=8000, minBW=100, maxBW=1000, minCoeff=10, maxCoeff=100, minG=0, maxG=0, minBiasLinNonLin=5, maxBiasLinNonLin=20, N_f=5, P=10, g_sd=2, SNRmin=10, SNRmax=40)

Store RawBoost parameters (paper defaults); raises if algo > 8.

Source code in kirad/augmentations.py

def __init__(
    self,
    algo=4,
    fs=16000,  # HARDCODED: assumes 16 kHz audio; maxF=8000 is its Nyquist
    nBands=5,
    minF=20,
    maxF=8000,
    minBW=100,
    maxBW=1000,
    minCoeff=10,
    maxCoeff=100,
    minG=0,
    maxG=0,
    minBiasLinNonLin=5,
    maxBiasLinNonLin=20,
    N_f=5,
    P=10,
    g_sd=2,
    SNRmin=10,
    SNRmax=40,
):
    """Store RawBoost parameters (paper defaults); raises if algo > 8."""

    if algo > 8:
        raise ValueError(f"Invalid algo: {algo}")

    self.algo = algo
    self.fs = fs

    # Notch-filter / LnL parameters.
    self.nBands = nBands
    self.minF = minF
    self.maxF = maxF
    self.minBW = minBW
    self.maxBW = maxBW
    self.minCoeff = minCoeff
    self.maxCoeff = maxCoeff
    self.minG = minG
    self.maxG = maxG
    self.minBias = minBiasLinNonLin
    self.maxBias = maxBiasLinNonLin
    self.N_f = N_f

    # ISD parameters.
    self.P = P
    self.g_sd = g_sd

    # SSI SNR range (dB).
    self.SNRmin = SNRmin
    self.SNRmax = SNRmax

__call__(x)

Augment one waveform [T]; returns a tensor with the input dtype and device.

Note: calls x.numpy(), so x must be a CPU tensor.

Source code in kirad/augmentations.py

def __call__(self, x: torch.tensor):
    """Augment one waveform [T]; returns a tensor with the input dtype and device.

    Note: calls x.numpy(), so x must be a CPU tensor.
    """

    # Work in numpy on CPU; remember dtype/device to restore them at the end.
    orig_dtype, orig_device = x.dtype, x.device
    x = x.numpy()

    # convenience function aliases
    LnL = lambda sig: LnL_convolutive_noise(
        sig,
        self.N_f,
        self.nBands,
        self.minF,
        self.maxF,
        self.minBW,
        self.maxBW,
        self.minCoeff,
        self.maxCoeff,
        self.minG,
        self.maxG,
        self.minBias,
        self.maxBias,
        self.fs,
    )

    ISD = lambda sig: ISD_additive_noise(sig, self.P, self.g_sd)

    SSI = lambda sig: SSI_additive_noise(
        sig,
        self.SNRmin,
        self.SNRmax,
        self.nBands,
        self.minF,
        self.maxF,
        self.minBW,
        self.maxBW,
        self.minCoeff,
        self.maxCoeff,
        self.minG,
        self.maxG,
        self.fs,
    )

    # ----- Algorithms -----
    if self.algo == 1:  # LnL
        x = LnL(x)

    if self.algo == 2:  # ISD
        x = ISD(x)

    if self.algo == 3:  # SSI
        x = SSI(x)

    if self.algo == 4:  # LnL → ISD → SSI
        x = SSI(ISD(LnL(x)))

    if self.algo == 5:  # LnL → ISD
        x = ISD(LnL(x))

    if self.algo == 6:  # LnL → SSI
        x = SSI(LnL(x))

    if self.algo == 7:  # ISD → SSI
        x = SSI(ISD(x))

    if self.algo == 8:  # parallel LnL + ISD
        x = LnL(x) + ISD(x)

    return torch.tensor(x, dtype=orig_dtype, device=orig_device)

randRange(x1, x2, integer)

Uniform random value in [x1, x2), cast to int if integer.

Source code in kirad/augmentations.py

def randRange(x1, x2, integer):
    """Uniform random value in [x1, x2), cast to int if `integer`."""
    y = np.random.uniform(low=x1, high=x2, size=(1,))
    if integer:
        y = int(y)
    return y

normWav(x, always)

Peak-normalize to [-1, 1] always, or only when the peak exceeds 1.

Source code in kirad/augmentations.py

def normWav(x, always):
    """Peak-normalize to [-1, 1] always, or only when the peak exceeds 1."""
    if always:
        x = x / np.amax(abs(x))
    elif np.amax(abs(x)) > 1:
        x = x / np.amax(abs(x))
    return x

genNotchCoeffs(nBands, minF, maxF, minBW, maxBW, minCoeff, maxCoeff, minG, maxG, fs)

Random FIR filter: cascade of nBands random notch (band-stop) filters, random gain.

Source code in kirad/augmentations.py

def genNotchCoeffs(
    nBands, minF, maxF, minBW, maxBW, minCoeff, maxCoeff, minG, maxG, fs
):
    """Random FIR filter: cascade of nBands random notch (band-stop) filters, random gain."""
    # Cascade nBands random notch filters by convolving their impulse responses.
    b = 1
    for i in range(0, nBands):
        # Random centre frequency, bandwidth and tap count for this notch.
        fc = randRange(minF, maxF, 0)
        bw = randRange(minBW, maxBW, 0)
        c = randRange(minCoeff, maxCoeff, 1)

        # Tap count must be odd for a band-stop FIR.
        if c / 2 == int(c / 2):
            c = c + 1
        # Band edges around fc, clamped inside (0, Nyquist).
        f1 = fc - bw / 2
        f2 = fc + bw / 2
        if f1 <= 0:
            f1 = 1 / 1000
        if f2 >= fs / 2:
            f2 = fs / 2 - 1 / 1000
        # Design the band-stop FIR and fold it into the cascade.
        b = np.convolve(
            signal.firwin(c, [float(f1), float(f2)], window="hamming", fs=fs), b
        )

    # Normalize peak frequency response to 1, then apply random gain G (dB).
    G = randRange(minG, maxG, 0)
    _, h = signal.freqz(b, 1, fs=fs)
    b = pow(10, G / 20) * b / np.amax(abs(h))
    return b

filterFIR(x, b)

Apply FIR filter b to x, compensating the filter delay (same output length).

Source code in kirad/augmentations.py

def filterFIR(x, b):
    """Apply FIR filter b to x, compensating the filter delay (same output length)."""
    N = b.shape[0] + 1
    xpad = np.pad(x, (0, N), "constant")
    y = signal.lfilter(b, 1, xpad)
    y = y[int(N / 2) : int(y.shape[0] - N / 2)]
    return y

LnL_convolutive_noise(x, N_f, nBands, minF, maxF, minBW, maxBW, minCoeff, maxCoeff, minG, maxG, minBiasLinNonLin, maxBiasLinNonLin, fs)

LnL: sum of random FIR-filtered powers x, x^2, ..., x^N_f (channel distortion).

Source code in kirad/augmentations.py

def LnL_convolutive_noise(
    x,
    N_f,
    nBands,
    minF,
    maxF,
    minBW,
    maxBW,
    minCoeff,
    maxCoeff,
    minG,
    maxG,
    minBiasLinNonLin,
    maxBiasLinNonLin,
    fs,
):
    """LnL: sum of random FIR-filtered powers x, x^2, ..., x^N_f (channel distortion)."""
    y = [0] * x.shape[0]
    # Sum FIR-filtered powers x^1..x^N_f of the signal.
    for i in range(0, N_f):
        # From the 2nd power on, lower the gain range by the bias.
        if i == 1:
            minG = minG - minBiasLinNonLin
            maxG = maxG - maxBiasLinNonLin
        b = genNotchCoeffs(
            nBands, minF, maxF, minBW, maxBW, minCoeff, maxCoeff, minG, maxG, fs
        )
        y = y + filterFIR(np.power(x, (i + 1)), b)
    # Remove DC offset and peak-normalize if above 1.
    y = y - np.mean(y)
    y = normWav(y, 0)
    return y

ISD_additive_noise(x, P, g_sd)

ISD: perturb up to P% of random samples by signal-proportional impulsive noise.

Source code in kirad/augmentations.py

def ISD_additive_noise(x, P, g_sd):
    """ISD: perturb up to P% of random samples by signal-proportional impulsive noise."""
    # Random percentage (up to P) of samples to corrupt.
    beta = randRange(0, P, 0)

    # Pick random sample positions to perturb.
    y = copy.deepcopy(x)
    x_len = x.shape[0]
    n = int(x_len * (beta / 100))
    p = np.random.permutation(x_len)[:n]
    # Impulsive noise g_sd * x * (u1 * u2), u in [-1, 1]; then peak-normalize.
    f_r = np.multiply(
        ((2 * np.random.rand(p.shape[0])) - 1), ((2 * np.random.rand(p.shape[0])) - 1)
    )
    r = g_sd * x[p] * f_r
    y[p] = x[p] + r
    y = normWav(y, 0)
    return y

SSI_additive_noise(x, SNRmin, SNRmax, nBands, minF, maxF, minBW, maxBW, minCoeff, maxCoeff, minG, maxG, fs)

SSI: add coloured Gaussian noise at a random SNR in [SNRmin, SNRmax] dB.

Source code in kirad/augmentations.py

def SSI_additive_noise(
    x,
    SNRmin,
    SNRmax,
    nBands,
    minF,
    maxF,
    minBW,
    maxBW,
    minCoeff,
    maxCoeff,
    minG,
    maxG,
    fs,
):
    """SSI: add coloured Gaussian noise at a random SNR in [SNRmin, SNRmax] dB."""
    # White Gaussian noise, coloured by a random notch FIR and peak-normalized.
    noise = np.random.normal(0, 1, x.shape[0])
    b = genNotchCoeffs(
        nBands, minF, maxF, minBW, maxBW, minCoeff, maxCoeff, minG, maxG, fs
    )
    noise = filterFIR(noise, b)
    noise = normWav(noise, 1)
    # Scale noise to a random SNR (dB) relative to the signal energy.
    SNR = randRange(SNRmin, SNRmax, 0)
    noise = (
        noise / np.linalg.norm(noise, 2) * np.linalg.norm(x, 2) / 10.0 ** (0.05 * SNR)
    )
    x = x + noise
    return x