Mirko HollerandClaude Opus 5.5
ffcc879d42
Brings the LamNI beamtime-2 line (fixes/lamni_during_beamtime_2, contained in this branch) and the Keithley 2400 work up to main so it can go in by fast-forward. Conflicts resolved: - omny_general_tools.py: keep stitch_tile_offsets/lamni_smallest_tile_position_count (branch) and read_escape_suffix (main) - flomni.py: import both helpers; state keeps recent_projection_gaps and tomo_completed - tomo_params.py: _job_tooltip(job, registry=None, profile=None) combines the branch's command-step formatting and main's per-profile field filtering; call site passes both by keyword - test_rt_flomni.py: keep both new tests Co-Authored-By: Claude Opus 5.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_bl_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_shared.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.rt_feedback_enable_with_reset()
|
|
|
|
return bec, flomni
|
|
|
|
|
|
def shutdown(bec) -> None:
|
|
bec.shutdown()
|
|
bec._client._reset_singleton()
|