Signal Synthesis

Parameterized synthetic waveforms as numpy arrays or WaveformData

Synthetic signal generation: parameterized waveforms as numpy arrays or WaveformData.

Public API for users who want synthetic data for testing and analysis, and the engine behind MockConnection's state-coupled waveform synthesis. Kinds live in a dispatch table of generator functions -- adding a kind is one new generator plus docs and tests; the mock coupling and code-conversion layers are kind-agnostic.

SignalSpec dataclass

SignalSpec(kind: str = 'sine', frequency: float = 1000.0, amplitude: float = 1.0, offset: float = 0.0, phase: float = 0.0, duty: float = 0.5, noise_rms: float = 0.0, seed: Optional[int] = None, drift_amplitude: float = 0.0, drift_frequency: float = 0.1, glitch_rate: float = 0.0, glitch_amplitude: float = 0.0, ringing_frequency: float = 0.0, ringing_damping: float = 5000.0, end_frequency: float = 10000.0, sweep_time: float = 0.01, sweep_log: bool = False, tau: float = 0.0001, pulse_width: float = 0.0002, edge_time: float = 1e-05, harmonics: Tuple[float, ...] = (0.1, 0.05), jitter_rms: float = 0.0, jitter_seed: Optional[int] = None, clip_level: float = 0.0, clip_softness: float = 0.0, distortion_h2: float = 0.0, distortion_h3: float = 0.0)

Parameters of one synthetic signal.

Attributes:

Name	Type	Description
kind	str	One of "sine", "square", "triangle", "ramp", "dc", "noise", "chirp", "exponential", "pulse", "multitone".
frequency	float	Repetition rate in Hz (periodic kinds only).
amplitude	float	Peak amplitude in volts (Vpp = 2*amplitude); for "noise", the standard deviation. Ignored for "dc".
offset	float	DC offset in volts, added to every kind ("dc" outputs exactly this level).
phase	float	Phase in radians (periodic kinds only).
duty	float	High fraction of a "square" period, 0 < duty < 1 (pulse/PWM).
noise_rms	float	Std-dev of additive Gaussian noise laid on any kind.
seed	Optional[int]	None for fresh randomness per call; an int for reproducibility.
drift_amplitude	float	Volts of slow baseline wander (0 = off).
drift_frequency	float	Hz of that wander; only used when drift_amplitude > 0.
glitch_rate	float	Mean glitches per second (0 = off).
glitch_amplitude	float	Volts, peak height of a glitch.
ringing_frequency	float	Hz of post-edge oscillation (0 = off). Ringing is an EDGE impairment: it is PHYSICALLY meaningful on kinds with fast edges ("square", "pulse", or a pulse-like "ramp"). It is not, however, a no-op elsewhere -- edges are found as any nonzero sample-to-sample change, not only as a discontinuity, so on a continuous kind ("sine", "chirp", "exponential", "multitone") it acts as a derivative-weighted filter whose magnitude scales with the signal's slew rate: measurable, but usually small. Only "dc", whose sample-to-sample differences are all zero, is a true no-op.
ringing_damping	float	Decay rate per second of that oscillation; only used when ringing_frequency > 0. Defaults away from 0 for the same reason drift_frequency does: undamped ringing (decay rate 0) never actually decays, so the kernel would run for the entire buffer on every edge -- quadratic in the number of edges once ringing_frequency is switched on the most natural way, by setting only that field.
end_frequency	float	"chirp" sweep stop frequency in Hz.
sweep_time	float	"chirp" seconds per sweep, after which it retraces.
sweep_log	bool	"chirp" sweeps logarithmically rather than linearly.
tau	float	"exponential" RC time constant in seconds.
pulse_width	float	"pulse" 50%-to-50% width in seconds (FWHM), matching the instrument convention and the threshold the repo's timing analyzer measures at. The flat top therefore runs for pulse_width - edge_time. "pulse" ignores duty.
edge_time	float	"pulse" 0-to-100% transition time in seconds; 0 gives an ideal instantaneous edge.
harmonics	Tuple[float, ...]	"multitone" relative amplitudes of the 2nd, 3rd, ... harmonic. amplitude is the FUNDAMENTAL's amplitude, so the peak of the sum is higher; that is deliberate, since normalizing would make the THD of a multitone depend on its harmonic set.
jitter_rms	float	Std-dev of period-to-period timing jitter, in seconds (0 = off). PERIODIC_KINDS only (sine, square, triangle, ramp, multitone, exponential, pulse): "dc"/"noise" have no cycle structure and "chirp" has no stable period, so on those three kinds this field is a no-op rather than an error. Each cycle BOUNDARY gets its own independent Gaussian time-shift, and every sample's shift is the LINEAR INTERPOLATION between the two boundaries straddling it -- a continuous warp, not a per-cycle hard step -- so an enabled ringing_frequency keys on the actual jittered edge, not the nominal one, with no special-casing needed. What you actually measure back depends on WHERE your kind's measurable edge sits within its cycle, because the interpolation blends less of the "wrong" neighbor the closer the edge sits to a boundary. Let f be that edge's fractional position in [0, 1) (f=0 is the cycle boundary itself; _cycle_fraction computes the same fraction for the underlying generator): - An edge AT the cycle boundary (f=0 -- the ascending v50 crossing of "sine", "square", "multitone", and "pulse" (whose edge sits within edge_time/2 of the boundary) at their default phase=0.0) gets the full, unblended shift from the boundary on each side, so period[n] = T + delta[n+1] - delta[n] exactly as a per-cycle-constant model would give: measures jitter_rms almost exactly (verified empirically: ratio 0.94-1.08 across trials at 1-5% of the period). - An edge at some other in-cycle fraction f gets a BLENDED shift and measures LESS. Two verified examples: "triangle"'s ascending crossing sits at f=0.25 (ratio measured ~0.66-0.68, close to the 0.6614 the formula below predicts); "ramp"'s zero crossing -- despite the ramp's own discontinuity sitting at the boundary -- is itself a continuous ramp, so its v50 crossing actually sits at f=0.5, cycle CENTER, the point of MAXIMUM attenuation (ratio measured ~0.50-0.52, matching the formula's minimum of 0.5 almost exactly). Both are consistent with Var(period)/jitter_rms**2 = 0.5*[(1-2f)**2 + f**2 + (1-f)**2], which is 1.0 at f=0 and a minimum of 0.25 (ratio 0.5) at f=0.5. The internal per-boundary draw uses sigma = jitter_rms / sqrt(2), which is what keeps the common f=0 case exactly calibrated. See docs/superpowers/specs/2026-08-28-signal-timing-jitter-design.md for the full derivation. stream() chunk-boundary continuity: a cycle straddling a stream() chunk boundary now draws the SAME deltas regardless of which chunk's call renders it -- see jitter_seed below and docs/superpowers/specs/2026-08-28-jitter-stream-continuity-design.md.
jitter_seed	Optional[int]	Decouples jitter's per-boundary randomness from seed. None (the default) falls back to seed, so a one-shot synthesize()/make_waveform() call behaves exactly as if this field didn't exist. stream() auto-fills it with the ORIGINAL, pre-chunk-bump seed on every chunk whose spec left it None -- seed itself keeps bumping per chunk (so noise/glitches keep varying chunk-to-chunk, unaffected), but every chunk of one stream() run now shares the same jitter entropy source, which is what makes a cycle straddling a chunk boundary resolve to identical boundary deltas no matter which chunk renders it. Set explicitly on a spec passed to stream() to opt out of the auto-fill and choose jitter's entropy source independently of seed.
clip_level	float	Volts, symmetric clipping/saturation threshold (0 = off). Models a non-linear output stage (an amplifier or probe front-end driven into its rails) clipping whatever signal reaches it, so -- unlike every kind-specific field above -- it is KIND-AGNOSTIC, same as noise_rms/drift_amplitude/glitch_rate: it applies equally to "sine", "dc", a jittered/ringing edge, or anything else this module can produce. It is also the LAST impairment synthesize() applies, after drift, glitches, and noise, so a noisy sample that happens to land past the rail gets flattened too, exactly as a real saturating stage would clip the noise riding on its input -- deliberate, not a bug. See clip_softness for the shape of the clip curve.
clip_softness	float	0 = hard clip (an exact flat top at ±clip_level, np.clip); blends toward clip_level * tanh(v / clip_level) soft saturation as it approaches 1.0, which is pure tanh. The blend is linear in the two curves' own output, not in some other parameterization: clip_level * ((1 - clip_softness) * hard + clip_softness * soft). Must be in [0, 1]; only used when clip_level > 0.
distortion_h2	float	Fraction of amplitude added as 2nd-harmonic content via Chebyshev waveshaping (0 = off). Uses the CHEBYSHEV polynomial T_2(u) = 2*u**2 - 1, not a naive u**2 term, because of the identity T_n(cos(theta)) = cos(n*theta): fed a pure sinusoid u = sin(theta), T_2 produces EXACTLY distortion_h2 fraction of clean 2nd-harmonic content with no leakage into DC or the 3rd harmonic. A naive u**2 polynomial does not have that property -- sin(theta)**2 == 0.5*(1 - cos(2*theta)) -- so a "2nd harmonic" built that way would drag a DC offset along with it, a measurable-but-wrong artifact this module works hard to avoid elsewhere (see _multitone's coherent harmonic series, built the same way for the same reason: THD analysis needs a signal with a known-correct answer). It is KIND-AGNOSTIC and applied to samples as they stand after drift/glitches/noise, same as clip_level -- so those impairments' contribution gets the same nonlinear coloring a real amplifier stage would give them -- but it runs BEFORE clip_level: a real non-linear gain stage's waveshaping happens upstream of a separate rail-limiting stage, so distortion is free to push samples further from zero and clip is what actually enforces the rails on the result. Normalized against samples - offset, NOT samples -- offset is a static bias, not part of the oscillating carrier the Chebyshev identity above is about, so a nonzero offset does not disturb the "EXACTLY clean" guarantee (a naive normalization against raw samples would leak offset into the fundamental and even into the other harmonic's bin). "Pure harmonic content" in the strict Fourier sense is only exact when samples is a pure sinusoid plus offset ("sine", any offset); on any other kind (or with drift/glitches/noise enabled) the same waveshaper still applies to whatever samples holds, coloring that kind's own harmonic content rather than producing a textbook 2nd harmonic. Requires amplitude != 0 (see _validate): the waveshaper normalizes by amplitude before applying T_2/T_3, so a zero amplitude would divide by zero -- this matters because e.g. kind="dc" legitimately ignores amplitude entirely and callers routinely set it to 0 there.
distortion_h3	float	Fraction of amplitude added as 3rd-harmonic content via Chebyshev waveshaping (0 = off), using T_3(u) = 4*u**3 - 3*u for the same reason as distortion_h2 -- see that entry for the full derivation and the naive-polynomial cross-leakage this avoids. distortion_h2 and distortion_h3 are independent and additive: T_2 and T_3 applied to the same pure sinusoid land in different, non-overlapping FFT bins (2*frequency and 3*frequency respectively, with nothing at DC or at each other's bin), so enabling both together colors the signal with both harmonics and no cross-talk between them.

SuperposedSignal dataclass

SuperposedSignal(components: Tuple[SignalSpec, ...], dut: Optional[Any] = None)

Two or more independently-synthesized signals summed into one trace.

Each component keeps its own full SignalSpec -- kind, impairments, seed -- and is synthesized in isolation; the combined waveform is their plain elementwise sum, with no state shared between components. Useful for modeling e.g. a tone riding on an independently-seeded noise floor, or two unrelated tones summed onto one channel.

Attributes:

Name	Type	Description
components	Tuple[SignalSpec, ...]	Two or more SignalSpecs to sum. synthesize_combined() and make_waveform_combined() dispatch each one exactly as synthesize()/make_waveform() would on its own, so a bad component parameter surfaces as the same InvalidParameterError it always has.
dut	Optional[Any]	Optional device-under-test model (e.g. dut.RCLowPass), applied to the SUMMED signal rather than to any one component. Mirrors connection/mock/loopback.py's AwgLoopback.dut: stored here so a caller can carry a DUT alongside a signal source, but only connection/mock/synth.py's raw_volts actually applies it (it is the only layer that knows the sample rate and can render the filter's lead-in).

synthesize

synthesize(spec: SignalSpec, sample_rate: float, n_points: int, t0: float = 0.0) -> np.ndarray

Generate voltage samples for a signal spec.

Parameters:

Name	Type	Description	Default
spec	SignalSpec	Signal parameters.	required
sample_rate	float	Samples per second.	required
n_points	int	Number of samples.	required
t0	float	Time of the first sample in seconds (shifts periodic signals).	0.0

Returns:

Type	Description
ndarray	float64 voltage array of length n_points.

Source code in scpi_control/signal_synth.py

def synthesize(spec: SignalSpec, sample_rate: float, n_points: int, t0: float = 0.0) -> np.ndarray:
    """Generate voltage samples for a signal spec.

    Args:
        spec: Signal parameters.
        sample_rate: Samples per second.
        n_points: Number of samples.
        t0: Time of the first sample in seconds (shifts periodic signals).

    Returns:
        float64 voltage array of length n_points.
    """
    _validate(spec, sample_rate, n_points)
    rng = np.random.default_rng(spec.seed)
    # `t` stays the NOMINAL time array all the way through this function --
    # drift below is deliberately a function of absolute time, unwarped. Jitter
    # (_apply_jitter) is applied only at the two _GENERATORS[...] call sites
    # below (the plain one and, inside the ringing branch, the extended-window
    # one), so ringing's own edge detection sees the jittered edge position,
    # not the nominal one.
    t = t0 + np.arange(n_points) / sample_rate
    if spec.ringing_frequency > 0:
        # A damped sinusoid triggered at each edge. Real probe/scope front-ends
        # ring after a fast transition; this is what gives overshoot/preshoot
        # measurements something real to measure. Applied to the base signal
        # BEFORE drift and glitches: ringing is part of the signal's own edge
        # response, not a baseline wander or an additive event.
        #
        # I3: this must be a function of ABSOLUTE TIME, like drift, not of the
        # current buffer alone -- np.diff() cannot see an edge across a
        # stream() chunk boundary, so an edge landing right at a boundary used
        # to get no ringing at all, and one near a chunk's end had its ring
        # truncated. Fixed the same way drift is continuous: render
        # `decay_len` extra samples BEFORE t0, detect edges (and let their
        # ringing spill forward) across that whole extended window, then slice
        # the prepended samples back off. The generator is called exactly
        # once (over the extended window) rather than once for the plain
        # buffer and again for the extended one, so a stochastic kind (e.g.
        # "noise") does not draw from `rng` twice.
        decay_len = min(_MAX_RINGING_KERNEL_SAMPLES, max(1, int(sample_rate / max(spec.ringing_damping, 1e-9) * 5)))
        # Built as t0 + (index - decay_len) / sample_rate, NOT (t0 - decay_len /
        # sample_rate) + index / sample_rate -- the two are mathematically equal
        # but round differently in float64. With the latter, this chunk's t0
        # (itself computed elsewhere as start_time + produced / sample_rate) and
        # this expression's own "t0 - decay_len/sample_rate" partial sum
        # accumulate rounding error differently than a neighboring chunk's
        # equivalent sample does, so the same absolute instant can land a few
        # ULP apart depending on which chunk computed it -- enough to flip which
        # side of a razor's-edge comparison (e.g. square wave's `< duty`) a
        # sample falls on, right when frequency and chunk_size divide evenly
        # (verified: this happened at every chunk boundary in the drift-style
        # continuity test below until reordered this way).
        t_ext = t0 + (np.arange(n_points + decay_len) - decay_len) / sample_rate
        samples_ext = _GENERATORS[spec.kind](spec, _apply_jitter(spec, t_ext), rng) + spec.offset
        # ANY nonzero sample-to-sample change is an edge here, not just a
        # discontinuity: on a continuous kind every sample qualifies, so this
        # becomes a derivative-weighted filter rather than a no-op. That is
        # defensible as a band-limited edge response and is documented as such
        # on SignalSpec.ringing_frequency -- it is not a special case to strip.
        edges = np.flatnonzero(np.diff(samples_ext))
        if edges.size:
            # 5 time constants of decay, in samples. `max(spec.ringing_damping, 1e-9)`
            # only guards the division against damping == 0 (undamped ringing) --
            # it must NOT clamp small-but-nonzero damping up to some larger floor,
            # or slow decay would be truncated before it actually decays, showing
            # up as a discontinuity where the kernel window ends. `max(1, ...)`
            # then guards int() truncating to 0 for very heavy damping.
            # `_MAX_RINGING_KERNEL_SAMPLES` is a defensive backstop, not the
            # normal control on cost -- M8 gives ringing_damping a sensible
            # nonzero default so ordinary use never approaches it; it only
            # bounds the (user-opted-into) case of an explicit near-zero
            # damping, which would otherwise make the kernel unboundedly long.
            total = n_points + decay_len
            tail = np.arange(decay_len) / sample_rate
            kernel = np.sin(2 * np.pi * spec.ringing_frequency * tail) * np.exp(-spec.ringing_damping * tail)
            response = np.zeros(total)
            for edge in edges:
                step = samples_ext[edge + 1] - samples_ext[edge]
                end = min(total, edge + 1 + decay_len)
                response[edge + 1 : end] += 0.5 * step * kernel[: end - edge - 1]
            samples = samples_ext[decay_len:] + response[decay_len:]
        else:
            samples = samples_ext[decay_len:]
    else:
        samples = _GENERATORS[spec.kind](spec, _apply_jitter(spec, t), rng) + spec.offset
    if spec.drift_amplitude > 0:
        # Time-based, NOT a random walk: stream() re-seeds per chunk, so a walk
        # would reset at every chunk boundary and a live view would show sawtooth
        # jumps. Deriving drift from absolute time keeps it continuous for free.
        samples = samples + spec.drift_amplitude * np.sin(2 * np.pi * spec.drift_frequency * t)
    if spec.glitch_rate > 0 and spec.glitch_amplitude > 0:
        glitch_rng = _impairment_rng(spec.seed, 1)
        expected = spec.glitch_rate * n_points / sample_rate
        count = glitch_rng.poisson(expected)
        if count:
            positions = glitch_rng.integers(0, n_points, size=count)
            signs = glitch_rng.choice(np.array([-1.0, 1.0]), size=count)
            samples = samples.copy()
            # np.add.at, NOT samples[positions] += ... -- fancy-index += applies
            # only ONCE per repeated index, so duplicate glitch positions would be
            # silently dropped and the glitch rate would come out low.
            np.add.at(samples, positions, signs * spec.glitch_amplitude)
    if spec.noise_rms > 0:
        samples = samples + rng.normal(0.0, spec.noise_rms, n_points)
    if spec.distortion_h2 > 0 or spec.distortion_h3 > 0:
        # Chebyshev waveshaping, deliberately NOT a naive u**2/u**3 polynomial:
        # T_n(cos(theta)) == cos(n*theta), so feeding a pure sinusoid through
        # T_2/T_3 yields EXACTLY distortion_h2/distortion_h3 fraction of clean
        # 2nd/3rd harmonic with no cross-leakage into DC or into each other --
        # sin(theta)**2, by contrast, is 0.5*(1 - cos(2*theta)), so a naive
        # square term would drag a DC offset along with the "2nd harmonic" it
        # was meant to add. Applied to `samples` as they stand after
        # drift/glitches/noise -- kind-agnostic, same as clip_level below, so
        # whatever reaches this stage gets the same nonlinear coloring a real
        # amplifier stage would give it -- but deliberately BEFORE clip_level:
        # a real non-linear gain stage's waveshaping happens upstream of a
        # separate rail-limiting stage, so this is free to push samples
        # further from zero and clip is what actually enforces the rails on
        # the result. Kind-agnostic in the same sense clip_level is: it
        # applies to any kind's `samples`, not just periodic ones, though
        # "pure harmonic content" in the strict Fourier sense is only exact
        # for a pure sinusoid. See SignalSpec.distortion_h2/distortion_h3.
        #
        # Normalized against (samples - offset), NOT samples: offset is a
        # static bias baked into `samples` by the generator step above, not
        # part of the oscillating carrier the Chebyshev identity is about. Left
        # in, a nonzero offset shifts u away from cos(theta) and the "EXACTLY
        # clean" guarantee above silently breaks -- verified numerically: a 2 V
        # sine with offset=0.5 and distortion_h2=0.3 came out with a 2.6 V
        # fundamental (not 2.0 V) and, with distortion_h3 alone, a spurious
        # nonzero 2nd-harmonic bin that shouldn't exist at all. Subtracting
        # spec.offset here (and it alone -- not drift/glitches/noise, which
        # SHOULD get colored per the "whatever reaches this stage" reasoning
        # above) restores the exact-harmonic guarantee for the common
        # sine+offset case. Each term is guarded behind its own field so a
        # single-harmonic call (the common case, and both gallery demos) does
        # not pay for computing the unused one.
        u = (samples - spec.offset) / spec.amplitude
        correction = 0.0
        if spec.distortion_h2 > 0:
            correction = correction + spec.distortion_h2 * (2.0 * u**2 - 1.0)
        if spec.distortion_h3 > 0:
            correction = correction + spec.distortion_h3 * (4.0 * u**3 - 3.0 * u)
        samples = samples + spec.amplitude * correction
    if spec.clip_level > 0:
        # LAST impairment, deliberately: a real saturating output stage clips
        # whatever reaches it, drift/glitches/noise included -- so noise near
        # the rail gets visibly flattened too. See SignalSpec.clip_level.
        # clip_softness=0 -> exact np.clip (flat-topped hard clip);
        # clip_softness=1 -> exact clip_level*tanh(u) soft saturation;
        # anything between is the linear blend of those two curves' outputs.
        u = samples / spec.clip_level
        hard = np.clip(u, -1.0, 1.0)
        soft = np.tanh(u)
        samples = spec.clip_level * ((1.0 - spec.clip_softness) * hard + spec.clip_softness * soft)
    return samples

synthesize_combined

synthesize_combined(signal: SuperposedSignal, sample_rate: float, n_points: int, t0: float = 0.0) -> np.ndarray

Generate voltage samples for a SuperposedSignal: the sum of its components.

Each component is synthesized independently -- via synthesize(), so it keeps its own full impairments (noise, drift, glitches, jitter, seed) -- and the results are summed elementwise. No state is shared between components. A bad component parameter is caught by synthesize()'s own validate(), the same way it always is; the SuperposedSignal itself was already validated at construction (SuperposedSignal.__post_init_).

Parameters:

Name	Type	Description	Default
signal	SuperposedSignal	The components to sum.	required
sample_rate	float	Samples per second.	required
n_points	int	Number of samples.	required
t0	float	Time of the first sample in seconds (shifts periodic signals).	0.0

Returns:

Type	Description
ndarray	float64 voltage array of length n_points.

Source code in scpi_control/signal_synth.py

def synthesize_combined(signal: SuperposedSignal, sample_rate: float, n_points: int, t0: float = 0.0) -> np.ndarray:
    """Generate voltage samples for a SuperposedSignal: the sum of its components.

    Each component is synthesized independently -- via synthesize(), so it
    keeps its own full impairments (noise, drift, glitches, jitter, seed) -- and
    the results are summed elementwise. No state is shared between components.
    A bad component parameter is caught by synthesize()'s own _validate(), the
    same way it always is; the SuperposedSignal itself was already validated at
    construction (SuperposedSignal.__post_init__).

    Args:
        signal: The components to sum.
        sample_rate: Samples per second.
        n_points: Number of samples.
        t0: Time of the first sample in seconds (shifts periodic signals).

    Returns:
        float64 voltage array of length n_points.
    """
    total = synthesize(signal.components[0], sample_rate, n_points, t0=t0)
    for component in signal.components[1:]:
        total = total + synthesize(component, sample_rate, n_points, t0=t0)
    return total

stream

stream(spec: SignalSpec, sample_rate: float, chunk_size: int, *, start_time: float = 0.0, duration: Optional[float] = None, realtime: bool = False) -> Iterator[np.ndarray]

Yield phase-continuous voltage chunks for live/continuous simulation.

Parameters:

Name	Type	Description	Default
spec	SignalSpec	Signal parameters. A seeded spec uses seed + chunk_index per chunk (reproducible run-to-run, non-repeating across chunks); seed=None re-rolls noise freshly every chunk.	required
sample_rate	float	Samples per second.	required
chunk_size	int	Samples per yielded chunk.	required
start_time	float	Time of the very first sample in seconds.	0.0
duration	Optional[float]	None streams forever (stop by breaking out); a positive number bounds the stream to round(duration * sample_rate) samples, truncating the final chunk.	None
realtime	bool	When True, chunks arrive at wall-clock rate (chunk k is withheld until k * chunk_size / sample_rate seconds after the first chunk); scheduling is absolute, so timing error never accumulates, and a consumer slower than real time simply never waits.	False

Returns:

Type	Description
Iterator[ndarray]	Iterator of float64 voltage arrays. Validation errors raise at call
Iterator[ndarray]	time, before the first chunk.

Source code in scpi_control/signal_synth.py

def stream(
    spec: SignalSpec,
    sample_rate: float,
    chunk_size: int,
    *,
    start_time: float = 0.0,
    duration: Optional[float] = None,
    realtime: bool = False,
) -> Iterator[np.ndarray]:
    """Yield phase-continuous voltage chunks for live/continuous simulation.

    Args:
        spec: Signal parameters. A seeded spec uses seed + chunk_index per
            chunk (reproducible run-to-run, non-repeating across chunks);
            seed=None re-rolls noise freshly every chunk.
        sample_rate: Samples per second.
        chunk_size: Samples per yielded chunk.
        start_time: Time of the very first sample in seconds.
        duration: None streams forever (stop by breaking out); a positive
            number bounds the stream to round(duration * sample_rate) samples,
            truncating the final chunk.
        realtime: When True, chunks arrive at wall-clock rate (chunk k is
            withheld until k * chunk_size / sample_rate seconds after the
            first chunk); scheduling is absolute, so timing error never
            accumulates, and a consumer slower than real time simply never
            waits.

    Returns:
        Iterator of float64 voltage arrays. Validation errors raise at call
        time, before the first chunk.
    """
    _validate(spec, sample_rate, chunk_size)
    if duration is not None and duration <= 0:
        raise exceptions.InvalidParameterError(f"duration must be positive: {duration}")
    total = None if duration is None else int(round(duration * sample_rate))

    def _chunks() -> Iterator[np.ndarray]:
        produced = 0
        index = 0
        wall_start = None
        while total is None or produced < total:
            n = chunk_size if total is None else min(chunk_size, total - produced)
            if spec.seed is None:
                chunk_spec = spec
            else:
                # jitter_seed=spec.seed (the pre-bump value) only when the caller
                # left it None -- so every chunk of this run shares the same
                # jitter entropy source (closing the stream()-chunk-boundary
                # discontinuity, see SignalSpec.jitter_seed) while `seed` itself
                # keeps bumping per chunk, unchanged, so noise/glitches keep
                # their existing non-repeating-across-chunks behavior. A caller
                # who set jitter_seed explicitly keeps their own choice.
                jitter_seed = spec.jitter_seed if spec.jitter_seed is not None else spec.seed
                chunk_spec = replace(spec, seed=spec.seed + index, jitter_seed=jitter_seed)
            chunk = synthesize(chunk_spec, sample_rate, n, t0=start_time + produced / sample_rate)
            if realtime:
                if wall_start is None:
                    wall_start = time.monotonic()
                else:
                    delay = wall_start + produced / sample_rate - time.monotonic()
                    if delay > 0:
                        time.sleep(delay)
            yield chunk
            produced += n
            index += 1

    return _chunks()

make_waveform

make_waveform(spec: SignalSpec, sample_rate: float, n_points: int, channel: int = 1)

Generate a WaveformData ready for analysis, saving, or reporting.

Source code in scpi_control/signal_synth.py

def make_waveform(spec: SignalSpec, sample_rate: float, n_points: int, channel: int = 1):
    """Generate a WaveformData ready for analysis, saving, or reporting."""
    # Function-level import: keeps `import scpi_control.connection` (which pulls
    # the mock package, which pulls this module) from importing waveform.py
    # mid-initialization.
    from scpi_control.waveform import WaveformData

    voltage = synthesize(spec, sample_rate, n_points)
    time = np.arange(n_points) / sample_rate
    return WaveformData(time=time, voltage=voltage, channel=channel, sample_rate=sample_rate)

make_waveform_combined

make_waveform_combined(signal: SuperposedSignal, sample_rate: float, n_points: int, channel: int = 1) -> WaveformData

Generate a WaveformData from a SuperposedSignal, mirroring make_waveform().

Source code in scpi_control/signal_synth.py

def make_waveform_combined(signal: SuperposedSignal, sample_rate: float, n_points: int, channel: int = 1) -> "WaveformData":
    """Generate a WaveformData from a SuperposedSignal, mirroring make_waveform()."""
    # Function-level import: see make_waveform()'s identical import above.
    from scpi_control.waveform import WaveformData

    voltage = synthesize_combined(signal, sample_rate, n_points)
    time = np.arange(n_points) / sample_rate
    return WaveformData(time=time, voltage=voltage, channel=channel, sample_rate=sample_rate)

See Also

• Waveform - Waveform acquisition and data handling
• Analysis - Signal analysis (FFT, THD, SNR)