diff --git a/superxas_bec/device_configs/laser_delay_sim.yaml b/superxas_bec/device_configs/laser_delay_sim.yaml index 1f2ebf0..fb24026 100644 --- a/superxas_bec/device_configs/laser_delay_sim.yaml +++ b/superxas_bec/device_configs/laser_delay_sim.yaml @@ -13,7 +13,7 @@ laser_delay: readoutPriority: baseline description: Simulated laser delay in ns (no EPICS) - deviceClass: superxas_bec.devices.laser_delay.SimLaserDelay + deviceClass: superxas_bec.devices.laser_delay_sim.SimLaserDelay deviceConfig: step_sleep: 0.01 calibration_file: ~/.superxas_bec/laser_delay_sim_calibration.json diff --git a/superxas_bec/devices/laser_delay.py b/superxas_bec/devices/laser_delay.py index 999e97c..9b78289 100644 --- a/superxas_bec/devices/laser_delay.py +++ b/superxas_bec/devices/laser_delay.py @@ -48,7 +48,6 @@ from __future__ import annotations import contextlib import datetime import functools -import inspect import json import math import os @@ -347,10 +346,6 @@ class _HardwareEpicsSignal(_DeviceWritesOnly, EpicsSignal): """EPICS signal of the delay card or the phase shifter (written only by the device).""" -class _HardwareSoftSignal(_DeviceWritesOnly, CheckedSignal): - """Soft stand-in for a hardware PV in the simulation (written only by the device).""" - - class ComputedPhaseSignal(SignalRO): """Read-only signal whose value is computed by the parent device on every read. @@ -1543,222 +1538,3 @@ class LaserDelay(PSIDeviceBase, PositionerBase): pass raise return True - - -def _sim_range(name: str, low: float, high: float) -> Callable[[object], None]: - def check(value): - try: - number = float(value) - except (TypeError, ValueError) as exc: - raise ValueError(f"{name} must be a number, got {value!r}") from exc - if not low <= number <= high: - raise ValueError(f"{name} must be within [{low}, {high}], got {value!r}") - - return check - - -class SimLaserDelay(LaserDelay): - """LaserDelay without EPICS, with a simulated oscilloscope (local demo BEC, tests). - - Soft signals replace the five PVs; the ``*_pv`` arguments are accepted and ignored. They refuse - what the IOCs are assumed to refuse: |V1|, |V2| above 5 V and FINE outside 0..199. Start values: - laser period 2.0002 µs, delay card 150 / 37 (300.37 ns), V1 = 1, V2 = 0. - - The simulated laser emits a pulse train with the cavity period, locked to the RF: turning the - phase shifter by ``x`` ns (tracked continuously, also across whole RF periods) shifts the train - by ``x`` ns. The delay card triggers the laser, which fires the first pulse of the train at or - after the trigger. ``sim_scope`` is the X-ray minus laser time an oscilloscope would show; it - starts at -0.88 ns, so ``calibrate(-0.88)`` (or - ``calibrate(dev.laser_delay.sim_scope.get())``) is a correct calibration. After any move - ``sim_scope`` should equal the readback within 0.01 ns. ``sim_response`` is a pump-probe signal: - a step of width ``sim_rise`` ns at ``sim_time_zero`` (scope time), plus noise. - - ``sim_set_hardware(...)`` changes the simulated PVs as somebody outside BEC would (caput, an IOC - reboot), to try the safety checks. - - Args: - sim_cavity_period (float | None): cavity period of the simulated laser in ns. None: the - device's ``cavity_period``. A different value shows what a wrong ``cavity_period`` - does. - sim_time_zero (float): scope delay of the pump-probe step in ns. - sim_rise (float): width of the step in ns. - sim_noise (float): standard deviation of the noise on ``sim_response``. - """ - - USER_ACCESS = LaserDelay.USER_ACCESS + ["sim_set_hardware"] - #: The simulation may ramp faster than the beamline (demo, tests). - MIN_STEP_SLEEP = 0.0 - - delay_coarse = Cpt( - _HardwareSoftSignal, - value=150, - kind=Kind.normal, - validator=_sim_range("delay_coarse", 0, 65535), - doc="Delay card, units of 2 ns", - ) - delay_fine = Cpt( - _HardwareSoftSignal, - value=37, - kind=Kind.normal, - validator=_sim_range("delay_fine", 0, DELAY_CARD_FINE_PER_COARSE - 1), - doc="Delay card, units of 0.01 ns", - ) - evr_period = Cpt(Signal, value=2.0002, kind=Kind.config, doc="EVR laser period in µs") - phase_v1 = Cpt( - _HardwareSoftSignal, - value=1.0, - kind=Kind.normal, - validator=_sim_range("phase_v1", -PHASE_VOLTAGE_LIMIT, PHASE_VOLTAGE_LIMIT), - doc="Phase shifter V1", - ) - phase_v2 = Cpt( - _HardwareSoftSignal, - value=0.0, - kind=Kind.normal, - validator=_sim_range("phase_v2", -PHASE_VOLTAGE_LIMIT, PHASE_VOLTAGE_LIMIT), - doc="Phase shifter V2", - ) - sim_scope = Cpt( - ComputedPhaseSignal, - compute="read_sim_scope", - kind=Kind.normal, - doc="Simulated oscilloscope: X-ray minus laser in ns", - ) - sim_response = Cpt( - ComputedPhaseSignal, - compute="read_sim_response", - kind=Kind.hinted, - doc="Simulated pump-probe signal", - ) - - #: Simulated X-ray arrival and laser pulse train (ns, delay card time scale). With the start - #: values a pulse sits half a cavity period after the trigger, and the scope shows -0.88 ns. - _SIM_XRAY_TIME = 305.495 - _SIM_TRAIN_OFFSET = 6.125 - - def __init__( - self, - prefix: str = "", - *, - name: str, - sim_cavity_period: float | None = None, - sim_time_zero: float = 0.0, - sim_rise: float = 0.3, - sim_noise: float = 0.01, - **kwargs, - ): - super().__init__(prefix=prefix, name=name, **kwargs) - self._sim_cavity_period = sim_cavity_period - self._sim_time_zero = float(sim_time_zero) - self._sim_rise = float(sim_rise) - self._sim_noise = float(sim_noise) - self._sim_rng = np.random.default_rng() - self._sim_lock = threading.Lock() - self._sim_angle = math.atan2(self.phase_v2.get(), self.phase_v1.get()) - self._sim_phase_turned = 0.0 # unwrapped phase shift in ns since the start - self.phase_v1.subscribe(self._sim_track_phase, run=False) - self.phase_v2.subscribe(self._sim_track_phase, run=False) - - def sim_set_hardware( - self, - delay_coarse: int | None = None, - delay_fine: int | None = None, - phase_v1: float | None = None, - phase_v2: float | None = None, - ) -> None: - """Change the simulated PVs as somebody outside BEC would (caput, IOC reboot). - - A voltage jump is followed by the simulated laser like any other phase change. - """ - with self._hardware_writes(): - for key, value in ( - ("delay_coarse", delay_coarse), - ("delay_fine", delay_fine), - ("phase_v1", phase_v1), - ("phase_v2", phase_v2), - ): - if value is not None: - getattr(self, key).put(value) - - def _sim_track_phase(self, *_, **__) -> None: - """Follow the phase shifter continuously, counting whole RF periods (the laser lock).""" - with self._sim_lock: - v1, v2 = self.phase_v1.get(), self.phase_v2.get() - if math.hypot(v1, v2) == 0: - return # no phase defined; the lock keeps the last one - angle = math.atan2(v2, v1) - turned = (angle - self._sim_angle + math.pi) % (2 * math.pi) - math.pi - self._sim_angle = angle - self._sim_phase_turned += float(self.rf_period.get()) * turned / (2 * math.pi) - - def on_connected(self) -> None: - """ - Called after the device is connected and its signals are connected. - Default values for signals should be set here. - """ - # Real IOCs and the laser keep their state across a BEC restart; soft signals do not. The - # calibration file therefore also keeps the simulated state; bring it back first, then - # restore the calibration like the real device. - try: - content = self._read_saved_calibration() - sim = (content or {}).get("sim") - if sim: - self.sim_set_hardware(**sim.get("hardware", {})) - with ( - self._sim_lock - ): # the puts above were tracked as turns; reset to the saved state - self._sim_angle = math.atan2(self.phase_v2.get(), self.phase_v1.get()) - self._sim_phase_turned = float(sim["phase_turned"]) - except Exception as exc: # pylint: disable=broad-except - logger.error(f"{self.name}: could not restore the simulated hardware: {exc}") - super().on_connected() - - def _extra_saved_state(self) -> dict: - hardware = { - "delay_coarse": int(self.delay_coarse.get()), - "delay_fine": int(self.delay_fine.get()), - "phase_v1": float(self.phase_v1.get()), - "phase_v2": float(self.phase_v2.get()), - } - with self._sim_lock: - return {"sim": {"phase_turned": self._sim_phase_turned, "hardware": hardware}} - - def read_sim_scope(self) -> float: - """X-ray minus laser in ns, as the oscilloscope would show it.""" - cavity = self._sim_cavity_period or float(self.cavity_period.get()) - trigger = delay_card_time(self.delay_coarse.get(), self.delay_fine.get()) - with self._sim_lock: - train = self._SIM_TRAIN_OFFSET + self._sim_phase_turned - laser = train + math.ceil((trigger - train) / cavity) * cavity - return self._SIM_XRAY_TIME - laser - - def read_sim_response(self) -> float: - """Pump-probe signal: step at ``sim_time_zero`` (scope time) plus noise.""" - x = (self.read_sim_scope() - self._sim_time_zero) / (math.sqrt(2) * self._sim_rise) - noise = self._sim_rng.normal(0, self._sim_noise) if self._sim_noise > 0 else 0.0 - return 0.5 * (1 + math.erf(x)) + noise - - -def _merged_signature(init: Callable, base_init: Callable) -> inspect.Signature: - """``init``'s own parameters plus the keyword parameters of ``base_init``. - - BEC's device server only passes ``deviceConfig`` keys (and ``device_manager``/``scan_info``) - that appear in the signature of the device class itself, so a subclass ``__init__`` that - forwards ``**kwargs`` has to show the base class parameters. - """ - own = [p for p in inspect.signature(init).parameters.values() if p.kind != p.VAR_KEYWORD] - names = {p.name for p in own} - base = [ - p - for p in inspect.signature(base_init).parameters.values() - if p.kind == p.KEYWORD_ONLY and p.name not in names - ] - var_kw = [ - p for p in inspect.signature(base_init).parameters.values() if p.kind == p.VAR_KEYWORD - ] - return inspect.signature(init).replace(parameters=own + base + var_kw) - - -SimLaserDelay.__init__.__signature__ = _merged_signature( - SimLaserDelay.__init__, LaserDelay.__init__ -) diff --git a/superxas_bec/devices/laser_delay_sim.py b/superxas_bec/devices/laser_delay_sim.py new file mode 100644 index 0000000..d0cf2be --- /dev/null +++ b/superxas_bec/devices/laser_delay_sim.py @@ -0,0 +1,252 @@ +""" +Simulated laser delay for SuperXAS: ``LaserDelay`` without EPICS, with a simulated laser and +oscilloscope. For a local demo BEC and tests only; the beamline device is +``superxas_bec.devices.laser_delay.LaserDelay``. +""" + +from __future__ import annotations + +import inspect +import math +import threading +from typing import Callable + +import numpy as np +from bec_lib.logger import bec_logger +from ophyd import Component as Cpt +from ophyd import Kind, Signal + +from superxas_bec.devices.laser_delay import ( + DELAY_CARD_FINE_PER_COARSE, + PHASE_VOLTAGE_LIMIT, + CheckedSignal, + ComputedPhaseSignal, + LaserDelay, + _DeviceWritesOnly, + delay_card_time, +) + +logger = bec_logger.logger + + +class _HardwareSoftSignal(_DeviceWritesOnly, CheckedSignal): + """Soft stand-in for a hardware PV in the simulation (written only by the device).""" + + +def _sim_range(name: str, low: float, high: float) -> Callable[[object], None]: + def check(value): + try: + number = float(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{name} must be a number, got {value!r}") from exc + if not low <= number <= high: + raise ValueError(f"{name} must be within [{low}, {high}], got {value!r}") + + return check + + +class SimLaserDelay(LaserDelay): + """LaserDelay without EPICS, with a simulated oscilloscope (local demo BEC, tests). + + Soft signals replace the five PVs; the ``*_pv`` arguments are accepted and ignored. They refuse + what the IOCs are assumed to refuse: |V1|, |V2| above 5 V and FINE outside 0..199. Start values: + laser period 2.0002 µs, delay card 150 / 37 (300.37 ns), V1 = 1, V2 = 0. + + The simulated laser emits a pulse train with the cavity period, locked to the RF: turning the + phase shifter by ``x`` ns (tracked continuously, also across whole RF periods) shifts the train + by ``x`` ns. The delay card triggers the laser, which fires the first pulse of the train at or + after the trigger. ``sim_scope`` is the X-ray minus laser time an oscilloscope would show; it + starts at -0.88 ns, so ``calibrate(-0.88)`` (or + ``calibrate(dev.laser_delay.sim_scope.get())``) is a correct calibration. After any move + ``sim_scope`` should equal the readback within 0.01 ns. ``sim_response`` is a pump-probe signal: + a step of width ``sim_rise`` ns at ``sim_time_zero`` (scope time), plus noise. + + ``sim_set_hardware(...)`` changes the simulated PVs as somebody outside BEC would (caput, an IOC + reboot), to try the safety checks. + + Args: + sim_cavity_period (float | None): cavity period of the simulated laser in ns. None: the + device's ``cavity_period``. A different value shows what a wrong ``cavity_period`` + does. + sim_time_zero (float): scope delay of the pump-probe step in ns. + sim_rise (float): width of the step in ns. + sim_noise (float): standard deviation of the noise on ``sim_response``. + """ + + USER_ACCESS = LaserDelay.USER_ACCESS + ["sim_set_hardware"] + #: The simulation may ramp faster than the beamline (demo, tests). + MIN_STEP_SLEEP = 0.0 + + delay_coarse = Cpt( + _HardwareSoftSignal, + value=150, + kind=Kind.normal, + validator=_sim_range("delay_coarse", 0, 65535), + doc="Delay card, units of 2 ns", + ) + delay_fine = Cpt( + _HardwareSoftSignal, + value=37, + kind=Kind.normal, + validator=_sim_range("delay_fine", 0, DELAY_CARD_FINE_PER_COARSE - 1), + doc="Delay card, units of 0.01 ns", + ) + evr_period = Cpt(Signal, value=2.0002, kind=Kind.config, doc="EVR laser period in µs") + phase_v1 = Cpt( + _HardwareSoftSignal, + value=1.0, + kind=Kind.normal, + validator=_sim_range("phase_v1", -PHASE_VOLTAGE_LIMIT, PHASE_VOLTAGE_LIMIT), + doc="Phase shifter V1", + ) + phase_v2 = Cpt( + _HardwareSoftSignal, + value=0.0, + kind=Kind.normal, + validator=_sim_range("phase_v2", -PHASE_VOLTAGE_LIMIT, PHASE_VOLTAGE_LIMIT), + doc="Phase shifter V2", + ) + sim_scope = Cpt( + ComputedPhaseSignal, + compute="read_sim_scope", + kind=Kind.normal, + doc="Simulated oscilloscope: X-ray minus laser in ns", + ) + sim_response = Cpt( + ComputedPhaseSignal, + compute="read_sim_response", + kind=Kind.hinted, + doc="Simulated pump-probe signal", + ) + + #: Simulated X-ray arrival and laser pulse train (ns, delay card time scale). With the start + #: values a pulse sits half a cavity period after the trigger, and the scope shows -0.88 ns. + _SIM_XRAY_TIME = 305.495 + _SIM_TRAIN_OFFSET = 6.125 + + def __init__( + self, + prefix: str = "", + *, + name: str, + sim_cavity_period: float | None = None, + sim_time_zero: float = 0.0, + sim_rise: float = 0.3, + sim_noise: float = 0.01, + **kwargs, + ): + super().__init__(prefix=prefix, name=name, **kwargs) + self._sim_cavity_period = sim_cavity_period + self._sim_time_zero = float(sim_time_zero) + self._sim_rise = float(sim_rise) + self._sim_noise = float(sim_noise) + self._sim_rng = np.random.default_rng() + self._sim_lock = threading.Lock() + self._sim_angle = math.atan2(self.phase_v2.get(), self.phase_v1.get()) + self._sim_phase_turned = 0.0 # unwrapped phase shift in ns since the start + self.phase_v1.subscribe(self._sim_track_phase, run=False) + self.phase_v2.subscribe(self._sim_track_phase, run=False) + + def sim_set_hardware( + self, + delay_coarse: int | None = None, + delay_fine: int | None = None, + phase_v1: float | None = None, + phase_v2: float | None = None, + ) -> None: + """Change the simulated PVs as somebody outside BEC would (caput, IOC reboot). + + A voltage jump is followed by the simulated laser like any other phase change. + """ + with self._hardware_writes(): + for key, value in ( + ("delay_coarse", delay_coarse), + ("delay_fine", delay_fine), + ("phase_v1", phase_v1), + ("phase_v2", phase_v2), + ): + if value is not None: + getattr(self, key).put(value) + + def _sim_track_phase(self, *_, **__) -> None: + """Follow the phase shifter continuously, counting whole RF periods (the laser lock).""" + with self._sim_lock: + v1, v2 = self.phase_v1.get(), self.phase_v2.get() + if math.hypot(v1, v2) == 0: + return # no phase defined; the lock keeps the last one + angle = math.atan2(v2, v1) + turned = (angle - self._sim_angle + math.pi) % (2 * math.pi) - math.pi + self._sim_angle = angle + self._sim_phase_turned += float(self.rf_period.get()) * turned / (2 * math.pi) + + def on_connected(self) -> None: + """ + Called after the device is connected and its signals are connected. + Default values for signals should be set here. + """ + # Real IOCs and the laser keep their state across a BEC restart; soft signals do not. The + # calibration file therefore also keeps the simulated state; bring it back first, then + # restore the calibration like the real device. + try: + content = self._read_saved_calibration() + sim = (content or {}).get("sim") + if sim: + self.sim_set_hardware(**sim.get("hardware", {})) + with ( + self._sim_lock + ): # the puts above were tracked as turns; reset to the saved state + self._sim_angle = math.atan2(self.phase_v2.get(), self.phase_v1.get()) + self._sim_phase_turned = float(sim["phase_turned"]) + except Exception as exc: # pylint: disable=broad-except + logger.error(f"{self.name}: could not restore the simulated hardware: {exc}") + super().on_connected() + + def _extra_saved_state(self) -> dict: + hardware = { + "delay_coarse": int(self.delay_coarse.get()), + "delay_fine": int(self.delay_fine.get()), + "phase_v1": float(self.phase_v1.get()), + "phase_v2": float(self.phase_v2.get()), + } + with self._sim_lock: + return {"sim": {"phase_turned": self._sim_phase_turned, "hardware": hardware}} + + def read_sim_scope(self) -> float: + """X-ray minus laser in ns, as the oscilloscope would show it.""" + cavity = self._sim_cavity_period or float(self.cavity_period.get()) + trigger = delay_card_time(self.delay_coarse.get(), self.delay_fine.get()) + with self._sim_lock: + train = self._SIM_TRAIN_OFFSET + self._sim_phase_turned + laser = train + math.ceil((trigger - train) / cavity) * cavity + return self._SIM_XRAY_TIME - laser + + def read_sim_response(self) -> float: + """Pump-probe signal: step at ``sim_time_zero`` (scope time) plus noise.""" + x = (self.read_sim_scope() - self._sim_time_zero) / (math.sqrt(2) * self._sim_rise) + noise = self._sim_rng.normal(0, self._sim_noise) if self._sim_noise > 0 else 0.0 + return 0.5 * (1 + math.erf(x)) + noise + + +def _merged_signature(init: Callable, base_init: Callable) -> inspect.Signature: + """``init``'s own parameters plus the keyword parameters of ``base_init``. + + BEC's device server only passes ``deviceConfig`` keys (and ``device_manager``/``scan_info``) + that appear in the signature of the device class itself, so a subclass ``__init__`` that + forwards ``**kwargs`` has to show the base class parameters. + """ + own = [p for p in inspect.signature(init).parameters.values() if p.kind != p.VAR_KEYWORD] + names = {p.name for p in own} + base = [ + p + for p in inspect.signature(base_init).parameters.values() + if p.kind == p.KEYWORD_ONLY and p.name not in names + ] + var_kw = [ + p for p in inspect.signature(base_init).parameters.values() if p.kind == p.VAR_KEYWORD + ] + return inspect.signature(init).replace(parameters=own + base + var_kw) + + +SimLaserDelay.__init__.__signature__ = _merged_signature( + SimLaserDelay.__init__, LaserDelay.__init__ +) diff --git a/tests/tests_devices/test_laser_delay.py b/tests/tests_devices/test_laser_delay.py index f2a9a4f..0c98d8e 100644 --- a/tests/tests_devices/test_laser_delay.py +++ b/tests/tests_devices/test_laser_delay.py @@ -37,7 +37,6 @@ from superxas_bec.devices.laser_delay import ( LaserDelayHardwareError, LaserDelayNotCalibratedError, LaserDelayPersistenceError, - SimLaserDelay, delay_card_setting, fold_phase_delta, phase_from_voltages, @@ -45,6 +44,7 @@ from superxas_bec.devices.laser_delay import ( voltages_for_phase, wrap_phase, ) +from superxas_bec.devices.laser_delay_sim import SimLaserDelay CONFIG_DIR = Path(__file__).resolve().parents[2] / "superxas_bec" / "device_configs" SCOPE = 546.04 - 546.92 @@ -301,7 +301,7 @@ def test_yaml_entries_map_onto_constructor_arguments(): entry = standard["laser_delay"] assert entry["deviceClass"] == "superxas_bec.devices.laser_delay.LaserDelay" assert entry["onFailure"] == "buffer" - assert sim["laser_delay"]["deviceClass"] == "superxas_bec.devices.laser_delay.SimLaserDelay" + assert sim["laser_delay"]["deviceClass"] == "superxas_bec.devices.laser_delay_sim.SimLaserDelay" for config in (entry, sim["laser_delay"]): with mock.patch.object(ophyd, "cl") as mock_cl: mock_cl.get_pv = MockPV