1 / 7
rx-controls-suite · primer, not a talk
It's the loop you already wrote
— minus the bookkeeping
One real piece of beamline logic, shown three ways
demo/synchrotron-beamline/guarded_scan.py · beam-loss recovery
← rx-controls-suite
2 / 7
00 · The Scenario Beam Drops. Shutter Must Close. Then reopen — automatically
The rule
"Beam current is polled once a second. Once it drops below 50 mA, close the shutter — and reopen it the instant beam recovers, without anyone watching a terminal."
  • Write the PV only on a transition — not every poll tick, or you flood the control system with redundant writes, forever.
  • Never miss the recovery — reopening has to happen on its own.
Every control-room engineer has written a version of this. It's usually a thread, a global, and a comparison against the last-known value.
Where this lives
The chain, top to bottom
SourceBeamCurrent · 1 Hz Tango poll · shared
↓
mapcurrent ≥ 50 mA → bool
↓
distinct_until_changedonly pass state changes
↓
flat_mapwrite_pv(PV_SHUTTER, …)
This is the shutter supervisor from the suite's combined storage-ring / beamline demo — a Tango-controlled ring feeding an EPICS-controlled beamline in one process. Nothing on this page is a toy example.
3 / 7
01 · Imperative Style vs. Reactive Same Logic, Side By Side Line for line, nothing new
Imperative style — thread + global + comparison
Python · the loop everyone already writes
_shutter_open = None   # must remember it yourself
def shutter_supervisor_loop():
    global _shutter_open
    while not _stop.is_set():
        current = controller.read_attribute(
            "BeamCurrent").value
        ok = current >= MIN_BEAM_CURRENT

        if ok != _shutter_open:      # manual "changed?"
            shutter_pv.put(1 if ok else 0)
            _shutter_open = ok

        time.sleep(1.0)   # match every poller, by hand

threading.Thread(
    target=shutter_supervisor_loop, daemon=True).start()
Works — and is where the bugs live: the global, the thread nobody join()s, and a second consumer of BeamCurrent means a second poll, or a hand-rolled callback list.
Reactive — guarded_scan.py, verbatim
Python · demo/synchrotron-beamline/guarded_scan.py
# health: shared 1 Hz poll (facility.py: ring_health())
# every consumer subscribes to the same stream —
# there is no second poll to open.
supervisor_disp = health.pipe(
    ops.map(lambda h: h.current >= MIN_BEAM_CURRENT),
    ops.distinct_until_changed(),
    ops.flat_map(lambda ok: write_pv(
        PV_SHUTTER, 1 if ok else 0, ctx
    )),
).subscribe(
    on_next=lambda v: print(
        "Shutter OPENED" if v else "Shutter CLOSED"
    ),
    scheduler=scheduler,
)
✓  .map = the ok = current >= ... line. .distinct_until_changed() is the if ok != _shutter_open check — no global required. .flat_map(write_pv(...)) is the write. One-to-one, no new concepts.
4 / 7
02 · Watch It Run The Same Chain, As A Marble Diagram Watch it loop
health — BeamCurrent, 1 Hz Tango poll, shared
92mA
90mA
88mA
41mA
35mA
38mA
55mA
91mA
↓.map(lambda h: h.current >= MIN_BEAM_CURRENT)
is_beam_ok — every reading becomes a bool
T
T
T
F
F
F
T
T
↓.distinct_until_changed().flat_map(write_pv(PV_SHUTTER, ...))
PV_SHUTTER writes — only transitions reach the device
OPEN
CLOSE
OPEN
beam OK
beam lost
PV write — state actually changed
distinct_until_changed swallowed it — no write
5 / 7
03 · See It Live Real Dashboard, Same Pipeline Not a diagram — the actual system
Live synchrotron-beamline dashboard: Tango storage ring (left) and EPICS tomography beamline (right), cycling through nominal, beam-loss, and vacuum-burst scenarios.
demo/synchrotron-beamline · live_dashboard.py · recorded run, cycling through fault scenarios
Run python inject_fault.py beam_loss in a second terminal during a scan: ring current falls below 50 mA — the bad marble, for real. Shutter flips to CLOSED within one poll tick, scan pauses (gated on the same health stream). inject_fault.py nominal reverses it — no restart needed.
6 / 7
04 · Try It Yourself Ten Minutes, Docker Run the exact pipeline above
Shell · live simulation, storage ring + beamline
cd demo/synchrotron-beamline
docker compose up -d --build
python guarded_scan.py --ascii

# in a second terminal, trip the beam mid-scan:
python inject_fault.py beam_loss      # shutter closes, scan pauses
python inject_fault.py nominal        # shutter reopens, scan resumes
  • Orbit-drift flagging — every frame tagged with quality, derived from Tango
  • Vacuum-burst abort — an interlock stream feeds take_until, scan terminates cleanly
  • Backpressure — one share()'d stream, split into HDF5 (every frame) and a sampled live display
7 / 7
rx-controls-suite
Same operators.
Every control system.
Full walkthrough
demo/synchrotron-beamline/README.md
The suite-wide picture
rx-controls-suite · top-level README
Same operators across EPICS, Tango, and TINE — in Java, Python, and C++
← back to talks
← talks