From 368db4f8ef39501b794f9f6bdd92b4bdbc84cffb Mon Sep 17 00:00:00 2001 From: menzel Date: Mon, 17 Aug 2026 18:36:04 +0200 Subject: [PATCH] fix(canon): make the probe fully usable without installing anything Both fixes came out of testing the workflow actually used at the beamline -- copying the probe to a console and running it with the system python3. Install-free transport import ----------------------------- Step 6 (which exercises the shipped XCTransport) was unreachable in practice: the ordinary package import goes through canon/__init__.py, which imports the device class and therefore ophyd, so `python -m csaxs_bec.devices.canon.probe` required the whole BEC stack -- ophyd, ophyd_devices, bec_lib -- just to make one HTTP call from a beamline console. The transport itself is standard-library only; only the package __init__ chain pulls ophyd in. _import_xctransport() now tries three routes in order of how much environment each needs, ending with `transport/` imported as a top-level package, which sidesteps canon/__init__.py entirely. Copying the canon/ directory and running it on a bare system python3 (verified on 3.9) now gets all six steps with nothing installed. Copying probe.py alone still works and still reports SKIP for step 6; the message now says how to get it (copy the directory), not just that it failed. Misleading summary line ----------------------- summarise() reported "step 6 was skipped" whenever xc_connect was absent from the report -- including when the probe stopped at step 1 with no TCP connection and never reached step 6 at all. It now remarks on step 6 only if that step actually ran and opted out, and quotes the recorded reason. Tests: 78 (was 76). The new pair covers the install-free import route (in a subprocess with no package context, so the fallback is genuinely exercised) and the summary no longer commenting on a step it never reached. Co-Authored-By: Claude Opus 5 (1M context) --- csaxs_bec/devices/canon/probe.py | 73 ++++++++++++++++++++----- tests/tests_devices/test_canon_probe.py | 49 +++++++++++++++++ 2 files changed, 107 insertions(+), 15 deletions(-) diff --git a/csaxs_bec/devices/canon/probe.py b/csaxs_bec/devices/canon/probe.py index b6afc0cd..86eecf48 100644 --- a/csaxs_bec/devices/canon/probe.py +++ b/csaxs_bec/devices/canon/probe.py @@ -279,19 +279,15 @@ class Probe: def step_xc_session(self) -> None: """Try the implemented transport as-is, so we know if the guesses happened to fit.""" print("\n[6] XCTransport.connect() -- does the shipped implementation work as-is?") - try: - from .transport.xc import XCTransport # pylint: disable=import-outside-toplevel - except ImportError: - try: - from csaxs_bec.devices.canon.transport.xc import ( # noqa: E501 pylint: disable=import-outside-toplevel - XCTransport, - ) - except ImportError: - print(" SKIP running as a standalone file -- the shipped transport is not") - print(" importable here. Step 5 above already answers the question;") - print(" re-run from a checkout to also validate XCTransport itself.") - self.report["steps"]["xc_connect"] = {"ok": None, "reason": "standalone"} - return + transport_cls = _import_xctransport() + if transport_cls is None: + print(" SKIP the shipped transport is not reachable from here. Step 5 above") + print(" already answers 'can a session be opened'; to validate") + print(" XCTransport itself, copy the whole canon/ directory (not just") + print(" this file) -- no install needed -- or run from a BEC checkout.") + self.report["steps"]["xc_connect"] = {"ok": None, "reason": "transport unavailable"} + return + XCTransport = transport_cls # noqa: N806 (class bound at runtime) user, password = self.auth if self.auth else (None, None) transport = XCTransport( host=self.host, port=self.port, username=user, password=password, timeout=_TIMEOUT @@ -345,11 +341,58 @@ class Probe: print("VERDICT: reachable over HTTP, but no candidate CGI path answered.") print(" The protocol differs from our assumptions -- harvest it from") print(" the camera's own web UI. Next: COMMISSIONING.md Stage 0.5.3.") - if steps.get("xc_connect", {}).get("ok") is None: - print(" (step 6 was skipped -- standalone file, not a checkout.)") + # Only remark on step 6 if it actually ran and opted out -- not when the probe + # stopped earlier and never reached it. + if "xc_connect" in steps and steps["xc_connect"].get("ok") is None: + print(f" (step 6 skipped: {steps['xc_connect'].get('reason')}.)") print("=" * 72) +def _import_xctransport(): + """Locate ``XCTransport``, trying progressively less-demanding routes. + + Ordered by how much environment each needs: + + 1. the normal package import -- requires the full BEC stack, because + ``canon/__init__.py`` pulls in ophyd via the device class; + 2. the same, by absolute name, when this file runs as a plain script; + 3. **``transport/`` as a top-level package**, sidestepping ``canon/__init__.py`` + entirely. The transport is standard-library only, so this works with nothing + installed -- copy the ``canon/`` directory to a beamline machine and step 6 + still runs on the system ``python3``. + + Returns the class, or ``None`` if none of the routes work. + """ + import os # pylint: disable=import-outside-toplevel + + try: + from .transport.xc import XCTransport # pylint: disable=import-outside-toplevel + + return XCTransport + except ImportError: + pass + try: + from csaxs_bec.devices.canon.transport.xc import ( # pylint: disable=import-outside-toplevel + XCTransport, + ) + + return XCTransport + except ImportError: + pass + + here = os.path.dirname(os.path.abspath(__file__)) + if not os.path.isdir(os.path.join(here, "transport")): + return None + if here not in sys.path: + sys.path.insert(0, here) + try: + from transport.xc import XCTransport # pylint: disable=import-outside-toplevel + + return XCTransport + except ImportError: + return None + + def _parse_session_id(text: str, set_cookie: str) -> str | None: """Extract a session id from an ``open.cgi`` response (body, else Set-Cookie). diff --git a/tests/tests_devices/test_canon_probe.py b/tests/tests_devices/test_canon_probe.py index 97c4a6e0..bfe3d13f 100644 --- a/tests/tests_devices/test_canon_probe.py +++ b/tests/tests_devices/test_canon_probe.py @@ -17,6 +17,11 @@ what a PASS looks like before hardware day: from __future__ import annotations import base64 +import importlib +import os +import pathlib +import subprocess +import sys import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -232,6 +237,50 @@ def test_session_id_parsing_covers_body_and_set_cookie(): assert _parse_session_id("model=CR-N300\n", "") is None +def test_transport_import_has_an_install_free_route(): + """Step 6 must be reachable without ophyd/BEC installed. + + `canon/__init__.py` imports the device class, which pulls in ophyd -- so the + ordinary package import needs the whole stack. The last resort imports + `transport/` as a top-level package instead, which is stdlib-only, so a copied + `canon/` directory works on a bare beamline python3. + """ + from csaxs_bec.devices.canon.probe import _import_xctransport + + assert _import_xctransport() is not None + + canon_dir = pathlib.Path( + importlib.import_module("csaxs_bec.devices.canon.probe").__file__ + ).parent + # The fallback route itself: transport as a top-level package, no canon/__init__. + proc = subprocess.run( + [ + sys.executable, + "-c", + "import sys; sys.path.insert(0, sys.argv[1]);" + "from transport.xc import XCTransport;" + "print(XCTransport.__name__)", + str(canon_dir), + ], + capture_output=True, + text=True, + env={"PATH": os.environ.get("PATH", "")}, # no PYTHONPATH: no package context + check=False, + ) + assert proc.returncode == 0, proc.stderr + assert "XCTransport" in proc.stdout + + +def test_summary_does_not_claim_step_6_skipped_when_it_never_ran(capsys): + """A probe that stops at step 1 must not report on a step it never reached.""" + probe = Probe("127.0.0.1", 9, "http", None) + probe.report["steps"]["tcp"] = {"ok": False} + probe.summarise() + out = capsys.readouterr().out + assert "no TCP connection" in out + assert "step 6" not in out + + def test_probe_main_runs_end_to_end_and_writes_json(camera, tmp_path): out = tmp_path / "report.json" rc = main(