feat(panda): line trigger stays armed like the DDG, every edge starts a line
CI for csaxs_bec / test (push) Successful in 3m30s
CI for csaxs_bec / test (push) Successful in 3m30s
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01M8zoDyJxbBLmjZCEPUo5FM
This commit is contained in:
Mirko HollerandClaude Opus 5.5 committed
1 parent
763c8e0c27
commit
b518e657cf
4 files changed
+281
-179
No files matched your search
@@ -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.
|
||||
@@ -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<B
|
||||
|
||||
*METADATA.LAYOUT<
|
||||
{"BITS": {"x": -700.0, "y": 250.0},
|
||||
"COUNTER8": {"x": 620.0, "y": 340.0},
|
||||
"LUT7": {"x": 50.0, "y": 80.0},
|
||||
"COUNTER6": {"x": -150.0, "y": -200.0},
|
||||
"COUNTER7": {"x": 200.0, "y": 450.0},
|
||||
"COUNTER8": {"x": 550.0, "y": 400.0},
|
||||
"LUT7": {"x": -150.0, "y": 150.0},
|
||||
"LUT8": {"x": -450.0, "y": 80.0},
|
||||
"PULSE3": {"x": 320.0, "y": -120.0},
|
||||
"PULSE4": {"x": 320.0, "y": 180.0},
|
||||
"SRGATE1": {"x": -200.0, "y": 230.0},
|
||||
"PULSE3": {"x": 200.0, "y": -100.0},
|
||||
"PULSE4": {"x": 200.0, "y": 200.0},
|
||||
"SRGATE1": {"x": -450.0, "y": 330.0},
|
||||
"TTLIN3": {"x": -700.0, "y": 0.0},
|
||||
"TTLOUT2": {"x": 620.0, "y": -120.0},
|
||||
"TTLOUT3": {"x": 620.0, "y": 120.0}}
|
||||
"TTLOUT2": {"x": 550.0, "y": -100.0},
|
||||
"TTLOUT3": {"x": 550.0, "y": 150.0}}
|
||||
|
||||
*METADATA.EXPORTS<
|
||||
@@ -1,28 +1,32 @@
|
||||
"""
|
||||
PandABox as a line trigger delay generator (stand-in for DDG1/DDG2 in test setups).
|
||||
|
||||
A rising edge on the line trigger input (TTLIN3, e.g. the Aerotech PSO output) starts
|
||||
Like the DDG it stays armed: every line trigger edge (TTLIN3, e.g. the Aerotech PSO output)
|
||||
starts one line
|
||||
- PULSE3 -> 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
|
||||
@@ -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():
|
||||
|
||||
Reference in new issue
Block a user