csaxs_bec/omny_e2e_tests/_bootstrap.py

"""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_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.feedback_enable_with_reset()

    return bec, flomni


def shutdown(bec) -> None:
    bec.shutdown()
    bec._client._reset_singleton()