Mirko HollerandClaude Sonnet 5
ff50711dc9
Sections A-C of TOMO_QUEUE_TESTING.md's checklist, run against a live sim session: params-restored-per-job, legacy queue migration, empty/all-done no-ops, start_index semantics, exception/SIGKILL crash-resume, resume- before-fresh ordering, and concurrent queue edits from a second client. All 10 tests pass together with no cross-test interference. The flomni_sim fixture and _bootstrap.py factor out what it takes to construct a live Flomni against the sim (builtins/reload bootstrap, side-effect neutralization, RT-feedback/fsamx setup) for reuse by both in-process tests and the SIGKILL-based subprocess tests. Also adds the AI_docs/ handoff docs (testing checklist results, command- jobs plan with the action-registry decisions) that this work updated. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
"""Shared bootstrap for constructing a live Flomni against the running sim.
|
|
|
|
Used both by conftest.py (in-process, section A) and by
|
|
_run_queue_subprocess.py (spawned as a separate OS process, section B/C) --
|
|
pulled out here so both paths for "build a real Flomni pointed at the sim"
|
|
share one definition of what real-instrument setup and side-effect
|
|
neutralization is needed. See TOMO_QUEUE_TESTING.md section 3.
|
|
|
|
Deliberately does NOT call bec.config.load_demo_config()/
|
|
update_session_with_file(): assumes the running sim session already has
|
|
device_configs/simulated_omny/simulated_flomni.yaml loaded.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import builtins
|
|
import importlib
|
|
|
|
from bec_ipython_client import BECIPythonClient
|
|
from bec_lib.messaging_services import SciLogMessagingService
|
|
from bec_lib.redis_connector import RedisConnector
|
|
from bec_lib.service_config import ServiceConfig
|
|
|
|
from csaxs_bec.bec_ipython_client.plugins.flomni import flomni as flomni_module
|
|
from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
|
|
FlomniWebpageGenerator,
|
|
)
|
|
from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import OMNYTools
|
|
|
|
|
|
class _NoOpSciLogMessage:
|
|
"""Chainable no-op standing in for a real SciLog message object."""
|
|
|
|
def add_text(self, *a, **k):
|
|
return self
|
|
|
|
def add_tags(self, *a, **k):
|
|
return self
|
|
|
|
def add_attachment(self, *a, **k):
|
|
return self
|
|
|
|
def send(self, *a, **k):
|
|
return self
|
|
|
|
|
|
def neutralize_side_effects() -> None:
|
|
"""Patch away operator-facing/networked side effects of Flomni().
|
|
|
|
Plain class-attribute assignment, not pytest's monkeypatch: every
|
|
caller (in-process fixture, subprocess script) wants the same patches
|
|
for the life of the process, so there's nothing to revert.
|
|
|
|
- OMNYTools.yesno -> always answers yes, so tomo_queue_execute()'s
|
|
confirmation prompt doesn't hang a headless run.
|
|
- FlomniWebpageGenerator.start -> no-op. Flomni.__init__ normally starts
|
|
this unconditionally, which opens a local HTTP server and spawns
|
|
background threads that POST to a real external URL.
|
|
- SciLogMessagingService.new -> no-op. tomo_scan() unconditionally sends
|
|
a SciLog summary at the end of every scan; SciLog isn't
|
|
configured/reachable in this dev environment.
|
|
"""
|
|
OMNYTools.yesno = lambda self, *a, **k: True
|
|
FlomniWebpageGenerator.start = lambda self: None
|
|
SciLogMessagingService.new = lambda self, *a, **k: _NoOpSciLogMessage()
|
|
|
|
|
|
def build_flomni(services_config_path):
|
|
"""Connect to the already-running sim and return (bec, flomni)."""
|
|
config = ServiceConfig(services_config_path)
|
|
bec = BECIPythonClient(config, RedisConnector, forced=True, wait_for_server=True)
|
|
bec.start()
|
|
|
|
# flomni.py's module-level `umv` resolves `scans` from the module's own
|
|
# globals, populated only by the `if builtins.__dict__.get("bec") is not
|
|
# None:` block at import time. Set builtins first, then reload so it
|
|
# re-runs (needed even on a fresh process: importing flomni_module above
|
|
# ran that block once already, before bec existed).
|
|
builtins.__dict__["bec"] = bec
|
|
builtins.__dict__["dev"] = bec.device_manager.devices
|
|
builtins.__dict__["scans"] = bec.scans
|
|
importlib.reload(flomni_module)
|
|
builtins.__dict__["umv"] = flomni_module.umv
|
|
|
|
neutralize_side_effects()
|
|
|
|
flomni = flomni_module.Flomni(bec)
|
|
|
|
# RT interferometer feedback + fsamx-in-position setup, mirroring
|
|
# Flomni._align_setup(). Only intervene when actually needed: fsamx is
|
|
# normally left read_only (locked) by a *previous*
|
|
# feedback_enable_with_reset() call (its own last step, so the RT
|
|
# controller's PID correction owns fine positioning going forward), and
|
|
# a move request against a read_only device raises DisabledDeviceError.
|
|
# feedback_enable_with_reset() itself is expensive (re-zeros the
|
|
# interferometers) -- skip it entirely if feedback is already running,
|
|
# which it normally is after the first caller in a test session sets it
|
|
# up (device-server state persists across client (re)connections).
|
|
rtx = bec.device_manager.devices.rtx
|
|
if not rtx.controller.feedback_is_running():
|
|
fsamx = bec.device_manager.devices.fsamx
|
|
fsamx_in = fsamx.user_parameter.get("in")
|
|
if abs(fsamx.readback.get() - fsamx_in) > 0.3:
|
|
fsamx.read_only = False
|
|
flomni_module.umv(fsamx, fsamx_in)
|
|
flomni.feedback_enable_with_reset()
|
|
|
|
return bec, flomni
|
|
|
|
|
|
def shutdown(bec) -> None:
|
|
bec.shutdown()
|
|
bec._client._reset_singleton()
|