From d4c8381d6030c07dec15b37434d1ec1828f0f073 Mon Sep 17 00:00:00 2001 From: Pierre Raybaut <1311787+PierreRaybaut@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:35:45 +0200 Subject: [PATCH] feat: add an oscilloscope simulator tool Assisted-by: Claude Opus 5.5 --- CHANGELOG.md | 1 + README.md | 2 + doc/oscilloscope-simulator.md | 47 +++ doc/simulation.md | 2 + .../adapters/desktop.py | 6 + .../adapters/web.py | 6 + .../core/oscilloscope.py | 280 +++++++++++++++++ .../simulator.py | 284 ++++++++++++++++++ tests/unit/test_oscilloscope_simulator.py | 185 ++++++++++++ 9 files changed, 813 insertions(+) create mode 100644 doc/oscilloscope-simulator.md create mode 100644 src/datalab_pulse_characterization/core/oscilloscope.py create mode 100644 src/datalab_pulse_characterization/simulator.py create mode 100644 tests/unit/test_oscilloscope_simulator.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c4d6c8..312b42b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,3 +36,4 @@ All notable changes to this project will be documented in this file. - Let DataLab's generic launcher run the recipes on Desktop, replacing the two-channel role dialog; reference and measured channels are still proposed from their labels. - Pair the laser warm-up demonstration with both the stability and single-channel campaign methods; example parameter values are now keyed by method. - Document how to set shot numbers and channel labels on your own signals with DataLab's Add metadata dialog (`doc/preparing-signals.md`). +- Add an oscilloscope simulator tool, in DataLab Desktop and DataLab-Web: a laser pulse, an amplifier step or a delayed pulse pair seen through a time base, vertical scales and an n-bit ADC, with a live view, and acquisitions tagged for the Pulse methods (`doc/oscilloscope-simulator.md`). diff --git a/README.md b/README.md index 66eb79e..20f38fa 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,8 @@ The plugin offers five methods, each with its own generated demonstration: The new methods are documented in [`doc/shot-stability.md`](doc/shot-stability.md), [`doc/step-response.md`](doc/step-response.md), [`doc/two-channel-delay.md`](doc/two-channel-delay.md) and [`doc/pulse-height-spectrum.md`](doc/pulse-height-spectrum.md). Every demonstration is validated against its simulator truth in `tests/validation`. None of these conventions claims compliance with a standard. +In DataLab Desktop and DataLab-Web, the **Oscilloscope simulator** tool acquires laser pulses, amplifier steps or delayed pulse pairs through a time base, vertical scales and an n-bit ADC, with a live view of the screen. Its acquisitions carry the shot and channel metadata the methods expect. See [`doc/oscilloscope-simulator.md`](doc/oscilloscope-simulator.md). + ## Development ```bash diff --git a/doc/oscilloscope-simulator.md b/doc/oscilloscope-simulator.md new file mode 100644 index 0000000..932a355 --- /dev/null +++ b/doc/oscilloscope-simulator.md @@ -0,0 +1,47 @@ +# Oscilloscope Simulator + +The Pulse application includes a simulated oscilloscope. It is a DataLab tool: open it from the Pulse page of the **Applications** catalog (**Tools** section), or from the **Plugins > Pulse & Transient Characterization** menu of the signal panel. It works the same way in DataLab Desktop and DataLab-Web. + +The window shows the oscilloscope screen on the left and the settings on the right. Every change of a setting refreshes the screen. **Live** refreshes it continuously: each refresh is a new trigger, so jitter and noise show as on a real oscilloscope. **Acquire** records a series of triggers and adds them to the workspace, in a new group per acquisition; the window stays open for the next acquisition. + +## Settings + +Connect a source first (**Source** tab), then set the oscilloscope (**Oscilloscope** tab). Settings that do not apply to the selected source are greyed out. + +- **Source.** + - Laser pulse seen by a photodiode: Gaussian or asymmetric pulse, with an optional linear drift of its time and amplitude over one acquisition. + - Amplifier step response: response of a second-order system, set by its natural frequency and damping ratio. + - Pulse pair: a reference pulse on CH1 and the delayed, wider pulse of a detector on CH2. + - For every source: amplitude, amplitude jitter and trigger jitter. The pulse pair adds the detector delay, gain, pulse width and its own jitter. +- **Oscilloscope.** + - Horizontal: time base (ns/div, ten divisions), sample rate and trigger position. The trigger is at t = 0; the trigger position is the part of the record before it. + - Vertical: scale (V/div, eight divisions) and offset of CH1 and CH2, ADC resolution and input noise. The offset is the voltage at the center of the screen. + - Acquisition: number of triggers per acquisition, and the random seed. + +The ADC codes the screen only: a trace that leaves the screen is clipped, then quantized on 2^bits levels. The summary below the screen gives the peak of each channel and its number of clipped samples. Times are in ns and voltages in V, for every source. + +The seed and the acquisition number fix the noise and jitter of each acquisition: the same settings give the same acquisitions in a new session. + +One acquisition is limited to 5 million samples (all signals together), which keeps DataLab-Web within the browser memory. + +## Acquired Signals + +Signals are named `CH1 shot 001`, `CH1 shot 002`, and so on; with the pulse pair, the CH2 signals follow the CH1 ones. Each signal carries the metadata the Pulse methods expect: + +- `plugin.org.datalab.pulse-characterization.shot`: the trigger number, from 1; +- `plugin.org.datalab.pulse-characterization.channel`: `CH1` or `CH2`; +- `plugin.org.datalab.pulse-characterization.synthetic` and `.simulation_seed`, which record the simulated origin. + +So an acquisition can be analyzed right away: + +| Source | Methods | +| --- | --- | +| Laser pulse | Single-channel pulse campaign, shot-to-shot stability | +| Amplifier step response | Step response | +| Pulse pair | Two-channel delay and jitter | + +The two-channel method measures the delay between 50 % crossings. With the default asymmetric detector pulse, this delay is shorter than the delay between pulse peaks set in the source. + +## Limits + +The trigger is external and synchronized with the source: there is no trigger level, slope or rearm logic. There is no bandwidth limit, no probe model and no acquisition mode other than a series of single triggers. The simulator helps learning the methods and checking a workflow; it does not replace measurements on a real bench. diff --git a/doc/simulation.md b/doc/simulation.md index 36a6843..ca3cdbd 100644 --- a/doc/simulation.md +++ b/doc/simulation.md @@ -5,6 +5,8 @@ exact settings used for every waveform. It is designed to exercise the normal Sigima-backed analysis path with reproducible drift, jitter, noise, and quality failures. It is not a calibrated oscilloscope model. +In DataLab, the oscilloscope simulator tool acquires pulses, amplifier steps and delayed pulse pairs interactively, through a time base and a digitizing front end. See [`oscilloscope-simulator.md`](oscilloscope-simulator.md). + ## Waveform Model Every acquisition uses the same evenly sampled X axis. Regular shots alternate diff --git a/src/datalab_pulse_characterization/adapters/desktop.py b/src/datalab_pulse_characterization/adapters/desktop.py index 88ec28e..1610a53 100644 --- a/src/datalab_pulse_characterization/adapters/desktop.py +++ b/src/datalab_pulse_characterization/adapters/desktop.py @@ -25,6 +25,7 @@ TWO_CHANNEL_DEMO, materialize_generated_example, ) +from ..simulator import OSCILLOSCOPE_SIMULATOR_TOOL, OscilloscopeSimulator from ..workflow import ( PULSE_CAMPAIGN_RECIPE, PULSE_HEIGHT_RECIPE, @@ -59,6 +60,7 @@ class PulseTransientCharacterizationPlugin(PluginBase): ) RECIPES = WORKFLOW_RECIPES EXAMPLES = DEMO_EXAMPLES + TOOLS = (OSCILLOSCOPE_SIMULATOR_TOOL,) WELCOME_TILES = ( WelcomeTile( id="application", @@ -81,6 +83,10 @@ def materialize_example(cls, example_id: str) -> PluginExampleData | None: cls.get_example(example_id) return materialize_generated_example(example_id) + def oscilloscope_simulator(self) -> OscilloscopeSimulator: + """Return a new oscilloscope simulator.""" + return OscilloscopeSimulator() + def open_demo_campaign(self) -> PluginExample | None: """Open the demonstration campaign from the plugin menu.""" return self.launch_example(PULSE_DEMO.id) diff --git a/src/datalab_pulse_characterization/adapters/web.py b/src/datalab_pulse_characterization/adapters/web.py index f963efc..fff60dc 100644 --- a/src/datalab_pulse_characterization/adapters/web.py +++ b/src/datalab_pulse_characterization/adapters/web.py @@ -24,6 +24,7 @@ build_simulated_campaign, materialize_generated_example, ) +from ..simulator import OSCILLOSCOPE_SIMULATOR_TOOL, OscilloscopeSimulator from ..workflow import ( PULSE_CAMPAIGN_RECIPE, PulseCampaignRecipeParameters, @@ -55,10 +56,15 @@ class PulseTransientCharacterizationWebPlugin(PluginBase): ) RECIPES = WORKFLOW_RECIPES EXAMPLES = DEMO_EXAMPLES + TOOLS = (OSCILLOSCOPE_SIMULATOR_TOOL,) def create_actions(self) -> None: """Application actions are provided by DataLab-Web's generic host.""" + def oscilloscope_simulator(self) -> OscilloscopeSimulator: + """Return a new oscilloscope simulator.""" + return OscilloscopeSimulator() + @classmethod def materialize_example(cls, example_id: str) -> PluginExampleData | None: """Build one deterministic generated example in memory.""" diff --git a/src/datalab_pulse_characterization/core/oscilloscope.py b/src/datalab_pulse_characterization/core/oscilloscope.py new file mode 100644 index 0000000..79f46a3 --- /dev/null +++ b/src/datalab_pulse_characterization/core/oscilloscope.py @@ -0,0 +1,280 @@ +"""Simulated oscilloscope: pulse sources, time base and digitizing front end.""" + +from __future__ import annotations + +import dataclasses +import enum +import math + +import numpy as np + +from .step_response import second_order_step + +#: Horizontal and vertical divisions of the oscilloscope screen +HORIZONTAL_DIVISIONS = 10 +VERTICAL_DIVISIONS = 8 +#: Longest record, in samples +MAX_RECORD_LENGTH = 1_000_000 +#: Ratio between a Gaussian's FWHM and its standard deviation +FWHM_PER_SIGMA = 2.0 * math.sqrt(2.0 * math.log(2.0)) +#: Rise and fall width factors of asymmetric pulses +ASYMMETRIC_RISE_FACTOR = 0.65 +ASYMMETRIC_FALL_FACTOR = 1.6 + + +def _check_positive(value: float, name: str) -> None: + """Validate a positive finite number.""" + if not math.isfinite(value) or value <= 0.0: + raise ValueError(f"{name} must be positive") + + +def _check_nonnegative(value: float, name: str) -> None: + """Validate a non-negative finite number.""" + if not math.isfinite(value) or value < 0.0: + raise ValueError(f"{name} must be non-negative") + + +@dataclasses.dataclass(frozen=True) +class ScopeChannel: + """Vertical settings of one oscilloscope channel. + + The screen shows ``VERTICAL_DIVISIONS`` divisions centered on + ``offset_v``; the ADC codes this range. + """ + + volts_per_div: float = 0.1 + offset_v: float = 0.0 + + def __post_init__(self) -> None: + """Validate the channel settings.""" + _check_positive(self.volts_per_div, "Volts per division") + if not math.isfinite(self.offset_v): + raise ValueError("Offset must be finite") + + @property + def screen_range(self) -> tuple[float, float]: + """Return the lowest and highest voltages coded by the ADC.""" + half_range = 0.5 * VERTICAL_DIVISIONS * self.volts_per_div + return self.offset_v - half_range, self.offset_v + half_range + + +@dataclasses.dataclass(frozen=True) +class OscilloscopeSettings: + """Time base, channels and ADC of the oscilloscope (times in ns).""" + + time_per_div_ns: float = 10.0 + sample_rate_gsps: float = 5.0 + trigger_position: float = 0.2 + ch1: ScopeChannel = ScopeChannel(0.1, 0.3) + ch2: ScopeChannel = ScopeChannel(0.1, 0.3) + adc_bits: int = 8 + noise_v: float = 0.005 + + def __post_init__(self) -> None: + """Validate the time base and the ADC.""" + _check_positive(self.time_per_div_ns, "Time per division") + _check_positive(self.sample_rate_gsps, "Sample rate") + if not 0.0 <= self.trigger_position < 1.0: + raise ValueError("Trigger position must be between 0 and 100 %") + if not 4 <= self.adc_bits <= 16: + raise ValueError("ADC resolution must be between 4 and 16 bits") + _check_nonnegative(self.noise_v, "Noise") + if not 21 <= self.record_length <= MAX_RECORD_LENGTH: + raise ValueError( + f"The record must hold between 21 and {MAX_RECORD_LENGTH} samples " + f"(now {self.record_length}): change the time base or the sample " + "rate" + ) + + @property + def record_length(self) -> int: + """Return the number of samples of one record.""" + duration = HORIZONTAL_DIVISIONS * self.time_per_div_ns + return int(round(duration * self.sample_rate_gsps)) + + def time_axis(self) -> np.ndarray: + """Return sample times, in ns, the trigger being at t = 0.""" + count = self.record_length + start = -self.trigger_position * count / self.sample_rate_gsps + return start + np.arange(count) / self.sample_rate_gsps + + def digitize(self, y: np.ndarray, channel: ScopeChannel) -> np.ndarray: + """Clip a waveform to the screen of a channel and quantize it.""" + low, high = channel.screen_range + step = (high - low) / (2**self.adc_bits - 1) + codes = np.clip(np.round((y - low) / step), 0, 2**self.adc_bits - 1) + return low + codes * step + + +class SourceKind(str, enum.Enum): + """Signal source connected to the oscilloscope.""" + + LASER_PULSE = "laser-pulse" + STEP = "step" + PULSE_PAIR = "pulse-pair" + + +@dataclasses.dataclass(frozen=True) +class PulseSource: + """Source connected to the oscilloscope, triggered at t = 0 (times in ns). + + Common settings apply to every source. A laser pulse has a Gaussian or + asymmetric profile and may drift over an acquisition. An amplifier step + is the response of a second-order system. A pulse pair feeds a reference + pulse to CH1 and the delayed pulse of a detector to CH2. + """ + + kind: SourceKind = SourceKind.LASER_PULSE + amplitude_v: float = 0.5 + amplitude_jitter_percent: float = 1.0 + timing_jitter_ns: float = 0.05 + pulse_width_ns: float = 2.0 + asymmetric: bool = False + timing_drift_ns: float = 0.0 + amplitude_drift_percent: float = 0.0 + natural_frequency_mhz: float = 80.0 + damping_ratio: float = 0.35 + delay_ns: float = 37.5 + detector_gain: float = 0.5 + detector_width_ns: float = 6.0 + detector_jitter_ns: float = 0.2 + + def __post_init__(self) -> None: + """Validate the source settings.""" + object.__setattr__(self, "kind", SourceKind(self.kind)) + for name, value in ( + ("Amplitude", self.amplitude_v), + ("Pulse width", self.pulse_width_ns), + ("Natural frequency", self.natural_frequency_mhz), + ("Detector gain", self.detector_gain), + ("Detector pulse width", self.detector_width_ns), + ): + _check_positive(value, name) + for name, value in ( + ("Amplitude jitter", self.amplitude_jitter_percent), + ("Timing jitter", self.timing_jitter_ns), + ("Detector jitter", self.detector_jitter_ns), + ): + _check_nonnegative(value, name) + for name, value in ( + ("Timing drift", self.timing_drift_ns), + ("Amplitude drift", self.amplitude_drift_percent), + ("Delay", self.delay_ns), + ): + if not math.isfinite(value): + raise ValueError(f"{name} must be finite") + if not 0.0 < self.damping_ratio < 1.0: + raise ValueError("Damping ratio must be between 0 and 1") + + @property + def channel_count(self) -> int: + """Return the number of channels fed by the source.""" + return 2 if self.kind is SourceKind.PULSE_PAIR else 1 + + def waveforms( + self, + t: np.ndarray, + rng: np.random.Generator, + campaign_fraction: float = 0.0, + ) -> tuple[np.ndarray, ...]: + """Return the noise-free waveform of each channel for one trigger. + + Args: + t: sample times, in ns + rng: random generator of this trigger + campaign_fraction: position of the trigger in the acquisition, + from -0.5 (first) to 0.5 (last), scaling the drifts + """ + trigger_jitter = rng.normal(0.0, self.timing_jitter_ns) + amplitude = self.amplitude_v * ( + 1.0 + + 0.01 * self.amplitude_drift_percent * campaign_fraction + + rng.normal(0.0, 0.01 * self.amplitude_jitter_percent) + ) + if self.kind is SourceKind.STEP: + natural_frequency = 1e-3 * self.natural_frequency_mhz + return ( + amplitude + * second_order_step( + t - trigger_jitter, natural_frequency, self.damping_ratio + ), + ) + sigma = self.pulse_width_ns / FWHM_PER_SIGMA + if self.kind is SourceKind.LASER_PULSE: + center = trigger_jitter + self.timing_drift_ns * campaign_fraction + rise, fall = ( + (ASYMMETRIC_RISE_FACTOR, ASYMMETRIC_FALL_FACTOR) + if self.asymmetric + else (1.0, 1.0) + ) + return (amplitude * _pulse(t, center, sigma * rise, sigma * fall),) + detector_center = ( + trigger_jitter + self.delay_ns + rng.normal(0.0, self.detector_jitter_ns) + ) + detector_sigma = self.detector_width_ns / FWHM_PER_SIGMA + return ( + amplitude * _pulse(t, trigger_jitter, sigma, sigma), + amplitude + * self.detector_gain + * _pulse( + t, + detector_center, + detector_sigma * ASYMMETRIC_RISE_FACTOR, + detector_sigma * ASYMMETRIC_FALL_FACTOR, + ), + ) + + +def _pulse(t: np.ndarray, center: float, rise: float, fall: float) -> np.ndarray: + """Return a unit-height pulse with Gaussian rising and falling edges.""" + scale = np.where(t < center, rise, fall) + return np.exp(-0.5 * np.square((t - center) / scale)) + + +def acquire_traces( + source: PulseSource, + scope: OscilloscopeSettings, + trigger_count: int, + seed: np.random.SeedSequence, +) -> tuple[np.ndarray, list[tuple[np.ndarray, ...]]]: + """Acquire digitized traces for successive triggers. + + Args: + source: source connected to the oscilloscope + scope: oscilloscope settings + trigger_count: number of triggers + seed: seed of the acquisition, spawning one stream per trigger + + Returns: + Sample times (ns), then the traces of each trigger, one per channel + """ + if trigger_count < 1: + raise ValueError("Trigger count must be positive") + t = scope.time_axis() + channels = (scope.ch1, scope.ch2)[: source.channel_count] + traces: list[tuple[np.ndarray, ...]] = [] + for index, child in enumerate(seed.spawn(trigger_count)): + rng = np.random.default_rng(child) + fraction = 0.0 if trigger_count == 1 else index / (trigger_count - 1) - 0.5 + waveforms = source.waveforms(t, rng, fraction) + traces.append( + tuple( + scope.digitize( + waveform + rng.normal(0.0, scope.noise_v, t.size), channel + ) + for waveform, channel in zip(waveforms, channels) + ) + ) + return t, traces + + +__all__ = [ + "HORIZONTAL_DIVISIONS", + "MAX_RECORD_LENGTH", + "VERTICAL_DIVISIONS", + "OscilloscopeSettings", + "PulseSource", + "ScopeChannel", + "SourceKind", + "acquire_traces", +] diff --git a/src/datalab_pulse_characterization/simulator.py b/src/datalab_pulse_characterization/simulator.py new file mode 100644 index 0000000..ce9bfed --- /dev/null +++ b/src/datalab_pulse_characterization/simulator.py @@ -0,0 +1,284 @@ +"""Host-neutral oscilloscope simulator, shared by the Desktop and Web adapters. + +The simulator behaves like an oscilloscope on a test bench: a source is +connected first, then the time base, channels and ADC are set. The live view +shows one trigger, and each acquisition adds one signal per trigger and +channel, tagged with the metadata the Pulse methods expect. +""" + +from __future__ import annotations + +import itertools + +import guidata.dataset as gds +import numpy as np +from datalab.plugin_instruments import ( + InstrumentAcquisition, + InstrumentFrame, + PluginInstrument, +) +from datalab.plugin_tools import PluginTool +from datalab.recipes import RecipeObjectType +from sigima.objects import SignalObj, create_signal + +from .core import metadata_key +from .core.oscilloscope import ( + OscilloscopeSettings, + PulseSource, + ScopeChannel, + SourceKind, + acquire_traces, +) +from .workflow import CHANNEL_METADATA_KEY, SHOT_METADATA_KEY + +#: Largest number of samples created by one acquisition (all signals together) +MAX_ACQUISITION_SAMPLES = 5_000_000 + +OSCILLOSCOPE_SIMULATOR_TOOL = PluginTool( + id="oscilloscope-simulator", + title="Oscilloscope simulator...", + description="Acquire pulses, steps or delayed pulse pairs with live view", + instrument="oscilloscope_simulator", + object_type=RecipeObjectType.SIGNAL, +) + +SOURCE_LABELS = { + SourceKind.LASER_PULSE.value: "Laser pulse (photodiode)", + SourceKind.STEP.value: "Amplifier step response", + SourceKind.PULSE_PAIR.value: "Pulse pair: reference and delayed detector", +} + +_SOURCE = gds.ValueProp(SourceKind.LASER_PULSE.value) + + +def _source_is(*kinds: SourceKind) -> gds.FuncProp: + """Return a property active when one of the given sources is connected.""" + values = {kind.value for kind in kinds} + return gds.FuncProp(_SOURCE, lambda source: source in values) + + +_LASER = _source_is(SourceKind.LASER_PULSE) +_STEP = _source_is(SourceKind.STEP) +_PAIR = _source_is(SourceKind.PULSE_PAIR) +_PULSES = _source_is(SourceKind.LASER_PULSE, SourceKind.PULSE_PAIR) + + +class OscilloscopeSimulatorSettings(gds.DataSet): + """Oscilloscope simulator""" + + _tabs = gds.BeginTabGroup("Settings") + + _source_tab = gds.BeginGroup("Source") + source = gds.ChoiceItem( + "Source", tuple(SOURCE_LABELS.items()), default=SourceKind.LASER_PULSE.value + ).set_prop("display", store=_SOURCE) + amplitude = gds.FloatItem("Amplitude", default=0.5, min=0.0, nonzero=True, unit="V") + amplitude_jitter = gds.FloatItem( + "Amplitude jitter", default=1.0, min=0.0, max=50.0, unit="% RMS" + ) + timing_jitter = gds.FloatItem( + "Trigger jitter", default=0.05, min=0.0, unit="ns RMS" + ) + pulse_width = gds.FloatItem( + "Pulse width (FWHM)", default=2.0, min=0.0, nonzero=True, unit="ns" + ).set_prop("display", active=_PULSES) + _laser = gds.BeginGroup("Laser pulse") + asymmetric = gds.BoolItem("Asymmetric profile", default=False).set_prop( + "display", active=_LASER + ) + timing_drift = gds.FloatItem( + "Timing drift", + default=0.0, + unit="ns", + help="Linear drift of the pulse time over one acquisition", + ).set_prop("display", active=_LASER) + amplitude_drift = gds.FloatItem( + "Amplitude drift", + default=0.0, + min=-90.0, + max=90.0, + unit="%", + help="Linear drift of the pulse amplitude over one acquisition", + ).set_prop("display", active=_LASER) + _laser_end = gds.EndGroup("Laser pulse") + _step = gds.BeginGroup("Amplifier step response") + natural_frequency = gds.FloatItem( + "Natural frequency", default=80.0, min=0.0, nonzero=True, unit="MHz" + ).set_prop("display", active=_STEP) + damping_ratio = gds.FloatItem( + "Damping ratio", default=0.35, min=0.01, max=0.99 + ).set_prop("display", active=_STEP) + _step_end = gds.EndGroup("Amplifier step response") + _pair = gds.BeginGroup("Pulse pair") + delay = gds.FloatItem("Detector delay", default=37.5, unit="ns").set_prop( + "display", active=_PAIR + ) + detector_gain = gds.FloatItem( + "Detector gain", default=0.5, min=0.0, nonzero=True + ).set_prop("display", active=_PAIR) + detector_width = gds.FloatItem( + "Detector pulse width (FWHM)", default=6.0, min=0.0, nonzero=True, unit="ns" + ).set_prop("display", active=_PAIR) + detector_jitter = gds.FloatItem( + "Detector jitter", default=0.2, min=0.0, unit="ns RMS" + ).set_prop("display", active=_PAIR) + _pair_end = gds.EndGroup("Pulse pair") + _source_tab_end = gds.EndGroup("Source") + + _scope_tab = gds.BeginGroup("Oscilloscope") + _horizontal = gds.BeginGroup("Horizontal") + time_per_div = gds.FloatItem( + "Time base", default=10.0, min=0.0, nonzero=True, unit="ns/div" + ) + sample_rate = gds.FloatItem( + "Sample rate", default=5.0, min=0.0, nonzero=True, max=1_000.0, unit="GS/s" + ) + trigger_position = gds.FloatItem( + "Trigger position", + default=20.0, + min=0.0, + max=99.0, + unit="%", + help="Part of the record before the trigger", + ) + _horizontal_end = gds.EndGroup("Horizontal") + _vertical = gds.BeginGroup("Vertical") + ch1_volts_per_div = gds.FloatItem( + "CH1 scale", default=0.1, min=0.0, nonzero=True, unit="V/div" + ) + ch1_offset = gds.FloatItem("CH1 offset", default=0.3, unit="V") + ch2_volts_per_div = gds.FloatItem( + "CH2 scale", default=0.1, min=0.0, nonzero=True, unit="V/div" + ).set_prop("display", active=_PAIR) + ch2_offset = gds.FloatItem("CH2 offset", default=0.3, unit="V").set_prop( + "display", active=_PAIR + ) + adc_bits = gds.IntItem("ADC resolution", default=8, min=4, max=16, unit="bits") + noise = gds.FloatItem("Noise", default=0.005, min=0.0, unit="V RMS") + _vertical_end = gds.EndGroup("Vertical") + _acquisition = gds.BeginGroup("Acquisition") + trigger_count = gds.IntItem( + "Triggers per acquisition", default=200, min=1, max=10_000 + ) + seed = gds.IntItem( + "Random seed", + default=20261007, + min=0, + help="Makes the acquisitions of a session reproducible", + ) + _acquisition_end = gds.EndGroup("Acquisition") + _scope_tab_end = gds.EndGroup("Oscilloscope") + + _tabs_end = gds.EndTabGroup("Settings") + + +class OscilloscopeSimulator(PluginInstrument): + """Oscilloscope simulator. + + The seed and the acquisition number fix the noise and jitter of each + acquisition; live traces use their own random streams. + """ + + live_interval_ms = 200 + + def __init__(self) -> None: + super().__init__(OscilloscopeSimulatorSettings()) + self._acquisition_count = 0 + self._previews = itertools.count(1) + + def source(self) -> PulseSource: + """Return the source set in the settings.""" + s = self.settings + return PulseSource( + kind=SourceKind(s.source), + amplitude_v=float(s.amplitude), + amplitude_jitter_percent=float(s.amplitude_jitter), + timing_jitter_ns=float(s.timing_jitter), + pulse_width_ns=float(s.pulse_width), + asymmetric=bool(s.asymmetric), + timing_drift_ns=float(s.timing_drift), + amplitude_drift_percent=float(s.amplitude_drift), + natural_frequency_mhz=float(s.natural_frequency), + damping_ratio=float(s.damping_ratio), + delay_ns=float(s.delay), + detector_gain=float(s.detector_gain), + detector_width_ns=float(s.detector_width), + detector_jitter_ns=float(s.detector_jitter), + ) + + def scope(self) -> OscilloscopeSettings: + """Return the oscilloscope set in the settings.""" + s = self.settings + return OscilloscopeSettings( + time_per_div_ns=float(s.time_per_div), + sample_rate_gsps=float(s.sample_rate), + trigger_position=float(s.trigger_position) / 100.0, + ch1=ScopeChannel(float(s.ch1_volts_per_div), float(s.ch1_offset)), + ch2=ScopeChannel(float(s.ch2_volts_per_div), float(s.ch2_offset)), + adc_bits=int(s.adc_bits), + noise_v=float(s.noise), + ) + + def preview(self) -> InstrumentFrame: + """Return the traces of one trigger, on the oscilloscope screen.""" + source, scope = self.source(), self.scope() + seed = np.random.SeedSequence( + (int(self.settings.seed), 2, next(self._previews)) + ) + t, (traces,) = acquire_traces(source, scope, 1, seed) + channels = (scope.ch1, scope.ch2)[: source.channel_count] + signals: list[SignalObj] = [] + parts: list[str] = [] + for index, (trace, channel) in enumerate(zip(traces, channels), start=1): + signals.append(create_signal(f"CH{index}", t, trace, units=("ns", "V"))) + low, high = channel.screen_range + clipped = np.count_nonzero((trace <= low) | (trace >= high)) + parts.append( + f"CH{index}: peak {np.max(trace):.3g} V, {clipped} clipped samples" + ) + lows, highs = zip(*(channel.screen_range for channel in channels)) + return InstrumentFrame(signals, " - ".join(parts), (min(lows), max(highs))) + + def acquire(self) -> InstrumentAcquisition: + """Acquire one signal per trigger and channel.""" + source, scope = self.source(), self.scope() + trigger_count = int(self.settings.trigger_count) + samples = trigger_count * scope.record_length * source.channel_count + if samples > MAX_ACQUISITION_SAMPLES: + raise ValueError( + f"This acquisition would create {samples / 1e6:.1f} million samples " + f"(limit: {MAX_ACQUISITION_SAMPLES / 1e6:g}): reduce the number of " + "triggers, the time base or the sample rate" + ) + seed_value = int(self.settings.seed) + acquisition_number = self._acquisition_count + 1 + seed = np.random.SeedSequence((seed_value, 1, acquisition_number)) + t, traces = acquire_traces(source, scope, trigger_count, seed) + signals: list[SignalObj] = [] + for channel_index in range(source.channel_count): + channel = f"CH{channel_index + 1}" + for shot, shot_traces in enumerate(traces, start=1): + signal = create_signal( + f"{channel} shot {shot:03d}", + t, + shot_traces[channel_index], + units=("ns", "V"), + ) + signal.metadata[SHOT_METADATA_KEY] = shot + signal.metadata[CHANNEL_METADATA_KEY] = channel + signal.metadata[metadata_key("synthetic")] = True + signal.metadata[metadata_key("simulation_seed")] = seed_value + signals.append(signal) + self._acquisition_count = acquisition_number + label = SOURCE_LABELS[source.kind.value].split(" (")[0].split(":")[0] + return InstrumentAcquisition( + f"Oscilloscope - {label} - acquisition {acquisition_number:03d}", signals + ) + + +__all__ = [ + "MAX_ACQUISITION_SAMPLES", + "OSCILLOSCOPE_SIMULATOR_TOOL", + "OscilloscopeSimulator", + "OscilloscopeSimulatorSettings", +] diff --git a/tests/unit/test_oscilloscope_simulator.py b/tests/unit/test_oscilloscope_simulator.py new file mode 100644 index 0000000..56779ce --- /dev/null +++ b/tests/unit/test_oscilloscope_simulator.py @@ -0,0 +1,185 @@ +"""Tests of the oscilloscope simulator instrument.""" + +from __future__ import annotations + +import numpy as np +import pytest +from datalab.plugin_instruments import InstrumentAcquisition, InstrumentFrame +from datalab.recipe_binding import ( + RecipeReadinessStatus, + assess_recipe_inputs, + create_recipe_parameters, +) +from datalab.recipes import RecipeExecutionContext + +from datalab_pulse_characterization.core.oscilloscope import ( + OscilloscopeSettings, + PulseSource, + ScopeChannel, + SourceKind, + acquire_traces, +) +from datalab_pulse_characterization.simulator import OscilloscopeSimulator +from datalab_pulse_characterization.workflow import ( + CHANNEL_METADATA_KEY, + PULSE_CAMPAIGN_RECIPE, + PULSE_STABILITY_RECIPE, + SHOT_METADATA_KEY, + STEP_RESPONSE_RECIPE, + TWO_CHANNEL_DELAY_RECIPE, + suggest_channel_roles, +) + + +def _ready(recipe, objects) -> bool: + """Return True if a recipe can run on objects with its default parameters.""" + parameters = create_recipe_parameters(recipe, {}) + readiness = assess_recipe_inputs(recipe, objects, parameters) + return readiness.status is RecipeReadinessStatus.READY + + +def _rows(outcome, result_id: str) -> dict[str, list[object]]: + """Return one metrics table by metric name.""" + table = next(item.value for item in outcome.results if item.id == result_id) + return {row[0]: row for row in table.data} + + +def test_time_base_places_the_trigger_and_the_adc_codes_the_screen() -> None: + """The record spans ten divisions; the ADC clips and quantizes the screen.""" + scope = OscilloscopeSettings(adc_bits=4, ch1=ScopeChannel(0.1, 0.0)) + t = scope.time_axis() + + assert scope.record_length == 500 + assert t[0] == pytest.approx(-20.0) + assert t[1] - t[0] == pytest.approx(0.2) + digitized = scope.digitize(np.linspace(-1.0, 1.0, 1001), scope.ch1) + assert digitized.min() == pytest.approx(-0.4) + assert digitized.max() == pytest.approx(0.4) + assert np.unique(digitized).size == 16 + for kwargs, message in ( + ({"trigger_position": 1.0}, "Trigger position"), + ({"adc_bits": 20}, "ADC resolution"), + ({"time_per_div_ns": 1e6}, "record must hold"), + ): + with pytest.raises(ValueError, match=message): + OscilloscopeSettings(**kwargs) + + +def test_sources_give_pulses_steps_and_delayed_pulse_pairs() -> None: + """Each source shapes its channels as connected to the oscilloscope.""" + scope = OscilloscopeSettings(noise_v=0.0, adc_bits=16) + seed = np.random.SeedSequence(1) + quiet = {"timing_jitter_ns": 0.0, "amplitude_jitter_percent": 0.0} + + t, ((laser,),) = acquire_traces(PulseSource(**quiet), scope, 1, seed) + assert t[np.argmax(laser)] == pytest.approx(0.0, abs=0.2) + assert laser.max() == pytest.approx(0.5, abs=0.005) + + step = PulseSource(kind=SourceKind.STEP, **quiet) + _t, ((response,),) = acquire_traces(step, scope, 1, seed) + assert response[t < 0].max() == pytest.approx(0.0, abs=0.005) + assert response.max() == pytest.approx(0.5 * 1.309, rel=0.01) + assert response[-1] == pytest.approx(0.5, abs=0.01) + + pair = PulseSource(kind=SourceKind.PULSE_PAIR, **quiet, detector_jitter_ns=0.0) + assert pair.channel_count == 2 + _t, ((reference, measured),) = acquire_traces(pair, scope, 1, seed) + assert t[np.argmax(reference)] == pytest.approx(0.0, abs=0.2) + assert t[np.argmax(measured)] == pytest.approx(37.5, abs=0.2) + assert measured.max() == pytest.approx(0.25, abs=0.005) + + +def test_live_view_shows_the_oscilloscope_screen() -> None: + """Live traces use the screen range of the channels and report clipping.""" + scope = OscilloscopeSimulator() + frame = scope.preview() + + assert isinstance(frame, InstrumentFrame) + (trace,) = frame.objects + assert trace.title == "CH1" + assert trace.xunit == "ns" + assert frame.value_range == pytest.approx((-0.1, 0.7)) + assert frame.summary.startswith("CH1: peak ") + assert not np.array_equal(trace.y, scope.preview().objects[0].y) + + scope.settings.source = SourceKind.PULSE_PAIR.value + scope.settings.ch2_offset = 0.0 + frame = scope.preview() + assert [signal.title for signal in frame.objects] == ["CH1", "CH2"] + assert frame.value_range == pytest.approx((-0.4, 0.7)) + assert "CH2: peak " in frame.summary + + +def test_acquisitions_are_tagged_and_reproducible() -> None: + """Shots carry their number and channel; the seed fixes each acquisition.""" + scope = OscilloscopeSimulator() + scope.settings.trigger_count = 5 + acquisition = scope.acquire() + + assert isinstance(acquisition, InstrumentAcquisition) + assert acquisition.group_title == "Oscilloscope - Laser pulse - acquisition 001" + assert [signal.title for signal in acquisition.objects] == [ + f"CH1 shot {shot:03d}" for shot in range(1, 6) + ] + assert [signal.metadata[SHOT_METADATA_KEY] for signal in acquisition.objects] == [ + 1, + 2, + 3, + 4, + 5, + ] + assert { + signal.metadata[CHANNEL_METADATA_KEY] for signal in acquisition.objects + } == {"CH1"} + second = scope.acquire() + assert second.group_title.endswith("acquisition 002") + assert not np.array_equal(acquisition.objects[0].y, second.objects[0].y) + + other = OscilloscopeSimulator() + other.settings.trigger_count = 5 + assert np.array_equal(other.acquire().objects[0].y, acquisition.objects[0].y) + + scope.settings.trigger_count = 10_000 + scope.settings.sample_rate = 100.0 + with pytest.raises(ValueError, match="million samples"): + scope.acquire() + + +def test_laser_and_step_acquisitions_suit_the_pulse_methods() -> None: + """Laser pulses suit the campaign and stability methods; steps are analyzed.""" + scope = OscilloscopeSimulator() + scope.settings.trigger_count = 50 + pulses = scope.acquire().objects + assert _ready(PULSE_CAMPAIGN_RECIPE, pulses) + assert _ready(PULSE_STABILITY_RECIPE, pulses) + + scope.settings.source = SourceKind.STEP.value + steps = scope.acquire().objects + assert _ready(STEP_RESPONSE_RECIPE, steps) + outcome = STEP_RESPONSE_RECIPE.run( + {"signals": steps}, + STEP_RESPONSE_RECIPE.parameter_class(), + RecipeExecutionContext(), + ) + rows = _rows(outcome, "step_metrics") + assert rows["Overshoot"][1] == pytest.approx(30.9, abs=1.5) + + +def test_pulse_pair_acquisitions_measure_the_detector_delay() -> None: + """The two-channel method measures the 50 % crossing delay of the pair.""" + scope = OscilloscopeSimulator() + scope.settings.source = SourceKind.PULSE_PAIR.value + scope.settings.trigger_count = 50 + signals = scope.acquire().objects + assert len(signals) == 100 + assert _ready(TWO_CHANNEL_DELAY_RECIPE, signals) + + outcome = TWO_CHANNEL_DELAY_RECIPE.run( + suggest_channel_roles(signals), + TWO_CHANNEL_DELAY_RECIPE.parameter_class(), + RecipeExecutionContext(), + ) + rows = _rows(outcome, "delay_summary") + # 50 % crossings: rise side of each pulse (detector rise factor 0.65) + expected = 37.5 - 0.65 * 6.0 / 2 + 2.0 / 2 + assert rows["Mean 50% (CFD) delay"][1] == pytest.approx(expected, abs=0.2)