From b6121cc3f045f833608c0c2244d2a2f4affd14df Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Sat, 3 Oct 2026 18:47:34 +0200 Subject: [PATCH] feat(devices): add opt-in shortest phase path to the laser delay --- .../device_configs/x10da_laser_delay.yaml | 1 + superxas_bec/devices/laser_delay.py | 175 +++++++++++++----- tests/tests_devices/test_laser_delay.py | 152 +++++++++++++++ .../tests_devices/test_laser_delay_golden.py | 69 ++++++- 4 files changed, 349 insertions(+), 48 deletions(-) diff --git a/superxas_bec/device_configs/x10da_laser_delay.yaml b/superxas_bec/device_configs/x10da_laser_delay.yaml index 08a5ca3..e464cd1 100644 --- a/superxas_bec/device_configs/x10da_laser_delay.yaml +++ b/superxas_bec/device_configs/x10da_laser_delay.yaml @@ -28,6 +28,7 @@ laser_delay: steps_per_rf_period: 36 step_sleep: 0.1 phase_reference: stored + phase_path: script # "shortest" is opt-in, see LASER_DELAY.md deviceTags: - laser - timing diff --git a/superxas_bec/devices/laser_delay.py b/superxas_bec/devices/laser_delay.py index 6bc714d..2d85813 100644 --- a/superxas_bec/devices/laser_delay.py +++ b/superxas_bec/devices/laser_delay.py @@ -17,6 +17,12 @@ How a delay ``t`` (X-ray minus laser, as read on the oscilloscope) is set: always starts at the phase read back from V1/V2 and walks in steps of ``sign(dP) * rf_period / steps_per_rf_period`` with ``step_sleep`` seconds between steps; the last step lands exactly on ``readback + dP``. Afterwards ``P`` is stored as the new ``phase_t``. +* **Phase path.** ``phase_path="script"`` (default) ramps ``dP`` as above, up to about two cavity + periods. ``phase_path="shortest"`` (opt-in) first folds ``dP`` by whole cavity periods into + ``(-cavity_period/2, +cavity_period/2]`` and stores ``P - k * cavity_period``. That is only the + same physical delay if the laser timing repeats every cavity period *and* the cavity period is a + whole number of RF periods: with ``rf_period=2.0`` and ``cavity_period=12.01`` each folded period + leaves the phase shifter 0.01 ns away from where the script would put it (see ``LASER_DELAY.md``). The stored phase is "unwrapped": the voltages only encode the phase modulo one RF period, while the stored value keeps track of how many RF periods the shifter has been turned. That is why the script @@ -71,6 +77,7 @@ DELAY_CARD_FINE_UNIT = 0.01 DELAY_CARD_FINE_DIVISOR = 0.009999 PHASE_REFERENCES = ("stored", "readback") +PHASE_PATHS = ("script", "shortest") class LaserDelayNotCalibratedError(RuntimeError): @@ -137,6 +144,20 @@ def wrap_phase(value: float, rf_period: float) -> float: return (value + rf_period / 2) % rf_period - rf_period / 2 +def fold_phase_delta(delta: float, cavity_period: float) -> tuple[float, int]: + """Fold a relative phase move by whole cavity periods into (-cavity/2, +cavity/2]. + + Returns: + tuple[float, int]: ``(delta - k * cavity_period, k)``. A move that is already inside the + interval is returned unchanged (``k == 0``, bit for bit). + """ + half = cavity_period / 2 + if -half < delta <= half: + return delta, 0 + wraps = math.ceil((delta - half) / cavity_period) + return delta - wraps * cavity_period, int(wraps) + + @dataclass(frozen=True) class LaserDelayCalibration: """Calibration of the delay, the four numbers of the script's ``Timing.offset`` file.""" @@ -159,6 +180,48 @@ class _Constants: steps_per_rf_period: int step_sleep: float phase_reference: str + phase_path: str + + +@dataclass(frozen=True) +class _PhasePlan: + """Phase shifter part of a move: ramp from ``start`` by ``delta``, then store ``stored_after``.""" + + target: float # the script's ForPhaseShifter, P + start: float # read back from V1/V2 + reference: float # what delta is measured from: stored phase or read-back + delta: float # relative ramp, folded for phase_path "shortest" + cavity_wraps: int # whole cavity periods folded away (0 for phase_path "script") + offsets: np.ndarray # intermediate ramp offsets, without the final step to delta + stored_after: float # new phase_t: P, or P - cavity_wraps * cavity_period + + @property + def n_puts(self) -> int: + """V1/V2 pairs written, including the final one.""" + return len(self.offsets) + 1 + + +def plan_phase_move( + target_t: float, calibration: LaserDelayCalibration, constants: _Constants, start: float +) -> _PhasePlan: + """Phase shifter ramp for a move to delay ``target_t``, starting at read-back phase ``start``.""" + phase_target = (-target_t) % constants.cavity_period - calibration.offset_phase_t + reference = calibration.phase_t if constants.phase_reference == "stored" else start + delta = phase_target - reference + wraps = 0 + if constants.phase_path == "shortest": + delta, wraps = fold_phase_delta(delta, constants.cavity_period) + offsets = phase_ramp_offsets(delta, constants.rf_period, constants.steps_per_rf_period) + stored_after = phase_target - wraps * constants.cavity_period if wraps else phase_target + return _PhasePlan( + target=phase_target, + start=start, + reference=reference, + delta=delta, + cavity_wraps=wraps, + offsets=offsets, + stored_after=stored_after, + ) # --------------------------------------------------------------------------------------------- # @@ -281,6 +344,11 @@ def _phase_reference(value): raise ValueError(f"phase_reference must be one of {PHASE_REFERENCES}, got {value!r}") +def _phase_path(value): + if value not in PHASE_PATHS: + raise ValueError(f"phase_path must be one of {PHASE_PATHS}, got {value!r}") + + # --------------------------------------------------------------------------------------------- # # Device # # --------------------------------------------------------------------------------------------- # @@ -314,6 +382,8 @@ class LaserDelay(PSIDeviceBase, PositionerBase): step_sleep (float): seconds between two phase ramp steps. phase_reference (str): ``"stored"`` (script behaviour) or ``"readback"``; what the relative phase move is computed from, see module docstring. + phase_path (str): ``"script"`` (default, script behaviour) or ``"shortest"``: fold the + relative phase move into half a cavity period either way. Opt-in, see module docstring. limits (list[float] | None): optional soft limits ``[low, high]`` in ns. """ @@ -385,6 +455,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): phase_reference = Cpt( CheckedSignal, value="stored", kind=Kind.config, validator=_phase_reference ) + phase_path = Cpt(CheckedSignal, value="script", kind=Kind.config, validator=_phase_path) # Soft limits (BEC publishes them as the device limits). Equal values mean no limits. low_limit_travel = Cpt( @@ -418,6 +489,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): steps_per_rf_period: int = 36, step_sleep: float = 0.1, phase_reference: str = "stored", + phase_path: str = "script", limits: list[float] | tuple[float, float] | None = None, scan_info: ScanInfo | None = None, device_manager: DeviceManagerBase | None = None, @@ -444,6 +516,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): self.steps_per_rf_period.put(steps_per_rf_period) self.step_sleep.put(float(step_sleep)) self.phase_reference.put(phase_reference) + self.phase_path.put(phase_path) if limits is not None: if len(limits) != 2: raise ValueError(f"limits must be [low, high], got {limits!r}") @@ -638,36 +711,48 @@ class LaserDelay(PSIDeviceBase, PositionerBase): def plan_move(self, position: float) -> dict: """Dry run of a move: what would be written, without writing anything. - Reads V1/V2 for the ramp start. Useful to compare with the script before the first move. + Reads V1/V2 for the ramp start. Useful to compare with the script before the first move, + and to see how long a move takes with the current ``phase_path``. Returns: - dict: ``delay_coarse``, ``delay_fine``, ``phase_target``, ``phase_start``, - ``phase_delta``, ``phase_steps`` (puts of V1/V2 incl. the last one) and + dict: ``delay_coarse``, ``delay_fine``, ``phase_target`` (the script's target ``P``), + ``phase_start``, ``phase_delta`` (the ramp actually done), ``phase_path``, + ``cavity_wraps`` (``k``: the stored phase after the move is ``P - k * cavity_period``; + always 0 for ``phase_path="script"``, counts the folds since the calibration for + ``"shortest"``), ``wrap_offset_ns`` (``-k * cavity_period`` modulo the RF period: how + far the phase shifter ends from where the script would put it; 0.01 ns per fold with + the script's 2.0/12.01 ns), ``phase_steps`` (puts of V1/V2 incl. the last one) and ``estimated_duration_s``. """ target = float(position) self.check_value(target) calibration = self._require_calibration() constants = self._constants() - total_t = -calibration.offset_t - target - coarse, fine = delay_card_setting(total_t, calibration.laser_period) - phase_target = (-target) % constants.cavity_period - calibration.offset_phase_t - phase_start = self._read_phase(constants.rf_period) - reference = calibration.phase_t if constants.phase_reference == "stored" else phase_start - delta = phase_target - reference - offsets = phase_ramp_offsets(delta, constants.rf_period, constants.steps_per_rf_period) - n_puts = len(offsets) + 1 + coarse, fine = delay_card_setting(-calibration.offset_t - target, calibration.laser_period) + plan = plan_phase_move( + target, calibration, constants, start=self._read_phase(constants.rf_period) + ) return { "target": target, "delay_coarse": coarse, "delay_fine": fine, - "phase_target": float(phase_target), - "phase_start": float(phase_start), - "phase_delta": float(delta), - "phase_steps": n_puts, - "estimated_duration_s": n_puts * constants.step_sleep, + "phase_target": float(plan.target), + "phase_start": float(plan.start), + "phase_delta": float(plan.delta), + "phase_path": constants.phase_path, + "cavity_wraps": plan.cavity_wraps, + "wrap_offset_ns": self._wrap_offset(plan.cavity_wraps, constants), + "phase_steps": plan.n_puts, + "estimated_duration_s": plan.n_puts * constants.step_sleep, } + @staticmethod + def _wrap_offset(cavity_wraps: int, constants: _Constants) -> float: + """Phase shifter offset in ns, modulo the RF period, left by folding whole cavity periods.""" + if cavity_wraps == 0: + return 0.0 + return float(wrap_phase(-cavity_wraps * constants.cavity_period, constants.rf_period)) + # ----------------------------------------------------------------------------------------- # # Readbacks used by the computed signals # # ----------------------------------------------------------------------------------------- # @@ -725,6 +810,7 @@ class LaserDelay(PSIDeviceBase, PositionerBase): steps_per_rf_period=int(self.steps_per_rf_period.get()), step_sleep=float(self.step_sleep.get()), phase_reference=str(self.phase_reference.get()), + phase_path=str(self.phase_path.get()), ) def _calibration(self) -> LaserDelayCalibration: @@ -832,24 +918,21 @@ class LaserDelay(PSIDeviceBase, PositionerBase): """The script's SetDelay: delay card, then the phase shifter ramp, then bookkeeping.""" if stop_event.is_set(): raise DeviceStoppedError(f"Device {self.name} has been stopped") - total_t = -calibration.offset_t - target - phase_target = (-target) % constants.cavity_period - calibration.offset_phase_t - coarse, fine = delay_card_setting(total_t, calibration.laser_period) + coarse, fine = delay_card_setting(-calibration.offset_t - target, calibration.laser_period) written_any = False done_offset = None - completed = False - reference = calibration.phase_t + plan = None try: self.delay_coarse.put(coarse) written_any = True self.delay_fine.put(fine) - phase_start = self._read_phase(constants.rf_period) - if constants.phase_reference == "readback": - reference = phase_start - else: - mismatch = wrap_phase(calibration.phase_t - phase_start, constants.rf_period) + plan = plan_phase_move( + target, calibration, constants, start=self._read_phase(constants.rf_period) + ) + if constants.phase_reference == "stored": + mismatch = wrap_phase(calibration.phase_t - plan.start, constants.rf_period) if abs(mismatch) > constants.rf_period / (2 * constants.steps_per_rf_period): logger.warning( f"{self.name}: stored phase and V1/V2 read-back differ by " @@ -857,37 +940,39 @@ class LaserDelay(PSIDeviceBase, PositionerBase): "The ramp starts at the read-back, so this move ends that far from the " "intended phase." ) - delta = phase_target - reference - offsets = phase_ramp_offsets(delta, constants.rf_period, constants.steps_per_rf_period) + wraps = "" + if plan.cavity_wraps: + wraps = ( + f"; shortest path, {plan.cavity_wraps:+d} cavity period(s) from the script's " + f"phase: {self._wrap_offset(plan.cavity_wraps, constants):+.4f} ns (mod RF)" + ) logger.info( f"{self.name}: move to {target} ns: DELAY={coarse}, FINE={fine}, phase " - f"{phase_start:.4f} -> +{delta:.4f} ns in {len(offsets) + 1} steps " - f"(~{(len(offsets) + 1) * constants.step_sleep:.1f} s)" + f"{plan.start:.4f} {plan.delta:+.4f} ns in {plan.n_puts} steps " + f"(~{plan.n_puts * constants.step_sleep:.1f} s){wraps}" ) - for offset in offsets: + for offset in plan.offsets: if stop_event.is_set(): raise DeviceStoppedError(f"Device {self.name} has been stopped") - self._write_phase(phase_start + offset, constants) + self._write_phase(plan.start + offset, constants) done_offset = offset if stop_event.wait(constants.step_sleep): raise DeviceStoppedError(f"Device {self.name} has been stopped") if stop_event.is_set(): raise DeviceStoppedError(f"Device {self.name} has been stopped") - self._write_phase(phase_start + delta, constants) - done_offset = delta - completed = True + self._write_phase(plan.start + plan.delta, constants) + done_offset = None + self.phase_t.put(float(plan.stored_after), internal=True) # The script also sleeps after the last step; a stop now no longer matters. stop_event.wait(constants.step_sleep) - finally: - if completed: - self.phase_t.put(float(phase_target), internal=True) - self._update_readback(target) - else: - if done_offset is not None: - # Keep stored and real (unwrapped) phase consistent for the next move. - self.phase_t.put(float(reference + done_offset), internal=True) - if written_any: - self._update_readback(math.nan) + self._update_readback(target) + except BaseException: + if done_offset is not None: + # Keep stored and real (unwrapped) phase consistent for the next move. + self.phase_t.put(float(plan.reference + done_offset), internal=True) + if written_any: + self._update_readback(math.nan) + raise class SimLaserDelay(LaserDelay): diff --git a/tests/tests_devices/test_laser_delay.py b/tests/tests_devices/test_laser_delay.py index 7e5ba9b..ab8770b 100644 --- a/tests/tests_devices/test_laser_delay.py +++ b/tests/tests_devices/test_laser_delay.py @@ -31,6 +31,7 @@ from superxas_bec.devices.laser_delay import ( LaserDelayNotCalibratedError, SimLaserDelay, delay_card_setting, + fold_phase_delta, phase_from_voltages, phase_ramp_offsets, voltages_for_phase, @@ -109,6 +110,7 @@ def test_default_pvs_constants_and_positioner_type(laser_delay): assert config["laser_delay_steps_per_rf_period"]["value"] == 36 assert config["laser_delay_step_sleep"]["value"] == 0.0 assert config["laser_delay_phase_reference"]["value"] == "stored" + assert config["laser_delay_phase_path"]["value"] == "script" assert get_device_base_class(laser_delay) == "positioner" assert laser_delay.hints == {"fields": ["laser_delay"]} assert laser_delay.egu == "ns" @@ -172,6 +174,7 @@ def test_config_overrides_pvs_and_constants(): "kwargs", [ {"phase_reference": "both"}, + {"phase_path": "fast"}, {"steps_per_rf_period": 0}, {"steps_per_rf_period": 2.5}, {"rf_period": -2.0}, @@ -198,6 +201,10 @@ def test_runtime_constant_update_is_validated(calibrated): with pytest.raises(ValueError): calibrated.phase_reference.set("sometimes").wait(timeout=2) assert calibrated._constants().phase_reference == "stored" + with pytest.raises(ValueError): + calibrated.phase_path.set("fastest").wait(timeout=2) + calibrated.phase_path.set("shortest").wait(timeout=2) + assert calibrated._constants().phase_path == "shortest" def test_runtime_limits_are_validated(calibrated): @@ -707,6 +714,151 @@ def test_first_move_after_calibration_can_ramp_two_cavity_periods(laser_delay): laser_delay.set(target).wait(timeout=10) +# --------------------------------------------------------------------------------------------- # +# phase_path "shortest" # +# --------------------------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + "delta, expected, wraps", + [ + (0.0, 0.0, 0), + (6.005, 6.005, 0), # +cavity/2 is inside + (-6.005, 6.005, -1), # -cavity/2 is not + (-6.0, -6.0, 0), + (6.006, 6.006 - 12.01, 1), + (16.13, 16.13 - 12.01, 1), + (11.51, 11.51 - 12.01, 1), + (-24.03, -24.03 + 2 * 12.01, -2), + (30.0, 30.0 - 2 * 12.01, 2), + ], +) +def test_fold_phase_delta(delta, expected, wraps): + folded, k = fold_phase_delta(delta, 12.01) + assert k == wraps and isinstance(k, int) + assert folded == pytest.approx(expected, abs=1e-12) + + +def test_fold_phase_delta_properties(): + rng = np.random.default_rng(7) + for delta in rng.uniform(-60, 60, 2000): + folded, k = fold_phase_delta(float(delta), 12.01) + assert -6.005 - 1e-9 < folded <= 6.005 + 1e-9 + assert folded == pytest.approx(delta - k * 12.01, abs=1e-12) + if -6.005 < delta <= 6.005: + assert folded == delta and k == 0 # unchanged, bit for bit + + +SHORTEST_TARGETS = [-5.0, 5.0, -5.0, 0.0, 0.5, 11.9, 12.2, -24.0, 37.25, -0.88, 400.0, -1500.3] + + +def run_targets(device: LaserDelay, targets: list) -> list[dict]: + """Plan and move to each target; return the plans with the final voltages and stored phase.""" + results = [] + for target in targets: + plan = device.plan_move(target) + device.set(target).wait(timeout=30) + plan["v1"], plan["v2"] = device.phase_v1.get(), device.phase_v2.get() + plan["stored"] = device.phase_t.get() + plan["mismatch"] = device.read_phase_mismatch() + plan["readback"] = device.readback.get() + results.append(plan) + return results + + +@pytest.mark.parametrize("rf_period, cavity_period", [(2.0, 12.01), (2.0014, 12.0084)]) +def test_shortest_path_ramps_at_most_half_a_cavity_period(rf_period, cavity_period): + """Same delay card as the script path, at most cavity/2 of phase per move (also the first move + after calibration), consistent bookkeeping, and the phase shifter ends where the script path + puts it, up to the documented wrap offset (0 when the cavity is a whole number of RF periods). + """ + kwargs = {"step_sleep": 0.0, "rf_period": rf_period, "cavity_period": cavity_period} + runs = {} + for path in ("script", "shortest"): + with patched_device(LaserDelay, name="laser_delay", phase_path=path, **kwargs) as device: + set_hardware(device) + device.calibrate(SCOPE) + runs[path] = run_targets(device, SHORTEST_TARGETS) + max_steps = math.ceil((cavity_period / 2) / (rf_period / 36)) + 1 + for script, short in zip(runs["script"], runs["shortest"]): + assert (short["delay_coarse"], short["delay_fine"]) == ( + script["delay_coarse"], + script["delay_fine"], + ) + assert short["phase_path"] == "shortest" and script["cavity_wraps"] == 0 + assert abs(short["phase_delta"]) <= cavity_period / 2 + 1e-9 + assert short["phase_steps"] <= max_steps + assert short["readback"] == short["target"] + assert short["mismatch"] == pytest.approx(0, abs=1e-9) + k = short["cavity_wraps"] + assert short["stored"] == pytest.approx(script["stored"] - k * cavity_period, abs=1e-9) + end_short = phase_from_voltages(short["v1"], short["v2"], rf_period) + end_script = phase_from_voltages(script["v1"], script["v2"], rf_period) + offset = wrap_phase(-k * cavity_period, rf_period) + assert short["wrap_offset_ns"] == pytest.approx(offset, abs=1e-12) + assert wrap_phase(end_short - end_script - offset, rf_period) == pytest.approx(0, abs=1e-9) + if cavity_period == 12.0084: + assert offset == pytest.approx(0, abs=1e-9) + else: + assert offset == pytest.approx(wrap_phase(-0.01 * k, 2.0), abs=1e-9) + first = runs["shortest"][0] + assert runs["script"][0]["phase_delta"] > cavity_period # script: long first move + assert first["phase_delta"] == pytest.approx( + runs["script"][0]["phase_delta"] - cavity_period, abs=1e-9 + ) + + +def test_shortest_path_is_the_script_path_when_no_fold_is_needed(): + """Moves that stay within half a cavity period write exactly what the script path writes.""" + targets = [1.5, 2.0, 3.0, 2.2, 7.0] + with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device: + set_hardware(device) + device.calibrate(SCOPE) + device.set(1.0).wait(timeout=10) + state = device.get_calibration() + hardware = { + "coarse": device.delay_coarse.get(), + "fine": device.delay_fine.get(), + "v1": device.phase_v1.get(), + "v2": device.phase_v2.get(), + } + log_script = record_puts(device) + for target in targets: + device.set(target).wait(timeout=10) + with patched_device( + LaserDelay, name="laser_delay", step_sleep=0.0, phase_path="shortest" + ) as device: + set_hardware(device, **hardware) + device.set_calibration(**state, current_delay=1.0) + log_short = record_puts(device) + for target in targets: + assert device.plan_move(target)["cavity_wraps"] == 0 + device.set(target).wait(timeout=10) + assert log_short == log_script + + +def test_stop_during_a_shortest_path_move_keeps_the_bookkeeping(laser_delay): + laser_delay.phase_path.put("shortest") + laser_delay.calibrate(SCOPE) + laser_delay.step_sleep.put(0.02) + log = record_puts(laser_delay) + plan = laser_delay.plan_move(-5.0) + assert plan["cavity_wraps"] == 1 and 0 < plan["phase_delta"] < 6.005 + status = laser_delay.set(-5.0) + assert wait_until(lambda: len(log) >= 2 + 2 * 5) + laser_delay.stop() + with pytest.raises(DeviceStoppedError): + status.wait(timeout=1) + assert wait_until(lambda: not laser_delay.moving) + assert laser_delay.read_phase_mismatch() == pytest.approx(0, abs=1e-9) + laser_delay.step_sleep.put(0.0) + laser_delay.set(-5.0).wait(timeout=10) + end = phase_from_voltages(laser_delay.phase_v1.get(), laser_delay.phase_v2.get(), 2.0) + expected = plan["phase_start"] + plan["phase_delta"] + assert wrap_phase(end - expected, 2.0) == pytest.approx(0, abs=1e-9) + assert laser_delay.phase_t.get() == pytest.approx(plan["phase_target"] - 12.01, abs=1e-9) + + @pytest.mark.parametrize("value", [float("nan"), float("inf"), "abc"]) def test_rejects_non_finite_targets(calibrated, value): with pytest.raises(ValueError): diff --git a/tests/tests_devices/test_laser_delay_golden.py b/tests/tests_devices/test_laser_delay_golden.py index 98c2881..68132f3 100644 --- a/tests/tests_devices/test_laser_delay_golden.py +++ b/tests/tests_devices/test_laser_delay_golden.py @@ -25,8 +25,10 @@ from ophyd_devices.tests.utils import patched_device from superxas_bec.devices.laser_delay import ( LaserDelay, delay_card_setting, + phase_from_voltages, phase_ramp_offsets, voltages_for_phase, + wrap_phase, ) REFERENCE_SCRIPT = Path(__file__).parent / "reference" / "DelayLaser.py.txt" @@ -182,9 +184,9 @@ def set_pv_values(device: LaserDelay, values: dict) -> None: signals[pvname]._read_pv.mock_data = value -def run_device(scenario: Scenario) -> dict: - """Run a scenario through the device.""" - with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0) as device: +def run_device(scenario: Scenario, **device_kwargs) -> dict: + """Run a scenario through the device (default configuration unless ``device_kwargs``).""" + with patched_device(LaserDelay, name="laser_delay", step_sleep=0.0, **device_kwargs) as device: set_pv_values(device, scenario.initial) log = record_puts(device) calibration = device.calibrate(scenario.scope) @@ -279,6 +281,67 @@ def test_device_matches_reference_script(scenario_id, monkeypatch, tmp_path): assert act_move["readback"] == exp_move["target"] +@pytest.mark.parametrize("scenario_id", list(SCENARIOS)) +def test_shortest_phase_path_against_reference_script(scenario_id, monkeypatch, tmp_path): + """Opt-in phase_path="shortest" on the golden scenarios. + + The delay card writes are the script's. Each ramp is at most half a cavity period and turns the + phase shifter by exactly the change of the stored phase, which stays the script's phase minus + whole cavity periods (k). Without hand-made voltage changes, every move therefore ends where + the script's move ends up to k x 12.01 ns of phase, i.e. -k x 0.01 ns modulo the 2 ns RF + period: the documented caveat of folding with RFPeriod 2.0 and CavityPeriod 12.01. + """ + scenario = SCENARIOS[scenario_id] + expected = run_reference(scenario, monkeypatch, tmp_path) + actual = run_device(scenario, phase_path="shortest") + for key, exp, act in zip( + ("laser_period", "offset_t", "offset_phase_t", "phase_t"), + expected["calibration"], + actual["calibration"], + ): + assert_same_value(exp, act, f"{scenario_id}: calibration {key}") + + max_pairs = math.ceil(6.005 / (2.0 / 36)) + 1 + touched = False + voltages = {"exp": (scenario.initial[V1], scenario.initial[V2])} + voltages["act"] = voltages["exp"] + stored = {"exp": expected["calibration"][3], "act": actual["calibration"][3]} + moves = iter(zip(expected["moves"], actual["moves"], strict=True)) + for step in scenario.steps: + if isinstance(step, dict): # both programs see the same hand-made change + voltages = { + key: (step.get(V1, v[0]), step.get(V2, v[1])) for key, v in voltages.items() + } + touched = True + continue + exp_move, act_move = next(moves) + context = f"{scenario_id}: move to {exp_move['target']}" + assert_same_puts(exp_move["puts"][:2], act_move["puts"][:2], f"{context}: delay card") + assert 1 <= (len(act_move["puts"]) - 2) // 2 <= max_pairs, context + for index in range(3): # laser_period, offset_t, offset_phase_t + assert_same_value(exp_move["state"][index], act_move["state"][index], context) + assert act_move["readback"] == exp_move["target"] + + wraps = (exp_move["state"][3] - act_move["state"][3]) / 12.01 + assert wraps == pytest.approx(round(wraps), abs=1e-9), context + turned = act_move["state"][3] - stored["act"] + assert abs(turned) <= 6.005 + 1e-9, context + start = {key: phase_from_voltages(*v, 2.0) for key, v in voltages.items()} + voltages = { + "exp": (exp_move["puts"][-2][1], exp_move["puts"][-1][1]), + "act": (act_move["puts"][-2][1], act_move["puts"][-1][1]), + } + end = {key: phase_from_voltages(*v, 2.0) for key, v in voltages.items()} + assert wrap_phase(end["act"] - start["act"] - turned, 2.0) == pytest.approx( + 0, abs=1e-9 + ), context + if not touched: + assert wrap_phase(end["act"] - end["exp"], 2.0) == pytest.approx( + wrap_phase(-0.01 * round(wraps), 2.0), abs=1e-9 + ), context + stored = {"exp": exp_move["state"][3], "act": act_move["state"][3]} + + def test_scenario_table_covers_the_edge_cases(monkeypatch, tmp_path): """Guard the table itself: both laser-period wraps, a cavity-period crossing and long phase ramps in both directions really occur."""