visualdynamics.decimate

decimate

Peak-keeping decimation: fewer points, with every peak kept.

A record of a million samples is too many to draw point for point, and keeping every n-th sample can skip the one large peak that was the reason to look at the record. Peak-keeping decimation cuts the samples into equal slices and keeps each slice's smallest and largest value at their own abscissa positions, so no resonance, dropout or shock is lost. It is the same method as pyqtgraph's peak downsampling, which the 2-D plot uses.

The report and the 3-D waterfall both use this module. It works on all of an object's records at once, because the waterfall passes it hundreds of them. It needs only numpy.

This thinning is for drawing, never for a record anything is computed from: its points sit at uneven times, and an anti-alias filter ahead of it would take off the peaks it exists to keep. A record taken to a lower sample rate for analysis goes through core.resample, which filters first.

Functions:

Name	Description
peak_decimate	x and y thinned to at most budget points that keep every
peak_decimate_rows	peak_decimate for every record of one object at once.
shared_peak_indices	The samples of one shared abscissa to keep, at most budget of
envelope_rows	Every record of one object thinned onto the same bins, at most

Functions:

peak_decimate

peak_decimate(x: ArrayLike, y: ArrayLike, budget: int) -> tuple[ndarray, ndarray]

x and y thinned to at most budget points that keep every extreme.

The samples are cut into budget // 2 slices, and each slice keeps its smallest and largest value. y must be real: a caller with complex data takes the component it wants first, because the extremes of a complex value are not defined. Input at or under the budget is returned as given.

NaN marks a gap (a specification is NaN outside its band) and is never kept as an extreme. A slice that is all NaN keeps its first point, so the gap is still drawn as a gap.

Source code in src/visualdynamics/decimate.py

def peak_decimate(x: ArrayLike, y: ArrayLike,
                  budget: int) -> tuple[np.ndarray, np.ndarray]:
    """`x` and `y` thinned to at most `budget` points that keep every
    extreme.

    The samples are cut into `budget // 2` slices, and each slice keeps
    its smallest and largest value. `y` must be real: a caller with
    complex data takes the component it wants first, because the
    extremes of a complex value are not defined. Input at or under the
    budget is returned as given.

    NaN marks a gap (a specification is NaN outside its band) and is
    never kept as an extreme. A slice that is all NaN keeps its first
    point, so the gap is still drawn as a gap.
    """
    x = np.asarray(x)
    y = np.asarray(y)
    if len(x) <= budget:
        return x, y
    keep = _extreme_indices(y[np.newaxis], budget // 2)[0]
    return x[keep], y[keep]

peak_decimate_rows

peak_decimate_rows(x: ArrayLike, rows: ArrayLike, budget: int) -> list[tuple[ndarray, ndarray]]

peak_decimate for every record of one object at once.

rows is a (records, samples) array on the shared abscissa x. The result is one (x, y) pair per record, and the records no longer share an abscissa: each keeps its own extremes at their own positions. The whole block is decimated in one vectorized pass.

Source code in src/visualdynamics/decimate.py

def peak_decimate_rows(x: ArrayLike, rows: ArrayLike,
                       budget: int) -> list[tuple[np.ndarray, np.ndarray]]:
    """`peak_decimate` for every record of one object at once.

    `rows` is a (records, samples) array on the shared abscissa `x`. The
    result is one (x, y) pair per record, and the records no longer share
    an abscissa: each keeps its own extremes at their own positions. The
    whole block is decimated in one vectorized pass.
    """
    x = np.asarray(x)
    rows = np.atleast_2d(np.asarray(rows))
    if rows.shape[1] <= budget:
        return [(x, rows[k]) for k in range(rows.shape[0])]
    kept = _extreme_indices(rows, budget // 2)
    return [(x[keep], rows[k][keep]) for k, keep in enumerate(kept)]

shared_peak_indices

shared_peak_indices(rows: ArrayLike, budget: int) -> ndarray

The samples of one shared abscissa to keep, at most budget of them, so that every row keeps its extremes.

peak_decimate_rows gives each row its own positions, which suits curves drawn on their own. Rows that share an abscissa on the page, a measurement and the limits shaded about it, must keep one set of positions between them, or a band is shaded between points that are not above each other. This keeps, for every slice of the samples, the smallest and largest value of every row, with the slices as fine as the budget allows, and the first and last sample so the extent is unchanged. The first missing sample of each gap is kept too, while the gaps fit a quarter of the budget, so a thinned curve breaks where its data does instead of being drawn across the gap. Rows with fewer than budget samples keep them all.

Parameters:

Name	Type	Description	Default
rows	array_like	(rows, samples) values on one abscissa; NaN marks a gap.	required
budget	int	The most samples to keep, at least two for each row and two for the ends.	required

Returns:

Type	Description
ndarray	Sorted sample indices.

Raises:

Type	Description
ValueError	If the budget cannot hold one slice's extremes for every row.

Source code in src/visualdynamics/decimate.py

def shared_peak_indices(rows: ArrayLike, budget: int) -> np.ndarray:
    """The samples of one shared abscissa to keep, at most `budget` of
    them, so that every row keeps its extremes.

    `peak_decimate_rows` gives each row its own positions, which suits
    curves drawn on their own. Rows that share an abscissa on the page,
    a measurement and the limits shaded about it, must keep one set of
    positions between them, or a band is shaded between points that are
    not above each other. This keeps, for every slice of the samples,
    the smallest and largest value of every row, with the slices as fine
    as the budget allows, and the first and last sample so the extent is
    unchanged. The first missing sample of each gap is kept too, while
    the gaps fit a quarter of the budget, so a thinned curve breaks
    where its data does instead of being drawn across the gap. Rows with
    fewer than `budget` samples keep them all.

    Parameters
    ----------
    rows : array_like
        (rows, samples) values on one abscissa; NaN marks a gap.
    budget : int
        The most samples to keep, at least two for each row and two for
        the ends.

    Returns
    -------
    numpy.ndarray
        Sorted sample indices.

    Raises
    ------
    ValueError
        If the budget cannot hold one slice's extremes for every row.
    """
    rows = np.atleast_2d(np.asarray(rows, dtype=float))
    count, samples = rows.shape
    budget = int(budget)
    if samples <= budget:
        return np.arange(samples)
    if budget < 2 * count + 2:
        raise ValueError(f'{budget} points cannot keep the extremes of '
                         f'{count} curves; at least {2 * count + 2} can')
    ends = np.array([0, samples - 1])
    # a slice's extremes skip a missing sample, so a lone NaN would be
    # thinned away and the line drawn straight across it; the start of
    # every gap is kept with the ends while there is room for it beside
    # the coarsest slicing's extremes
    missing = ~np.isfinite(rows)
    opened = missing.copy()
    opened[:, 1:] &= ~missing[:, :-1]
    gaps = np.flatnonzero(opened.any(axis=0))
    if len(gaps) <= min(budget // 4, budget - 2 * count - 2):
        ends = np.union1d(ends, gaps)

    def kept(bins):
        return np.unique(np.concatenate(
            [ends, *_extreme_indices(rows, bins)]))

    # the most slices whose extremes fit: rows that rise and fall
    # together share most of theirs, so the answer usually lies well
    # above budget / (2 * rows), and a bisection finds it in a dozen
    # vectorized passes
    low, high = 1, budget // 2
    best = kept(low)
    while low < high:
        middle = (low + high + 1) // 2
        trial = kept(middle)
        if len(trial) <= budget:
            low, best = middle, trial
        else:
            high = middle - 1
    return best

envelope_rows

envelope_rows(x: ArrayLike, rows: ArrayLike, bins: int) -> tuple[ndarray, ndarray]

Every record of one object thinned onto the same bins, at most bins of them: two points a bin, the bin's least and greatest value in the order they occurred, at the bin's first and last sample time.

peak_decimate_rows keeps each record's extremes at their own positions, so a page carrying two dozen records carries two dozen abscissas. An envelope on common bins lets a report's time-history page share one abscissa among its curves, and draws the same picture, since a bin is far narrower than a pixel at any zoom the page offers. Input with at most two points a bin is returned as given.

Returns (axis, envelope): the shared abscissa and a (records, points) array.

Source code in src/visualdynamics/decimate.py

def envelope_rows(x: ArrayLike, rows: ArrayLike, bins: int
                  ) -> tuple[np.ndarray, np.ndarray]:
    """Every record of one object thinned onto the same bins, at most
    `bins` of them: two points a bin, the bin's least and greatest value
    in the order they occurred, at the bin's first and last sample time.

    `peak_decimate_rows` keeps each record's extremes at their own
    positions, so a page carrying two dozen records carries two dozen
    abscissas. An envelope on common bins lets a report's time-history
    page share one abscissa among its curves, and draws the same
    picture, since a bin is far narrower than a pixel at any zoom the
    page offers. Input with at most two points a bin is returned as
    given.

    Returns ``(axis, envelope)``: the shared abscissa and a (records,
    points) array.
    """
    # The shared axis halved the size of a report's time-history page
    # (2026-09-20).
    x = np.asarray(x, dtype=float)
    rows = np.atleast_2d(np.asarray(rows, dtype=float))
    n, m = rows.shape
    if m <= 2 * bins:
        return x, rows
    chunk = -(-m // bins)
    real = int(-(-m // chunk))
    padded = np.full((n, real * chunk), np.nan)
    padded[:, :m] = rows
    shaped = padded.reshape(n, real, chunk)
    finite = np.isfinite(shaped)
    lo = np.argmin(np.where(finite, shaped, np.inf), axis=-1)
    hi = np.argmax(np.where(finite, shaped, -np.inf), axis=-1)
    least = np.take_along_axis(shaped, lo[..., None], axis=-1)[..., 0]
    most = np.take_along_axis(shaped, hi[..., None], axis=-1)[..., 0]
    any_finite = finite.any(axis=-1)
    least = np.where(any_finite, least, np.nan)
    most = np.where(any_finite, most, np.nan)
    # the extreme that came first goes at the bin's start
    first_is_low = lo <= hi
    start = np.where(first_is_low, least, most)
    end = np.where(first_is_low, most, least)
    starts = np.arange(real) * chunk
    ends = np.minimum(starts + chunk - 1, m - 1)
    axis = np.column_stack([x[starts], x[ends]]).ravel()
    out = np.empty((n, 2 * real))
    out[:, 0::2] = start
    out[:, 1::2] = end
    return axis, out