STFT and features

kudio.core.stft

STFT geometry as a value you can pass around and save.

A spectrogram and the waveform reconstructed from it must be computed with the same n_fft / hop_length / win_length / window. Passing those four to two separate functions and keeping them in sync by hand is how a producer and a consumer end up quietly disagreeing — the geometry belongs with the data that was made from it.

>>> stft = kudio.STFT(sr=16000, n_fft=512, hop_length=256)
>>> spec = stft.forward(y)                 # (frames, bins)
>>> back = stft.inverse(y, spec)           # phase taken from y
>>> stft.to_dict()                         # store next to a model checkpoint
{'sr': 16000, 'n_fft': 512, 'hop_length': 256, 'win_length': 512, 'window': 'hamming'}

A thin wrapper: waveform_to_spectrogram and spectrogram_to_waveform do the work.

Classes:

Name	Description
STFT	Immutable STFT settings, with the two transforms attached.

STFT dataclass

STFT(
    sr: int = 16000,
    n_fft: int = 512,
    hop_length: int = 256,
    win_length: Optional[int] = None,
    window: str = "hamming",
)

Immutable STFT settings, with the two transforms attached.

Parameters:

Name	Type	Description	Default
sr	int	sample rate the settings were chosen for. Carried so a stored geometry can be checked against the audio it is applied to; the transforms themselves do not resample.	16000
win_length	Optional[int]	defaults to n_fft.	None

Methods:

Name	Description
n_frames	Frames a waveform of n_samples will produce (centred STFT).
frame_times	Centre time, in seconds, of each frame.
forward	Waveform -> log-power spectrogram (frames, bins).
inverse	Log-power spectrogram -> waveform, reusing the phase of y_ref.
analyse	Waveform -> complex spectrogram (frames, bins).
synthesise	Complex spectrogram -> waveform, the inverse of analyse.
to_dict	Plain dict, ready for JSON/YAML next to a checkpoint.
from_dict	Rebuild the geometry to_dict stored.
matches	Is sr the rate this geometry was configured for?

Attributes:

Name	Type	Description
n_bins	int	Frequency bins produced by this geometry.

n_bins property

n_bins: int

Frequency bins produced by this geometry.

n_frames

n_frames(n_samples: int) -> int

Frames a waveform of n_samples will produce (centred STFT).

librosa pads n_fft // 2 on each side, so an odd n_fft gives one frame fewer than n // hop + 1 whenever n is a multiple of the hop.

Source code in kudio/core/stft.py

def n_frames(self, n_samples: int) -> int:
    """Frames a waveform of *n_samples* will produce (centred STFT).

    librosa pads ``n_fft // 2`` on each side, so an odd *n_fft* gives one
    frame fewer than ``n // hop + 1`` whenever *n* is a multiple of the hop.
    """
    padded = int(n_samples) + 2 * (self.n_fft // 2)
    return 1 + (padded - self.n_fft) // self.hop_length

frame_times

frame_times(n_frames: int) -> ndarray

Centre time, in seconds, of each frame.

Source code in kudio/core/stft.py

def frame_times(self, n_frames: int) -> np.ndarray:
    """Centre time, in seconds, of each frame."""
    return np.arange(n_frames) * self.hop_length / self.sr

forward

forward(
    y: ndarray,
    *,
    norm: bool = False,
    sequence: bool = False,
    context: int = 0,
    desired_length: Optional[int] = None,
) -> ndarray

Waveform -> log-power spectrogram (frames, bins).

Source code in kudio/core/stft.py

def forward(self, y: np.ndarray, *, norm: bool = False,
            sequence: bool = False, context: int = 0,
            desired_length: Optional[int] = None) -> np.ndarray:
    """Waveform -> log-power spectrogram ``(frames, bins)``."""
    from kudio.core.feature import waveform_to_spectrogram
    return waveform_to_spectrogram(
        y, sequence=sequence, forward_backward=context, norm=norm,
        hop_length=self.hop_length, n_fft=self.n_fft,
        win_length=self.win_length, window=self.window,
        desired_length=desired_length)

inverse

inverse(
    y_ref: ndarray, spec: ndarray, *, sequence: bool = False
) -> ndarray

Log-power spectrogram -> waveform, reusing the phase of y_ref.

Source code in kudio/core/stft.py

def inverse(self, y_ref: np.ndarray, spec: np.ndarray, *,
            sequence: bool = False) -> np.ndarray:
    """Log-power spectrogram -> waveform, reusing the phase of *y_ref*."""
    from kudio.core.feature import spectrogram_to_waveform
    return spectrogram_to_waveform(
        y_ref, spec, sequence=sequence, hop_length=self.hop_length,
        n_fft=self.n_fft, win_length=self.win_length, window=self.window)

analyse

analyse(y: ndarray) -> ndarray

Waveform -> complex spectrogram (frames, bins).

forward throws the phase away, which is right for a feature and wrong for anything that has to reconstruct the signal. Enhancement multiplies the complex spectrum by a real gain and inverts it, so it needs the phase kept — see kudio.spectral_enhance.

Source code in kudio/core/stft.py

def analyse(self, y: np.ndarray) -> np.ndarray:
    """Waveform -> **complex** spectrogram ``(frames, bins)``.

    :meth:`forward` throws the phase away, which is right for a feature and
    wrong for anything that has to reconstruct the signal. Enhancement
    multiplies the complex spectrum by a real gain and inverts it, so it
    needs the phase kept — see :func:`kudio.spectral_enhance`.
    """
    import librosa
    from kudio.core.feature import _mono
    spec = librosa.stft(np.asarray(_mono(y, "STFT.analyse()"), dtype=np.float32),
                        n_fft=self.n_fft, hop_length=self.hop_length,
                        win_length=self.win_length, window=self.window)
    return spec.T                              # (frames, bins), as kudio speaks

synthesise

synthesise(
    spec: ndarray, length: Optional[int] = None
) -> ndarray

Complex spectrogram -> waveform, the inverse of analyse.

length trims (or pads) to an exact sample count. Pass the input length and the round trip is sample-aligned, which is what an A/B or a training target needs.

Source code in kudio/core/stft.py

def synthesise(self, spec: np.ndarray,
               length: Optional[int] = None) -> np.ndarray:
    """Complex spectrogram -> waveform, the inverse of :meth:`analyse`.

    *length* trims (or pads) to an exact sample count. Pass the input
    length and the round trip is sample-aligned, which is what an A/B or a
    training target needs.
    """
    import librosa
    out = librosa.istft(np.asarray(spec).T, hop_length=self.hop_length,
                        win_length=self.win_length, window=self.window,
                        n_fft=self.n_fft, length=length)
    return np.asarray(out, dtype=np.float32)

to_dict

to_dict() -> Dict[str, Any]

Plain dict, ready for JSON/YAML next to a checkpoint.

Source code in kudio/core/stft.py

def to_dict(self) -> Dict[str, Any]:
    """Plain dict, ready for JSON/YAML next to a checkpoint."""
    return dataclasses.asdict(self)

from_dict classmethod

from_dict(values: Dict[str, Any]) -> 'STFT'

Rebuild the geometry to_dict stored.

Every key has to be there. A missing one taking its default is how a stored geometry and the code reading it end up disagreeing -- a config without hop_length would quietly analyse at 256.

Source code in kudio/core/stft.py

@classmethod
def from_dict(cls, values: Dict[str, Any]) -> "STFT":
    """Rebuild the geometry :meth:`to_dict` stored.

    Every key has to be there. A missing one taking its default is how a
    stored geometry and the code reading it end up disagreeing -- a config
    without ``hop_length`` would quietly analyse at 256.
    """
    known = {f.name for f in dataclasses.fields(cls)}
    unknown = set(values) - known
    if unknown:
        raise ValueError(
            f"unknown STFT key(s): {', '.join(sorted(unknown))}. "
            f"Valid keys: {', '.join(sorted(known))}")
    missing = known - set(values)
    if missing:
        raise ValueError(
            f"STFT key(s) missing: {', '.join(sorted(missing))}. A stored "
            f"geometry has to name all of {', '.join(sorted(known))}")
    return cls(**values)

matches

matches(sr: int) -> bool

Is sr the rate this geometry was configured for?

Source code in kudio/core/stft.py

def matches(self, sr: int) -> bool:
    """Is *sr* the rate this geometry was configured for?"""
    return int(sr) == int(self.sr)

kudio.core.feature

Feature extraction and spectrogram <-> waveform conversions.

v3 canonical names

================================= ============================================= canonical deprecated aliases ================================= ============================================= waveform_to_spectrogram w2s, wavform2spec spectrogram_to_waveform spec2wavform file_to_spectrogram f2s, wav2spec save_spectrogram_as_wave s2w, spec2wav mfcc w2mfcc, wav2mfcc mfcc_from_files concat_mfcc, concat_mfcc_ logspec_from_files concat_logspec, concat_logspec_, contextual_LogSpectrogram melspectrogram wav2mel ================================= =============================================

The aliases still work but emit a DeprecationWarning.

Classes:

Name	Description
ContextFrames	stack_context over a whole corpus, stacked a batch at a time.
Standardizer	Per-column mean/std normalisation, with the statistics saved alongside.

Functions:

Name	Description
wave_separate	Split an interleaved multi-channel waveform into per-channel arrays.
waveform_to_spectrogram	Waveform -> log-power spectrogram (frames, bins), log10(|X|^2).
spectrogram_to_waveform	Log-power spectrogram -> waveform, reusing the phase of reference y.
file_to_spectrogram	File -> log-power spectrogram (see waveform_to_spectrogram).
save_spectrogram_as_wave	Reconstruct a waveform from enhanced_spec (phase from noisy_file)
mfcc	Waveform -> MFCC, shaped (1, n_frames, n_mfcc). y is mono.
mfcc_from_files	Load each wave file and stack their MFCC matrices (n_files, frames, n_mfcc).
logspec_from_files	Convert each file to a log spectrogram and stack them.
melspectrogram	Audio -> stacked log-mel feature vectors (n_vectors, n_mels * n_frames).
denoise_wav2spec	Peak-normalized magnitude spectrogram + phase, for denoising models.
denoise_spec2wav	Inverse of denoise_wav2spec; output peak-normalized.
stack_context	Stack ±context neighbouring frames onto each frame.
frame_windows	Cut a feature matrix into fixed-length windows.
wave_slicing	Slice a waveform into overlapping frames (flattened if is_reshape).
features2matrix	Concatenate feature matrices; label rows by list index.

ContextFrames

ContextFrames(
    utterances: Union[ndarray, Sequence[ndarray]],
    context: int,
    pad: str = "edge",
)

stack_context over a whole corpus, stacked a batch at a time.

Stacking ±5 frames makes every row eleven rows wide, so a training set stacked up front takes eleven times the memory of its features. This keeps the features once -- each utterance padded at its own edges, so no window reaches into the next utterance -- and stacks only the rows asked for:

```pycon
>>> bank = kudio.ContextFrames(features, context=5)   # (frames, dim) each
>>> bank.shape
(707112, 2827)
>>> x = bank[batch]                # indices, a slice or a mask

```

Row i is row i of np.vstack([stack_context(f, context, pad) for f in features]), so a target built with np.vstack(targets) lines up with it index for index. A single (frames, dim) matrix is one utterance; a 3-D array is read as a stack of them.

Attributes:

Name	Type	Description
shape	Tuple[int, int]	(rows, dim * (2 * context + 1)), as np.vstack would give.

Source code in kudio/core/feature.py

def __init__(self, utterances: Union[np.ndarray, Sequence[np.ndarray]],
             context: int, pad: str = 'edge'):
    if context < 0:
        raise FeatureError(f"context must be >= 0, got {context}")
    if pad not in ('edge', 'zero'):
        raise FeatureError(f"pad must be 'edge' or 'zero', got {pad!r}")
    if isinstance(utterances, np.ndarray) and utterances.ndim == 2:
        utterances = [utterances]
    mats = [np.asarray(u, dtype=np.float32) for u in utterances]
    if not mats:
        raise FeatureError("ContextFrames needs at least one utterance")
    if any(m.ndim != 2 for m in mats) or len({m.shape[1] for m in mats}) != 1:
        raise FeatureError("every utterance must be (frames, dim) with one "
                           f"dim, got {sorted({m.shape for m in mats})}")
    self.context = int(context)
    self.dim = int(mats[0].shape[1])
    mode = 'edge' if pad == 'edge' else 'constant'
    padded, centres, offset = [], [], 0
    for m in mats:
        if not len(m):
            continue
        padded.append(np.pad(m, ((context, context), (0, 0)), mode=mode))
        centres.append(offset + context + np.arange(len(m)))
        offset += len(m) + 2 * context
    self._frames = (np.concatenate(padded) if padded
                    else np.empty((0, self.dim), np.float32))
    self._centres = (np.concatenate(centres) if centres
                     else np.empty(0, dtype=np.intp))
    self._window = np.arange(-context, context + 1)

shape property

shape: Tuple[int, int]

(rows, dim * (2 * context + 1)), as np.vstack would give.

Standardizer

Standardizer(
    mean: Optional[ndarray] = None,
    std: Optional[ndarray] = None,
)

Per-column mean/std normalisation, with the statistics saved alongside.

Fit on the training features, then reused verbatim at inference — the statistics are as much a part of a trained model as its weights, and a model loaded without them produces confident nonsense.

>>> std = kudio.Standardizer().fit(train_frames)
>>> x = std.transform(frames)
>>> frames_again = std.inverse(x)
>>> std.save('runs/exp1/stats.npz')

Methods:

Name	Description
fit	Learn each column's mean and standard deviation from frames
transform	(frames - mean) / std, column by column, as float32.
inverse	Undo transform — used to map a prediction back.
fit_transform	fit on frames, then transform them.
save	Write the statistics to an .npz file and return its path.
load	Read statistics save wrote. 'stats' finds stats.npz.

Source code in kudio/core/feature.py

def __init__(self, mean: Optional[np.ndarray] = None,
             std: Optional[np.ndarray] = None):
    self.mean = mean
    self.std = std

fit

fit(frames: ndarray) -> 'Standardizer'

Learn each column's mean and standard deviation from frames (rows, dim) -- the training set, never the data it is applied to.

Source code in kudio/core/feature.py

def fit(self, frames: np.ndarray) -> "Standardizer":
    """Learn each column's mean and standard deviation from *frames*
    ``(rows, dim)`` -- the training set, never the data it is applied to."""
    frames = np.asarray(frames, dtype=np.float32)
    if frames.ndim != 2:
        raise FeatureError(f"frames must be 2-D, got shape {frames.shape}")
    self.mean = frames.mean(axis=0)
    self.std = frames.std(axis=0) + 1e-8
    return self

transform

transform(frames: ndarray) -> ndarray

(frames - mean) / std, column by column, as float32.

Source code in kudio/core/feature.py

def transform(self, frames: np.ndarray) -> np.ndarray:
    """``(frames - mean) / std``, column by column, as float32."""
    self._check()
    return ((np.asarray(frames, dtype=np.float32) - self.mean)
            / self.std).astype(np.float32)

inverse

inverse(frames: ndarray) -> ndarray

Undo transform — used to map a prediction back.

Source code in kudio/core/feature.py

def inverse(self, frames: np.ndarray) -> np.ndarray:
    """Undo :meth:`transform` — used to map a prediction back."""
    self._check()
    return (np.asarray(frames, dtype=np.float32) * self.std
            + self.mean).astype(np.float32)

fit_transform

fit_transform(frames: ndarray) -> ndarray

fit on frames, then transform them.

Source code in kudio/core/feature.py

def fit_transform(self, frames: np.ndarray) -> np.ndarray:
    """:meth:`fit` on *frames*, then :meth:`transform` them."""
    return self.fit(frames).transform(frames)

save

save(path: PathLike) -> Path

Write the statistics to an .npz file and return its path.

numpy adds .npz to a name without it, so the path returned is the file actually written: save('stats') returns stats.npz.

Source code in kudio/core/feature.py

def save(self, path: PathLike) -> Path:
    """Write the statistics to an ``.npz`` file and return its path.

    numpy adds ``.npz`` to a name without it, so the path returned is the
    file actually written: ``save('stats')`` returns ``stats.npz``.
    """
    self._check()
    path = Path(path)
    if path.suffix != '.npz':
        path = path.parent / (path.name + '.npz')
    path.parent.mkdir(parents=True, exist_ok=True)
    np.savez(str(path), mean=self.mean, std=self.std)
    return path

load classmethod

load(path: PathLike) -> 'Standardizer'

Read statistics save wrote. 'stats' finds stats.npz.

Source code in kudio/core/feature.py

@classmethod
def load(cls, path: PathLike) -> "Standardizer":
    """Read statistics :meth:`save` wrote. ``'stats'`` finds ``stats.npz``."""
    path = Path(path)
    if not path.exists() and path.suffix != '.npz':
        path = path.parent / (path.name + '.npz')
    with np.load(str(path)) as data:
        return cls(mean=data['mean'], std=data['std'])

wave_separate

wave_separate(wave: ndarray, channels: int) -> list

Split an interleaved multi-channel waveform into per-channel arrays.

Source code in kudio/core/feature.py

def wave_separate(wave: np.ndarray, channels: int) -> list:
    """Split an interleaved multi-channel waveform into per-channel arrays."""
    if channels < 1:
        raise FeatureError(f"channels must be >= 1, got {channels}")
    return [wave[c::channels] for c in range(channels)]

waveform_to_spectrogram

waveform_to_spectrogram(
    y: ndarray,
    sequence: bool = False,
    forward_backward: int = 0,
    norm: bool = False,
    hop_length: int = 256,
    n_fft: int = 512,
    win_length: Optional[int] = None,
    window: str = "hamming",
    desired_length: Optional[int] = None,
) -> ndarray

Waveform -> log-power spectrogram (frames, bins), log10(|X|^2).

Parameters:

Name	Type	Description	Default
y	ndarray	a mono waveform, shape (n,).	required
sequence	bool	return (1, frames, dim) instead, for a sequence model.	False
forward_backward	int	stack this many context frames on each side, zero-padded (stack_context does the same for any feature).	0
norm	bool	standardise each frequency bin over this clip.	False
hop_length	int	samples between frames.	256
n_fft	int	FFT size; there are n_fft // 2 + 1 bins.	512
win_length	Optional[int]	window length, defaulting to n_fft.	None
window	str	any window name librosa takes.	'hamming'
desired_length	Optional[int]	cut the waveform to this many samples first.	None

Raises:

Type	Description
FeatureError	for anything but a 1-D waveform. A (frames, channels) array would be read as one short frame per sample, so pick a channel or mix down first.

Source code in kudio/core/feature.py

def waveform_to_spectrogram(y: np.ndarray,
                            sequence: bool = False,
                            forward_backward: int = 0,
                            norm: bool = False,
                            hop_length: int = 256,
                            n_fft: int = 512,
                            win_length: Optional[int] = None,
                            window: str = 'hamming',
                            desired_length: Optional[int] = None) -> np.ndarray:
    """Waveform -> log-power spectrogram ``(frames, bins)``, ``log10(|X|^2)``.

    :param y: a mono waveform, shape ``(n,)``.
    :param sequence: return ``(1, frames, dim)`` instead, for a sequence model.
    :param forward_backward: stack this many context frames on each side,
        zero-padded (:func:`stack_context` does the same for any feature).
    :param norm: standardise each frequency bin over this clip.
    :param hop_length: samples between frames.
    :param n_fft: FFT size; there are ``n_fft // 2 + 1`` bins.
    :param win_length: window length, defaulting to *n_fft*.
    :param window: any window name librosa takes.
    :param desired_length: cut the waveform to this many samples first.
    :raises FeatureError: for anything but a 1-D waveform. A
        ``(frames, channels)`` array would be read as one short frame per
        sample, so pick a channel or mix down first.
    """
    y = _mono(y, "waveform_to_spectrogram()")
    if desired_length and desired_length < len(y):
        y = y[:desired_length]
    D = librosa.stft(y, n_fft=n_fft, hop_length=hop_length,
                     win_length=win_length, window=window)
    D = D + _EPS
    Sxx = np.log10(np.abs(D) ** 2)

    if norm:
        # statistics in float64 -- a steady bin has a tiny std, and dividing
        # by it in float32 leaves the normalised mean visibly off zero --
        # then stored at the precision the audio came in at
        S64 = Sxx.astype(np.float64)
        Sxx_mean = np.mean(S64, axis=1, keepdims=True)
        Sxx_std = np.std(S64, axis=1, keepdims=True) + 1e-12
        Sxx_r = ((S64 - Sxx_mean) / Sxx_std).astype(Sxx.dtype)
    else:
        Sxx_r = np.array(Sxx)

    if forward_backward:
        stacked = stack_context(Sxx_r.T, forward_backward, pad='zero')
        if sequence:
            return stacked.reshape(1, *stacked.shape)
        return stacked

    Sxx_r = np.array(Sxx_r).T
    if sequence:
        return Sxx_r.reshape(1, *Sxx_r.shape)
    return Sxx_r

spectrogram_to_waveform

spectrogram_to_waveform(
    y: ndarray,
    enhanced_spec: ndarray,
    sequence: bool = False,
    hop_length: int = 256,
    n_fft: int = 512,
    win_length: Optional[int] = None,
    window: str = "hamming",
) -> ndarray

Log-power spectrogram -> waveform, reusing the phase of reference y.

enhanced_spec is (frames, bins) with one row per frame of y, the shape waveform_to_spectrogram gives for it. The result has exactly len(y) samples.

Source code in kudio/core/feature.py

def spectrogram_to_waveform(y: np.ndarray, enhanced_spec: np.ndarray,
                            sequence: bool = False, hop_length: int = 256,
                            n_fft: int = 512,
                            win_length: Optional[int] = None,
                            window: str = 'hamming') -> np.ndarray:
    """Log-power spectrogram -> waveform, reusing the phase of reference *y*.

    *enhanced_spec* is ``(frames, bins)`` with one row per frame of *y*, the
    shape :func:`waveform_to_spectrogram` gives for it. The result has exactly
    ``len(y)`` samples.
    """
    if sequence:
        enhanced_spec = enhanced_spec.squeeze()
    D = librosa.stft(np.asarray(y, dtype=np.float32), hop_length=hop_length,
                     n_fft=n_fft, win_length=win_length, window=window)
    D = D + _EPS
    enhanced_spec = np.asarray(enhanced_spec)
    if enhanced_spec.shape != D.T.shape:
        raise FeatureError(
            f"the spectrogram is {enhanced_spec.shape} but the reference gives "
            f"{D.T.shape} (frames, bins): it needs one row per frame of the "
            f"reference, with no context stacked and no window padding")
    phase = np.exp(1j * np.angle(D))
    magnitude = np.sqrt(10 ** enhanced_spec).T
    reverse = np.multiply(magnitude, phase)
    # length= has istft rebuild the last partial hop; trimming and padding
    # afterwards repeated one sample over the last len(y) % hop_length
    return librosa.istft(reverse, hop_length=hop_length, win_length=win_length,
                         window=window, n_fft=n_fft, length=len(y))

file_to_spectrogram

file_to_spectrogram(
    file: PathLike,
    sequence: bool = False,
    forward_backward: int = 0,
    norm: bool = False,
    hop_length: int = 256,
    n_fft: int = 512,
    win_length: Optional[int] = None,
    desired_length: Optional[int] = None,
    resample_rate: Optional[int] = None,
    window: str = "hamming",
) -> ndarray

File -> log-power spectrogram (see waveform_to_spectrogram).

window comes last, after the arguments callers already pass by position.

Source code in kudio/core/feature.py

def file_to_spectrogram(file: PathLike,
                        sequence: bool = False,
                        forward_backward: int = 0,
                        norm: bool = False,
                        hop_length: int = 256,
                        n_fft: int = 512,
                        win_length: Optional[int] = None,
                        desired_length: Optional[int] = None,
                        resample_rate: Optional[int] = None,
                        window: str = 'hamming') -> np.ndarray:
    """File -> log-power spectrogram (see :func:`waveform_to_spectrogram`).

    *window* comes last, after the arguments callers already pass by
    position.
    """
    y, _ = file_load(file, sr=resample_rate, mono=True)
    return waveform_to_spectrogram(y, sequence, forward_backward, norm,
                                   hop_length, n_fft, win_length, window,
                                   desired_length=desired_length)

save_spectrogram_as_wave

save_spectrogram_as_wave(
    wave_out_dir: PathLike,
    noisy_file: PathLike,
    enhanced_spec: ndarray,
    squeeze: bool = False,
    hop_length: int = 256,
    n_fft: int = 512,
    win_length: Optional[int] = None,
    window: str = "hamming",
) -> None

Reconstruct a waveform from enhanced_spec (phase from noisy_file) and write it to wave_out_dir as 16-bit PCM, clipped at full scale.

Source code in kudio/core/feature.py

def save_spectrogram_as_wave(wave_out_dir: PathLike, noisy_file: PathLike,
                             enhanced_spec: np.ndarray,
                             squeeze: bool = False, hop_length: int = 256,
                             n_fft: int = 512,
                             win_length: Optional[int] = None,
                             window: str = 'hamming') -> None:
    """Reconstruct a waveform from *enhanced_spec* (phase from *noisy_file*)
    and write it to *wave_out_dir* as 16-bit PCM, clipped at full scale."""
    y, rate = librosa.load(str(noisy_file), sr=None)
    y_out = spectrogram_to_waveform(y, enhanced_spec, squeeze, hop_length,
                                    n_fft, win_length, window)
    save_wave(wave_out_dir, y_out, rate)

mfcc

mfcc(y: ndarray, sr: int, n_mfcc: int = 40) -> ndarray

Waveform -> MFCC, shaped (1, n_frames, n_mfcc). y is mono.

Source code in kudio/core/feature.py

def mfcc(y: np.ndarray, sr: int, n_mfcc: int = 40) -> np.ndarray:
    """Waveform -> MFCC, shaped ``(1, n_frames, n_mfcc)``. *y* is mono."""
    y = _mono(y, "mfcc()")
    return librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc).T.reshape(1, -1, n_mfcc)

mfcc_from_files

mfcc_from_files(
    wav_list: Sequence,
    n_mfcc: int = 40,
    desired_samples: Optional[int] = None,
    desired_length: Optional[int] = None,
) -> ndarray

Load each wave file and stack their MFCC matrices (n_files, frames, n_mfcc).

Parameters:

Name	Type	Description	Default
desired_samples	Optional[int]	cut each file to this many samples.	None
desired_length	Optional[int]	cut each file to this many seconds (the spectrogram functions' desired_length counts samples).	None

Source code in kudio/core/feature.py

def mfcc_from_files(wav_list: Sequence, n_mfcc: int = 40,
                    desired_samples: Optional[int] = None,
                    desired_length: Optional[int] = None) -> np.ndarray:
    """Load each wave file and stack their MFCC matrices ``(n_files, frames, n_mfcc)``.

    :param desired_samples: cut each file to this many samples.
    :param desired_length: cut each file to this many **seconds** (the
        spectrogram functions' ``desired_length`` counts samples).
    """
    mfcc_list = []
    for wav_dir in tqdm(wav_list, desc="[kudio] MFCC"):
        y, sr = file_load(wav_dir)
        if desired_length:
            y = y[:int(round(sr * desired_length))]
        if desired_samples and desired_samples < len(y):
            y = y[:desired_samples]
        m = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=n_mfcc).T
        mfcc_list.append(m.reshape(1, m.shape[0], -1))
    return np.vstack(mfcc_list)

logspec_from_files

logspec_from_files(
    wav_list: Sequence,
    sequence: bool,
    forward_backward: int,
    norm: bool,
    hop_length: int = 256,
    n_fft: int = 512,
    win_length: Optional[int] = None,
    desired_samples: Optional[int] = None,
    window: str = "hamming",
) -> ndarray

Convert each file to a log spectrogram and stack them.

Source code in kudio/core/feature.py

def logspec_from_files(wav_list: Sequence,
                       sequence: bool,
                       forward_backward: int,
                       norm: bool,
                       hop_length: int = 256,
                       n_fft: int = 512,
                       win_length: Optional[int] = None,
                       desired_samples: Optional[int] = None,
                       window: str = 'hamming') -> np.ndarray:
    """Convert each file to a log spectrogram and stack them."""
    return np.vstack([
        file_to_spectrogram(file=file, sequence=sequence,
                            forward_backward=forward_backward, norm=norm,
                            hop_length=hop_length, n_fft=n_fft,
                            win_length=win_length, desired_length=desired_samples,
                            window=window)
        for file in tqdm(wav_list, desc='[kudio] log-spectrograms')
    ])

melspectrogram

melspectrogram(
    source: Union[PathLike, ndarray],
    n_mels: int = 64,
    n_frames: int = 5,
    n_fft: int = 1024,
    hop_length: int = 512,
    power: float = 2.0,
    *,
    sr: Optional[int] = None,
) -> ndarray

Audio -> stacked log-mel feature vectors (n_vectors, n_mels * n_frames).

source is a file path or a waveform; a waveform requires sr, the same way mfcc does. With a path, sr resamples the file to that rate, and leaving it out keeps the file's own.

The window is librosa's Hann, not the Hamming of STFT and waveform_to_spectrogram.

>>> feats = kudio.melspectrogram('clip.wav')
>>> feats = kudio.melspectrogram(y, sr=16000)

Source code in kudio/core/feature.py

def melspectrogram(source: Union[PathLike, np.ndarray], n_mels: int = 64,
                   n_frames: int = 5,
                   n_fft: int = 1024, hop_length: int = 512,
                   power: float = 2.0, *,
                   sr: Optional[int] = None) -> np.ndarray:
    """Audio -> stacked log-mel feature vectors ``(n_vectors, n_mels * n_frames)``.

    *source* is a file path **or** a waveform; a waveform requires *sr*, the
    same way :func:`mfcc` does. With a path, *sr* resamples the file to that
    rate, and leaving it out keeps the file's own.

    The window is librosa's Hann, not the Hamming of :class:`STFT` and
    :func:`waveform_to_spectrogram`.

    >>> feats = kudio.melspectrogram('clip.wav')            # doctest: +SKIP
    >>> feats = kudio.melspectrogram(y, sr=16000)           # doctest: +SKIP
    """
    dims = n_mels * n_frames
    if isinstance(source, (str, os.PathLike)):
        # sr used to be overwritten by the file's own rate without a word
        y, sr = file_load(source, sr=sr, mono=True)
    else:
        if sr is None:
            raise FeatureError("melspectrogram() needs sr= when given a waveform")
        y = np.asarray(_mono(source, "melspectrogram()"), dtype=np.float32)
    mel = librosa.feature.melspectrogram(y=y, sr=sr, n_fft=n_fft,
                                         hop_length=hop_length, n_mels=n_mels,
                                         power=power)
    log_mel = 20.0 / power * np.log10(np.maximum(mel, sys.float_info.epsilon))
    n_vectors = log_mel.shape[1] - n_frames + 1
    if n_vectors < 1:
        return np.empty((0, dims), dtype=np.float32)
    vectors = np.zeros((n_vectors, dims), dtype=np.float32)
    for t in range(n_frames):
        vectors[:, n_mels * t: n_mels * (t + 1)] = log_mel[:, t: t + n_vectors].T
    return vectors

denoise_wav2spec

denoise_wav2spec(
    waveform: ndarray,
) -> Tuple[ndarray, ndarray]

Peak-normalized magnitude spectrogram + phase, for denoising models.

The two come back in different layouts: the spectrogram as (1, frames, bins) and the phase as (bins, frames), the way denoise_spec2wav takes them back.

Raises:

Type	Description
FeatureError	for a silent clip, which has no peak to normalise by.

Source code in kudio/core/feature.py

def denoise_wav2spec(waveform: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
    """Peak-normalized magnitude spectrogram + phase, for denoising models.

    The two come back in different layouts: the spectrogram as
    ``(1, frames, bins)`` and the phase as ``(bins, frames)``, the way
    :func:`denoise_spec2wav` takes them back.

    :raises FeatureError: for a silent clip, which has no peak to normalise by.
    """
    waveform = _mono(waveform, "denoise_wav2spec()")
    peak = float(np.max(np.abs(waveform))) if waveform.size else 0.0
    if not peak > 0:
        raise FeatureError("denoise_wav2spec() got a silent clip; there is no "
                           "peak to normalise by")
    D = librosa.stft(waveform / peak,
                     n_fft=512, hop_length=256, win_length=512, window='hann')
    D += np.finfo(float).eps
    Sxx = np.abs(D)
    phase = np.angle(D)
    Sxx_mean = np.mean(Sxx, axis=1, keepdims=True)
    Sxx_std = np.std(Sxx, axis=1, keepdims=True) + 1e-12
    Sxx_r = (Sxx - Sxx_mean) / Sxx_std
    return np.reshape(Sxx_r.T, (1, Sxx_r.shape[1], Sxx_r.shape[0])), phase

denoise_spec2wav

denoise_spec2wav(spec: ndarray, phase: ndarray) -> ndarray

Inverse of denoise_wav2spec; output peak-normalized.

Source code in kudio/core/feature.py

def denoise_spec2wav(spec: np.ndarray, phase: np.ndarray) -> np.ndarray:
    """Inverse of :func:`denoise_wav2spec`; output peak-normalized."""
    spec = np.squeeze(spec).T
    pred_wav = librosa.istft(np.multiply(spec, np.exp(1j * phase)),
                             hop_length=256, win_length=512, window='hann')
    peak = float(np.max(np.abs(pred_wav))) if pred_wav.size else 0.0
    return pred_wav / peak if peak > 0 else pred_wav

stack_context

stack_context(
    frames: ndarray, context: int, pad: str = "edge"
) -> ndarray

Stack ±context neighbouring frames onto each frame.

(frames, dim) -> (frames, dim * (2 * context + 1)), one output row per input row: the edges are padded rather than dropped, so a prediction made from the result lines up 1:1 with the input.

Works on any feature matrix — MFCC, mel, log-spectrogram — unlike the forward_backward argument of waveform_to_spectrogram, which only reaches the spectrogram it computes itself.

Parameters:

Name	Type	Description	Default
pad	str	'edge' repeats the first/last frame; 'zero' pads with zeros, which is what forward_backward uses. Zero is not silence: in log10 power it is a power of 1, and after a Standardizer it is the mean frame. 'edge' avoids inventing a step at the boundary. >>> stack_context(np.zeros((10, 40)), context=2).shape (10, 200)	'edge'

Source code in kudio/core/feature.py

def stack_context(frames: np.ndarray, context: int,
                  pad: str = 'edge') -> np.ndarray:
    """Stack ``±context`` neighbouring frames onto each frame.

    ``(frames, dim)`` -> ``(frames, dim * (2 * context + 1))``, one output row
    per input row: the edges are padded rather than dropped, so a prediction
    made from the result lines up 1:1 with the input.

    Works on *any* feature matrix — MFCC, mel, log-spectrogram — unlike the
    ``forward_backward`` argument of :func:`waveform_to_spectrogram`, which
    only reaches the spectrogram it computes itself.

    :param pad: ``'edge'`` repeats the first/last frame; ``'zero'`` pads with
        zeros, which is what ``forward_backward`` uses. Zero is not silence: in
        log10 power it is a power of 1, and after a :class:`Standardizer` it is
        the mean frame. ``'edge'`` avoids inventing a step at the boundary.

    >>> stack_context(np.zeros((10, 40)), context=2).shape
    (10, 200)
    """
    frames = np.asarray(frames, dtype=np.float32)
    if frames.ndim != 2:
        raise FeatureError(f"frames must be 2-D, got shape {frames.shape}")
    if context < 0:
        raise FeatureError(f"context must be >= 0, got {context}")
    if context == 0:
        return frames.copy()                   # never the caller's own array
    if pad not in ('edge', 'zero'):
        raise FeatureError(f"pad must be 'edge' or 'zero', got {pad!r}")

    mode = 'edge' if pad == 'edge' else 'constant'
    padded = np.pad(frames, ((context, context), (0, 0)), mode=mode)
    width = 2 * context + 1
    n, dim = frames.shape
    out = np.empty((n, dim * width), dtype=np.float32)
    for i in range(width):
        out[:, i * dim:(i + 1) * dim] = padded[i:i + n]
    return out

frame_windows

frame_windows(
    frames: ndarray, n_frames: int, pad: str = "edge"
) -> ndarray

Cut a feature matrix into fixed-length windows.

(frames, dim) -> (n_windows, n_frames, dim). The tail is padded rather than dropped, so a clip shorter than one window still yields one — the shape sequence models are trained on.

>>> frame_windows(np.zeros((60, 40)), n_frames=16).shape
(4, 16, 40)

Source code in kudio/core/feature.py

def frame_windows(frames: np.ndarray, n_frames: int,
                  pad: str = 'edge') -> np.ndarray:
    """Cut a feature matrix into fixed-length windows.

    ``(frames, dim)`` -> ``(n_windows, n_frames, dim)``. The tail is padded
    rather than dropped, so a clip shorter than one window still yields one —
    the shape sequence models are trained on.

    >>> frame_windows(np.zeros((60, 40)), n_frames=16).shape
    (4, 16, 40)
    """
    frames = np.asarray(frames, dtype=np.float32)
    if frames.ndim != 2:
        raise FeatureError(f"frames must be 2-D, got shape {frames.shape}")
    if n_frames < 1:
        raise FeatureError(f"n_frames must be >= 1, got {n_frames}")
    if pad not in ('edge', 'zero'):
        raise FeatureError(f"pad must be 'edge' or 'zero', got {pad!r}")

    n, dim = frames.shape
    if n == 0:
        raise FeatureError("there are no frames to cut into windows; a clip "
                           "too short for its features gives an empty matrix")
    n_windows = max(1, int(np.ceil(n / n_frames)))
    missing = n_windows * n_frames - n
    if missing:
        mode = 'edge' if pad == 'edge' else 'constant'
        frames = np.pad(frames, ((0, missing), (0, 0)), mode=mode)
    return frames.reshape(n_windows, n_frames, dim)

wave_slicing

wave_slicing(
    y: ndarray,
    frame_length: int = 2048,
    hop_length: int = 512,
    is_reshape: bool = True,
) -> ndarray

Slice a waveform into overlapping frames (flattened if is_reshape).

Source code in kudio/core/feature.py

def wave_slicing(y: np.ndarray, frame_length: int = 2048,
                 hop_length: int = 512, is_reshape: bool = True) -> np.ndarray:
    """Slice a waveform into overlapping frames (flattened if *is_reshape*)."""
    frame_stack = librosa.util.frame(y, frame_length=frame_length,
                                     hop_length=hop_length)
    if is_reshape:
        frame_stack = frame_stack.T.reshape(-1)
    return frame_stack

features2matrix

features2matrix(
    features: Sequence[ndarray],
) -> Tuple[ndarray, ndarray]

Concatenate feature matrices; label rows by list index.

Source code in kudio/core/feature.py

def features2matrix(features: Sequence[np.ndarray]) -> Tuple[np.ndarray, np.ndarray]:
    """Concatenate feature matrices; label rows by list index."""
    feature_matrix = np.vstack(features)
    labels = np.concatenate([i * np.ones(len(f)) for i, f in enumerate(features)])
    return feature_matrix, labels

kudio.core.spectrogram

A spectrogram for audio that has not finished arriving.

kudio.STFT and kudio.waveform_to_spectrogram want the whole clip. That is right for a file and impossible for a microphone, and the usual workaround -- keep the last few seconds in a list and re-run the STFT over all of it several times a second -- is wasteful in a way that shows: at 16 kHz with a 3-second window and the default geometry, twenty redraws a second recompute 375 columns to gain about six, and 98% of that work redraws what was already on screen.

SpectrogramStream computes each column once, when the samples for it have arrived, and keeps the last N in a ring buffer. Feeding it a stream in whatever block sizes the device happens to deliver produces exactly the columns the offline transform would.

>>> spec = kudio.SpectrogramStream(sr=16000, seconds=3.0)
>>> while recorder.recording:
...     spec.push(recorder.drain())
...     draw(spec.columns())                             # (columns, bins), dB

Columns are not centred. A file transform pads the start so that column k is centred on sample k x hop; a stream cannot pad audio it has not heard yet, so column k here covers samples [k*hop, k*hop + n_fft) -- the center=False convention. The first column therefore needs a whole frame before it exists, which is latency_seconds.

Values are dBFS on a linear axis. Amplitude is normalised by the window, so a full-scale sine reads 0 dB whatever the frame size and window are. A mel axis sums bins through librosa's area-normalised filters, so there the number is band power in dB, not dBFS: a full-scale sine reads about -18 dB at 1 kHz. The number on screen then means the same thing after someone changes n_fft -- which it does not if the scaling is left to the FFT.

Classes:

Name	Description
SpectrogramStream	A rolling log-magnitude spectrogram, one column per hop.

SpectrogramStream

SpectrogramStream(
    sr: int,
    *,
    n_fft: int = 512,
    hop_length: Optional[int] = None,
    window: str = "hann",
    seconds: float = 3.0,
    top_db: float = 80.0,
    ref: float = 1.0,
    n_mels: int = 0,
    fmin: float = 0.0,
    fmax: Optional[float] = None,
)

A rolling log-magnitude spectrogram, one column per hop.

Parameters:

Name	Type	Description	Default
sr	int	sample rate of the audio that will be pushed.	required
n_fft	int	analysis frame, in samples. Also the latency.	512
hop_length	Optional[int]	samples between columns; defaults to n_fft // 4.	None
window	str	any window name scipy.signal.get_window accepts.	'hann'
seconds	float	how much history to keep on screen. Rounded to a whole number of columns, at least one.	3.0
top_db	float	dynamic range. Anything quieter than top_db below the reference is clamped to the floor, so an empty room draws as one flat colour instead of amplified dither.	80.0
ref	float	amplitude that reads 0 dB. 1.0 is dBFS.	1.0
n_mels	int	0 for a linear frequency axis, or a band count for a mel one. Speech at 16 kHz spends most of its energy in the bottom eighth of a linear axis; 64 mel bands spend about a third of the picture below 1 kHz.	0
fmin	float	lowest mel band, ignored when n_mels is 0.	0.0
fmax	Optional[float]	highest mel band; Nyquist when omitted.	None

Methods:

Name	Description
to_stft	The equivalent kudio.STFT geometry.
reset	Empty the window and the part-filled frame, keeping the settings.
push	Feed newly-arrived audio in; get the columns it completed back.
columns	The window as an image: (n_columns, bins), newest on the right.
latest	The most recent column, floor-filled if there is not one yet.

Attributes:

Name	Type	Description
n_bins	int	Rows in a column: mel bands, or linear frequency bins.
freqs	ndarray	Centre frequency of each row, in Hz. Mel bands included.
floor_db	float	What an empty column holds, and the clamp applied to every other.
latency_seconds	float	How long before the first column exists: one analysis frame.
column_seconds	float	Time between columns.
window_seconds	float	Span of columns -- the width of the picture, in seconds.
frames_seen	int	Columns computed since the last reset, including the ones
filled	int	Columns of real audio in the window. Below n_columns only

Source code in kudio/core/spectrogram.py

def __init__(self, sr: int, *, n_fft: int = 512,
             hop_length: Optional[int] = None, window: str = 'hann',
             seconds: float = 3.0, top_db: float = 80.0, ref: float = 1.0,
             n_mels: int = 0, fmin: float = 0.0,
             fmax: Optional[float] = None):
    if sr <= 0:
        raise FeatureError(f"SpectrogramStream needs a positive rate, got {sr}")
    if n_fft < 2:
        raise FeatureError(f"n_fft must be >= 2, got {n_fft}")
    hop = max(1, int(n_fft) // 4) if hop_length is None else int(hop_length)
    if hop < 1:
        raise FeatureError(f"hop_length must be >= 1, got {hop_length}")
    if seconds <= 0:
        raise FeatureError(f"seconds must be positive, got {seconds}")
    if top_db <= 0:
        raise FeatureError(f"top_db must be positive, got {top_db}")
    if ref <= 0:
        raise FeatureError(f"ref must be positive, got {ref}")

    self.sr = int(sr)
    self.n_fft = int(n_fft)
    self.hop_length = hop
    self.window = window
    self.top_db = float(top_db)
    self.ref = float(ref)
    self.n_mels = int(n_mels or 0)
    self.fmin = float(fmin)
    self.fmax = float(fmax) if fmax is not None else self.sr / 2.0

    self._window = _get_window(window, self.n_fft)
    wsum = float(np.sum(self._window)) or 1.0
    # Single-sided amplitude scaling. Every bin except DC and Nyquist has a
    # mirror holding half the energy, so those get the factor of two and
    # those two do not. Without it the level printed beside a peak depends
    # on n_fft and on the window, which makes it unreadable *as a level* --
    # and a meter you cannot read is worse than no meter.
    self._scale = np.full(self.n_fft // 2 + 1, 2.0 / wsum)
    self._scale[0] = 1.0 / wsum
    if self.n_fft % 2 == 0:
        self._scale[-1] = 1.0 / wsum

    self._mel_fb = None
    if self.n_mels:
        import librosa
        self._mel_fb = np.asarray(librosa.filters.mel(
            sr=self.sr, n_fft=self.n_fft, n_mels=self.n_mels,
            fmin=self.fmin, fmax=self.fmax), dtype=np.float64)
        # the band centres: mel_frequencies(n_mels + 2) spans fmin to
        # fmax, and the filters sit on its inner points
        self._freqs = np.asarray(librosa.mel_frequencies(
            n_mels=self.n_mels + 2, fmin=self.fmin, fmax=self.fmax)[1:-1],
            dtype=np.float64)
    else:
        self._freqs = np.fft.rfftfreq(self.n_fft, 1.0 / self.sr)

    self.n_columns = max(1, int(round(seconds * self.sr / self.hop_length)))
    self.reset()

n_bins property

n_bins: int

Rows in a column: mel bands, or linear frequency bins.

freqs property

freqs: ndarray

Centre frequency of each row, in Hz. Mel bands included.

floor_db property

floor_db: float

What an empty column holds, and the clamp applied to every other.

latency_seconds property

latency_seconds: float

How long before the first column exists: one analysis frame.

column_seconds property

column_seconds: float

Time between columns.

window_seconds property

window_seconds: float

Span of columns -- the width of the picture, in seconds.

frames_seen property

frames_seen: int

Columns computed since the last reset, including the ones that have already scrolled off.

filled property

filled: int

Columns of real audio in the window. Below n_columns only while the display is still filling up.

to_stft

to_stft() -> STFT

The equivalent kudio.STFT geometry.

Useful for recording what a live view was computed with, or for reproducing a moment from the file afterwards -- with the caveat at the top of this module: the offline transform centres its frames and this does not.

Source code in kudio/core/spectrogram.py

def to_stft(self) -> STFT:
    """The equivalent :class:`kudio.STFT` geometry.

    Useful for recording what a live view was computed with, or for
    reproducing a moment from the file afterwards -- with the caveat at the
    top of this module: the offline transform centres its frames and this
    does not.
    """
    from kudio.core.stft import STFT
    return STFT(sr=self.sr, n_fft=self.n_fft, hop_length=self.hop_length,
                win_length=self.n_fft, window=self.window)

reset

reset() -> None

Empty the window and the part-filled frame, keeping the settings.

Source code in kudio/core/spectrogram.py

def reset(self) -> None:
    """Empty the window and the part-filled frame, keeping the settings."""
    self._buffer = np.full((self.n_columns, self.n_bins), self.floor_db,
                           dtype=np.float32)
    self._write = 0
    self._filled = 0
    self._pending = np.zeros(0, dtype=np.float64)
    self._frames = 0

push

push(block: ndarray) -> ndarray

Feed newly-arrived audio in; get the columns it completed back.

Blocks may be any size, including zero: samples that do not fill a frame are held until the rest of it arrives, so the answer never depends on how the device chose to chop the stream up.

Feed this kudio.StreamRecorder.drain, not tail. The tail overlaps between calls, and every overlapping sample would be drawn twice, at the wrong time.

Returns:

Type	Description
ndarray	(new_columns, bins) in dB, oldest first. Empty when the block did not complete a frame. More than n_columns of them when the block was longer than the window -- all are returned, only the last n_columns are kept.

Source code in kudio/core/spectrogram.py

def push(self, block: np.ndarray) -> np.ndarray:
    """Feed newly-arrived audio in; get the columns it completed back.

    Blocks may be any size, including zero: samples that do not fill a
    frame are held until the rest of it arrives, so the answer never
    depends on how the device chose to chop the stream up.

    Feed this :meth:`kudio.StreamRecorder.drain`, **not**
    :meth:`~kudio.StreamRecorder.tail`. The tail overlaps between calls,
    and every overlapping sample would be drawn twice, at the wrong time.

    :returns: ``(new_columns, bins)`` in dB, oldest first. Empty when the
        block did not complete a frame. More than :attr:`n_columns` of them
        when the block was longer than the window -- all are returned, only
        the last :attr:`n_columns` are kept.
    """
    block = as_float(block)
    if block.ndim > 1:
        raise FeatureError(
            f"SpectrogramStream takes mono audio, got shape {block.shape}. "
            f"Mix down first (y.mean(axis=1)) -- flattening interleaved "
            f"channels would analyse a signal that does not exist.")
    block = block.astype(np.float64, copy=False).reshape(-1)
    if block.size:
        self._pending = np.concatenate([self._pending, block])

    columns = []
    while len(self._pending) >= self.n_fft:
        columns.append(self._column(self._pending[:self.n_fft]))
        self._pending = self._pending[self.hop_length:]
    if not columns:
        return np.zeros((0, self.n_bins), dtype=np.float32)

    new = np.asarray(columns, dtype=np.float32)
    self._store(new)
    self._frames += len(new)
    return new

columns

columns() -> ndarray

The window as an image: (n_columns, bins), newest on the right.

Always the full width, floor-filled on the left until enough audio has arrived. A display that grows from nothing rescales its own time axis on every frame, which reads as the picture jittering rather than as the recording starting.

Source code in kudio/core/spectrogram.py

def columns(self) -> np.ndarray:
    """The window as an image: ``(n_columns, bins)``, newest on the right.

    Always the full width, floor-filled on the left until enough audio has
    arrived. A display that grows from nothing rescales its own time axis
    on every frame, which reads as the picture jittering rather than as the
    recording starting.
    """
    if self._filled >= self.n_columns:
        return np.concatenate([self._buffer[self._write:],
                               self._buffer[:self._write]])
    out = np.full_like(self._buffer, self.floor_db)
    if self._filled:
        out[-self._filled:] = self._buffer[:self._filled]
    return out

latest

latest() -> ndarray

The most recent column, floor-filled if there is not one yet.

A single column is a spectrum: this is what an instantaneous frequency readout draws.

Source code in kudio/core/spectrogram.py

def latest(self) -> np.ndarray:
    """The most recent column, floor-filled if there is not one yet.

    A single column is a spectrum: this is what an instantaneous frequency
    readout draws.
    """
    if not self._filled:
        return np.full(self.n_bins, self.floor_db, dtype=np.float32)
    return self._buffer[(self._write - 1) % self.n_columns].copy()