bec/csaxs_bec
0
0 Fork 3
Code Issues 17 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

fix(canon): make the probe fully usable without installing anything

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
menzelandClaude Opus 5 committed 2026-08-31 16:41:06 +02:00
1 parent 7885b3f1d4
commit 368db4f8ef
2 files changed
+107 -15
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

csaxs_bec/devices/canon/probe.py
+58 -15
View File
@@ -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).
tests/tests_devices/test_canon_probe.py
+49
View File
@@ -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(