refactor(devices): move the simulated laser delay into its own module
This commit is contained in:
wyzula_j committed
1 parent
70c08e9c11
commit
e691a816ab
4 files changed
+255
-227
No files matched your search
@@ -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
|
||||
|
||||
@@ -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__
|
||||
)
|
||||
@@ -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__
|
||||
)
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user