From b518e657cf4ac2cd5d32b8b5846255c0ac12e6ed Mon Sep 17 00:00:00 2001 From: Mirko Holler Date: Wed, 28 Oct 2026 16:22:55 +0100 Subject: [PATCH] feat(panda): line trigger stays armed like the DDG, every edge starts a line configure_line() once, arm() once, then every line trigger edge (TTLIN3 or software) starts a line; a scan no longer re-arms per line. LUT7 accepts an edge only while armed and while no line is running (PULSE3/PULSE4 low), so edges during a line are ignored like on the DDG and never reach the PULSE blocks (a dropped trigger puts the FPGA 4.1 PULSE block into the replay state). COUNTER6/7 count raw edges and accepted lines, wait_lines_done() checks lines, gates and ignored edges. arm(one_shot=True) keeps the single-line mode. Design SASTT_line_trigger updated on omny-panda. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM --- csaxs_bec/devices/aerotech/README.md | 36 ++- .../panda_box/layouts/sastt_line_trigger.ini | 44 ++-- .../devices/panda_box/panda_line_trigger.py | 245 +++++++++++------- .../tests_devices/test_panda_line_trigger.py | 135 ++++++---- 4 files changed, 281 insertions(+), 179 deletions(-) diff --git a/csaxs_bec/devices/aerotech/README.md b/csaxs_bec/devices/aerotech/README.md index 5ef3ab79..0a5b6f50 100644 --- a/csaxs_bec/devices/aerotech/README.md +++ b/csaxs_bec/devices/aerotech/README.md @@ -99,28 +99,36 @@ Commissioning of the scan: Config `device_configs/sastt_aerotech_panda.yaml` = `sastt_aerotech.yaml` + device `pandatrig` (`devices/panda_box/panda_line_trigger.py`, PandA `omny-panda.psi.ch`). The PandA -replaces DDG1/DDG2: on the line trigger edge it sends a shutter pulse and n detector pulses, -8 ns resolution. +replaces DDG1/DDG2 and behaves like them: it stays armed, and every line trigger edge sends a +shutter pulse and n detector gates (8 ns resolution). A scan configures and arms once; the +lines run back to back without re-arming. | PandA | Use | Wiring | |-------|-----|--------| | TTLIN3 (50 Ω) | line trigger | Npaq J7 opto output: J7-25 (+5 V) -> J7-6, J7-7 -> TTLIN3, J7-24 -> TTLIN3 ground | -| TTLOUT2 | "shutter": trigger edge to end of the last exposure | scope | -| TTLOUT3 | "detector": n pulses, width exp_time - readout_time, period exp_time, first one shutter_delay after the edge | scope, later Npaq J12 pin 1 (+) / pin 2 (-) for position sampling | +| TTLOUT2 | "shutter": trigger edge to end of the last gate | scope | +| TTLOUT3 | "detector" gates (high during exposure): n pulses, width exp_time - readout_time, period exp_time, first one shutter_delay after the edge | scope, later Npaq J12 pin 1 (+) / pin 2 (-) for position sampling (rising edge = exposure start) | -Blocks: LUT8 = TTLIN3 | BITS.A (software trigger), SRGATE1 one-shot arm gate (set by -`arm_line`, reset by the first edge), LUT7 = LUT8 & SRGATE1 -> PULSE3 (shutter), PULSE4 -(detector), COUNTER8 counts the PULSE4 pulses. Design `SASTT_line_trigger` on the PandA web +Blocks: LUT8 = TTLIN3 | BITS.A (software trigger); SRGATE1 arm gate; LUT7 = rising edge of +LUT8 & SRGATE1 & no line running (PULSE3/PULSE4 low) -> PULSE3 (shutter), PULSE4 (gates). +An edge during a running line is ignored, like on the DDG. COUNTER6 counts raw edges, +COUNTER7 accepted lines, COUNTER8 gates. Design `SASTT_line_trigger` on the PandA web interface, copy in `devices/panda_box/layouts/sastt_line_trigger.ini`; the device also writes the wiring at connect. ```python -dev.pandatrig.arm_line(n_pulses=100, exp_time=0.01, readout_time=0.002, shutter_delay=0.005) -dev.pandatrig.software_trigger() # or the PSO edge on TTLIN3 -dev.pandatrig.wait_line_done(100, timeout=5) +dev.pandatrig.configure_line(n_pulses=100, exp_time=0.01, readout_time=0.002, shutter_delay=0.005) +dev.pandatrig.arm() # arm(one_shot=True): only the next edge +dev.pandatrig.software_trigger() # or the PSO edges on TTLIN3, one per line +dev.pandatrig.wait_lines_done(n_lines=1, timeout=5) +dev.pandatrig.line_status() # triggers, lines, pulses, dropped since arm() ``` -PandA FPGA 4.1 PULSE block: an ENABLE edge or a parameter write often replays the previous -pulse train without a trigger. The device keeps the PULSE blocks enabled and writes their -parameters only when they change (once per scan), then waits until such a train is over; -during that wait pulses can appear on TTLOUT2/3. +`wait_lines_done` fails on timeout, on a wrong number of lines or gates and on ignored +edges. + +PandA FPGA 4.1 PULSE block: an ENABLE edge, a parameter write or a dropped trigger often +makes it replay the previous pulse train without a trigger. The device keeps the PULSE +blocks enabled, writes their parameters only when they change (once per scan) and then waits +until such a train is over (pulses can appear on TTLOUT2/3 meanwhile), and the trigger path +filters out edges during a line, so no trigger is ever dropped by a PULSE block. diff --git a/csaxs_bec/devices/panda_box/layouts/sastt_line_trigger.ini b/csaxs_bec/devices/panda_box/layouts/sastt_line_trigger.ini index af2554ca..6e4bb072 100644 --- a/csaxs_bec/devices/panda_box/layouts/sastt_line_trigger.ini +++ b/csaxs_bec/devices/panda_box/layouts/sastt_line_trigger.ini @@ -522,7 +522,7 @@ SRGATE1.ENABLE=ONE SRGATE2.ENABLE=ZERO SRGATE3.ENABLE=ZERO SRGATE4.ENABLE=ZERO -SRGATE1.RST=LUT8.OUT +SRGATE1.RST=ZERO SRGATE2.RST=ZERO SRGATE3.RST=ZERO SRGATE4.RST=ZERO @@ -548,7 +548,7 @@ LUT3.INPC=ZERO LUT4.INPC=ZERO LUT5.INPC=ZERO LUT6.INPC=ZERO -LUT7.INPC=ZERO +LUT7.INPC=PULSE3.OUT LUT8.INPC=ZERO LUT1.INPA=ZERO LUT2.INPA=ZERO @@ -564,7 +564,7 @@ LUT3.INPD=ZERO LUT4.INPD=ZERO LUT5.INPD=ZERO LUT6.INPD=ZERO -LUT7.INPD=ZERO +LUT7.INPD=PULSE4.OUT LUT8.INPD=ZERO LUT1.INPE=ZERO LUT2.INPE=ZERO @@ -580,7 +580,7 @@ LUT3.FUNC=0x00000000 LUT4.FUNC=0x00000000 LUT5.FUNC=0x00000000 LUT6.FUNC=0x00000000 -LUT7.FUNC=A&B +LUT7.FUNC=A&B&~C&~D LUT8.FUNC=A|B LUT1.TYPEA=Input-Level LUT2.TYPEA=Input-Level @@ -588,7 +588,7 @@ LUT3.TYPEA=Input-Level LUT4.TYPEA=Input-Level LUT5.TYPEA=Input-Level LUT6.TYPEA=Input-Level -LUT7.TYPEA=Input-Level +LUT7.TYPEA=Pulse-On-Rising-Edge LUT8.TYPEA=Input-Level LUT1.TYPEC=Input-Level LUT2.TYPEC=Input-Level @@ -682,15 +682,15 @@ COUNTER3.ENABLE=ZERO COUNTER4.ENABLE=ZERO COUNTER5.ENABLE=ZERO COUNTER6.ENABLE=ZERO -COUNTER7.ENABLE=ONE -COUNTER8.ENABLE=ONE +COUNTER7.ENABLE=ZERO +COUNTER8.ENABLE=ZERO COUNTER1.TRIG=ZERO COUNTER2.TRIG=ZERO COUNTER3.TRIG=ZERO COUNTER4.TRIG=ZERO COUNTER5.TRIG=ZERO -COUNTER6.TRIG=ZERO -COUNTER7.TRIG=PULSE4.OUT +COUNTER6.TRIG=LUT8.OUT +COUNTER7.TRIG=LUT7.OUT COUNTER8.TRIG=PULSE4.OUT COUNTER1.TRIG_EDGE=Falling COUNTER2.TRIG_EDGE=Rising @@ -735,7 +735,7 @@ COUNTER8.STEP=1 PULSE1.PULSES=0 PULSE2.PULSES=5 PULSE3.PULSES=1 -PULSE4.PULSES=10 +PULSE4.PULSES=5 PULSE1.ENABLE=ONE PULSE2.ENABLE=ZERO PULSE3.ENABLE=ONE @@ -743,7 +743,7 @@ PULSE4.ENABLE=ONE PULSE1.DELAY=0 PULSE2.DELAY=0 PULSE3.DELAY=0 -PULSE4.DELAY=0.001 +PULSE4.DELAY=0 PULSE1.TRIG=ZERO PULSE2.TRIG=ZERO PULSE3.TRIG=LUT7.OUT @@ -754,12 +754,12 @@ PULSE3.TRIG_EDGE=Rising PULSE4.TRIG_EDGE=Rising PULSE1.WIDTH=0.001 PULSE2.WIDTH=0.002 -PULSE3.WIDTH=0.0108 -PULSE4.WIDTH=0.0008 +PULSE3.WIDTH=0.01 +PULSE4.WIDTH=0.002 PULSE1.STEP=0 PULSE2.STEP=0.0025 PULSE3.STEP=0 -PULSE4.STEP=0.001 +PULSE4.STEP=0.002 FILTER1.INP=ZERO FILTER2.INP=ZERO FILTER1.MODE=average @@ -866,14 +866,16 @@ PGEN2.TABLE TTLOUT2: "shutter", one pulse from the trigger to the end of the last exposure -- PULSE4 -> TTLOUT3: "detector", n pulses of width exp_time - readout_time every exp_time, - the first one shutter_delay after the trigger -COUNTER8 counts the PULSE4 pulses of the line. +- PULSE4 -> TTLOUT3: "detector" gates, n pulses of width exp_time - readout_time every + exp_time, the first one shutter_delay after the trigger +so a scan arms once and the lines run back to back. Trigger path: - LUT8 = TTLIN3 | BITS.A: raw line trigger, BITS.A is the software trigger -- SRGATE1: one-shot arm gate, set by arm_line() (FORCE_SET), reset by the first raw - trigger edge -- LUT7 = LUT8 & SRGATE1: only the first edge after arming reaches PULSE3/PULSE4 +- SRGATE1: arm gate, set by arm(); one_shot=True resets it with the first accepted edge +- LUT7 = rising edge of LUT8 & SRGATE1 & ~PULSE3.OUT & ~PULSE4.OUT: an edge is accepted + only while armed and while no line is running; edges during a line are ignored (the + DDG ignores them too) and never reach the PULSE blocks +Counters: COUNTER6 raw edges, COUNTER7 accepted lines, COUNTER8 detector pulses. -PULSE block problem (PandA FPGA 4.1, seen 2026-10-28): a rising edge of ENABLE or a -parameter write often replays a pulse train without any trigger (typically the previous -train). With unchanged parameters and ENABLE kept high, 200 triggered lines in 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 per line -only the counter is reset and the one-shot gate armed. Replayed trains during a parameter -change do appear on TTLOUT2/3 (the gate only protects the trigger path). +PULSE block problem (PandA FPGA 4.1, seen 2026-10-28): a rising edge of ENABLE, a parameter +write or a dropped trigger often makes the block replay the previous pulse train without +any trigger. With unchanged parameters and ENABLE kept high, hundreds of triggered lines in +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. 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). +Design SASTT_line_trigger on the PandA web interface, copy in layouts/sastt_line_trigger.ini. """ from __future__ import annotations @@ -92,8 +96,8 @@ class PandaControl: class PandaLineTrigger(Device): """ - PandABox line trigger: on the first trigger edge after arm_line(), a shutter pulse and - n detector pulses. + PandABox line trigger: while armed, every line trigger edge gives a shutter pulse and + n detector gates. Args: host: PandA host name. @@ -104,11 +108,12 @@ class PandaLineTrigger(Device): USER_ACCESS = [ "setup_layout", - "arm_line", + "configure_line", + "arm", "disarm", "software_trigger", "line_status", - "wait_line_done", + "wait_lines_done", "panda", ] @@ -116,9 +121,11 @@ class PandaLineTrigger(Device): SOFT_TRIGGER_BIT = "BITS.A" RAW_TRIGGER_LUT = "LUT8" ARM_GATE = "SRGATE1" - GATED_TRIGGER_LUT = "LUT7" + ACCEPTED_TRIGGER_LUT = "LUT7" SHUTTER_PULSE = "PULSE3" DETECTOR_PULSE = "PULSE4" + TRIGGER_COUNTER = "COUNTER6" + LINE_COUNTER = "COUNTER7" PULSE_COUNTER = "COUNTER8" SHUTTER_OUTPUT = "TTLOUT2" DETECTOR_OUTPUT = "TTLOUT3" @@ -159,8 +166,8 @@ class PandaLineTrigger(Device): super().stop(success=success) def _layout(self) -> list[tuple[str, str]]: - raw, gate, gated = self.RAW_TRIGGER_LUT, self.ARM_GATE, self.GATED_TRIGGER_LUT - sp, dp, cnt = self.SHUTTER_PULSE, self.DETECTOR_PULSE, self.PULSE_COUNTER + raw, gate, accepted = self.RAW_TRIGGER_LUT, self.ARM_GATE, self.ACCEPTED_TRIGGER_LUT + sp, dp = self.SHUTTER_PULSE, self.DETECTOR_PULSE fields = [ (f"{self.TRIGGER_INPUT}.TERM", self.input_termination), (self.SOFT_TRIGGER_BIT, "0"), @@ -169,30 +176,43 @@ class PandaLineTrigger(Device): (f"{raw}.FUNC", "A|B"), (f"{gate}.ENABLE", "ONE"), (f"{gate}.SET", "ZERO"), - (f"{gate}.RST", f"{raw}.OUT"), + (f"{gate}.RST", "ZERO"), (f"{gate}.RST_EDGE", "Rising"), (f"{gate}.WHEN_DISABLED", "Set output low"), (f"{gate}.FORCE_RST", ""), - (f"{gated}.INPA", f"{raw}.OUT"), - (f"{gated}.INPB", f"{gate}.OUT"), - (f"{gated}.FUNC", "A&B"), + (f"{accepted}.INPA", f"{raw}.OUT"), + (f"{accepted}.TYPEA", "Pulse-On-Rising-Edge"), + (f"{accepted}.INPB", f"{gate}.OUT"), + (f"{accepted}.TYPEB", "Input-Level"), + (f"{accepted}.INPC", f"{sp}.OUT"), + (f"{accepted}.TYPEC", "Input-Level"), + (f"{accepted}.INPD", f"{dp}.OUT"), + (f"{accepted}.TYPED", "Input-Level"), + (f"{accepted}.FUNC", "A&B&~C&~D"), ] for pulse in (sp, dp): fields += [ (f"{pulse}.ENABLE", "ONE"), - (f"{pulse}.TRIG", f"{gated}.OUT"), + (f"{pulse}.TRIG", f"{accepted}.OUT"), (f"{pulse}.TRIG_EDGE", "Rising"), (f"{pulse}.DELAY.UNITS", "s"), (f"{pulse}.WIDTH.UNITS", "s"), (f"{pulse}.STEP.UNITS", "s"), ] + for counter, source in ( + (self.TRIGGER_COUNTER, f"{raw}.OUT"), + (self.LINE_COUNTER, f"{accepted}.OUT"), + (self.PULSE_COUNTER, f"{dp}.OUT"), + ): + fields += [ + (f"{counter}.ENABLE", "ZERO"), + (f"{counter}.TRIG", source), + (f"{counter}.TRIG_EDGE", "Rising"), + (f"{counter}.DIR", "ZERO"), + (f"{counter}.START", "0"), + (f"{counter}.STEP", "1"), + ] fields += [ - (f"{cnt}.ENABLE", "ZERO"), - (f"{cnt}.TRIG", f"{dp}.OUT"), - (f"{cnt}.TRIG_EDGE", "Rising"), - (f"{cnt}.DIR", "ZERO"), - (f"{cnt}.START", "0"), - (f"{cnt}.STEP", "1"), (f"{self.SHUTTER_OUTPUT}.VAL", f"{sp}.OUT"), (f"{self.DETECTOR_OUTPUT}.VAL", f"{dp}.OUT"), ] @@ -210,49 +230,14 @@ class PandaLineTrigger(Device): return 0.1 return params["delay"] + params["n_pulses"] * params["step"] - def _set_pulse_params(self, params: dict) -> None: - """Write the PULSE parameters (only if changed) and wait until a replayed train is - over: a stable counter and both PULSE outputs low.""" - if params == self._pulse_params: - return - sp, dp = self.SHUTTER_PULSE, self.DETECTOR_PULSE - 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) - timeout = 2 * max(self._train_duration(self._pulse_params), self._train_duration(params)) - self._pulse_params = None - deadline = time.time() + timeout + 1.0 - last = None - while True: - time.sleep(0.02) - state = ( - self.panda.get(f"{self.PULSE_COUNTER}.OUT"), - self.panda.get(f"{sp}.OUT"), - self.panda.get(f"{dp}.OUT"), - ) - if state == last and state[1:] == ("0", "0"): - break - if time.time() > deadline: - raise PandaLineTriggerError( - f"{self.name}: PULSE blocks still active {timeout + 1:.1f} s after a" - " parameter change." - ) - last = state - self._pulse_params = params - - def arm_line( + def configure_line( self, n_pulses: int, exp_time: float, readout_time: float = 0.0, shutter_delay: float = 0.0 ) -> None: """ - Arm one line: on the next trigger edge, n_pulses detector pulses of width - exp_time - readout_time every exp_time, the first one shutter_delay after the edge. - The shutter pulse lasts from the edge to the end of the last exposure. Further - edges are ignored until the next arm_line(). + Pulses of one line: n_pulses detector gates of width exp_time - readout_time every + exp_time, the first one shutter_delay after the trigger edge. The shutter pulse + lasts from the edge to the end of the last gate. Written to the PandA only when the + values change; must not be called while lines are running. """ n_pulses = int(n_pulses) width = exp_time - readout_time @@ -264,48 +249,102 @@ class PandaLineTrigger(Device): ) if shutter_delay < 0: raise PandaLineTriggerError(f"shutter_delay must be >= 0, got {shutter_delay}.") + params = { + "n_pulses": n_pulses, + "width": width, + "step": exp_time, + "delay": shutter_delay, + "shutter_width": shutter_delay + (n_pulses - 1) * exp_time + width, + } + 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) + self._pulse_params = None + self._wait_pulses_idle( + 2 * max(self._train_duration(self._pulse_params), self._train_duration(params)) + 1 + ) + self._pulse_params = params + + def _wait_pulses_idle(self, timeout: float) -> None: + """Wait until a replayed train is over: stable counter and both PULSE outputs low.""" + sp, dp = self.SHUTTER_PULSE, self.DETECTOR_PULSE + deadline = time.time() + timeout + last = None + while True: + time.sleep(0.02) + state = ( + self.panda.get(f"{self.PULSE_COUNTER}.OUT"), + self.panda.get(f"{sp}.OUT"), + self.panda.get(f"{dp}.OUT"), + ) + if state == last and state[1:] == ("0", "0"): + return + if time.time() > deadline: + raise PandaLineTriggerError( + f"{self.name}: PULSE blocks still active {timeout:.1f} s after a parameter" + " change." + ) + last = state + + def arm(self, one_shot: bool = False) -> None: + """ + Reset the counters and accept line trigger edges: every edge while no line is + running starts a line, until disarm(). one_shot=True accepts only the next edge. + """ + if self._pulse_params is None: + raise PandaLineTriggerError(f"{self.name}: configure_line() before arm().") + gate = self.ARM_GATE + self.panda.set(f"{gate}.FORCE_RST", "") if self.panda.get(f"{self.RAW_TRIGGER_LUT}.OUT") != "0": raise PandaLineTriggerError( f"{self.name}: line trigger input {self.TRIGGER_INPUT} is high, cannot arm." ) - self._set_pulse_params( - { - "n_pulses": n_pulses, - "width": width, - "step": exp_time, - "delay": shutter_delay, - "shutter_width": shutter_delay + (n_pulses - 1) * exp_time + width, - } - ) - # rising edge of ENABLE resets the counter to START - cnt = self.PULSE_COUNTER - self.panda.set(f"{cnt}.ENABLE", "ZERO") - self.panda.set(f"{cnt}.ENABLE", "ONE") + self.panda.set(f"{gate}.RST", f"{self.ACCEPTED_TRIGGER_LUT}.OUT" if one_shot else "ZERO") + # rising edge of ENABLE resets a counter to START + for counter in (self.TRIGGER_COUNTER, self.LINE_COUNTER, self.PULSE_COUNTER): + self.panda.set(f"{counter}.ENABLE", "ZERO") + self.panda.set(f"{counter}.ENABLE", "ONE") self._dropped_at_arm = int(self.panda.get(f"{self.DETECTOR_PULSE}.DROPPED")) - self.panda.set(f"{self.ARM_GATE}.FORCE_SET", "") + self.panda.set(f"{gate}.FORCE_SET", "") def disarm(self) -> None: - """Ignore line trigger edges until the next arm_line() (a running line continues).""" + """Ignore line trigger edges until the next arm() (a running line continues).""" self.panda.set(f"{self.ARM_GATE}.FORCE_RST", "") self.panda.set(self.SOFT_TRIGGER_BIT, "0") def software_trigger(self) -> None: - """Start the armed line without an input edge.""" + """Line trigger edge from software (accepted like a TTLIN3 edge).""" self.panda.set(self.SOFT_TRIGGER_BIT, "1") self.panda.set(self.SOFT_TRIGGER_BIT, "0") def line_status(self) -> dict: - dp = self.DETECTOR_PULSE + """Counts since arm(): trigger edges, accepted lines, detector pulses.""" + sp, dp = self.SHUTTER_PULSE, self.DETECTOR_PULSE return { "armed": self.panda.get(f"{self.ARM_GATE}.OUT") == "1", - "pulses_sent": int(self.panda.get(f"{self.PULSE_COUNTER}.OUT")), - "queued": int(self.panda.get(f"{dp}.QUEUED")), + "busy": self.panda.get(f"{sp}.OUT") == "1" or self.panda.get(f"{dp}.OUT") == "1", + "triggers": int(self.panda.get(f"{self.TRIGGER_COUNTER}.OUT")), + "lines": int(self.panda.get(f"{self.LINE_COUNTER}.OUT")), + "pulses": int(self.panda.get(f"{self.PULSE_COUNTER}.OUT")), "dropped": int(self.panda.get(f"{dp}.DROPPED")) - self._dropped_at_arm, } - def wait_line_done(self, n_pulses: int, timeout: float, poll: float = 0.02) -> dict: - """Wait until the line was triggered and n_pulses detector pulses were sent.""" + def wait_lines_done(self, n_lines: int, timeout: float, poll: float = 0.02) -> dict: + """ + Wait until n_lines lines were triggered and all their pulses sent. Fails on timeout, + on more lines or pulses than expected and on ignored trigger edges (edge during a + running line). + """ + n_pulses = self._pulse_params["n_pulses"] if self._pulse_params else 0 start = time.time() while True: status = self.line_status() @@ -313,16 +352,26 @@ class PandaLineTrigger(Device): raise PandaLineTriggerError( f"{self.name}: {status['dropped']} trigger(s) dropped by {self.DETECTOR_PULSE}." ) - if not status["armed"] and status["pulses_sent"] >= n_pulses and not status["queued"]: + if ( + status["lines"] >= n_lines + and status["pulses"] >= status["lines"] * n_pulses + and not status["busy"] + ): break if time.time() - start > timeout: raise PandaLineTriggerError( - f"{self.name}: line not finished within {timeout} s: {status}. Was the line" - f" trigger received on {self.TRIGGER_INPUT}?" + f"{self.name}: {n_lines} line(s) not finished within {timeout} s: {status}." + f" Was the line trigger received on {self.TRIGGER_INPUT}?" ) time.sleep(poll) - if status["pulses_sent"] != n_pulses: + if status["lines"] != n_lines or status["pulses"] != n_lines * n_pulses: raise PandaLineTriggerError( - f"{self.name}: sent {status['pulses_sent']} pulses instead of {n_pulses}." + f"{self.name}: expected {n_lines} line(s) with {n_lines * n_pulses} pulses," + f" got {status['lines']} with {status['pulses']}." + ) + if status["triggers"] != status["lines"]: + raise PandaLineTriggerError( + f"{self.name}: {status['triggers'] - status['lines']} trigger edge(s) ignored" + f" (edge while a line was running or while not armed): {status}." ) return status diff --git a/tests/tests_devices/test_panda_line_trigger.py b/tests/tests_devices/test_panda_line_trigger.py index ee592398..f092da9a 100644 --- a/tests/tests_devices/test_panda_line_trigger.py +++ b/tests/tests_devices/test_panda_line_trigger.py @@ -8,12 +8,13 @@ from csaxs_bec.devices.panda_box.panda_line_trigger import ( class FakePanda(PandaControl): - """In-memory PandA: fields, the one-shot arm gate, the counter and a PULSE4 train.""" + """In-memory PandA: fields, arm gate, counters; a line runs completely at its trigger.""" def __init__(self): super().__init__("fake-panda") self.fields = {} self.writes = [] + self.busy = False # simulate a running line: edges are ignored def open(self): pass @@ -28,18 +29,23 @@ class FakePanda(PandaControl): return self.fields.get(command[:-1], "0") field, value = command.split("=", 1) self.writes.append(command) + f = self.fields if field == "SRGATE1.FORCE_SET": - self.fields["SRGATE1.OUT"] = "1" + f["SRGATE1.OUT"] = "1" elif field == "SRGATE1.FORCE_RST": - self.fields["SRGATE1.OUT"] = "0" - elif field == "COUNTER8.ENABLE" and value == "ONE": - self.fields["COUNTER8.OUT"] = "0" - elif field == "BITS.A" and value == "1" and self.fields.get("SRGATE1.OUT") == "1": - # first edge after arming: gate closes, the whole train is sent at once - self.fields["SRGATE1.OUT"] = "0" - self.fields["COUNTER8.OUT"] = self.fields.get("PULSE4.PULSES", "1") + f["SRGATE1.OUT"] = "0" + elif field.startswith("COUNTER") and field.endswith(".ENABLE") and value == "ONE": + f[field.replace("ENABLE", "OUT")] = "0" + elif field == "BITS.A" and value == "1": + f["COUNTER6.OUT"] = str(int(f.get("COUNTER6.OUT", "0")) + 1) + if f.get("SRGATE1.OUT") == "1" and not self.busy: + f["COUNTER7.OUT"] = str(int(f.get("COUNTER7.OUT", "0")) + 1) + pulses = int(f.get("COUNTER8.OUT", "0")) + int(f.get("PULSE4.PULSES", "1")) + f["COUNTER8.OUT"] = str(pulses) + if f.get("SRGATE1.RST") == "LUT7.OUT": + f["SRGATE1.OUT"] = "0" else: - self.fields[field] = value + f[field] = value return "" @@ -53,11 +59,13 @@ def trigger(): def test_setup_layout_wires_the_blocks(trigger): fields = trigger.panda.fields - assert fields["PULSE3.TRIG"] == "LUT7.OUT" - assert fields["PULSE4.TRIG"] == "LUT7.OUT" - assert fields["LUT7.FUNC"] == "A&B" + assert fields["PULSE3.TRIG"] == fields["PULSE4.TRIG"] == "LUT7.OUT" + assert fields["LUT7.FUNC"] == "A&B&~C&~D" + assert fields["LUT7.TYPEA"] == "Pulse-On-Rising-Edge" + assert (fields["LUT7.INPC"], fields["LUT7.INPD"]) == ("PULSE3.OUT", "PULSE4.OUT") assert fields["LUT8.INPA"] == "TTLIN3.VAL" - assert fields["SRGATE1.RST"] == "LUT8.OUT" + assert fields["COUNTER6.TRIG"] == "LUT8.OUT" + assert fields["COUNTER7.TRIG"] == "LUT7.OUT" assert fields["COUNTER8.TRIG"] == "PULSE4.OUT" assert fields["TTLOUT2.VAL"] == "PULSE3.OUT" assert fields["TTLOUT3.VAL"] == "PULSE4.OUT" @@ -65,8 +73,8 @@ def test_setup_layout_wires_the_blocks(trigger): assert fields["PULSE3.ENABLE"] == fields["PULSE4.ENABLE"] == "ONE" -def test_arm_line_sets_timing(trigger): - trigger.arm_line(10, exp_time=0.01, readout_time=0.002, shutter_delay=0.005) +def test_configure_line_sets_timing(trigger): + trigger.configure_line(10, exp_time=0.01, readout_time=0.002, shutter_delay=0.005) fields = trigger.panda.fields assert float(fields["PULSE4.DELAY"]) == pytest.approx(0.005) assert float(fields["PULSE4.WIDTH"]) == pytest.approx(0.008) @@ -74,66 +82,99 @@ def test_arm_line_sets_timing(trigger): assert fields["PULSE4.PULSES"] == "10" # shutter: from the trigger to the end of the last exposure assert float(fields["PULSE3.WIDTH"]) == pytest.approx(0.005 + 9 * 0.01 + 0.008) - assert trigger.line_status()["armed"] def test_pulse_parameters_are_written_only_on_change(trigger): - trigger.arm_line(10, exp_time=0.01) + trigger.configure_line(10, exp_time=0.01) trigger.panda.writes.clear() - trigger.arm_line(10, exp_time=0.01) - assert not [w for w in trigger.panda.writes if w.startswith(("PULSE3.", "PULSE4."))] - trigger.arm_line(11, exp_time=0.01) + trigger.configure_line(10, exp_time=0.01) + assert not trigger.panda.writes + trigger.configure_line(11, exp_time=0.01) assert "PULSE4.PULSES=11" in trigger.panda.writes def test_never_toggles_pulse_enable_after_setup(trigger): + trigger.configure_line(5, exp_time=0.01) trigger.panda.writes.clear() - trigger.arm_line(5, exp_time=0.01) + trigger.arm() trigger.software_trigger() - trigger.wait_line_done(5, timeout=1) + trigger.wait_lines_done(1, timeout=1) trigger.disarm() trigger.stop() - assert not [w for w in trigger.panda.writes if ".ENABLE" in w and "PULSE" in w] + assert not [w for w in trigger.panda.writes if w.startswith("PULSE")] -def test_line_runs_once_per_arm(trigger): - trigger.arm_line(5, exp_time=0.01) +def test_stays_armed_for_many_lines(trigger): + trigger.configure_line(5, exp_time=0.01) + trigger.arm() + for _ in range(4): + trigger.software_trigger() + status = trigger.wait_lines_done(4, timeout=1) + assert (status["lines"], status["pulses"], status["armed"]) == (4, 20, True) + + +def test_one_shot_accepts_one_edge(trigger): + trigger.configure_line(5, exp_time=0.01) + trigger.arm(one_shot=True) trigger.software_trigger() - assert trigger.wait_line_done(5, timeout=1)["pulses_sent"] == 5 - assert not trigger.line_status()["armed"] + assert trigger.wait_lines_done(1, timeout=1)["armed"] is False + trigger.software_trigger() + with pytest.raises(PandaLineTriggerError, match="1 trigger edge"): + trigger.wait_lines_done(1, timeout=1) -def test_wait_line_done_times_out_without_trigger(trigger): - trigger.arm_line(5, exp_time=0.01) +def test_edge_during_line_is_reported(trigger): + trigger.configure_line(5, exp_time=0.01) + trigger.arm() + trigger.software_trigger() + trigger.panda.busy = True + trigger.software_trigger() + trigger.panda.busy = False + with pytest.raises(PandaLineTriggerError, match="ignored"): + trigger.wait_lines_done(1, timeout=1) + + +def test_arm_needs_configure(trigger): + with pytest.raises(PandaLineTriggerError, match="configure_line"): + trigger.arm() + + +def test_wait_lines_done_times_out_without_trigger(trigger): + trigger.configure_line(5, exp_time=0.01) + trigger.arm() with pytest.raises(PandaLineTriggerError, match="not finished"): - trigger.wait_line_done(5, timeout=0.1, poll=0.01) + trigger.wait_lines_done(1, timeout=0.1, poll=0.01) -def test_wait_line_done_reports_wrong_count(trigger): - trigger.arm_line(5, exp_time=0.01) - trigger.software_trigger() - trigger.panda.fields["COUNTER8.OUT"] = "6" - with pytest.raises(PandaLineTriggerError, match="6 pulses instead of 5"): - trigger.wait_line_done(5, timeout=0.1, poll=0.01) +def test_wait_lines_done_reports_too_many_lines(trigger): + trigger.configure_line(5, exp_time=0.01) + trigger.arm() + for _ in range(3): + trigger.software_trigger() + with pytest.raises(PandaLineTriggerError, match="expected 2 line"): + trigger.wait_lines_done(2, timeout=0.1, poll=0.01) -def test_wait_line_done_reports_dropped(trigger): - trigger.arm_line(5, exp_time=0.01) +def test_wait_lines_done_reports_dropped(trigger): + trigger.configure_line(5, exp_time=0.01) + trigger.arm() trigger.panda.fields["PULSE4.DROPPED"] = "1" with pytest.raises(PandaLineTriggerError, match="dropped"): - trigger.wait_line_done(5, timeout=0.1, poll=0.01) + trigger.wait_lines_done(1, timeout=0.1, poll=0.01) def test_dropped_counts_only_since_arm(trigger): + trigger.configure_line(5, exp_time=0.01) trigger.panda.fields["PULSE4.DROPPED"] = "3" - trigger.arm_line(5, exp_time=0.01) + trigger.arm() assert trigger.line_status()["dropped"] == 0 def test_arm_refuses_high_trigger_input(trigger): + trigger.configure_line(5, exp_time=0.01) trigger.panda.fields["LUT8.OUT"] = "1" with pytest.raises(PandaLineTriggerError, match="is high"): - trigger.arm_line(5, exp_time=0.01) + trigger.arm() @pytest.mark.parametrize( @@ -144,16 +185,18 @@ def test_arm_refuses_high_trigger_input(trigger): ({"n_pulses": 5, "exp_time": 0.01, "shutter_delay": -0.001}, "shutter_delay"), ], ) -def test_arm_line_checks_arguments(trigger, kwargs, match): +def test_configure_line_checks_arguments(trigger, kwargs, match): with pytest.raises(PandaLineTriggerError, match=match): - trigger.arm_line(**kwargs) + trigger.configure_line(**kwargs) def test_disarm_closes_the_gate(trigger): - trigger.arm_line(5, exp_time=0.01) + trigger.configure_line(5, exp_time=0.01) + trigger.arm() trigger.disarm() trigger.software_trigger() - assert trigger.line_status() == {"armed": False, "pulses_sent": 0, "queued": 0, "dropped": 0} + status = trigger.line_status() + assert (status["armed"], status["lines"], status["pulses"]) == (False, 0, 0) def test_control_reply_parsing():