bec/csaxs_bec
0
0
Fork 2
Code Issues 17 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(flomni): add command jobs to the tomo queue

Browse Source 
Type-tags queue jobs with a "kind" field (back-compat: missing kind
defaults to "tomo"), and adds "command" jobs that reconfigure the
beamline (currently: absolute moves on an allow-listed device, plus an
optimize_idgap stub) between tomograms instead of running a scan.

Command jobs are added via tomo_queue_add_command() against a curated
named-action registry (_TOMO_QUEUE_ACTIONS) - not arbitrary code - so
every queued action is auditable and safe to persist across a kernel
restart. Validation is two-layer: schema checks at add time, and each
action re-checks its own arguments against live hardware at execution
time, which is the actual safety boundary. Crash-resume for a command
job re-runs the whole step sequence from the top if every step is
idempotent, or prompts the operator otherwise.

flomni_webpage_generator.py is guarded against command jobs, which
have no "params" key.

See AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md for the full design.
This commit is contained in:
Mirko Holler
2026-07-13 09:05:29 +02:00
parent ff50711dc9
commit 30b5119a2f
14 changed files with 283 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/SIMULATED_ENDSTATIONS.md → csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/SIMULATED_ENDSTATIONS.md
View File
csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+272 -5
View File
@@ -1778,6 +1778,9 @@ class Flomni(
# below); use tomo_progress_reset() if you want to explicitly clear
# stale progress without starting a new scan.
self._tomo_queue_proxy = _TomoQueueProxy(self.client)
# Published so a future GUI never needs a hardcoded mirror of the
# action registry -- see _describe_tomo_queue_actions().
self.client.set_global_var("tomo_queue_actions", self._describe_tomo_queue_actions())
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
FlomniWebpageGenerator,
)
@@ -3572,6 +3575,55 @@ class Flomni(
"corridor_size",
)
# Curated allow-list for the "move" queue action -- NOT "any device in
# dev". Only devices actually reconfigured between tomograms belong here;
# this is the difference between a reconfiguration tool and a remote
# control for the whole endstation. mokev/idgap are real front-end
# devices (bl_frontend.yaml, not the flomni sim config); ftray (sample
# transfer tray, harmless to move, untouched by tomo-scan code) is
# included so the move action has a sim-testable target.
_TOMO_QUEUE_MOVE_DEVICES = {
"mokev": {"label": "Energy", "unit": "keV"},
"idgap": {"label": "Undulator gap", "unit": "mm"},
"ftray": {"label": "Sample transfer tray (sim-testable)", "unit": "mm"},
}
# Named-action registry for tomo-queue command jobs. Each entry maps a
# name to a bound Flomni method (arbitrary, reviewed Python -- a
# "script") plus enough metadata to validate and render it generically:
# `func` is the bound method name, `idempotent` is the default safe-to-
# rerun-after-a-crash flag, `help` is operator-facing, `params` is a
# schema of the method's kwargs (possibly empty). Adding a new script is
# exactly: write a new bound method + one new entry here -- nothing else
# (tomo_queue_add_command/_show/_run_command_job, the tomo_queue_actions
# global var) needs to change. Deliberately minimal for now: no
# foptics_in/out, feye_in/out, or shutter actions yet, even though the
# first two already exist as callable methods -- those are future
# "dedicated scripts" to add later as their own reviewed changes.
_TOMO_QUEUE_ACTIONS = {
"move": {
"func": "_queue_action_move",
"idempotent": True, # absolute moves re-run harmlessly
"help": "Move device(s) to absolute position(s).",
"params": {"positions": {"type": "device_positions"}},
},
"optimize_idgap": {
"func": "optimize_idgap",
"idempotent": True,
"help": "Scan idgap over a search range and move to the peak. STUB: no-op.",
"params": {
"search_range": {
"type": "float",
"default": 0.5,
"min": 0.0,
"max": 2.0,
"unit": "mm",
"label": "Search range",
}
},
},
}
def tomo_queue_add(self, label: str = None) -> int:
"""Snapshot the currently set tomo parameters and append them as a
new job to the tomo queue (persisted, survives a kernel restart).
@@ -3595,6 +3647,7 @@ class Flomni(
params = {name: getattr(self, name) for name in self._TOMO_SCAN_PARAM_NAMES}
index = len(self._tomo_queue_proxy)
job = {
"kind": "tomo",
"id": uuid.uuid4().hex,
"label": label or f"job_{index + 1}",
"params": params,
@@ -3605,6 +3658,164 @@ class Flomni(
print(f"Added tomo queue job #{index} ({job['label']}).")
return index
def tomo_queue_add_command(self, steps, label: str = None, idempotent: bool = None) -> int:
"""Append a "command" job -- an ordered list of named-registry actions
that reconfigure the beamline (move a device, ...) -- to the tomo
queue.
Unlike tomo_queue_add(), this does not snapshot the current tomo scan
parameters; a command job has no "params" key at all.
Typical usage::
flomni.tomo_queue_add_command(
[{"action": "move", "kwargs": {"positions": {"mokev": 6.2}}},
{"action": "optimize_idgap", "kwargs": {"search_range": 0.5}}],
label="reconfigure to 6.2 keV",
)
Args:
steps: A single {"action": ..., "kwargs": {...}} dict, or a list
of them, run in sequence within this one job.
label: Optional name for the job, shown by tomo_queue_show().
Defaults to "job_<n>".
idempotent: Overrides the default of "every step's registry
idempotent flag, ANDed together". Set this explicitly if the
registry defaults don't capture whether this particular
sequence is safe to blindly re-run after a crash.
Returns:
The index of the newly added job.
"""
if isinstance(steps, dict):
steps = [steps]
if not steps:
raise ValueError("tomo_queue_add_command: at least one step is required.")
resolved_steps = []
all_idempotent = True
for step in steps:
action_name = step.get("action")
kwargs = step.get("kwargs", {})
self._validate_action_kwargs(action_name, kwargs)
all_idempotent = all_idempotent and self._TOMO_QUEUE_ACTIONS[action_name]["idempotent"]
resolved_steps.append({"action": action_name, "kwargs": kwargs})
try:
json.dumps(resolved_steps)
except (TypeError, ValueError) as exc:
raise ValueError(f"Command job steps must be JSON-serializable: {exc}") from exc
index = len(self._tomo_queue_proxy)
job = {
"kind": "command",
"id": uuid.uuid4().hex,
"label": label or f"job_{index + 1}",
"steps": resolved_steps,
"idempotent": all_idempotent if idempotent is None else idempotent,
"status": "pending",
"added_at": datetime.datetime.now().isoformat(),
}
self._tomo_queue_proxy.append(job)
print(f"Added command queue job #{index} ({job['label']}).")
return index
def _validate_action_kwargs(self, action_name: str, kwargs: dict) -> None:
"""Add-time schema validation for a command-job step's kwargs.
Ergonomic, not the safety boundary: catches a typo or an out-of-range
value now instead of at 2am. Each action method re-validates its own
arguments against live hardware at execution time -- that's the
actual safety boundary, since the schema and the hardware can
disagree (limits changed; a value arrived via the CLI, bypassing
this).
"""
spec = self._TOMO_QUEUE_ACTIONS.get(action_name)
if spec is None:
raise ValueError(f"Unknown queue action '{action_name}'.")
schema = spec["params"]
for key in kwargs:
if key not in schema:
raise ValueError(f"Action '{action_name}' has no parameter '{key}'.")
for name, field in schema.items():
if name not in kwargs:
continue
value = kwargs[name]
field_type = field.get("type")
if field_type == "float":
if not isinstance(value, (int, float)) or isinstance(value, bool):
raise ValueError(f"{action_name}.{name} must be a number, got {value!r}.")
lo, hi = field.get("min"), field.get("max")
if lo is not None and value < lo:
raise ValueError(f"{action_name}.{name}={value} is below the minimum {lo}.")
if hi is not None and value > hi:
raise ValueError(f"{action_name}.{name}={value} is above the maximum {hi}.")
elif field_type == "device_positions":
if not isinstance(value, dict) or not value:
raise ValueError(
f"{action_name}.{name} must be a non-empty {{device: target}} dict."
)
for device_name in value:
if device_name not in self._TOMO_QUEUE_MOVE_DEVICES:
raise ValueError(
f"Queue move: device '{device_name}' is not queue-movable."
)
def _queue_action_move(self, positions: dict) -> None:
"""positions: {device_name: absolute_target, ...}. Absolute umv only.
Re-checks the allow-list at execution time (not just add time, see
_validate_action_kwargs) -- the actual safety boundary, since a value
could have arrived via the CLI, bypassing tomo_queue_add_command's
validation.
"""
args = []
for name, target in positions.items():
if name not in self._TOMO_QUEUE_MOVE_DEVICES:
raise ValueError(f"Queue move: device '{name}' is not queue-movable.")
if name not in dev:
raise ValueError(f"Queue move: unknown device '{name}'.")
args += [dev[name], target]
umv(*args)
def optimize_idgap(self, search_range: float = 0.5) -> None:
"""Scan idgap over +/- search_range around its current position and
move to the peak of <signal not yet decided>.
STUB -- not implemented. The peak-detection signal/device is not yet
decided (see AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md section 10).
This is the ONLY place that needs to change to make this a real
action: the registry entry, resume policy, and (eventually) GUI
wiring already treat it as a normal idempotent compound action, so
dropping in the real scan-and-move-to-peak logic here (or swapping
which device/signal it reads) is a self-contained change.
"""
print(
f"optimize_idgap(search_range={search_range}): not yet implemented -- "
"peak-detection signal not yet decided. No-op."
)
def _describe_tomo_queue_actions(self) -> dict:
"""Serializable description of the action registry + move allow-list.
Published as the ``tomo_queue_actions`` global var (see __init__) so
the GUI (or any other client) can build its dropdowns/forms from it
directly -- no hardcoded mirror to drift, which is exactly the
problem the QUEUE_PARAM_NAMES mirror between this file and
tomo_params.py already has.
"""
return {
"actions": {
name: {
"idempotent": spec["idempotent"],
"help": spec["help"],
"params": spec["params"],
}
for name, spec in self._TOMO_QUEUE_ACTIONS.items()
},
"move_devices": dict(self._TOMO_QUEUE_MOVE_DEVICES),
}
def tomo_queue_delete(self, *indices: int) -> None:
"""Delete one or more jobs from the tomo queue by index.
@@ -3633,6 +3844,14 @@ class Flomni(
print("Tomo queue is empty.")
return jobs
for i, job in enumerate(jobs):
if job.get("kind", "tomo") == "command":
steps_str = " > ".join(
f"{step['action']}{step['kwargs']}" if step["kwargs"] else step["action"]
for step in job["steps"]
)
idem = "idem" if job.get("idempotent") else "NOT idem"
print(f"[{i}] {job['status']:>10s} {job['label']:<20s} CMD {steps_str} [{idem}]")
continue
p = job["params"]
print(
f"[{i}] {job['status']:>10s} {job['label']:<20s} "
@@ -3693,6 +3912,13 @@ class Flomni(
job by its id (indices can shift if the queue is reordered):
flomni._tomo_queue_proxy.update_by_id(job_id, status="done")
Command jobs (job["kind"] == "command", added by
tomo_queue_add_command()) are dispatched to _run_command_job()
instead of tomo_scan()/tomo_scan_resume() -- see its docstring for
their own resume policy. A job with no "kind" key is treated as
"tomo" (back-compat for queues persisted before command jobs
existed).
Args:
start_index: Position in the *current* queue order below which
fresh "pending" jobs are ignored. Defaults to 0. Resumable
@@ -3747,18 +3973,22 @@ class Flomni(
job_id = job["id"]
label = job["label"]
kind = job.get("kind", "tomo")
resume_job = job["status"] in ("incomplete", "running")
print(f"\n=== Tomo queue job: {label} ===")
for name, value in job["params"].items():
setattr(self, name, value)
self._tomo_queue_proxy.update_by_id(job_id, status="running")
try:
if resume_job:
self.tomo_scan_resume()
if kind == "command":
self._run_command_job(job, resume_job)
else:
self.tomo_scan()
for name, value in job["params"].items():
setattr(self, name, value)
if resume_job:
self.tomo_scan_resume()
else:
self.tomo_scan()
except Exception as exc:
self._tomo_queue_proxy.update_by_id(job_id, status="incomplete")
print(f"Tomo queue job '{label}' did not complete: {exc}")
@@ -3771,6 +4001,43 @@ class Flomni(
print("\nTomo queue finished -- all jobs done.")
def _run_command_job(self, job: dict, resume_job: bool) -> None:
"""Run (or resume) a "command" job's steps in order.
No per-step resume cursor -- deliberately: tracking which step of a
multi-step job completed means more persisted state and more edge
cases than this warrants. Instead the job-level "idempotent" flag
decides:
- idempotent: silently re-run every step from the top. Harmless --
absolute moves and in/out actions re-run cleanly.
- not idempotent: prompt the operator rather than silently re-run,
which could be actively wrong for a relative/stateful step. "no"
leaves the steps un-run; the caller (tomo_queue_execute) still
marks the job "done" either way, matching the same edge case tomo
jobs have -- a job that actually finished but crashed before its
status hit "done".
"""
label = job["label"]
if resume_job:
if job.get("idempotent"):
print(
f"Command job '{label}' may have partially run before the crash "
"-- idempotent, re-running from the top."
)
else:
if not self.OMNYTools.yesno(
f"Command job '{label}' may have partially run before the crash "
"(not every step is idempotent). Re-run it from the top?",
"n",
):
print(f"Skipping '{label}' -- it will be marked done without re-running.")
return
for step in job["steps"]:
spec = self._TOMO_QUEUE_ACTIONS[step["action"]]
method = getattr(self, spec["func"])
method(**step["kwargs"])
def rt_off(self):
dev.rtx.enabled = False
dev.rty.enabled = False
csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py
+11 -2
View File
@@ -2845,6 +2845,7 @@ function renderTomoQueue(jobs){{
jobs.forEach((job,i)=>{{
const status=job.status||'pending';
const label=job.label||('Job '+(i+1));
const isCommand=job.kind==='command';
const params=job.params||{{}};
const key=(job.added_at||'')+'|'+label;
presentKeys.add(key);
@@ -2855,7 +2856,13 @@ function renderTomoQueue(jobs){{
shouldOpen=true;
autoOpened.add(key);
}}
const paramRows=buildParamRows(params);
// Command jobs have no "params" key -- render their step sequence
// instead of an (otherwise blank) params table.
const paramRows=isCommand?'':buildParamRows(params);
const stepsSummary=isCommand
?(job.steps||[]).map(s=>s.action+(s.kwargs&&Object.keys(s.kwargs).length?JSON.stringify(s.kwargs):'')).join(' → ')
:'';
const idemBadge=isCommand?(job.idempotent?' [idem]':' [NOT idem]'):'';
const addedAt=job.added_at
?'Added '+new Date(job.added_at).toLocaleString([],{{month:'short',day:'numeric',hour:'2-digit',minute:'2-digit'}})
:'';
@@ -2866,7 +2873,9 @@ function renderTomoQueue(jobs){{
+'<span class="tq-badge tq-'+esc(status)+'">'+esc(status)+'</span>'
+'</summary>'
+'<div class="tq-details">'
+(paramRows?'<table class="tq-params">'+paramRows+'</table>':'')
+(isCommand
?'<div class="tq-steps">CMD '+esc(stepsSummary)+esc(idemBadge)+'</div>'
:(paramRows?'<table class="tq-params">'+paramRows+'</table>':''))
+(addedAt?'<div class="tq-added">'+esc(addedAt)+'</div>':'')
+'</div>'
+'</details>';
tests/e2e/_bootstrap.py → omny_e2e_tests/_bootstrap.py
View File
tests/e2e/_queue_helpers.py → omny_e2e_tests/_queue_helpers.py
View File
tests/e2e/_run_queue_subprocess.py → omny_e2e_tests/_run_queue_subprocess.py
View File
tests/e2e/conftest.py → omny_e2e_tests/conftest.py
View File
tests/sim_flomni_harness.py → omny_e2e_tests/sim_flomni_harness.py
View File
tests/sim_lamni_omny_harness.py → omny_e2e_tests/sim_lamni_omny_harness.py
View File
tests/e2e/test_smoke.py → omny_e2e_tests/test_smoke.py
View File
tests/e2e/test_tomo_queue.py → omny_e2e_tests/test_tomo_queue.py
View File
tests/e2e/test_tomo_queue_concurrency.py → omny_e2e_tests/test_tomo_queue_concurrency.py
View File
tests/e2e/test_tomo_queue_crash.py → omny_e2e_tests/test_tomo_queue_crash.py
View File
tests/e2e/test_tomo_queue_crash_subprocess.py → omny_e2e_tests/test_tomo_queue_crash_subprocess.py
View File