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)

Parameters of one synthetic signal.

Attributes:

Name	Type	Description
kind	str	One of "sine", "square", "triangle", "ramp", "dc", "noise".
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.

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 = t0 + np.arange(n_points) / sample_rate
    samples = _GENERATORS[spec.kind](spec, t, rng) + spec.offset
    if spec.noise_rms > 0:
        samples = samples + rng.normal(0.0, spec.noise_rms, n_points)
    return samples

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)
            chunk_spec = spec if spec.seed is None else replace(spec, seed=spec.seed + index)
            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)

See Also

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