From 30b5119a2f8824087dd4e970ead66d752c54252e Mon Sep 17 00:00:00 2001 From: Mirko Holler Date: Mon, 13 Jul 2026 09:05:29 +0200 Subject: [PATCH] feat(flomni): add command jobs to the tomo queue 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. --- .../flomni/AI_docs}/SIMULATED_ENDSTATIONS.md | 0 .../plugins/flomni/flomni.py | 277 +++++++++++++++++- .../flomni/flomni_webpage_generator.py | 13 +- {tests/e2e => omny_e2e_tests}/_bootstrap.py | 0 .../e2e => omny_e2e_tests}/_queue_helpers.py | 0 .../_run_queue_subprocess.py | 0 {tests/e2e => omny_e2e_tests}/conftest.py | 0 .../sim_flomni_harness.py | 0 .../sim_lamni_omny_harness.py | 0 {tests/e2e => omny_e2e_tests}/test_smoke.py | 0 .../e2e => omny_e2e_tests}/test_tomo_queue.py | 0 .../test_tomo_queue_concurrency.py | 0 .../test_tomo_queue_crash.py | 0 .../test_tomo_queue_crash_subprocess.py | 0 14 files changed, 283 insertions(+), 7 deletions(-) rename {docs => csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs}/SIMULATED_ENDSTATIONS.md (100%) rename {tests/e2e => omny_e2e_tests}/_bootstrap.py (100%) rename {tests/e2e => omny_e2e_tests}/_queue_helpers.py (100%) rename {tests/e2e => omny_e2e_tests}/_run_queue_subprocess.py (100%) rename {tests/e2e => omny_e2e_tests}/conftest.py (100%) rename {tests => omny_e2e_tests}/sim_flomni_harness.py (100%) rename {tests => omny_e2e_tests}/sim_lamni_omny_harness.py (100%) rename {tests/e2e => omny_e2e_tests}/test_smoke.py (100%) rename {tests/e2e => omny_e2e_tests}/test_tomo_queue.py (100%) rename {tests/e2e => omny_e2e_tests}/test_tomo_queue_concurrency.py (100%) rename {tests/e2e => omny_e2e_tests}/test_tomo_queue_crash.py (100%) rename {tests/e2e => omny_e2e_tests}/test_tomo_queue_crash_subprocess.py (100%) diff --git a/docs/SIMULATED_ENDSTATIONS.md b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/SIMULATED_ENDSTATIONS.md similarity index 100% rename from docs/SIMULATED_ENDSTATIONS.md rename to csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/SIMULATED_ENDSTATIONS.md diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py index c8b0fb0..2429b28 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py @@ -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_". + 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 . + + 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 diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py index 4c59eff..a3b7198 100644 --- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py +++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py @@ -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){{ +''+esc(status)+'' +'' +'
' - +(paramRows?''+paramRows+'
':'') + +(isCommand + ?'
CMD '+esc(stepsSummary)+esc(idemBadge)+'
' + :(paramRows?''+paramRows+'
':'')) +(addedAt?'
'+esc(addedAt)+'
':'') +'
' +''; diff --git a/tests/e2e/_bootstrap.py b/omny_e2e_tests/_bootstrap.py similarity index 100% rename from tests/e2e/_bootstrap.py rename to omny_e2e_tests/_bootstrap.py diff --git a/tests/e2e/_queue_helpers.py b/omny_e2e_tests/_queue_helpers.py similarity index 100% rename from tests/e2e/_queue_helpers.py rename to omny_e2e_tests/_queue_helpers.py diff --git a/tests/e2e/_run_queue_subprocess.py b/omny_e2e_tests/_run_queue_subprocess.py similarity index 100% rename from tests/e2e/_run_queue_subprocess.py rename to omny_e2e_tests/_run_queue_subprocess.py diff --git a/tests/e2e/conftest.py b/omny_e2e_tests/conftest.py similarity index 100% rename from tests/e2e/conftest.py rename to omny_e2e_tests/conftest.py diff --git a/tests/sim_flomni_harness.py b/omny_e2e_tests/sim_flomni_harness.py similarity index 100% rename from tests/sim_flomni_harness.py rename to omny_e2e_tests/sim_flomni_harness.py diff --git a/tests/sim_lamni_omny_harness.py b/omny_e2e_tests/sim_lamni_omny_harness.py similarity index 100% rename from tests/sim_lamni_omny_harness.py rename to omny_e2e_tests/sim_lamni_omny_harness.py diff --git a/tests/e2e/test_smoke.py b/omny_e2e_tests/test_smoke.py similarity index 100% rename from tests/e2e/test_smoke.py rename to omny_e2e_tests/test_smoke.py diff --git a/tests/e2e/test_tomo_queue.py b/omny_e2e_tests/test_tomo_queue.py similarity index 100% rename from tests/e2e/test_tomo_queue.py rename to omny_e2e_tests/test_tomo_queue.py diff --git a/tests/e2e/test_tomo_queue_concurrency.py b/omny_e2e_tests/test_tomo_queue_concurrency.py similarity index 100% rename from tests/e2e/test_tomo_queue_concurrency.py rename to omny_e2e_tests/test_tomo_queue_concurrency.py diff --git a/tests/e2e/test_tomo_queue_crash.py b/omny_e2e_tests/test_tomo_queue_crash.py similarity index 100% rename from tests/e2e/test_tomo_queue_crash.py rename to omny_e2e_tests/test_tomo_queue_crash.py diff --git a/tests/e2e/test_tomo_queue_crash_subprocess.py b/omny_e2e_tests/test_tomo_queue_crash_subprocess.py similarity index 100% rename from tests/e2e/test_tomo_queue_crash_subprocess.py rename to omny_e2e_tests/test_tomo_queue_crash_subprocess.py