Rooms and channels

kudio.core.room

Rooms: describe one, apply it, measure what it did.

Noise is additive: kudio.Synthesizer mixes it at a chosen SNR, and kudio.spectral_enhance takes it back out. Reverberation is the other kind, and it is convolutive: the room does not add anything, it smears what was already there across time. This module describes a room, puts a clip in it, and measures one from its impulse response.

>>> ir = kudio.rir(16000, rt60=0.6, drr_db=6.0, seed=0)
>>> wet = kudio.apply_rir(speech, ir)
>>> print(kudio.rt60(ir, 16000))
T30 0.60 s (EDT 0.55 s)  ·  C50 +11.7 dB  ·  DRR +6.0 dB  ·  fit 1.000

Removing it is a different problem and is not here. Spectral subtraction has nothing to subtract, and the single-channel predictive methods did not survive measurement. What this module is for is making controlled reverberation, for augmentation, and measuring it when you have an impulse response.

Everything here is mono: a (frames, channels) array is refused rather than read as one interleaved signal twice as long.

The room here is a statistical one, not a real one. Exponentially decaying noise is the textbook model of a diffuse tail and it is what an augmentation pipeline wants: one knob, reproducible, and the answer known in advance. It has no early-reflection pattern, so it will not stand in for a measured impulse response when the shape of the early energy is what matters. Feed apply_rir a real one when you have it.

Every measurement says how much to believe it. A decay curve that is not straight -- too short a tail, noise reaching the floor, a recording that was never impulsive -- produces a T30 anyway. Reverberation.fit is the R² of the line that was fitted, and it is the difference between "this room rings for 600 ms" and "this arithmetic ran".

Classes:

Name	Description
Reverberation	What a room does, measured from its impulse response.

Functions:

Name	Description
rir	A synthetic room impulse response: exponentially decaying noise.
apply_rir	Put y in the room described by ir.
schroeder_curve	Backward-integrated energy decay of ir, in dB, starting at 0.
rt60	Measure a room from its impulse response, to ISO 3382-1.

Attributes:

Name	Type	Description
MAX_EDT_RATIO		How far EDT and T30 may disagree before the measurement stops counting as

MAX_EDT_RATIO module-attribute

MAX_EDT_RATIO = 2.0

How far EDT and T30 may disagree before the measurement stops counting as one. A sanity bound, not a figure from the standard -- see Reverberation.reliable.

Reverberation dataclass

Reverberation(
    t20: float,
    t30: float,
    edt: float,
    c50: float,
    c80: float,
    drr: float,
    fit: float,
    sr: int,
)

What a room does, measured from its impulse response.

Parameters:

Name	Type	Description	Default
t20	float	reverberation time from the -5 to -25 dB decay, in seconds.	required
t30	float	the same from -5 to -35 dB. The usual figure quoted as "the RT60", and the one seconds returns.	required
edt	float	early decay time -- the first 10 dB, extrapolated. Closer to what a listener calls liveness, and it disagrees with T30 in a room that is not diffuse.	required
c50	float	clarity for speech, in dB: energy in the first 50 ms against everything after it. Positive means the direct sound wins.	required
c80	float	the same split at 80 ms, the figure used for music.	required
drr	float	direct-to-reverberant ratio in dB, taking +/-2.5 ms around the peak as the direct sound.	required
fit	float	R² of the straight line fitted to the T30 decay. Read this before quoting the rest.	required

Methods:

Name	Description
reliable	Was this an impulse response, and was its decay straight?

Attributes:

Name	Type	Description
seconds	float	The reverberation time, in seconds -- T30 by convention.

seconds property

seconds: float

The reverberation time, in seconds -- T30 by convention.

reliable

reliable(
    minimum: float = 0.9, edt_ratio: float = MAX_EDT_RATIO
) -> bool

Was this an impulse response, and was its decay straight?

Two questions, because the R² alone answers only the second. Backward integration makes any signal decay monotonically, and over a long clip that decay can be quite straight -- this repo's speech fixture fits at R² 0.94 and comes back with a 3.9 second "reverberation time".

What gives it away is that its EDT and T30 disagree by a factor of eleven. In a real room the first 10 dB and the 5-to-35 dB span describe the same tail and land within a few per cent of each other, so a large disagreement means the two spans were measuring different things and neither of them was a room. edt_ratio is a sanity bound and not a figure from the standard: over 40 seeds of rir(16000, 0.6) the ratio runs 0.87 to 1.16 at drr_db=6 and 0.95 to 1.09 at 0, and published halls stay well inside 0.8 to 1.3.

A close microphone has no EDT to compare: the direct sound alone carries the first 10 dB of the decay, so the EDT is nan. That says nothing against the tail, and the decay is then judged on its fit alone.

Source code in kudio/core/room.py

def reliable(self, minimum: float = 0.9,
             edt_ratio: float = MAX_EDT_RATIO) -> bool:
    """Was this an impulse response, and was its decay straight?

    Two questions, because the R² alone answers only the second. Backward
    integration makes *any* signal decay monotonically, and over a long
    clip that decay can be quite straight -- this repo's speech fixture
    fits at R² 0.94 and comes back with a 3.9 second "reverberation time".

    What gives it away is that its EDT and T30 disagree by a factor of
    eleven. In a real room the first 10 dB and the 5-to-35 dB span describe
    the same tail and land within a few per cent of each other, so a large
    disagreement means the two spans were measuring different things and
    neither of them was a room. *edt_ratio* is a sanity bound and not a
    figure from the standard: over 40 seeds of ``rir(16000, 0.6)`` the
    ratio runs 0.87 to 1.16 at ``drr_db=6`` and 0.95 to 1.09 at 0, and
    published halls stay well inside 0.8 to 1.3.

    A close microphone has no EDT to compare: the direct sound alone
    carries the first 10 dB of the decay, so the EDT is ``nan``. That says
    nothing against the tail, and the decay is then judged on its fit
    alone.
    """
    if not (np.isfinite(self.fit) and self.fit >= minimum):
        return False
    if not np.isfinite(self.t30) or self.t30 <= 0:
        return False
    if not np.isfinite(self.edt):
        return True
    ratio = self.edt / self.t30
    return 1.0 / edt_ratio <= ratio <= edt_ratio

rir

rir(
    sr: int,
    rt60: float = 0.5,
    *,
    drr_db: float = 0.0,
    seconds: Optional[float] = None,
    seed: Optional[int] = None,
) -> ndarray

A synthetic room impulse response: exponentially decaying noise.

Parameters:

Name	Type	Description	Default
rt60	float	the tail this room is asked for, in seconds -- the time the envelope takes to fall 60 dB.	0.5
drr_db	float	how far the direct path sits above the reverberant energy, in dB. This is distance in a readable unit: a close microphone is +10 dB or more, a metre or two away is around 0, and the far end of a hall is negative. Solved for exactly, so rt60 measures back what was asked for.	0.0
seconds	Optional[float]	how long to make it. Defaults to 1.5x rt60, which holds the decay well past the 35 dB that T30 needs.	None
seed	Optional[int]	makes the room reproducible, the way kudio.Synthesizer makes a mixture reproducible. >>> ir = kudio.rir(16000, rt60=0.6, drr_db=6.0, seed=0) >>> room = kudio.rt60(ir, 16000) >>> round(room.t30, 2), round(room.drr, 1) (0.6, 6.0)	None

Source code in kudio/core/room.py

def rir(sr: int, rt60: float = 0.5, *, drr_db: float = 0.0,
        seconds: Optional[float] = None,
        seed: Optional[int] = None) -> np.ndarray:
    """A synthetic room impulse response: exponentially decaying noise.

    :param rt60: the tail this room is asked for, in seconds -- the time the
        envelope takes to fall 60 dB.
    :param drr_db: how far the direct path sits above the reverberant energy,
        in dB. This is distance in a readable unit: a close microphone is
        +10 dB or more, a metre or two away is around 0, and the far end of a
        hall is negative. Solved for exactly, so :func:`rt60` measures back
        what was asked for.
    :param seconds: how long to make it. Defaults to 1.5x *rt60*, which holds
        the decay well past the 35 dB that T30 needs.
    :param seed: makes the room reproducible, the way
        :class:`kudio.Synthesizer` makes a mixture reproducible.

    >>> ir = kudio.rir(16000, rt60=0.6, drr_db=6.0, seed=0)
    >>> room = kudio.rt60(ir, 16000)
    >>> round(room.t30, 2), round(room.drr, 1)               # doctest: +SKIP
    (0.6, 6.0)
    """
    if sr <= 0:
        raise FeatureError(f"rir() needs a positive rate, got {sr}")
    if rt60 <= 0:
        raise FeatureError(f"rt60 must be positive, got {rt60}")

    length = float(seconds) if seconds else rt60 * 1.5
    n = int(round(length * sr))
    if n < 2:
        raise FeatureError(
            f"an impulse response of {length:g}s at {sr} Hz is {n} samples; "
            f"raise rt60 or pass a longer seconds=")

    rng = np.random.default_rng(seed)
    t = np.arange(n, dtype=np.float64) / sr
    # -60 dB of amplitude at t = rt60, which is what the name means
    envelope = 10.0 ** (-3.0 * t / rt60)
    tail = rng.standard_normal(n) * envelope
    tail[0] = 0.0                       # sample 0 belongs to the direct path

    # Solve for the direct sample that produces the DRR asked for, using the
    # same +/-2.5 ms split the measurement uses. Anything else would give a
    # knob whose number does not come back out.
    half = int(round(_DIRECT_MS / 1000.0 * sr))
    inside = float(np.sum(tail[1:half + 1] ** 2))
    outside = float(np.sum(tail[half + 1:] ** 2))
    wanted = outside * 10.0 ** (drr_db / 10.0)
    if wanted <= inside:
        # The early reflections inside the direct window already exceed the
        # ratio asked for; a room this diffuse cannot be built this way.
        raise FeatureError(
            f"drr_db={drr_db:g} is below what this tail already carries in its "
            f"first {_DIRECT_MS:g} ms "
            f"({10 * np.log10(max(inside, _EPS) / max(outside, _EPS)):.1f} dB). "
            f"Ask for a higher drr_db, or a longer rt60, whose tail puts less "
            f"of its energy inside that window.")
    tail[0] = float(np.sqrt(wanted - inside))
    return (tail / (np.max(np.abs(tail)) + _EPS)).astype(np.float32)

apply_rir

apply_rir(
    y: ndarray,
    ir: ndarray,
    *,
    trim: bool = True,
    normalise: bool = True,
) -> ndarray

Put y in the room described by ir.

Parameters:

Name	Type	Description	Default
trim	bool	return the same number of samples that went in. The convolution is longer by the length of the tail, and a clip that grows every time it is processed stops lining up with everything measured against it.	True
normalise	bool	keep the output's peak where the input's was. A room with a loud direct path has a gain, and an augmentation pipeline that also changes the level is measuring two things at once.	True

Source code in kudio/core/room.py

def apply_rir(y: np.ndarray, ir: np.ndarray, *,
              trim: bool = True, normalise: bool = True) -> np.ndarray:
    """Put *y* in the room described by *ir*.

    :param trim: return the same number of samples that went in. The
        convolution is longer by the length of the tail, and a clip that grows
        every time it is processed stops lining up with everything measured
        against it.
    :param normalise: keep the output's peak where the input's was. A room
        with a loud direct path has a gain, and an augmentation pipeline that
        also changes the level is measuring two things at once.
    """
    y = _mono("apply_rir()", y, "audio")
    ir = _mono("apply_rir()", ir, "impulse response")
    if y.size == 0 or ir.size == 0:
        raise FeatureError("apply_rir() needs audio and an impulse response, "
                           f"got {y.size} and {ir.size} samples")

    from scipy.signal import fftconvolve
    wet = fftconvolve(y, ir)
    if trim:
        wet = wet[:len(y)]
    if normalise:
        peak_in = float(np.max(np.abs(y)))
        peak_out = float(np.max(np.abs(wet)))
        if peak_out > 0 and peak_in > 0:
            wet = wet * (peak_in / peak_out)
    return wet.astype(np.float32)

schroeder_curve

schroeder_curve(ir: ndarray) -> ndarray

Backward-integrated energy decay of ir, in dB, starting at 0.

Schroeder's method: the decay a room would show if you averaged infinitely many noise bursts, obtained from one impulse by integrating its energy from the end backwards. It is what makes a single measurement readable.

Source code in kudio/core/room.py

def schroeder_curve(ir: np.ndarray) -> np.ndarray:
    """Backward-integrated energy decay of *ir*, in dB, starting at 0.

    Schroeder's method: the decay a room would show if you averaged infinitely
    many noise bursts, obtained from one impulse by integrating its energy from
    the end backwards. It is what makes a single measurement readable.
    """
    ir = _mono("schroeder_curve()", ir, "impulse response")
    if ir.size == 0:
        raise FeatureError("schroeder_curve() needs an impulse response")
    energy = np.cumsum((ir ** 2)[::-1])[::-1]
    total = energy[0]
    if total <= 0:
        return np.full(len(ir), -np.inf)
    with np.errstate(divide='ignore'):
        return 10.0 * np.log10(np.maximum(energy / total, _EPS))

rt60

rt60(ir: ndarray, sr: int) -> Reverberation

Measure a room from its impulse response, to ISO 3382-1.

T20, T30 and EDT are read off one backward-integrated decay curve (schroeder_curve); C50, C80 and DRR are energy ratios taken from the response around its peak. The response is measured from its onset, the first sample within 20 dB of the peak, so leading silence -- time of flight, converter latency -- is not counted as decay.

It does not find or remove a noise floor. A recorded response whose tail reaches the floor less than 35 dB down has no honest T30, and Reverberation.reliable cannot tell: check the raw response first.

>>> room = kudio.rt60(ir, 16000)
>>> room.seconds, room.reliable()
(0.6, True)

Source code in kudio/core/room.py

def rt60(ir: np.ndarray, sr: int) -> Reverberation:
    """Measure a room from its impulse response, to ISO 3382-1.

    T20, T30 and EDT are read off one backward-integrated decay curve
    (:func:`schroeder_curve`); C50, C80 and DRR are energy ratios taken from
    the response around its peak. The response is measured from its onset,
    the first sample within 20 dB of the peak, so leading silence -- time of
    flight, converter latency -- is not counted as decay.

    It does not find or remove a noise floor. A recorded response whose tail
    reaches the floor less than 35 dB down has no honest T30, and
    :meth:`Reverberation.reliable` cannot tell: check the raw response first.

    >>> room = kudio.rt60(ir, 16000)
    >>> room.seconds, room.reliable()                        # doctest: +SKIP
    (0.6, True)
    """
    if sr <= 0:
        raise FeatureError(f"rt60() needs a positive rate, got {sr}")
    ir = _mono("rt60()", ir, "impulse response")
    if ir.size < 2:
        raise FeatureError(f"rt60() needs an impulse response, got {ir.size} "
                           f"sample(s)")
    ir = ir[_onset(ir):]

    curve = schroeder_curve(ir)
    t20, _ = _decay_time(curve, sr, _T20_RANGE)
    t30, fit = _decay_time(curve, sr, _T30_RANGE)
    edt, _ = _decay_time(curve, sr, _EDT_RANGE)

    peak = int(np.argmax(np.abs(ir)))
    half = int(round(_DIRECT_MS / 1000.0 * sr))
    lo, hi = max(0, peak - half), min(len(ir), peak + half + 1)
    direct = float(np.sum(ir[lo:hi] ** 2))
    reverberant = float(np.sum(ir ** 2) - direct)
    drr = (10.0 * np.log10(max(direct, _EPS) / reverberant)
           if reverberant > 0 else float('inf'))

    return Reverberation(t20=t20, t30=t30, edt=edt,
                         c50=_clarity(ir, sr, _C50_MS),
                         c80=_clarity(ir, sr, _C80_MS),
                         drr=drr, fit=fit, sr=int(sr))

kudio.effects.channel

What the transmission did to it: companding, quantisation, packet loss.

Speech arrives degraded in three ways. Noise is additive -- kudio.Synthesizer mixes it at a chosen SNR. A room is convolutive -- kudio.apply_rir smears the signal across time. The third is the channel: a codec quantises, a link drops packets, a converter throws bits away. The telephone band is a filter, but the codec and the lost packets cannot be written as y + n or y * h, and they are what this module adds.

>>> phone = kudio.telephone(speech, sr)          # 300-3400 Hz, 8 kHz, G.711
>>> lossy = kudio.dropouts(speech, sr, loss=0.05, seed=0)
>>> rough = kudio.bit_depth(speech, bits=8)

This is the augmentation that matters for anything that will meet a telephone line or a conferencing codec, and it is well established for exactly that (Audio Codec Simulation based Data Augmentation for Telephony Speech Recognition, APSIPA 2019; packet-loss augmentation for contact-centre ASR, Applied Sciences 2022).

Companding is the interesting one. mu_law spends its 8 bits logarithmically, so the quantisation noise follows the signal down instead of sitting at a fixed floor: on this repo's fixture it holds ~37 dB of SNR over a 24 dB range of levels, where linear 8-bit loses the textbook 6 dB per halving. That is the whole reason telephony uses it, and it is what the tests check.

These are models, not bit-exact codecs. mu_law and a_law implement the G.711 companding curves quantised to 8 bits, not the standard's piecewise-linear segment layout. Measured against Python's own audioop -- the reference, and removed in 3.13, which is why it cannot be a dependency -- the round-trip SNR comes out within about 0.9 dB of it over a 24 dB range of levels, and single samples within a few hundredths of full scale. For augmentation that is the same thing; for a bit-exact decoder it is not, and this does not claim to be one.

Functions:

Name	Description
independent_burst	The mean run length that independent loss at rate loss produces.
mu_law	Push y through the G.711 mu-law companding curve and back.
a_law	The same, through the A-law curve -- G.711 as Europe specifies it.
bit_depth	Requantise y to bits, linearly.
dropouts	Drop packets, the way a bad link does.
telephone	Everything a phone line does, in the order it does it.
channel_spec	Normalise spec -- a preset name, a dict, or None -- to a dict.
channel_tag	A short, stable description of spec, for a manifest or a filename.
apply_channel	Run y through a whole link in the order a link does it.

Attributes:

Name	Type	Description
MU_LAW_MU		G.711's mu for the North American / Japanese law.
A_LAW_A		G.711's A for the European law.
PACKET_MS		Default packet length, in milliseconds. 20 ms is what G.711 over RTP uses,
TELEPHONE_BAND		The passband a telephone line gives you, in Hz, and the rate it runs at.
CHANNEL_DEFAULTS		What a channel can be asked for, and what it defaults to. One dictionary,
CHANNEL_PRESETS		The named links, so the common cases are one word.

MU_LAW_MU module-attribute

MU_LAW_MU = 255.0

G.711's mu for the North American / Japanese law.

A_LAW_A module-attribute

A_LAW_A = 87.6

G.711's A for the European law.

PACKET_MS module-attribute

PACKET_MS = 20.0

Default packet length, in milliseconds. 20 ms is what G.711 over RTP uses, so a lost packet is 20 ms of nothing.

TELEPHONE_BAND module-attribute

TELEPHONE_BAND = (300.0, 3400.0)

The passband a telephone line gives you, in Hz, and the rate it runs at.

TELEPHONE_SR module-attribute

TELEPHONE_SR = 8000

The rate a telephone line runs at, in Hz.

CONCEALMENT module-attribute

CONCEALMENT = ('silence', 'hold')

How a lost packet is filled in.

CHANNEL_DEFAULTS module-attribute

CHANNEL_DEFAULTS = {
    "band": False,
    "codec": "none",
    "bits": None,
    "loss": 0.0,
    "burst": None,
    "packet_ms": PACKET_MS,
    "conceal": "silence",
}

What a channel can be asked for, and what it defaults to. One dictionary, so a corpus builder, the command line and a GUI all describe a link the same way instead of each inventing an argument order.

CHANNEL_PRESETS module-attribute

CHANNEL_PRESETS = {
    "none": {},
    "telephone": {"band": True, "codec": "mu_law"},
    "telephone_a": {"band": True, "codec": "a_law"},
    "mu_law": {"codec": "mu_law"},
    "a_law": {"codec": "a_law"},
    "voip": {
        "codec": "mu_law",
        "loss": 0.05,
        "burst": 4.0,
        "conceal": "hold",
    },
}

The named links, so the common cases are one word.

independent_burst

independent_burst(loss: float) -> float

The mean run length that independent loss at rate loss produces.

Worth a function because the obvious guess is 1 and it is wrong. Independent draws do land side by side sometimes, so the runs already average 1 / (1 - loss) -- 1.05 packets at 5% loss, 1.25 at 20%. Asking dropouts for burst=1 is therefore asking for something less clustered than the default, not for the default.

>>> kudio.independent_burst(0.05)
1.0526315789473684

Source code in kudio/effects/channel.py

def independent_burst(loss: float) -> float:
    """The mean run length that *independent* loss at rate *loss* produces.

    Worth a function because the obvious guess is 1 and it is wrong.
    Independent draws do land side by side sometimes, so the runs already
    average ``1 / (1 - loss)`` -- 1.05 packets at 5% loss, 1.25 at 20%.
    Asking :func:`dropouts` for ``burst=1`` is therefore asking for something
    *less* clustered than the default, not for the default.

    >>> kudio.independent_burst(0.05)
    1.0526315789473684
    """
    if not 0.0 <= loss < 1.0:
        raise FeatureError(f"loss must be a fraction in [0, 1), got {loss}")
    return 1.0 / (1.0 - loss)

mu_law

mu_law(y: ndarray, *, bits: int = 8) -> ndarray

Push y through the G.711 mu-law companding curve and back.

Parameters:

Name	Type	Description	Default
bits	int	code word size. 8 is G.711; fewer is a cruder link and a way to ask how far a model's robustness reaches. The point of companding is that the quantisation noise tracks the signal: loud passages get coarse steps and quiet ones get fine steps, so the signal-to-noise ratio stays roughly constant instead of collapsing on anything quiet. bit_depth is the same number of bits spent linearly, and is the comparison that shows what it buys. >>> kudio.mu_law(speech)	8

Source code in kudio/effects/channel.py

def mu_law(y: np.ndarray, *, bits: int = 8) -> np.ndarray:
    """Push *y* through the G.711 mu-law companding curve and back.

    :param bits: code word size. 8 is G.711; fewer is a cruder link and a way
        to ask how far a model's robustness reaches.

    The point of companding is that the quantisation noise tracks the signal:
    loud passages get coarse steps and quiet ones get fine steps, so the
    signal-to-noise ratio stays roughly constant instead of collapsing on
    anything quiet. :func:`bit_depth` is the same number of bits spent
    linearly, and is the comparison that shows what it buys.

    >>> kudio.mu_law(speech)                                 # doctest: +SKIP
    """
    y = _mono("mu_law()", y)
    if bits < 2:
        raise FeatureError(f"bits must be >= 2, got {bits}")
    y = np.clip(y, -1.0, 1.0)

    compressed = np.sign(y) * np.log1p(MU_LAW_MU * np.abs(y)) \
        / np.log1p(MU_LAW_MU)
    coded = _quantise(compressed, bits)
    expanded = np.sign(coded) * ((1.0 + MU_LAW_MU) ** np.abs(coded) - 1.0) \
        / MU_LAW_MU
    return expanded.astype(np.float32)

a_law

a_law(y: ndarray, *, bits: int = 8) -> ndarray

The same, through the A-law curve -- G.711 as Europe specifies it.

A-law is slightly finer near silence and slightly coarser at the top than mu-law. The difference is audible only on very quiet material; it is here because a corpus recorded off a European line went through this one.

Source code in kudio/effects/channel.py

def a_law(y: np.ndarray, *, bits: int = 8) -> np.ndarray:
    """The same, through the A-law curve -- G.711 as Europe specifies it.

    A-law is slightly finer near silence and slightly coarser at the top than
    mu-law. The difference is audible only on very quiet material; it is here
    because a corpus recorded off a European line went through this one.
    """
    y = _mono("a_law()", y)
    if bits < 2:
        raise FeatureError(f"bits must be >= 2, got {bits}")
    y = np.clip(y, -1.0, 1.0)

    magnitude = np.abs(y)
    denominator = 1.0 + np.log(A_LAW_A)
    low = magnitude < 1.0 / A_LAW_A
    compressed = np.where(
        low,
        A_LAW_A * magnitude / denominator,
        (1.0 + np.log(np.maximum(A_LAW_A * magnitude, 1e-300))) / denominator)
    coded = _quantise(np.sign(y) * compressed, bits)

    size = np.abs(coded)
    low = size < 1.0 / denominator
    expanded = np.where(
        low,
        size * denominator / A_LAW_A,
        np.exp(size * denominator - 1.0) / A_LAW_A)
    return (np.sign(coded) * expanded).astype(np.float32)

bit_depth

bit_depth(
    y: ndarray,
    bits: int = 8,
    *,
    dither: bool = False,
    seed: Optional[int] = None,
) -> ndarray

Requantise y to bits, linearly.

Parameters:

Name	Type	Description	Default
dither	bool	add triangular (TPDF) dither two steps wide, peak to peak, before rounding. It raises the noise floor and makes the error's mean and power independent of the signal, which is what stops quiet passages sounding granular rather than merely noisy. Pass seed with it, or every call dithers differently. A full-scale sine comes back at about 6.02 * bits + 1.76 dB of SNR, which is the textbook figure and what the tests assert.	False

Source code in kudio/effects/channel.py

def bit_depth(y: np.ndarray, bits: int = 8, *,
              dither: bool = False, seed: Optional[int] = None) -> np.ndarray:
    """Requantise *y* to *bits*, linearly.

    :param dither: add triangular (TPDF) dither two steps wide, peak to
        peak, before rounding. It raises the noise floor and makes the error's
        mean and power independent of the signal, which is what stops quiet
        passages sounding granular rather than merely noisy. Pass *seed* with
        it, or every call dithers differently.

    A full-scale sine comes back at about ``6.02 * bits + 1.76`` dB of SNR,
    which is the textbook figure and what the tests assert.
    """
    y = _mono("bit_depth()", y)
    if bits < 2:
        raise FeatureError(f"bits must be >= 2, got {bits}")
    y = np.clip(y, -1.0, 1.0)

    if dither:
        # the grid's own step: _quantise has 2**(bits-1) - 1 levels each side
        # of zero, and 2 / 2**bits is half of that at 2 bits
        step = 1.0 / (2 ** (bits - 1) - 1)
        rng = np.random.default_rng(seed)
        # triangular: the difference of two uniforms of one step each, two
        # steps peak to peak -- the width that takes the signal out of the
        # error's mean and power; one step only loosened the tie
        y = y + (rng.random(len(y)) - rng.random(len(y))) * step
        y = np.clip(y, -1.0, 1.0)
    return _quantise(y, bits).astype(np.float32)

dropouts

dropouts(
    y: ndarray,
    sr: int,
    *,
    loss: float = 0.05,
    packet_ms: float = PACKET_MS,
    conceal: str = "silence",
    burst: Optional[float] = None,
    seed: Optional[int] = None,
    return_spans: bool = False,
) -> Union[
    ndarray, Tuple[ndarray, List[Tuple[float, float]]]
]

Drop packets, the way a bad link does.

Parameters:

Name	Type	Description	Default
loss	float	fraction of packets to lose. Achieved exactly -- round(loss * packets) of them go, chosen without replacement -- so the number asked for is the number a corpus actually carries.	0.05
packet_ms	float	packet length. 20 ms is G.711 over RTP.	PACKET_MS
conceal	str	'silence' leaves a hole; 'hold' repeats the last packet that arrived, which is the crudest concealment a decoder does and sounds very different from a hole.	'silence'
burst	Optional[float]	mean run length, in packets. None (the default) loses them independently. Real links lose them in runs -- a fade, a queue overflowing, a handover -- and a run is far harder to conceal than the same number of packets scattered about, because there is no recent speech left to repeat.	None
seed	Optional[int]	which packets go, reproducibly.	None
return_spans	bool	also return the lost packets as (start, end) seconds, one per packet, so an experiment can say where the damage was. Packets in a burst come back as neighbouring spans rather than one merged span; merging is a line of downstream code and unmerging is not. burst=1 is not the default in disguise. Independent loss already averages independent_burst -- 1 / (1 - loss) -- because independent draws sometimes land side by side, so burst=1 asks for runs strictly of one and is less clustered than leaving it alone. >>> lossy = kudio.dropouts(speech, sr, loss=0.05, seed=0) >>> bursty = kudio.dropouts(speech, sr, loss=0.05, burst=6, seed=0) >>> lossy, gaps = kudio.dropouts(speech, sr, loss=0.05, seed=0, ... return_spans=True)	False

Source code in kudio/effects/channel.py

def dropouts(y: np.ndarray, sr: int, *, loss: float = 0.05,
             packet_ms: float = PACKET_MS, conceal: str = 'silence',
             burst: Optional[float] = None, seed: Optional[int] = None,
             return_spans: bool = False
             ) -> Union[np.ndarray, Tuple[np.ndarray, List[Tuple[float, float]]]]:
    """Drop packets, the way a bad link does.

    :param loss: fraction of packets to lose. Achieved exactly -- ``round(loss
        * packets)`` of them go, chosen without replacement -- so the number
        asked for is the number a corpus actually carries.
    :param packet_ms: packet length. 20 ms is G.711 over RTP.
    :param conceal: ``'silence'`` leaves a hole; ``'hold'`` repeats the last
        packet that arrived, which is the crudest concealment a decoder does
        and sounds very different from a hole.
    :param burst: mean run length, in packets. ``None`` (the default) loses
        them independently. Real links lose them in runs -- a fade, a queue
        overflowing, a handover -- and a run is far harder to conceal than
        the same number of packets scattered about, because there is no
        recent speech left to repeat.
    :param seed: which packets go, reproducibly.
    :param return_spans: also return the lost packets as ``(start, end)``
        seconds, one per packet, so an experiment can say where the damage
        was. Packets in a burst come back as neighbouring spans rather than
        one merged span; merging is a line of downstream code and unmerging
        is not.

    ``burst=1`` is not the default in disguise. Independent loss already
    averages :func:`independent_burst` -- ``1 / (1 - loss)`` -- because
    independent draws sometimes land side by side, so ``burst=1`` asks for
    runs strictly of one and is *less* clustered than leaving it alone.

    >>> lossy = kudio.dropouts(speech, sr, loss=0.05, seed=0)
    >>> bursty = kudio.dropouts(speech, sr, loss=0.05, burst=6, seed=0)
    >>> lossy, gaps = kudio.dropouts(speech, sr, loss=0.05, seed=0,
    ...                              return_spans=True)         # doctest: +SKIP
    """
    y = _mono("dropouts()", y)
    if sr <= 0:
        raise FeatureError(f"dropouts() needs a positive rate, got {sr}")
    if not 0.0 <= loss <= 1.0:
        raise FeatureError(f"loss must be a fraction in [0, 1], got {loss}")
    if packet_ms <= 0:
        raise FeatureError(f"packet_ms must be positive, got {packet_ms}")
    if conceal not in CONCEALMENT:
        raise FeatureError(
            f"unknown concealment {conceal!r}; choose from "
            f"{', '.join(CONCEALMENT)}")
    if burst is not None and burst < 1.0:
        raise FeatureError(
            f"burst is a run length in packets and cannot be under 1, got "
            f"{burst}. Independent loss is burst=None, and already runs to "
            f"{independent_burst(min(loss, 0.999)):.2f} packets at this rate.")

    size = max(1, int(round(packet_ms / 1000.0 * sr)))
    packets = int(np.ceil(len(y) / size))
    out = y.astype(np.float32, copy=True)
    spans: List[Tuple[float, float]] = []
    if packets == 0 or loss == 0.0:
        return (out, spans) if return_spans else out

    rng = np.random.default_rng(seed)
    n_lost = min(int(round(loss * packets)), packets)
    if n_lost == 0:
        return (out, spans) if return_spans else out
    if burst is None:
        lost = np.sort(rng.choice(packets, size=n_lost, replace=False))
    else:
        lost = _burst_runs(packets, n_lost, float(burst), rng)

    for index in lost:
        start, stop = index * size, min((index + 1) * size, len(y))
        if conceal == 'hold' and start > 0:
            # repeat the previous packet, tiled if this one runs short
            previous = out[max(0, start - size):start]
            if previous.size:
                out[start:stop] = np.resize(previous, stop - start)
            else:
                out[start:stop] = 0.0
        else:
            out[start:stop] = 0.0
        spans.append((start / sr, stop / sr))

    return (out, spans) if return_spans else out

telephone

telephone(
    y: ndarray,
    sr: int,
    *,
    codec: str = "mu_law",
    band: Tuple[float, float] = TELEPHONE_BAND,
) -> ndarray

Everything a phone line does, in the order it does it.

Band-limit to 300-3400 Hz, run at 8 kHz, compand to 8 bits, come back. The result is returned at the rate it went in and the same length, so it drops straight into a dataset beside its own source.

Parameters:

Name	Type	Description	Default
codec	str	'mu_law', 'a_law' or 'none' for the band and the rate without the companding. The bandwidth is the part that survives being measured: kudio.audio_report reports content up to about 3.8 kHz afterwards -- the 3.4 kHz band edge plus the filter's roll-off -- and flags the file as band-limited, which is exactly what it is. >>> kudio.telephone(speech, 16000)	'mu_law'

Source code in kudio/effects/channel.py

def telephone(y: np.ndarray, sr: int, *, codec: str = 'mu_law',
              band: Tuple[float, float] = TELEPHONE_BAND) -> np.ndarray:
    """Everything a phone line does, in the order it does it.

    Band-limit to 300-3400 Hz, run at 8 kHz, compand to 8 bits, come back.
    The result is returned at the rate it went in and the same length, so it
    drops straight into a dataset beside its own source.

    :param codec: ``'mu_law'``, ``'a_law'`` or ``'none'`` for the band and the
        rate without the companding.

    The bandwidth is the part that survives being measured:
    ``kudio.audio_report`` reports content up to about 3.8 kHz afterwards --
    the 3.4 kHz band edge plus the filter's roll-off -- and flags the file as
    band-limited, which is exactly what it is.

    >>> kudio.telephone(speech, 16000)                       # doctest: +SKIP
    """
    y = _mono("telephone()", y)
    if sr <= 0:
        raise FeatureError(f"telephone() needs a positive rate, got {sr}")
    if codec not in ('mu_law', 'a_law', 'none'):
        raise FeatureError(
            f"unknown codec {codec!r}; choose mu_law, a_law or none")
    if y.size == 0:
        return y.astype(np.float32)

    from kudio.core.io import resample
    from kudio.effects.filters import bandpass

    low, high = band
    # Band-limit at the original rate first: the anti-alias filter of the
    # resampler is not the telephone's passband, and doing it afterwards would
    # measure the resampler rather than the line.
    narrow = bandpass(y.astype(np.float32), sr, low=low,
                      high=min(high, sr / 2 * 0.99))
    if sr != TELEPHONE_SR:
        narrow = resample(narrow, sr, TELEPHONE_SR)

    if codec == 'mu_law':
        narrow = mu_law(narrow)
    elif codec == 'a_law':
        narrow = a_law(narrow)

    if sr != TELEPHONE_SR:
        narrow = resample(narrow, TELEPHONE_SR, sr)
    # resampling twice rarely lands on the same sample count
    out = np.zeros(len(y), dtype=np.float32)
    keep = min(len(out), len(narrow))
    out[:keep] = narrow[:keep]
    return out

channel_spec

channel_spec(spec: Optional[Union[str, dict]]) -> dict

Normalise spec -- a preset name, a dict, or None -- to a dict.

Parameters:

Name	Type	Description	Default
spec	Optional[Union[str, dict]]	None or 'none' for no channel; one of CHANNEL_PRESETS ('telephone', 'voip', 'mu_law', ...); or a dict of any of CHANNEL_DEFAULTS. A preset with overrides is a dict with a 'preset' key. Returns {} for "no channel", which is falsy, so callers can write if spec: and mean it. >>> kudio.channel_spec('telephone')['codec'] 'mu_law' >>> kudio.channel_spec({'preset': 'voip', 'loss': 0.1})['loss'] 0.1	required

Source code in kudio/effects/channel.py

def channel_spec(spec: Optional[Union[str, dict]]) -> dict:
    """Normalise *spec* -- a preset name, a dict, or ``None`` -- to a dict.

    :param spec: ``None`` or ``'none'`` for no channel; one of
        :data:`CHANNEL_PRESETS` (``'telephone'``, ``'voip'``, ``'mu_law'``,
        ...); or a dict of any of :data:`CHANNEL_DEFAULTS`. A preset with
        overrides is a dict with a ``'preset'`` key.

    Returns ``{}`` for "no channel", which is falsy, so callers can write
    ``if spec:`` and mean it.

    >>> kudio.channel_spec('telephone')['codec']
    'mu_law'
    >>> kudio.channel_spec({'preset': 'voip', 'loss': 0.1})['loss']
    0.1
    """
    if spec is None:
        return {}
    if isinstance(spec, str):
        if spec not in CHANNEL_PRESETS:
            raise FeatureError(
                f"unknown channel {spec!r}; choose from "
                f"{', '.join(sorted(CHANNEL_PRESETS))}, or pass a dict")
        spec = dict(CHANNEL_PRESETS[spec])
    elif isinstance(spec, dict):
        spec = dict(spec)
        preset = spec.pop('preset', None)
        if preset is not None:
            base = channel_spec(preset)
            base.update(spec)
            spec = base
    else:
        raise FeatureError(
            f"channel must be a preset name, a dict or None, got "
            f"{type(spec).__name__}")

    unknown = set(spec) - set(CHANNEL_DEFAULTS)
    if unknown:
        raise FeatureError(
            f"unknown channel setting(s) {', '.join(sorted(unknown))}; "
            f"known: {', '.join(sorted(CHANNEL_DEFAULTS))}")

    out = dict(CHANNEL_DEFAULTS)
    out.update(spec)
    if out['codec'] not in ('mu_law', 'a_law', 'none'):
        raise FeatureError(
            f"unknown codec {out['codec']!r}; choose mu_law, a_law or none")
    if out['conceal'] not in CONCEALMENT:
        raise FeatureError(
            f"unknown concealment {out['conceal']!r}; choose from "
            f"{', '.join(CONCEALMENT)}")
    if not 0.0 <= float(out['loss']) <= 1.0:
        raise FeatureError(
            f"loss must be a fraction in [0, 1], got {out['loss']}")
    if out['bits'] is not None and int(out['bits']) < 2:
        raise FeatureError(f"bits must be >= 2, got {out['bits']}")
    # checked here rather than on the first file, so Synthesizer(channel=...)
    # refuses a bad link when it is built
    if out['burst'] is not None and float(out['burst']) < 1:
        raise FeatureError(f"burst is a mean run length in packets, at least 1, "
                           f"got {out['burst']}")
    if not float(out['packet_ms']) > 0:
        raise FeatureError(f"packet_ms must be > 0, got {out['packet_ms']}")

    # Nothing asked for is not a channel. Saying so here means one falsy
    # check downstream instead of five truthy ones.
    if (not out['band'] and out['codec'] == 'none'
            and out['bits'] is None and not float(out['loss'])):
        return {}
    return out

channel_tag

channel_tag(spec: Optional[Union[str, dict]]) -> str

A short, stable description of spec, for a manifest or a filename.

Readable rather than exhaustive: it is what went on the corpus label, and a reader should be able to tell two runs apart at a glance.

>>> kudio.channel_tag('telephone')
'telephone'
>>> kudio.channel_tag({'codec': 'mu_law', 'loss': 0.05, 'burst': 4})
'mu_law+loss5%x4'

Source code in kudio/effects/channel.py

def channel_tag(spec: Optional[Union[str, dict]]) -> str:
    """A short, stable description of *spec*, for a manifest or a filename.

    Readable rather than exhaustive: it is what went on the corpus label, and
    a reader should be able to tell two runs apart at a glance.

    >>> kudio.channel_tag('telephone')
    'telephone'
    >>> kudio.channel_tag({'codec': 'mu_law', 'loss': 0.05, 'burst': 4})
    'mu_law+loss5%x4'
    """
    spec = channel_spec(spec)
    if not spec:
        return ''
    parts = []
    if spec['band']:
        # named as the presets are: 'banda_law' read as one made-up word
        parts.append({'mu_law': 'telephone', 'a_law': 'telephone_a'}.get(
            spec['codec'], 'band'))
    elif spec['codec'] != 'none':
        parts.append(spec['codec'])
    if spec['bits'] is not None:
        parts.append(f"{int(spec['bits'])}bit")
    loss = float(spec['loss'])
    if loss:
        tag = f"loss{loss * 100:g}%"
        if spec['burst']:
            tag += f"x{float(spec['burst']):g}"
        if spec['conceal'] != 'silence':
            tag += f"-{spec['conceal']}"
        parts.append(tag)
    return '+'.join(parts)

apply_channel

apply_channel(
    y: ndarray,
    sr: int,
    spec: Optional[Union[str, dict]],
    seed: Optional[int] = None,
) -> ndarray

Run y through a whole link in the order a link does it.

Band and rate first, then the codec, then the converter, then the wire -- packets are lost from the encoded signal, not from the microphone, so the loss goes last. Same length and same rate as it went in.

>>> kudio.apply_channel(mix, sr, 'telephone')
>>> kudio.apply_channel(mix, sr, {'preset': 'voip', 'loss': 0.1}, seed=7)
...

Source code in kudio/effects/channel.py

def apply_channel(y: np.ndarray, sr: int, spec: Optional[Union[str, dict]],
                  seed: Optional[int] = None) -> np.ndarray:
    """Run *y* through a whole link in the order a link does it.

    Band and rate first, then the codec, then the converter, then the wire --
    packets are lost from the encoded signal, not from the microphone, so the
    loss goes last. Same length and same rate as it went in.

    >>> kudio.apply_channel(mix, sr, 'telephone')          # doctest: +SKIP
    >>> kudio.apply_channel(mix, sr, {'preset': 'voip', 'loss': 0.1}, seed=7)
    ...                                                    # doctest: +SKIP
    """
    spec = channel_spec(spec)
    out = _mono("apply_channel()", y).astype(np.float32)
    if not spec:
        return out
    if spec['band']:
        out = telephone(out, sr, codec=spec['codec'])
    elif spec['codec'] == 'mu_law':
        out = mu_law(out)
    elif spec['codec'] == 'a_law':
        out = a_law(out)

    if spec['bits'] is not None:
        out = bit_depth(out, int(spec['bits']))
    if float(spec['loss']):
        out = dropouts(out, sr, loss=float(spec['loss']),
                       packet_ms=float(spec['packet_ms']),
                       conceal=spec['conceal'], burst=spec['burst'],
                       seed=seed)
    return out