visualdynamics.core.sine

sine

Sine sweep specifications: named tones, each its own time-boxed sweep.

Phase B of the sine sweep arc (PLAN.md). A sine test's requirement is not a curve over one abscissa — it is a set of tones, each a breakpoint table with its own sweep law and its own start time, played simultaneously and only partially overlapping in the ordinary case (measured on the Phase A fixture: four tones playing 0-15, 2-13, 3-14 and 1-8 s of one 15 s environment). That shape does not fit a DataArray, so it gets its own class rather than a forced one.

The sweep law here is the one the Phase A runs pinned against the controller's own records, not assumed from documentation:

• segment_rate[i] governs the segment from frequency[i] to frequency[i+1] — the leading convention, matched to 0.0009 Hz over a 14 s sweep.
• A linear segment's rate is in Hz/s; a logarithmic segment's rate is in oct/min (one octave at rate 10 transited in 5.87 s ~= 6 s; the controller's own docstrings disagree with each other and the measurement settled it).
• A descending segment takes a negative rate. A rate whose sign contradicts its breakpoints is refused by name here, because the controller refuses the same thing as a negative sweep time — at initialization, deep in its log.

Amplitude between breakpoints interpolates linearly in time, which is linear in frequency on a linear segment and linear in log-frequency on a logarithmic one — target() interpolates that way per segment, so the compliance curve is the curve the controller was actually chasing.

One more alignment fact for the extraction to lean on, measured on the Phase A fixture: the controller ramps each tone up for the environment's ramp_time before sweeping, so the recorded trajectory leaves its start frequency at start_time + ramp_time — all four fixture tones rebuilt against the controller's own record to 0.0000 Hz with exactly that lag.

Everything is SI at rest, like every other object in core.

Classes:

Name	Description
SineTone	One tone: breakpoints, a sweep law, and its own window in time.
SineSweepSpecification	What a sine test was controlled to: tones over control channels.
SineTarget	One tone's requirement laid over frequency, as a curve.
SineLevel	A sine sweep's measured level: amplitude against frequency.
SineLevelSet	One extraction, one object: the tones' levels, grouped the way
SineExtraction	How a sweep's levels are read out of a recording — the settings

Functions:

Name	Description
find_tone	Where a tone's sweep begins in a recording, by matched filter.
find_environment	Where the tones' shared clock starts, by joint matched filter.
vold_kalman	Every tone's complex envelope at every sample, solved jointly.
sample_levels	The levels read at windows stretches of each tone, cheaply.
scatter_db	The predicted standard deviation of a reading, in dB, from the
suggest_cycles	The smoothing the data asks for: the first rung of
extract_sine	Read each tone's level out of a recording, against its own sweep.

Classes

SineTone

SineTone(name: str, start_time: float, frequency: Any, amplitude: Any, segment_type: Any, segment_rate: Any, phase: Any = None, warning_lower: Any = None, warning_upper: Any = None, abort_lower: Any = None, abort_upper: Any = None)

One tone: breakpoints, a sweep law, and its own window in time.

Attributes: name: What the controller called it ('Sine Tone 1'). start_time: Seconds after the environment started that this tone turns on. Tones are silent outside their own span — never held at an end frequency. frequency: Breakpoints, Hz, shape (n,). Ascending or descending per segment; each segment's rate sign must agree. amplitude: Target amplitude at each breakpoint per control channel, shape (n, m), SI. phase: Radians at each breakpoint per channel, shape (n, m). segment_type: (n-1,), LINEAR or LOG per segment. segment_rate: (n-1,), Hz/s for a linear segment, oct/min for a logarithmic one; negative for a descending segment. warning_lower, warning_upper, abort_lower, abort_upper: Band curves at the breakpoints, shape (n, m), or None where the specification carries no such limit.

Methods:

Name	Description
segment_seconds	How long each segment takes, from its span and its rate.
span	When this tone plays, in environment time.
trajectory	Frequency versus time over the tone's own span.
argument	The cosine argument over the tone's span: 2piintegral(f),
frequency_at	The instantaneous frequency, Hz, at seconds t from the
phase_at	The cosine argument, radians, at seconds t from the
grid	The sample times trajectory(dt) lays down, from sample
grid_at	The sample times trajectory(dt) lays down, at the given
target	The specified level at given frequencies, shape (len, m).

Source code in src/visualdynamics/core/sine.py

def __init__(self, name: str, start_time: float,
             frequency: Any, amplitude: Any,
             segment_type: Any, segment_rate: Any,
             phase: Any = None,
             warning_lower: Any = None, warning_upper: Any = None,
             abort_lower: Any = None, abort_upper: Any = None) -> None:
    self.name: str = str(name)
    self.start_time: float = float(start_time)
    self.frequency: np.ndarray = np.asarray(frequency,
                                            dtype=np.float64)
    self.amplitude: np.ndarray = np.atleast_2d(
        np.asarray(amplitude, dtype=np.float64))
    n = len(self.frequency)
    if self.amplitude.shape[0] != n:
        raise ValueError(
            f'{self.name}: {n} breakpoints but amplitude has '
            f'{self.amplitude.shape[0]} rows')
    self.phase: np.ndarray = (
        np.zeros_like(self.amplitude) if phase is None
        else np.atleast_2d(np.asarray(phase, dtype=np.float64)))
    self.segment_type: np.ndarray = np.asarray(segment_type,
                                               dtype=np.int64)
    self.segment_rate: np.ndarray = np.asarray(segment_rate,
                                               dtype=np.float64)
    if len(self.segment_type) != n - 1 or len(self.segment_rate) != n - 1:
        raise ValueError(
            f'{self.name}: {n} breakpoints take {n - 1} segments; '
            f'got {len(self.segment_type)} types and '
            f'{len(self.segment_rate)} rates')
    self.limits: dict[str, np.ndarray] = {}
    for limit, values in (('warning_lower', warning_lower),
                          ('warning_upper', warning_upper),
                          ('abort_lower', abort_lower),
                          ('abort_upper', abort_upper)):
        if values is None:
            continue
        values = np.atleast_2d(np.asarray(values, dtype=np.float64))
        if values.shape != self.amplitude.shape:
            raise ValueError(
                f'{self.name}: {limit} has shape {values.shape}, '
                f'expected {self.amplitude.shape} to match the '
                'amplitude')
        self.limits[limit] = values
    self._check_rates()

Methods:

segment_seconds

segment_seconds() -> ndarray

How long each segment takes, from its span and its rate.

Source code in src/visualdynamics/core/sine.py

def segment_seconds(self) -> np.ndarray:
    """How long each segment takes, from its span and its rate."""
    seconds = np.empty(len(self.segment_rate))
    for i in range(len(seconds)):
        f0, f1 = self.frequency[i], self.frequency[i + 1]
        if self.segment_type[i] == LINEAR:
            seconds[i] = (f1 - f0) / self.segment_rate[i]
        else:
            # oct/min, measured — see the module docstring
            seconds[i] = np.log2(f1 / f0) / self.segment_rate[i] * 60.0
    return seconds

span

span() -> tuple[float, float]

When this tone plays, in environment time.

Source code in src/visualdynamics/core/sine.py

def span(self) -> tuple[float, float]:
    """When this tone plays, in environment time."""
    return (self.start_time, self.start_time + self.duration())

trajectory

trajectory(dt: float) -> tuple[ndarray, ndarray]

Frequency versus time over the tone's own span.

Returns (t, f): t in seconds from the tone's start (add start_time for environment time), f in Hz. This is the rebuild that matched the controller's own record to 0.0009 Hz.

Source code in src/visualdynamics/core/sine.py

def trajectory(self, dt: float) -> tuple[np.ndarray, np.ndarray]:
    """Frequency versus time over the tone's own span.

    Returns (t, f): t in seconds from the tone's start (add
    `start_time` for environment time), f in Hz. This is the
    rebuild that matched the controller's own record to 0.0009 Hz.
    """
    seconds = self.segment_seconds()
    pieces_t: list[np.ndarray] = [np.array([0.0])]
    pieces_f: list[np.ndarray] = [self.frequency[:1]]
    start = 0.0
    for i, length in enumerate(seconds):
        n = max(round(length / dt), 1)
        t = start + length * np.arange(1, n + 1) / n
        f0, f1 = self.frequency[i], self.frequency[i + 1]
        fraction = np.arange(1, n + 1) / n
        if self.segment_type[i] == LINEAR:
            f = f0 + (f1 - f0) * fraction
        else:
            f = f0 * (f1 / f0) ** fraction
        pieces_t.append(t)
        pieces_f.append(f)
        start += length
    return np.concatenate(pieces_t), np.concatenate(pieces_f)

argument

argument(dt: float) -> ndarray

The cosine argument over the tone's span: 2piintegral(f), at the sample times trajectory returns — phase_at on them, so a slice of the sweep (argument_slice) is the same numbers as the whole (2026-10-01).

Source code in src/visualdynamics/core/sine.py

def argument(self, dt: float) -> np.ndarray:
    """The cosine argument over the tone's span: 2*pi*integral(f),
    at the sample times `trajectory` returns — `phase_at` on them,
    so a slice of the sweep (`argument_slice`) is the same numbers
    as the whole (2026-10-01)."""
    t, _f = self.trajectory(dt)
    return self.phase_at(t)

frequency_at

frequency_at(t: Any) -> ndarray

The instantaneous frequency, Hz, at seconds t from the tone's start — the same law trajectory lays on its grid, evaluated anywhere, so a slice of a long sweep costs the slice.

Parameters:

Name	Type	Description	Default
t	array - like	Seconds from the tone's start.	required

Returns:

Type	Description
ndarray

Source code in src/visualdynamics/core/sine.py

def frequency_at(self, t: Any) -> np.ndarray:
    """The instantaneous frequency, Hz, at seconds `t` from the
    tone's start — the same law `trajectory` lays on its grid,
    evaluated anywhere, so a slice of a long sweep costs the slice.

    Parameters
    ----------
    t : array-like
        Seconds from the tone's start.

    Returns
    -------
    ndarray
    """
    t = np.asarray(t, dtype=np.float64)
    f = np.full(t.shape, float(self.frequency[-1]))
    f[t <= 0.0] = float(self.frequency[0])
    for start, length, f0, f1, log in self._segments():
        inside = (t > start) & (t <= start + length)
        fraction = (t[inside] - start) / length
        f[inside] = (f0 * (f1 / f0) ** fraction if log
                     else f0 + (f1 - f0) * fraction)
    return f

phase_at

phase_at(t: Any) -> ndarray

The cosine argument, radians, at seconds t from the tone's start: 2*pi times the integral of the frequency law, segment by segment in closed form — a linear sweep's chirp, a log sweep's exponential.

Parameters:

Name	Type	Description	Default
t	array - like	Seconds from the tone's start.	required

Returns:

Type	Description
ndarray

Source code in src/visualdynamics/core/sine.py

def phase_at(self, t: Any) -> np.ndarray:
    """The cosine argument, radians, at seconds `t` from the
    tone's start: 2*pi times the integral of the frequency law,
    segment by segment in closed form — a linear sweep's chirp, a
    log sweep's exponential.

    Parameters
    ----------
    t : array-like
        Seconds from the tone's start.

    Returns
    -------
    ndarray
    """
    t = np.asarray(t, dtype=np.float64)
    phase = np.zeros(t.shape)
    carried = 0.0            # cycles at the start of each segment
    for start, length, f0, f1, log in self._segments():
        local = np.clip(t - start, 0.0, length)
        if log:
            ratio = f1 / f0
            cycles = f0 * length / np.log(ratio) * (ratio ** (local / length) - 1.0)
            whole = f0 * length / np.log(ratio) * (ratio - 1.0)
        else:
            cycles = f0 * local + (f1 - f0) * local ** 2 / (2.0 * length)
            whole = (f0 + f1) * length / 2.0
        phase += cycles
        carried += whole
    # past the last segment the phase holds, as the frequency does
    over = t > sum(length for _s, length, *_r in self._segments())
    phase[over] = carried
    return 2.0 * np.pi * phase

grid

grid(dt: float, first: int, last: int) -> ndarray

The sample times trajectory(dt) lays down, from sample first to last (exclusive): the whole grid's own numbers, for a slice of it.

Parameters:

Name	Type	Description	Default
dt	float	The sample interval.	required
first	int	The slice of the sweep's samples.	required
last	int	The slice of the sweep's samples.	required

Returns:

Type	Description
ndarray

Source code in src/visualdynamics/core/sine.py

def grid(self, dt: float, first: int, last: int) -> np.ndarray:
    """The sample times `trajectory(dt)` lays down, from sample
    `first` to `last` (exclusive): the whole grid's own numbers,
    for a slice of it.

    Parameters
    ----------
    dt : float
        The sample interval.
    first, last : int
        The slice of the sweep's samples.

    Returns
    -------
    ndarray
    """
    # the grid is one sample at t=0, then each segment's samples
    # evenly over its own length — rebuilt per segment so a slice
    # never allocates the whole
    edges = []
    for start, length, *_rest in self._segments():
        n = max(round(length / dt), 1)
        edges.append((start, length, n))
    total = 1 + sum(n for _s, _l, n in edges)
    first, last = max(int(first), 0), min(int(last), total)
    out = np.empty(max(last - first, 0))
    if last <= first:
        return out
    index = 0
    if first == 0:
        out[0] = 0.0
        index = 1
    offset = 1
    for start, length, n in edges:
        lo, hi = max(first, offset), min(last, offset + n)
        if hi > lo:
            k = np.arange(lo - offset + 1, hi - offset + 1)
            out[index:index + hi - lo] = start + length * k / n
            index += hi - lo
        offset += n
    return out

grid_at

grid_at(dt: float, samples: Any) -> ndarray

The sample times trajectory(dt) lays down, at the given sample indices — grid for a handful of samples picked out of a long sweep, without laying the whole sweep down.

Parameters:

Name	Type	Description	Default
dt	float	The sample interval.	required
samples	array-like of int	Sample indices into the sweep's grid.	required

Returns:

Type	Description
ndarray

Source code in src/visualdynamics/core/sine.py

def grid_at(self, dt: float, samples: Any) -> np.ndarray:
    """The sample times `trajectory(dt)` lays down, at the given
    sample indices — `grid` for a handful of samples picked out of
    a long sweep, without laying the whole sweep down.

    Parameters
    ----------
    dt : float
        The sample interval.
    samples : array-like of int
        Sample indices into the sweep's grid.

    Returns
    -------
    ndarray
    """
    samples = np.asarray(samples, dtype=np.int64)
    out = np.zeros(samples.shape)
    offset = 1
    for start, length, *_rest in self._segments():
        n = max(round(length / dt), 1)
        inside = (samples >= offset) & (samples < offset + n)
        out[inside] = start + length * (samples[inside] - offset + 1) / n
        offset += n
    return out

target

target(frequencies: Any, curve: str = 'amplitude') -> ndarray

The specified level at given frequencies, shape (len, m).

curve is 'amplitude' or one of the limit names. Amplitude interpolates linearly in time — linear in frequency on a linear segment, linear in log-frequency on a logarithmic one — because that is the target the controller chases. Frequencies the tone never sweeps come back NaN: the specification says nothing there, and NaN says so where zero would lie.

Source code in src/visualdynamics/core/sine.py

def target(self, frequencies: Any,
           curve: str = 'amplitude') -> np.ndarray:
    """The specified level at given frequencies, shape (len, m).

    `curve` is 'amplitude' or one of the limit names. Amplitude
    interpolates linearly in time — linear in frequency on a
    linear segment, linear in log-frequency on a logarithmic one —
    because that is the target the controller chases. Frequencies
    the tone never sweeps come back NaN: the specification says
    nothing there, and NaN says so where zero would lie.
    """
    values = (self.amplitude if curve == 'amplitude'
              else self.limits.get(curve))
    if values is None:
        raise ValueError(f'{self.name} carries no {curve}')
    frequencies = np.asarray(frequencies, dtype=np.float64)
    out = np.full((len(frequencies), values.shape[1]), np.nan)
    for i in range(len(self.segment_rate)):
        f0, f1 = self.frequency[i], self.frequency[i + 1]
        lo, hi = min(f0, f1), max(f0, f1)
        inside = (frequencies >= lo) & (frequencies <= hi)
        if not inside.any():
            continue
        if self.segment_type[i] == LINEAR:
            fraction = (frequencies[inside] - f0) / (f1 - f0)
        else:
            fraction = (np.log(frequencies[inside] / f0)
                        / np.log(f1 / f0))
        out[inside] = (values[i]
                       + (values[i + 1] - values[i]) * fraction[:, None])
    return out

SineSweepSpecification

SineSweepSpecification(tones: Sequence[SineTone], response_dof: Sequence[str], ordinate_dim: str = 'acceleration', ordinate_unit: str | None = None, comment: str = '')

What a sine test was controlled to: tones over control channels.

One object per sine environment, mirroring the controller's file: the tones (each with its own breakpoints, sweep law, bands and start time) and the control DOFs their amplitude columns belong to. Simultaneous tones are the ordinary case, not a variant.

Attributes: tones: The SineTone list, in the file's order. response_dof: The control channel DOF strings — the columns of every tone's amplitude table, in order. ordinate_dim: What the amplitudes are ('acceleration' for a controller run on accelerometers). ordinate_unit: The SI unit the amplitudes are stored in, or None while undeclared — the same convention every DataArray follows. comment: One line about the set as a whole.

Methods:

Name	Description
span	When any tone is playing: first start to last end.
tone_curve	One tone's requirement as a plottable, comparable curve.

Source code in src/visualdynamics/core/sine.py

def __init__(self, tones: Sequence[SineTone],
             response_dof: Sequence[str],
             ordinate_dim: str = 'acceleration',
             ordinate_unit: str | None = None,
             comment: str = '') -> None:
    self.tones: list[SineTone] = list(tones)
    self.response_dof: list[str] = list(response_dof)
    m = len(self.response_dof)
    for tone in self.tones:
        if tone.amplitude.shape[1] != m:
            raise ValueError(
                f'{tone.name} carries {tone.amplitude.shape[1]} '
                f'amplitude columns for {m} control channels')
    names = [tone.name for tone in self.tones]
    if len(set(names)) != len(names):
        raise ValueError(f'tone names repeat: {sorted(names)}')
    self.ordinate_dim: str = ordinate_dim
    self.ordinate_unit: str | None = ordinate_unit
    self.comment: str = comment

Methods:

span

span() -> tuple[float, float]

When any tone is playing: first start to last end.

Source code in src/visualdynamics/core/sine.py

def span(self) -> tuple[float, float]:
    """When any tone is playing: first start to last end."""
    starts, ends = zip(*(tone.span() for tone in self.tones))
    return (min(starts), max(ends))

tone_curve

tone_curve(name: str, lines: int = 400) -> SineTarget

One tone's requirement as a plottable, comparable curve.

The abscissa follows the tone's own sweep — lines points spaced the way the sweep dwells, ascending whichever way it swept — and the bands ride along as Bounded limits.

Source code in src/visualdynamics/core/sine.py

def tone_curve(self, name: str, lines: int = 400) -> SineTarget:
    """One tone's requirement as a plottable, comparable curve.

    The abscissa follows the tone's own sweep — `lines` points
    spaced the way the sweep dwells, ascending whichever way it
    swept — and the bands ride along as Bounded limits.
    """
    tone = self.tone(name)
    _t, f = tone.trajectory(tone.duration() / lines)
    order = np.argsort(f)
    frequencies = f[order]
    target = tone.target(frequencies)
    limits = {limit: tone.target(frequencies, curve=limit).T
              for limit in tone.limits}
    return SineTarget(
        abscissa=frequencies, ordinate=target.T,
        response_dof=list(self.response_dof),
        ordinate_dim=[self.ordinate_dim] * len(self.response_dof),
        ordinate_unit=[self.ordinate_unit] * len(self.response_dof),
        comment=[f'{name} requirement at {dof}'
                 for dof in self.response_dof],
        tone=name, **limits)

SineTarget

SineTarget(*args: Any, tone: str = '', **kwargs: Any)

Bases: Bounded, Spectrum

One tone's requirement laid over frequency, as a curve.

What SineSweepSpecification.tone_curve hands the plot and the report: the tone's amplitude interpolated along its own sweep, with the warning and abort bands as the four limit curves every Bounded object carries — so the comparison against an extracted level draws with the same zones, the same shading and the same pairing a random specification gets, from the same machinery. Derived on demand; the specification object stays the one source.

Source code in src/visualdynamics/core/sine.py

def __init__(self, *args: Any, tone: str = '', **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self.tone: str = str(tone)

SineLevel

SineLevel(*args: Any, tone: str = '', onset: float = 0.0, seconds: Any = None, floor: Any = None, below_floor: Any = None, drift_hz: float = 0.0, **kwargs: Any)

Bases: Spectrum

A sine sweep's measured level: amplitude against frequency.

What extract_sine reads out of a recording for one tone — the demodulated complex amplitude of that tone at each control channel, sampled along the sweep and laid over frequency. The magnitude is the tracked amplitude the controller was steering; the angle is the phase relative to the reconstructed sweep argument, meaningful between channels rather than absolutely.

The frequency coverage is the coverage: a run stopped early, or a tone the instructions windowed, extracts fewer lines, and the comparison against the specification sees exactly how much of the required range was actually run rather than being told everything was.

Attributes: tone: The specification tone this level was extracted for. onset: Seconds into the recording where the tone's sweep proper was found (matched filter, or the caller's override) — reported so the alignment is auditable. seconds: When each line was measured, in recording seconds — the sweep's own clock, one entry per abscissa line, which is what lets the level stand on the 3-D stage without the specification beside it. None on a level from before the clock was kept. floor: The noise floor beside each reading, channels × lines, in the level's own units: the amplitude a tone would need to stand clear of the noise the smoothing let through (2026-09-30). None on a level from before it was kept. below_floor: Channels × lines, True where the tone was under its floor and the reading is reported at the floor rather than at zero — a mark on the plot and a line the comparison does not judge. None on an older level. drift_hz: The sweep-clock correction applied at the end of the tone's sweep, in Hz, when the recording's sweep had drifted from the specification's; 0 when none was needed.

Attributes:

Name	Type	Description
resolved	ndarray	Channels × lines, True where the tone stood above its floor

Source code in src/visualdynamics/core/sine.py

def __init__(self, *args: Any, tone: str = '', onset: float = 0.0,
             seconds: Any = None, floor: Any = None,
             below_floor: Any = None, drift_hz: float = 0.0,
             **kwargs: Any) -> None:
    super().__init__(*args, **kwargs)
    self.tone: str = str(tone)
    self.onset: float = float(onset)
    self.seconds: np.ndarray | None = (
        None if seconds is None
        else np.asarray(seconds, dtype=np.float64))
    self.floor: np.ndarray | None = (
        None if floor is None else np.asarray(floor, dtype=np.float64))
    self.below_floor: np.ndarray | None = (
        None if below_floor is None
        else np.asarray(below_floor, dtype=bool))
    self.drift_hz: float = float(drift_hz)

Attributes

resolved property

resolved: ndarray

Channels × lines, True where the tone stood above its floor — every line, on a level that kept no floor.

SineLevelSet

SineLevelSet(levels: Sequence[SineLevel], cycles: float | None = None)

One extraction, one object: the tones' levels, grouped the way the specification groups its tones (Brandon, 2026-08-22).

Each tone sweeps its own frequencies on its own clock, so the levels stay separate SineLevels inside — different abscissas cannot share a DataArray — but the project holds one thing, it expands into one row per tone, and picking rows plots a subset, exactly as the specification does.

Attributes: levels: The per-tone SineLevels, in the specification's order. cycles: The smoothing the levels were read with, in cycles of each tone's instantaneous frequency; None on a set from before it was kept.

Source code in src/visualdynamics/core/sine.py

def __init__(self, levels: Sequence[SineLevel],
             cycles: float | None = None) -> None:
    self.levels: list[SineLevel] = list(levels)
    self.cycles: float | None = None if cycles is None else float(cycles)
    names = [level.tone for level in self.levels]
    if len(set(names)) != len(names):
        raise ValueError(f'tone names repeat: {sorted(names)}')

SineExtraction dataclass

SineExtraction(cycles: float | None = None, target_db: float = TARGET_SCATTER_DB, refine: bool = True, chosen: float | None = None)

How a sweep's levels are read out of a recording — the settings that ride the time history, the way Averaging and Filtering do (the sine view sets them; Extract Sine Levels reads whatever is there).

cycles is the smoothing: how many cycles of the tone's own instantaneous frequency each reading averages over. Shorter follows a resonance closely; longer holds the random environment out of the reading. None means automatic: the extraction samples the recording at BASE_CYCLES, measures how much noise the smoothing lets through against the tone it finds, and climbs CYCLES_LADDER until the predicted scatter of the readings is under target_db (Brandon, 2026-09-30: a 0.5 g sweep under a 2.3 g RMS random environment read with a 17 dB spread and a fifth of its points at zero at ten cycles; the lever is the smoothing, and the data can say how much). chosen records what the automatic picked, so the setting read back says what was used.

refine corrects the specification's sweep clock against the recording: the envelope's residual phase is fitted and the sweep argument moved by it, then solved again, so a controller whose sweep drifted from the commanded rate over a long run does not read low once the drift leaves the filter's bandwidth.

Frozen like the other settings, and for the same reason: the staleness fingerprint is the fields, and a mutable setting would be a fingerprint that lies.

Methods:

Name	Description
effective_cycles	The smoothing that applies: the one set, else the one the
describe	One line for a status bar.

Attributes:

Name	Type	Description
automatic	bool	Whether the smoothing is chosen from the data.

Attributes

automatic property

automatic: bool

Whether the smoothing is chosen from the data.

Methods:

effective_cycles

effective_cycles() -> float | None

The smoothing that applies: the one set, else the one the automatic chose, else None until an extraction has run.

Source code in src/visualdynamics/core/sine.py

def effective_cycles(self) -> float | None:
    """The smoothing that applies: the one set, else the one the
    automatic chose, else None until an extraction has run."""
    return self.cycles if self.cycles is not None else self.chosen

describe

describe() -> str

One line for a status bar.

Source code in src/visualdynamics/core/sine.py

def describe(self) -> str:
    """One line for a status bar."""
    if self.cycles is not None:
        return f'{self.cycles:g} cycles of smoothing'
    if self.chosen is not None:
        return (f'automatic smoothing, {self.chosen:g} cycles chosen '
                f'for {self.target_db:g} dB')
    return f'automatic smoothing for {self.target_db:g} dB'

Functions:

find_tone

find_tone(records: ndarray, dt: float, tone: SineTone, search: tuple[float, float] | None = None) -> float

Where a tone's sweep begins in a recording, by matched filter.

Correlates the reconstructed sweep template against every record and sums the correlation power across them — the processing gain over a whole sweep is what finds a tone under a random excitation much louder than it (measured: a clean find under +15.6 dB of random). search bounds the onset in seconds when the caller knows roughly where to look. Returns the onset in seconds.

Source code in src/visualdynamics/core/sine.py

def find_tone(records: np.ndarray, dt: float, tone: SineTone,
              search: tuple[float, float] | None = None) -> float:
    """Where a tone's sweep begins in a recording, by matched filter.

    Correlates the reconstructed sweep template against every record
    and sums the correlation power across them — the processing gain
    over a whole sweep is what finds a tone under a random excitation
    much louder than it (measured: a clean find under +15.6 dB of
    random). `search` bounds the onset in seconds when the caller
    knows roughly where to look. Returns the onset in seconds.
    """
    records = _records(records)
    score = _tone_score(records, dt, tone)
    if search is not None:
        lo = max(round(search[0] / dt), 0)
        hi = min(round(search[1] / dt) + 1, len(score))
        if lo >= hi:
            raise ValueError(f'{tone.name}: the search window '
                             f'{search} holds no onset candidates')
        return float((lo + int(np.argmax(score[lo:hi]))) * dt)
    return float(int(np.argmax(score)) * dt)

find_environment

find_environment(records: ndarray, dt: float, tones: Sequence[SineTone], ticker=None) -> float

Where the tones' shared clock starts, by joint matched filter.

Every tone in one environment begins at its own start_time on one clock, so there is one unknown — the clock's position in the recording — and every tone's correlation votes on it at its own lag. The sharp votes carry the ambiguous ones: a near-dwell or a log sweep that mis-locks alone (measured: half a second off in a four-tone mix) is pinned by the linear sweeps beside it. Returns the clock origin in seconds; tone i's sweep begins at origin + start_time_i.

Source code in src/visualdynamics/core/sine.py

def find_environment(records: np.ndarray, dt: float,
                     tones: Sequence[SineTone], ticker=None) -> float:
    """Where the tones' shared clock starts, by joint matched filter.

    Every tone in one environment begins at its own `start_time` on
    one clock, so there is one unknown — the clock's position in the
    recording — and every tone's correlation votes on it at its own
    lag. The sharp votes carry the ambiguous ones: a near-dwell or a
    log sweep that mis-locks alone (measured: half a second off in a
    four-tone mix) is pinned by the linear sweeps beside it. Returns
    the clock origin in seconds; tone i's sweep begins at
    `origin + start_time_i`.
    """
    records = _records(records)
    joint = None
    if ticker is not None:
        ticker.add(len(tones))
    for tone in tones:
        lag = round(tone.start_time / dt)
        score = _tone_score(records, dt, tone)
        if ticker is not None:
            ticker.tick()
        if len(score) <= lag:
            continue
        vote = score[lag:]
        # summed as it comes, one vote alive at a time (2026-10-01)
        if joint is None:
            joint = vote.copy()
        else:
            length = min(len(joint), len(vote))
            joint = joint[:length] + vote[:length]
    if joint is None:
        raise ValueError('no tone fits the recording at its own start '
                         'time; nothing to align')
    return float(int(np.argmax(joint)) * dt)

vold_kalman

vold_kalman(signal: ndarray, arguments: Sequence[ndarray], frequencies: Sequence[ndarray], starts: Sequence[int], dt: float, cycles: float = BASE_CYCLES, chunk: int | None = None) -> list[ndarray]

Every tone's complex envelope at every sample, solved jointly.

Solved in chunks (CHUNK samples, or chunk), each with a margin of MARGIN_WINDOWS smoothing windows of the slowest tone active at its edges solved beyond it and discarded: the penalty's memory is a few windows, so the interior is the whole-record answer and the memory is the chunk's, not the record's. _vold_kalman_whole is the solve itself, on one span; _solve_piece cuts one piece out and solves it, which the sampled reading uses too.

Source code in src/visualdynamics/core/sine.py

def vold_kalman(signal: np.ndarray, arguments: Sequence[np.ndarray],
                frequencies: Sequence[np.ndarray],
                starts: Sequence[int], dt: float,
                cycles: float = BASE_CYCLES,
                chunk: int | None = None) -> list[np.ndarray]:
    """Every tone's complex envelope at every sample, solved jointly.

    Solved in chunks (`CHUNK` samples, or `chunk`), each with a margin
    of `MARGIN_WINDOWS` smoothing windows of the slowest tone active
    at its edges solved beyond it and discarded: the penalty's memory
    is a few windows, so the interior is the whole-record answer and
    the memory is the chunk's, not the record's. `_vold_kalman_whole`
    is the solve itself, on one span; `_solve_piece` cuts one piece
    out and solves it, which the sampled reading uses too.
    """
    signal = np.asarray(signal, dtype=float)
    spans = _spans(arguments, starts, len(signal))
    lo = min(a for a, _b in spans)
    hi = max(b for _a, b in spans)
    if hi - lo <= 0:
        raise ValueError('no tone overlaps the recording')
    chunk = CHUNK if chunk is None else int(chunk)
    if hi - lo <= chunk:
        return _vold_kalman_whole(signal, arguments, frequencies, starts,
                                  dt, cycles)
    envelopes = [np.zeros(b - a, dtype=complex) for a, b in spans]
    start = lo
    while start < hi:
        end = min(start + chunk, hi)
        solved = _solve_piece(signal, arguments, frequencies, spans, dt,
                              cycles, start, end)
        for k, (a, _b) in enumerate(spans):
            piece = solved[k]
            if piece is None:
                continue
            entry, envelope = piece
            keep_lo = max(start, entry)
            keep_hi = min(end, entry + len(envelope))
            if keep_hi > keep_lo:
                envelopes[k][keep_lo - a:keep_hi - a] = \
                    envelope[keep_lo - entry:keep_hi - entry]
        start = end
    return envelopes

sample_levels

sample_levels(history: Any, specification: SineSweepSpecification, cycles: float = BASE_CYCLES, windows: int = SAMPLE_WINDOWS, tones: Sequence[str] | None = None, onsets: dict[str, float] | None = None, ticker: Any = None) -> list[dict[str, Any]]

The levels read at windows stretches of each tone, cheaply.

The sine view's preview and the automatic smoothing's measurement: rather than solve the whole record, solve one smoothing window's worth at each of windows evenly spaced points along each tone (with the chunked solve's margin around it), and read the amplitude and the noise floor at the center. Costs a few short solves per channel whatever the record's length, so it can run on every edit of the setting.

Returns, per wanted tone, a dict: 'tone', 'frequency' (the sampled points), 'seconds', 'amplitude' and 'floor' (channels × points, debiased the way the full extraction is), 'scatter_db' (channels × points, the predicted standard deviation of a reading at this smoothing, from the floor against the amplitude; infinite where the tone is under the floor), and 'cycles'.

Source code in src/visualdynamics/core/sine.py

def sample_levels(history: Any, specification: SineSweepSpecification,
                  cycles: float = BASE_CYCLES,
                  windows: int = SAMPLE_WINDOWS,
                  tones: Sequence[str] | None = None,
                  onsets: dict[str, float] | None = None,
                  ticker: Any = None) -> list[dict[str, Any]]:
    """The levels read at `windows` stretches of each tone, cheaply.

    The sine view's preview and the automatic smoothing's measurement:
    rather than solve the whole record, solve one smoothing window's
    worth at each of `windows` evenly spaced points along each tone
    (with the chunked solve's margin around it), and read the
    amplitude and the noise floor at the center. Costs a few short
    solves per channel whatever the record's length, so it can run on
    every edit of the setting.

    Returns, per wanted tone, a dict: 'tone', 'frequency' (the sampled
    points), 'seconds', 'amplitude' and 'floor' (channels × points,
    debiased the way the full extraction is), 'scatter_db' (channels ×
    points, the predicted standard deviation of a reading at this
    smoothing, from the floor against the amplitude; infinite where
    the tone is under the floor), and 'cycles'.
    """
    dt = _even_steps(np.asarray(history.abscissa, dtype=float), None)
    rows = _control_rows(history, specification)
    signals = [history.ordinate[row] for row in rows]
    wanted = (specification.tones if tones is None
              else [specification.tone(name) for name in tones])
    laid = _lay_tones(signals, specification, wanted, onsets, dt, cycles,
                      ticker)
    centers, pieces = [], []
    for placed in laid:
        first = int(placed.window_at(0) / 2.0)
        last = placed.n - 1 - int(placed.window_at(placed.n - 1) / 2.0)
        count = max(1, min(windows, placed.n))
        chosen = (np.linspace(first, last, count).astype(np.int64)
                  if last > first else np.array([placed.n // 2]))
        centers.append(chosen)
        own = []
        for c in chosen:
            half = int(placed.window_at(c) / 2.0) + 1
            own.append((placed.start + c - half, placed.start + c + half + 1))
        pieces.append(own)
    amplitude, floor, below, _slopes = _read_grouped(
        signals, laid, dt, cycles, centers, pieces_by_tone=pieces,
        ticker=ticker)
    out = []
    for k, placed in enumerate(laid):
        seconds = placed.seconds_at(centers[k])
        # a reading under its floor is nothing seen; a weak reading that
        # stood above it is still a reading, and says its own scatter
        magnitude = np.where(below[k], 0.0, np.abs(amplitude[k]))
        out.append({'tone': placed.tone.name,
                    'frequency': placed.tone.frequency_at(seconds),
                    'seconds': placed.onset + centers[k] * dt,
                    'amplitude': magnitude,
                    'floor': floor[k],
                    'scatter_db': scatter_db(magnitude, floor[k]),
                    'cycles': float(cycles)})
    return out

scatter_db

scatter_db(amplitude: ndarray, floor: ndarray) -> ndarray

The predicted standard deviation of a reading, in dB, from the debiased amplitude and the noise floor beside it.

The envelope is the tone plus a complex noise of power floor²; half of that power lies along the tone, so the amplitude's standard deviation is floor/√2, and in decibels of the amplitude 8.686 times their ratio. Infinite where the tone is under the floor — there is no reading to scatter around.

Source code in src/visualdynamics/core/sine.py

def scatter_db(amplitude: np.ndarray, floor: np.ndarray) -> np.ndarray:
    """The predicted standard deviation of a reading, in dB, from the
    debiased amplitude and the noise floor beside it.

    The envelope is the tone plus a complex noise of power floor²;
    half of that power lies along the tone, so the amplitude's
    standard deviation is floor/√2, and in decibels of the amplitude
    8.686 times their ratio. Infinite where the tone is under the
    floor — there is no reading to scatter around.
    """
    amplitude = np.asarray(amplitude, dtype=float)
    floor = np.asarray(floor, dtype=float)
    with np.errstate(divide='ignore', invalid='ignore'):
        ratio = np.where(amplitude > 0.0, floor / np.sqrt(2.0)
                         / np.maximum(amplitude, 1e-300), np.inf)
    return 20.0 / np.log(10.0) * ratio

suggest_cycles

suggest_cycles(history: Any, specification: SineSweepSpecification, target_db: float = TARGET_SCATTER_DB, tones: Sequence[str] | None = None, onsets: dict[str, float] | None = None, ticker: Any = None) -> float

The smoothing the data asks for: the first rung of CYCLES_LADDER at which the predicted scatter of the readings is under target_db (one standard deviation).

Measured at BASE_CYCLES with sample_levels, where the noise the smoothing lets through is read beside the tone it finds; the scatter falls as the square root of the smoothing, so the rung follows from one measurement. The median over the sampled points and channels speaks for a tone — a resonance or a dropout should not set the smoothing for a whole sweep — and the widest-asking tone speaks for the recording, since one setting rides it. The ladder is capped where a tone's smoothing window at its lowest frequency would reach a quarter of its own span: past that the reading is the sweep's mean, not a level along it. A tone under the floor at the base smoothing asks for the top of the ladder.

Source code in src/visualdynamics/core/sine.py

def suggest_cycles(history: Any, specification: SineSweepSpecification,
                   target_db: float = TARGET_SCATTER_DB,
                   tones: Sequence[str] | None = None,
                   onsets: dict[str, float] | None = None,
                   ticker: Any = None) -> float:
    """The smoothing the data asks for: the first rung of
    `CYCLES_LADDER` at which the predicted scatter of the readings is
    under `target_db` (one standard deviation).

    Measured at `BASE_CYCLES` with `sample_levels`, where the noise
    the smoothing lets through is read beside the tone it finds; the
    scatter falls as the square root of the smoothing, so the rung
    follows from one measurement. The median over the sampled points
    and channels speaks for a tone — a resonance or a dropout should
    not set the smoothing for a whole sweep — and the widest-asking
    tone speaks for the recording, since one setting rides it. The
    ladder is capped where a tone's smoothing window at its lowest
    frequency would reach a quarter of its own span: past that the
    reading is the sweep's mean, not a level along it. A tone under
    the floor at the base smoothing asks for the top of the ladder.
    """
    sampled = sample_levels(history, specification, BASE_CYCLES,
                            tones=tones, onsets=onsets, ticker=ticker)
    wanted = (specification.tones if tones is None
              else [specification.tone(name) for name in tones])
    asked = BASE_CYCLES
    cap = CYCLES_LADDER[-1]
    for reading, tone in zip(sampled, wanted):
        scatter = reading['scatter_db']
        finite = scatter[np.isfinite(scatter)]
        # the median reading under the floor: the tone was not seen
        # at the base smoothing, and only the longest smoothing can say
        if finite.size < scatter.size / 2.0:
            need = cap
        else:
            # the square-root law, with room: the scatter in decibels
            # grows faster than the noise-to-tone ratio once the noise
            # is a fair fraction of the tone, and the rung the ratio
            # alone chose for a 1.39 dB reading measured 1.25 dB
            # (2026-09-30); half again on the smoothing brings it under
            need = (1.5 * BASE_CYCLES
                    * (float(np.median(finite)) / target_db) ** 2)
        asked = max(asked, need)
        lowest = max(float(np.min(tone.frequency)), 1e-9)
        cap = min(cap, lowest * tone.duration() / 4.0)
    rung = next((c for c in CYCLES_LADDER if c >= asked), CYCLES_LADDER[-1])
    return float(max(BASE_CYCLES, min(rung, cap)))

extract_sine

extract_sine(history: Any, specification: SineSweepSpecification, tones: Sequence[str] | None = None, onsets: dict[str, float] | None = None, cycles: float | None = None, points_per_window: float = 2.0, refine: bool = True, target_db: float = TARGET_SCATTER_DB, workers: int | None = None, progress: Callable[[int, int], None] | None = None) -> SineLevelSet

Read each tone's level out of a recording, against its own sweep.

Reconstruct every wanted tone's sweep from the specification's breakpoints, find where the environment's clock begins in the recording (joint matched filter; onsets overrides per tone name, and the found value rides the result as .onset), then solve for every tone's complex envelope on every control channel at once with the second-order Vold-Kalman filter (vold_kalman): the record modeled as the sum of the tones on their known sweeps, each envelope held to a slow curve over cycles cycles of its own instantaneous frequency. Joint, so crossing sweeps are separated by their frequency histories rather than each reading the other as noise — the documented limit of the tracking demodulation this replaced (2026-09-03). Solved one chunk of the record at a time (_read_levels), so the memory is a chunk's whatever the record's length (2026-10-01).

cycles None is automatic (suggest_cycles): the smoothing is climbed until the predicted scatter of the readings is under target_db, measured from the recording itself. The smoothing used rides the result as SineLevelSet.cycles.

The magnitude is debiased: a noisy envelope's magnitude reads high, so the noise power left in the residual around each reading, scaled by the filter's equivalent averaging length, is subtracted from the squared magnitude before the square root — a planted amplitude under 4x its own RMS of noise reads back within a fraction of a dB. Where the tone is under that noise the reading is reported at the floor and flagged (SineLevel.below_floor) rather than at zero. The controller's own live tracker carries the raw bias, which is worth remembering when the two are compared.

With refine, the sweep clock is checked against the recording: each tone's envelope phase slope is fitted chunk by chunk (_clock_correction) and, where the implied frequency offset reaches REFINE_FRACTION of the filter's bandwidth at either end of the sweep, the argument is moved by it and the solve repeated, up to three times. The offset applied at the end of each tone's sweep rides the result as SineLevel.drift_hz, zero when none was needed.

workers is how many processes the pieces are spread over (Brandon, 2026-10-01: use the cores): None picks the cores up to MAX_WORKERS, and spreads only when the work is more than PARALLEL_FLOOR samples times channels, since the workers take a second each to start; 1 does everything here. The numbers are the same either way — each piece and channel is one task, and a task is the serial code.

The envelope is read every window/points_per_window along the sweep, one (almost) independent reading each. Returns a SineLevelSet — one object, one SineLevel per tone inside, frequencies ascending whichever way the tone swept, each line stamped with the second it was measured. A recording that ends before a tone does yields the lines it reached — the coverage the comparison reports.

Source code in src/visualdynamics/core/sine.py

def extract_sine(history: Any, specification: SineSweepSpecification,
                 tones: Sequence[str] | None = None,
                 onsets: dict[str, float] | None = None,
                 cycles: float | None = None,
                 points_per_window: float = 2.0,
                 refine: bool = True,
                 target_db: float = TARGET_SCATTER_DB,
                 workers: int | None = None,
                 progress: Callable[[int, int], None] | None = None
                 ) -> SineLevelSet:
    """Read each tone's level out of a recording, against its own sweep.

    Reconstruct every wanted tone's sweep from the specification's
    breakpoints, find where the environment's clock begins in the
    recording (joint matched filter; `onsets` overrides per tone name,
    and the found value rides the result as `.onset`), then solve for
    every tone's complex envelope on every control channel at once
    with the second-order **Vold-Kalman filter** (`vold_kalman`): the
    record modeled as the sum of the tones on their known sweeps, each
    envelope held to a slow curve over `cycles` cycles of its own
    instantaneous frequency. Joint, so crossing sweeps are separated
    by their frequency histories rather than each reading the other
    as noise — the documented limit of the tracking demodulation this
    replaced (2026-09-03). Solved one chunk of the record at a time
    (`_read_levels`), so the memory is a chunk's whatever the record's
    length (2026-10-01).

    `cycles` None is **automatic** (`suggest_cycles`): the smoothing
    is climbed until the predicted scatter of the readings is under
    `target_db`, measured from the recording itself. The smoothing
    used rides the result as `SineLevelSet.cycles`.

    The magnitude is *debiased*: a noisy envelope's magnitude reads
    high, so the noise power left in the residual around each
    reading, scaled by the filter's equivalent averaging length, is
    subtracted from the squared magnitude before the square root — a
    planted amplitude under 4x its own RMS of noise reads back within
    a fraction of a dB. Where the tone is under that noise the reading
    is reported at the floor and flagged (`SineLevel.below_floor`)
    rather than at zero. The controller's own live tracker carries the
    raw bias, which is worth remembering when the two are compared.

    With `refine`, the sweep clock is checked against the recording:
    each tone's envelope phase slope is fitted chunk by chunk
    (`_clock_correction`) and, where the implied frequency offset
    reaches `REFINE_FRACTION` of the filter's bandwidth at either end
    of the sweep, the argument is moved by it and the solve repeated,
    up to three times. The offset applied at the end of each tone's
    sweep rides the result as `SineLevel.drift_hz`, zero when none was
    needed.

    `workers` is how many processes the pieces are spread over
    (Brandon, 2026-10-01: use the cores): None picks the cores up to
    `MAX_WORKERS`, and spreads only when the work is more than
    `PARALLEL_FLOOR` samples times channels, since the workers take a
    second each to start; 1 does everything here. The numbers are the
    same either way — each piece and channel is one task, and a task
    is the serial code.

    The envelope is read every window/`points_per_window` along the
    sweep, one (almost) independent reading each. Returns a
    `SineLevelSet` — one object, one `SineLevel` per tone inside,
    frequencies ascending whichever way the tone swept, each line
    stamped with the second it was measured. A recording that ends
    before a tone does yields the lines it reached — the coverage the
    comparison reports.
    """
    dt = _even_steps(np.asarray(history.abscissa, dtype=float), None)
    rows = _control_rows(history, specification)
    signals = [history.ordinate[row] for row in rows]
    ticker = Ticker(progress)
    if cycles is None:
        cycles = suggest_cycles(history, specification, target_db,
                                tones=tones, onsets=onsets, ticker=ticker)
    cycles = float(cycles)
    wanted = (specification.tones if tones is None
              else [specification.tone(name) for name in tones])
    laid = _lay_tones(signals, specification, wanted, onsets, dt, cycles,
                      ticker)
    centers = [_centers(placed, points_per_window) for placed in laid]

    span = max(p.end for p in laid) - min(p.start for p in laid)
    count = _workers(workers)
    if workers is None and span * len(rows) < PARALLEL_FLOOR:
        count = 1
    pool = None
    if count > 1:
        import multiprocessing
        from concurrent.futures import ProcessPoolExecutor

        # spawned, on every platform: a forked worker inherits the
        # parent's BLAS threads and locks and can hang in them
        pool = ProcessPoolExecutor(
            max_workers=count, mp_context=multiprocessing.get_context('spawn'))
    try:
        amplitude, floor, below, _slopes = _passes(
            signals, laid, dt, cycles, centers, refine, pool, ticker)
    finally:
        if pool is not None:
            pool.shutdown(wait=True)
    out = []
    for k, placed in enumerate(laid):
        frequencies = placed.tone.frequency_at(placed.seconds_at(centers[k]))
        seconds = placed.onset + centers[k] * dt
        order = np.argsort(frequencies)
        dims = [history.ordinate_dim[row] for row in rows]
        units = [history.ordinate_unit[row] for row in rows]
        out.append(SineLevel(
            abscissa=frequencies[order], ordinate=amplitude[k][:, order],
            response_dof=list(specification.response_dof),
            ordinate_dim=dims, ordinate_unit=units,
            comment=[f'{placed.tone.name} at {dof}'
                     for dof in specification.response_dof],
            tone=placed.tone.name, onset=float(placed.onset),
            seconds=seconds[order],
            floor=floor[k][:, order],
            below_floor=below[k][:, order],
            drift_hz=placed.drift_hz))
    return SineLevelSet(out, cycles=cycles)