fix(panda): write layout and pulse fields only when they differ

A replayed PULSE train also came 1.3 s after an unchanged parameter write by a new
device instance (connect / configure_line), outside the 1 s quiet wait, and blocked
the first line trigger. Fields are now read first and written only if the PandA holds
a different value, so a reconnect on a configured PandA writes nothing; the quiet wait
after a real change is 2 s.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM

csaxs_bec/devices/panda_box/panda_line_trigger.py

@@ -22,9 +22,11 @@ any trigger. With unchanged parameters and ENABLE kept high, hundreds of trigger
 a row were correct. So the PULSE blocks stay enabled, their parameters are only written
 when they change (once per scan), followed by a wait until a replayed train is over, and
 edges during a line are filtered out before the PULSE blocks. Replayed trains during a
-parameter change do appear on TTLOUT2/3. The replayed train starts only ~0.3-0.6 s after
+parameter change do appear on TTLOUT2/3. The replayed train starts only ~0.3-1.3 s after
 the write (seen 2026-10-29), so configure_line waits until both PULSE outputs stayed low
-for PARAM_QUIET_TIME.
+for PARAM_QUIET_TIME. All fields are read first and written only if the PandA holds a
+different value: a new device instance (BEC restart, reconnect) on an already configured
+PandA writes nothing, so it cannot cause a replay.
 
 Only the control port (8888) is used: the fields are written by setup_layout() at connect,
 the PCAP data capture is not touched. Timing resolution 8 ns (125 MHz FPGA clock).
@@ -37,6 +39,7 @@ import socket
 import threading
 import time
 
+import numpy as np
 from bec_lib.logger import bec_logger
 from ophyd import Device
 
@@ -131,8 +134,8 @@ class PandaLineTrigger(Device):
     TRIGGER_COUNTER = "COUNTER6"
     LINE_COUNTER = "COUNTER7"
     PULSE_COUNTER = "COUNTER8"
-    # a replayed train starts up to ~0.6 s after a parameter write
-    PARAM_QUIET_TIME = 1.0
+    # a replayed train starts up to ~1.3 s after a parameter write
+    PARAM_QUIET_TIME = 2.0
     SHUTTER_OUTPUT = "TTLOUT2"
     DETECTOR_OUTPUT = "TTLOUT3"
 
@@ -231,9 +234,29 @@ class PandaLineTrigger(Device):
 
     def setup_layout(self) -> None:
         """Write the wiring of the line trigger blocks (other blocks are not touched)."""
-        for field, value in self._layout():
-            self.panda.set(field, value)
+        written = [field for field, value in self._layout() if self._set_if_changed(field, value)]
         self._pulse_params = None
+        pulses = (self.SHUTTER_PULSE, self.DETECTOR_PULSE)
+        if any(field.split(".")[0] in pulses for field in written):
+            self._wait_pulses_idle(2 * self._train_duration(None) + 1)
+
+    def _set_if_changed(self, field: str, value) -> bool:
+        """
+        Write a field only if the PandA holds a different value (fewer writes, fewer replayed
+        PULSE trains). Actions (empty value, e.g. FORCE_RST) are always sent. Returns True
+        if the field was written.
+        """
+        value = str(value)
+        if value != "":
+            current = self.panda.get(field)
+            try:
+                if np.isclose(float(current), float(value), rtol=1e-9, atol=1e-12):
+                    return False
+            except ValueError:
+                if current == value:
+                    return False
+        self.panda.set(field, value)
+        return True
 
     @staticmethod
     def _train_duration(params: dict | None) -> float:
@@ -270,19 +293,24 @@ class PandaLineTrigger(Device):
         if params == self._pulse_params:
             return
         sp, dp = self.SHUTTER_PULSE, self.DETECTOR_PULSE
-        self.panda.set(f"{self.ARM_GATE}.FORCE_RST", "")
-        self.panda.set(f"{dp}.DELAY", f"{params['delay']:.9f}")
-        self.panda.set(f"{dp}.WIDTH", f"{params['width']:.9f}")
-        self.panda.set(f"{dp}.STEP", f"{params['step']:.9f}")
-        self.panda.set(f"{dp}.PULSES", params["n_pulses"])
-        self.panda.set(f"{sp}.DELAY", "0")
-        self.panda.set(f"{sp}.WIDTH", f"{params['shutter_width']:.9f}")
-        self.panda.set(f"{sp}.STEP", "0")
-        self.panda.set(f"{sp}.PULSES", 1)
+        fields = [
+            (f"{dp}.DELAY", f"{params['delay']:.9f}"),
+            (f"{dp}.WIDTH", f"{params['width']:.9f}"),
+            (f"{dp}.STEP", f"{params['step']:.9f}"),
+            (f"{dp}.PULSES", params["n_pulses"]),
+            (f"{sp}.DELAY", "0"),
+            (f"{sp}.WIDTH", f"{params['shutter_width']:.9f}"),
+            (f"{sp}.STEP", "0"),
+            (f"{sp}.PULSES", 1),
+        ]
+        previous = self._pulse_params
         self._pulse_params = None
-        self._wait_pulses_idle(
-            2 * max(self._train_duration(self._pulse_params), self._train_duration(params)) + 1
-        )
+        self.panda.set(f"{self.ARM_GATE}.FORCE_RST", "")
+        written = [field for field, value in fields if self._set_if_changed(field, value)]
+        if written:
+            self._wait_pulses_idle(
+                2 * max(self._train_duration(previous), self._train_duration(params)) + 1
+            )
         self._pulse_params = params
 
     def _wait_pulses_idle(self, timeout: float) -> None:

tests/tests_devices/test_panda_line_trigger.py

@@ -241,3 +241,17 @@ def test_configure_line_waits_for_a_late_replayed_train(trigger):
     trigger.configure_line(5, 0.01)
     # the train ended at ~0.1 s, then 0.2 s quiet
     assert time.time() - t0 >= 0.28
+
+
+def test_unchanged_fields_are_not_written(trigger):
+    """A new device instance on a configured PandA writes nothing (no replayed trains)."""
+    trigger.configure_line(5, 0.01, 0.002)
+    fields = dict(trigger.panda.fields)
+    again = PandaLineTrigger(name="pandatrig", host="fake-panda")
+    again.PARAM_QUIET_TIME = 0.01
+    again.panda = FakePanda()
+    again.panda.fields = fields
+    again.wait_for_connection()
+    again.configure_line(5, 0.01, 0.002)
+    values = [w for w in again.panda.writes if not w.split("=", 1)[1] == ""]
+    assert values == []