diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ac3cec..b6b070b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,29 @@ Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow ## [Unreleased] ### Added +- `analysis/worm_short.py` - C. elegans Short from an AML18 time-lapse + surveyed by `aml18_survey`: neurons-only opening, brightfield with RFP glow, + trails and a live distance chart for up to three featured worms (auto-picked + adults, named by speed), and every track at once. Original synthesised music. + The speed claim uses the smaller of the raw and median-filtered ratios. +- `analysis/physarum_short.py` - the whole-dish Physarum Short: time-lapse + with a live on-agar area curve, then a wipe to the green new-growth map. + Reuses the combined analysis folder's alignment so frames match what was + measured. Captions are the 23-25 Sep 2026 experiment's. +- `analysis/short_video.py` - shared pieces for vertical YouTube Shorts: + 1080x1920 canvas, fonts, centred text, and a writer that pipes frames into + the bundled ffmpeg (libx264, CRF 19) with the WAV soundtrack muxed in the + same pass. `--preview`-style stills instead of a render for layout checks. +- NIS bridge (not yet run inside NIS - see `docs/nis-bridge.md`): + `acquisition/nis_bridge/bridge_job.py`, a JOBS Python task that serves NIS + macro calls (status, relative XY move, 4x/10x objective change, ND2 + capture, saved ND experiment run/finish) on 127.0.0.1:8766; + `acquisition/backends/nis_bridge.py`, its client and command line; and + `mcp_server/server_nis.py` (`confocal-mcp-nis`), a separate MCP server + with confirm-gated tools. The e-stop is enforced in both client and + bridge; `server_loop` is unchanged. +- `acquisition/calibration/nis_port_probe.py` - minimal JOBS task showing + that a port bound inside `nis_ar.exe` is reachable from outside. - `analysis/make_soundtrack.py` - original ambient soundtrack synthesised with numpy (no samples, so no licensing questions), stretched to any length. `--movie run.mp4` sizes it to the movie and muxes it on as AAC @@ -77,6 +100,16 @@ Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow back to `~/.confocal-mcp` with a warning on stderr. Setting `CONFOCAL_MCP_DATA_DIR` is still honoured as-is, and a normal source checkout still resolves to the checkout directory. +- The `analysis/` scripts imported `scipy`, `scikit-image`, `nd2`, + `imageio-ffmpeg` and `opencv-python`, none of which the package declared, so + a fresh install could not run them. They are now the `analysis` optional + group, which `all` (and so `requirements.txt`) includes. +- `python -m acquisition.orchestration.stage_positions` ended by loading + `protocols/example_protocol.yaml`, which is not in this repo, so the demo + always failed. That step is removed. Code comments that pointed to files in + ConfocalOrchestrator (`docs/microscope-notes.md`, `run_protocol.py`, the + calibration tests) now say so, and those citing old names + (`mcp_server/server.py`, `acquisition_tools.py`) use the current ones. ## [0.1.0] - 2026-08-31 diff --git a/README.md b/README.md index 9e70adb..8646c9a 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,21 @@ python -m timelapse.scheduler --backend mock --model-trigger ... # Claude judg python -m timelapse.frame_audit FRAME_DIR --timestamps times.csv --around 25h --window 1h ``` +## NIS bridge (`confocal-mcp-nis`) - not yet tested on the scope + +A separate MCP server for what only NIS-Elements can do: the AX confocal +lasers and saved ND experiments. A JOBS Python task inside NIS +(`acquisition/nis_bridge/bridge_job.py`) serves NIS macro calls on +`127.0.0.1:8766`. The `confocal-mcp-nis` server talks to it and adds a +confirm gate, the e-stop and a capture-folder rule. `server_loop` is +unchanged. Setup, safety rules and the on-scope test plan: +[docs/nis-bridge.md](docs/nis-bridge.md). + +``` +python -m acquisition.backends.nis_bridge status # by hand, while the bridge job runs +confocal-mcp-nis # the MCP server +``` + ## Tests ``` @@ -85,10 +100,11 @@ point. Dependencies are split into groups so a machine only pulls what it needs: | Group | Pulls in | For | |---|---|---| -| *(core)* | `mcp`, `Pillow`, `PyYAML` | the MCP server against the **mock** stage | +| *(core)* | `mcp`, `Pillow`, `PyYAML`, `numpy` | the MCP server against the **mock** stage | | `camera` | `harvesters`, `genicam`, `opencv-python` | real Baumer camera capture | | `sdk` | `pywin32` | real Ti2 stage control (Windows) | | `harness` | `anthropic`, `python-dotenv` | the standalone Claude loops in `harness/` | +| `analysis` | `scipy`, `scikit-image`, `nd2`, `imageio-ffmpeg`, `opencv-python` | the offline scripts in `analysis/` (source checkout only) | | `all` | everything above | a full workstation | The real hardware backends are imported lazily, so a core-only install runs diff --git a/acquisition/backends/baumer_genicam.py b/acquisition/backends/baumer_genicam.py index 46182bb..c5fb09c 100644 --- a/acquisition/backends/baumer_genicam.py +++ b/acquisition/backends/baumer_genicam.py @@ -10,7 +10,7 @@ # (Baumer Camera Explorer + Toshiba Teli GenICam SDK). Originally added # just to exercise the rest of the pipeline (focus_check, dashboard) # against real frames while N-SPARC capture was blocked on NIS-Elements -# licensing/Jobs work (see docs/microscope-notes.md's "Image Capture" +# licensing/Jobs work (see ConfocalOrchestrator's docs/microscope-notes.md, "Image Capture" # investigation for that history). # # 2026-08-17: now the PRIMARY capture path by deliberate decision - the diff --git a/acquisition/backends/nis_bridge.py b/acquisition/backends/nis_bridge.py new file mode 100644 index 0000000..c0c9f61 --- /dev/null +++ b/acquisition/backends/nis_bridge.py @@ -0,0 +1,128 @@ +# nis_bridge.py +# ------------------------------------------------------------ +# Client for the NIS bridge (acquisition/nis_bridge/bridge_job.py), the +# JOBS task that exposes NIS-Elements macro calls on 127.0.0.1:8766 while +# it runs. Used by the NIS MCP server; also a command line for testing by +# hand: +# +# python -m acquisition.backends.nis_bridge status +# python -m acquisition.backends.nis_bridge move 100 0 +# python -m acquisition.backends.nis_bridge objective "Plan Fluor 10x Ph1 DLL" +# python -m acquisition.backends.nis_bridge capture D:\...\check.nd2 +# python -m acquisition.backends.nis_bridge nd-run "C. elegans 3h" +# python -m acquisition.backends.nis_bridge nd-finish +# python -m acquisition.backends.nis_bridge shutdown +# +# The e-stop is checked HERE as well as in the bridge: a refused request +# never leaves this process, so a stopped microscope does not depend on +# the NIS side having been updated. Stopping an ND run (nd-finish) and +# shutting the bridge down are always allowed - they only end things. +# +# CONFOCAL_NIS_BRIDGE_URL overrides the address (default +# http://127.0.0.1:8766). +# ------------------------------------------------------------ + +from __future__ import annotations + +import argparse +import json +import os +import urllib.error +import urllib.request + +from acquisition import estop + +DEFAULT_URL = "http://127.0.0.1:8766" + + +class BridgeUnavailable(RuntimeError): + """The bridge job is not running (or not reachable).""" + + +class NISBridge: + def __init__(self, url: str | None = None, timeout_s: float = 30.0): + self.url = (url or os.environ.get("CONFOCAL_NIS_BRIDGE_URL") or DEFAULT_URL).rstrip("/") + self.timeout_s = timeout_s + + def _call(self, method: str, path: str, body: dict | None = None, + timeout_s: float | None = None) -> dict: + data = json.dumps(body or {}).encode() if method == "POST" else None + req = urllib.request.Request(self.url + path, data=data, method=method, + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=timeout_s or self.timeout_s) as r: + return json.loads(r.read()) + except urllib.error.HTTPError as e: + try: + return {"http_status": e.code, **json.loads(e.read())} + except ValueError: + return {"http_status": e.code, "error": str(e)} + except (urllib.error.URLError, ConnectionError, TimeoutError) as e: + raise BridgeUnavailable( + f"NIS bridge not reachable at {self.url} ({e}). Is the bridge job " + "running in NIS-Elements? See docs/nis-bridge.md.") from e + + # Read-only + def status(self) -> dict: + return self._call("GET", "/status") + + # Motion / acquisition: e-stop checked before anything is sent + def move_relative(self, dx_um: float, dy_um: float) -> dict: + estop.check() + return self._call("POST", "/move", {"dx": dx_um, "dy": dy_um}) + + def change_objective(self, name: str) -> dict: + estop.check() + return self._call("POST", "/objective", {"name": name}) + + def capture(self, path: str) -> dict: + estop.check() + return self._call("POST", "/capture", {"path": path}, timeout_s=300) + + def nd_run(self, experiment: str, timeout_s: float = 7200.0) -> dict: + # ND_RunExperiment may block until the run ends (unverified), so + # the answer can take as long as the run itself. + estop.check() + return self._call("POST", "/nd/run", {"experiment": experiment}, timeout_s=timeout_s) + + # Always allowed: these only end things + def nd_finish(self) -> dict: + return self._call("POST", "/nd/finish") + + def shutdown(self) -> dict: + return self._call("POST", "/shutdown") + + +def main() -> None: + ap = argparse.ArgumentParser(description="Talk to the NIS bridge job by hand.") + ap.add_argument("--url", help=f"bridge address (default {DEFAULT_URL})") + sub = ap.add_subparsers(dest="cmd", required=True) + sub.add_parser("status") + m = sub.add_parser("move", help="relative XY move in um (max 1000 per axis)") + m.add_argument("dx", type=float); m.add_argument("dy", type=float) + o = sub.add_parser("objective", help="exact name as listed by status (4x/10x only)") + o.add_argument("name") + c = sub.add_parser("capture", help="capture one image and save it as .nd2") + c.add_argument("path") + r = sub.add_parser("nd-run", help="load a saved ND experiment by name and run it") + r.add_argument("experiment") + sub.add_parser("nd-finish", help="end the running ND experiment after this loop") + sub.add_parser("shutdown", help="stop the bridge job") + a = ap.parse_args() + + b = NISBridge(a.url) + try: + out = {"status": lambda: b.status(), + "move": lambda: b.move_relative(a.dx, a.dy), + "objective": lambda: b.change_objective(a.name), + "capture": lambda: b.capture(a.path), + "nd-run": lambda: b.nd_run(a.experiment), + "nd-finish": lambda: b.nd_finish(), + "shutdown": lambda: b.shutdown()}[a.cmd]() + except (estop.EStopEngaged, BridgeUnavailable) as e: + raise SystemExit(str(e)) + print(json.dumps(out, indent=2, ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/acquisition/backends/nis_mock.py b/acquisition/backends/nis_mock.py index bd4ab50..87c73e6 100644 --- a/acquisition/backends/nis_mock.py +++ b/acquisition/backends/nis_mock.py @@ -1,11 +1,11 @@ # nis_mock.py # ------------------------------------------------------------ # Mock simulator for the NIS-Elements Python API ('nis' module), so -# acquisition scripts (stage_positions.py, run_protocol.py, etc.) can be +# acquisition scripts (stage_positions.py here, ConfocalOrchestrator's run_protocol.py) can be # developed and tested off the microscope PC. # # The real `nis` module only exists inside the NIS-Elements Python -# environment on the microscope PC (see nis_jobs_connection_test.py). This mock +# environment on the microscope PC (see ConfocalOrchestrator's nis_jobs_connection_test.py). This mock # reproduces the small subset of that API used by ConfocalOrchestrator - # XY/Z stage position, movement, and abort checks - as plain in-memory # state, so it can run anywhere. @@ -26,7 +26,7 @@ from acquisition.paths import captures_dir as _captures_dir, data_root as _data_root -# Stage travel limits from the Nikon Ti2-E spec (see docs/microscope-notes.md), +# Stage travel limits from the Nikon Ti2-E spec (stroke X +/-57 mm, Y +/-36.5 mm), # converted from mm to microns to match the units used by the NIS API. X_LIMIT_UM = 57_000.0 Y_LIMIT_UM = 36_500.0 @@ -41,7 +41,7 @@ MAX_Z_STEP_UM = 50.0 # Realistic movement delay for XY_Move, based on the Ti2-E's documented max -# XY stage speed (docs/microscope-notes.md: "Max speed: 25mm/sec"). The +# XY stage speed (Nikon Ti2-E spec: "Max speed: 25mm/sec"). The # focus (Z) drive's speed isn't documented, so Z_Move uses a small fixed # placeholder delay instead of a physics-based one. XY_MAX_SPEED_UM_PER_SEC = 25_000.0 @@ -64,7 +64,7 @@ class _MockContext: """Mock of the NIS-Elements Jobs 'ctx' context object - only the abort-check method ConfocalOrchestrator actually uses (see - run_protocol.py's should_abort(), which calls ctx.shouldAbort()). + ConfocalOrchestrator's run_protocol.py should_abort(), which calls ctx.shouldAbort()). Always reports False - there is no UI to click Abort from in the mock. """ @@ -82,12 +82,12 @@ class MockNIS: Also exposes capture() (simulated image capture) and .ctx.shouldAbort() (mock Jobs context), for parity with the rest of the real API surface - ConfocalOrchestrator's acquisition scripts use - see run_protocol.py. + ConfocalOrchestrator's acquisition scripts use - see its run_protocol.py. NOTE on capture(): unlike XY_Move/Z_Move/XY_GetPosition/Z_GetPosition - (all confirmed against docs/microscope-notes.md's documented API), + (all confirmed against the API documented in ConfocalOrchestrator's docs/microscope-notes.md), the real capture function's name and signature are NOT confirmed yet - - run_protocol.py's capture_image() flags this as a TODO and guesses + ConfocalOrchestrator's run_protocol.py capture_image() flags this as a TODO and guesses `nis.Capture()` as a placeholder. `capture()` here is written to match this project's explicit spec for the mock, not a confirmed real signature - expect to rename/adjust it once the real one is confirmed. diff --git a/acquisition/backends/nis_sdk.py b/acquisition/backends/nis_sdk.py index 9866be6..981a74d 100644 --- a/acquisition/backends/nis_sdk.py +++ b/acquisition/backends/nis_sdk.py @@ -3,31 +3,31 @@ # Ti2 ActiveX SDK backend for stage control - real hardware via # win32com.client.Dispatch(NkTi2Ax.NikonTi2AxAutoConnectMicroscope.CLSID), # the same connection pattern confirmed working in -# acquisition/calibration/nikon_connection_test.py against the Ti2-E Device Simulator. +# ConfocalOrchestrator's acquisition/calibration/nikon_connection_test.py against the Ti2-E Device Simulator. # # ConfocalOrchestrator has two stage-control backends, both exposing the # same shape of interface so orchestration/stage_positions.py can swap # between them via its `backend` parameter - see nis_mock.py for the # other one ("mock"). # This is the "sdk" backend: direct ActiveX bindings, now that Nikon has -# approved SDK access (see docs/microscope-notes.md's "SDK Status") - +# approved SDK access (see ConfocalOrchestrator's docs/microscope-notes.md, "SDK Status") - # confirmed end-to-end against the Ti2-E Device Simulator, 2026-07-27. # # CONFIRMED PROPERTIES (from .venv/Lib/site-packages/NkTi2Ax.py, the # generated bindings for the SDK's own type library - the same file that -# defines iTURRET1POS/Turret1Pos, confirmed working in calibration/nikon_connection_test.py): +# defines iTURRET1POS/Turret1Pos, confirmed working in ConfocalOrchestrator's calibration/nikon_connection_test.py): # iXPOSITION / iYPOSITION / iZPOSITION - direct properties, readable and # writable, same shape as iTURRET1POS. # XPosition / YPosition / ZPosition - child settings objects (.Value/ # .Lower/.Higher), same shape as Turret1Pos. Read-verified against the -# Ti2-E Device Simulator via acquisition/calibration/nikon_stage_test.py - +# Ti2-E Device Simulator via ConfocalOrchestrator's acquisition/calibration/nikon_stage_test.py - # both forms returned identical values. # # UNITS (inferred, not stated anywhere explicit - the bindings just # declare a plain integer VARIANT, no unit metadata): cross-referencing # the simulator's reported Lower/Higher travel limits against -# docs/microscope-notes.md's documented hardware spec ("Stroke X: +# the Nikon Ti2-E hardware spec ("Stroke X: # +/-57mm, Y: +/-36.5mm ... Focusing: min increment 0.01um, 10mm stroke"): # X: Lower/Higher = +/-570000 -> 0.1um/count exactly reproduces +/-57mm # Z: Lower/Higher = 0..1000000 -> 0.01um/count exactly reproduces the @@ -41,7 +41,7 @@ # So: X/Y properties are in units of 0.1um ("decimicrons"), Z is in units # of 0.01um ("centimicrons"). XY_GetPosition/XY_Move/Z_GetPosition/Z_Move # below convert to/from plain microns at their boundary so callers -# (StagePositionManager, run_protocol.py) never see raw counts. +# (StagePositionManager, mcp_server/loop_tools.py) never see raw counts. # ------------------------------------------------------------ import queue @@ -92,7 +92,7 @@ def to_plain_float(value) -> float: """Convert a numpy scalar (or anything float-like) to a plain Python float. - Matches the same convention used in calibration/nis_jobs_connection_test.py/orchestration/stage_positions.py - + Matches the same convention used in orchestration/stage_positions.py (and ConfocalOrchestrator's nis_jobs_connection_test.py) - values passed to a COM property setter must be plain Python numbers, not numpy types. """ @@ -170,7 +170,7 @@ class NISSdk: Every instance shares the same underlying COM connection (see _ComThread above) - constructing NISSdk() repeatedly (once per MCP - tool call, as acquisition_tools.py does) is cheap and does not open a + tool call, as mcp_server/loop_tools.py does) is cheap and does not open a new connection each time. """ @@ -351,7 +351,7 @@ def nudge_pfs_offset(self, delta_counts: float) -> dict: mode, a different property/method entirely, or there may be an interlock not exposed by NkTi2Ax's type library. Confirm the correct procedure with Nikon's SDK docs or a Ti2 SDK-experienced - contact (see docs/microscope-notes.md) before relying on this - + contact (see ConfocalOrchestrator's docs/microscope-notes.md) before relying on this - do not attempt to fix by further trial-and-error against real hardware. Failure mode observed so far is safe (no motion, no error) - not a functional feature yet, but not a hazard either. diff --git a/acquisition/nis_bridge/__init__.py b/acquisition/nis_bridge/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/acquisition/nis_bridge/bridge_job.py b/acquisition/nis_bridge/bridge_job.py new file mode 100644 index 0000000..7456a4f --- /dev/null +++ b/acquisition/nis_bridge/bridge_job.py @@ -0,0 +1,280 @@ +# IMPORTANT: 'limjob' must be imported like this (not from nor as) +try: + import limjob +except ImportError: # off the microscope (tests): the module is only needed for type hints + limjob = None + +# bridge_job.py +# ------------------------------------------------------------ +# The NIS-Elements side of the NIS bridge: paste this file into a JOBS +# Python task and run the job. While the job runs, NIS answers a small +# JSON-over-HTTP API on 127.0.0.1:8766, which acquisition/backends/ +# nis_bridge.py (and through it the NIS MCP server) talks to. +# +# GET /status positions, objective, ND run state +# POST /move {"dx", "dy"} relative XY move, um +# POST /objective {"name"} ChangeObjective, allow-listed names only +# POST /capture {"path"} Capture, save as ND2, close the document +# POST /nd/run {"experiment"} ND_LoadExperiment + ND_RunExperiment(0) +# POST /nd/finish ND_FinishExperiment (stop after this loop) +# POST /shutdown stop serving and end the job +# +# WHY A BRIDGE AND NOT THE MCP SERVER ITSELF INSIDE NIS. Nothing has to be +# installed into NIS's own Python (Nikon pins its packages), a crash in our +# code cannot take NIS down with it, and the MCP server keeps the e-stop, +# confirm gate and tests it already has. This file only translates HTTP +# requests into NIS macro calls. +# +# HOW NIS IS CALLED. NIS macro functions are exported by g5_regprocs.dll +# and called through ctypes - the way NIS's own limpy.macro calls +# WaitText. Signatures are from the NIS 6.20 macro reference +# (Docs/nis/eng_ar on the microscope PC). A macro char* is wchar_t* in +# this build (limpy passes WaitText's text as c_wchar_p). +# +# THREADING. NIS is not known to be thread-safe, so every NIS call runs on +# the job's own thread in run(); the HTTP threads only queue requests and +# wait for the answer. A call that takes long (ND_RunExperiment may block +# until the run ends - unverified) delays the requests queued behind it. +# +# SAFETY, enforced here as well as in the client: +# * motion and acquisition only on POST, so a browser cannot trigger them +# * nothing moves or starts while the e-stop file exists (the same file +# acquisition/estop.py uses) +# * XY moves are relative and capped at MAX_STEP_UM; Z is never moved +# * objective changes only to 4x/10x names (ALLOWED_MAGNIFICATIONS): +# working distances of 20 and 4 mm cannot reach the dish +# * serves on 127.0.0.1 only +# ------------------------------------------------------------ +import ctypes as ct +import http.server +import json +import os +import queue +import re +import threading +import time +from pathlib import Path + +PORT = 8766 # 8765 is the plain port probe (nis_port_probe.py) +VERSION = 1 +MAX_STEP_UM = 1000.0 +ALLOWED_MAGNIFICATIONS = (4, 10) +# "4x" / "10x" as a whole magnification - not the 4x inside "14x" or "40x" +_MAG = re.compile(r"(? bool: + try: + return ESTOP.exists() + except OSError: + return True # cannot tell -> treat as engaged, like estop.py + + +class Bridge: + """The operations, as plain methods returning JSON-able dicts.""" + + def __init__(self, macro: Macro): + self.m = macro + + def status(self) -> dict: + x, y, z = ct.c_double(), ct.c_double(), ct.c_double() + rc_xy = self.m.StgGetPosXY(ct.byref(x), ct.byref(y)) + rc_z = self.m.StgGetPosZ(ct.byref(z), 0) + cur = ct.create_unicode_buffer(255) + self.m.GetCurrentObjName(cur) + names = {} + for i in range(0, self.m.Stg_GetNosepiecePositions() + 1): # index base unverified + b = ct.create_unicode_buffer(255) + if self.m.Stg_GetNosepieceObjectiveName(i, b, 255) == 1 and b.value: + names[str(i)] = b.value + return {"version": VERSION, "pid": os.getpid(), + "xy_um": [x.value, y.value], "xy_rc": rc_xy, + "z_um": z.value, "z_rc": rc_z, + "nosepiece_position": self.m.Stg_GetNosepiecePosition(), + "current_objective": cur.value, "nosepiece_objectives": names, + "nd_running": bool(self.m.ND_IsInExperimentCapture()), + "estop_engaged": _estop_engaged()} + + def _refuse_if_stopped(self, what: str) -> dict | None: + if _estop_engaged(): + return {"error": f"e-stop engaged - {what} refused", "estop_engaged": True} + return None + + def move(self, dx: float, dy: float) -> dict: + if (r := self._refuse_if_stopped("move")): + return r + if max(abs(dx), abs(dy)) > MAX_STEP_UM: + return {"error": f"step over {MAX_STEP_UM} um refused"} + before = self.status()["xy_um"] + rc = self.m.StgMoveXY(float(dx), float(dy), MOVE_RELATIVE) + return {"rc": rc, "before_um": before, "after_um": self.status()["xy_um"]} + + def objective(self, name: str) -> dict: + if (r := self._refuse_if_stopped("objective change")): + return r + mags = {float(m) for m in _MAG.findall(name)} + if len(mags) != 1 or mags.pop() not in ALLOWED_MAGNIFICATIONS: + return {"error": f"only {ALLOWED_MAGNIFICATIONS}x objectives are allowed, got {name!r}"} + before = self.status() + rc = self.m.ChangeObjective(name) + after = self.status() + return {"rc": rc, "before": before["current_objective"], "after": after["current_objective"], + "nosepiece_before": before["nosepiece_position"], + "nosepiece_after": after["nosepiece_position"]} + + def capture(self, path: str) -> dict: + if (r := self._refuse_if_stopped("capture")): + return r + if not path.lower().endswith(".nd2"): + return {"error": "path must end in .nd2"} + if os.path.exists(path): + return {"error": f"{path} already exists - refusing to overwrite"} + rc_cap = self.m.Capture() + rc_save = self.m.ImageSaveAs(path, ND2_ALL_LAYERS, 0) + rc_close = self.m.CloseCurrentDocument(QUERYSAVE_NO) + return {"rc_capture": rc_cap, "rc_save": rc_save, "rc_close": rc_close, + "path": path, "saved": os.path.exists(path)} + + def nd_run(self, experiment: str) -> dict: + if (r := self._refuse_if_stopped("ND run")): + return r + if self.m.ND_IsInExperimentCapture(): + return {"error": "an ND experiment is already running"} + rc_load = self.m.ND_LoadExperiment(experiment) + if rc_load != 1: + return {"error": f"ND_LoadExperiment({experiment!r}) returned {rc_load}", "rc_load": rc_load} + t0 = time.time() + rc_run = self.m.ND_RunExperiment(0) # 0: save to disk, do not open + return {"rc_load": rc_load, "rc_run": rc_run, + "call_returned_after_s": round(time.time() - t0, 1), + "nd_running": bool(self.m.ND_IsInExperimentCapture())} + + def nd_finish(self) -> dict: + # Always allowed, e-stop or not: it only ends a run early. + rc = self.m.ND_FinishExperiment() + return {"rc": rc, "nd_running": bool(self.m.ND_IsInExperimentCapture())} + + +# Routes: (method, path) -> function(bridge, body) -> dict +ROUTES = { + ("GET", "/status"): lambda b, body: b.status(), + ("POST", "/move"): lambda b, body: b.move(float(body.get("dx", 0)), float(body.get("dy", 0))), + ("POST", "/objective"): lambda b, body: b.objective(str(body.get("name", ""))), + ("POST", "/capture"): lambda b, body: b.capture(str(body.get("path", ""))), + ("POST", "/nd/run"): lambda b, body: b.nd_run(str(body.get("experiment", ""))), + ("POST", "/nd/finish"): lambda b, body: b.nd_finish(), +} + + +def make_handler(jobs: "queue.Queue", stop: threading.Event, timeout_s: float = 7200.0): + class Handler(http.server.BaseHTTPRequestHandler): + def _answer(self, code: int, out: dict) -> None: + body = (json.dumps(out) + "\n").encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _dispatch(self, method: str) -> None: + path = self.path.split("?")[0] + if method == "POST" and path == "/shutdown": + stop.set() + return self._answer(200, {"stopping": True}) + fn = ROUTES.get((method, path)) + if fn is None: + return self._answer(404, {"error": f"no route {method} {path}", + "routes": [f"{m} {p}" for m, p in ROUTES]}) + try: + n = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(n) or b"{}") if n else {} + except ValueError: + return self._answer(400, {"error": "body is not JSON"}) + box = queue.Queue() + jobs.put((fn, body, box)) + try: + self._answer(200, box.get(timeout=timeout_s)) + except queue.Empty: + self._answer(504, {"error": "timed out waiting for NIS"}) + + def do_GET(self): + self._dispatch("GET") + + def do_POST(self): + self._dispatch("POST") + + def log_message(self, *args): # keep the JOBS log readable + pass + + return Handler + + +def serve(bridge: Bridge, port: int = PORT, should_abort=lambda: False, + stop: threading.Event | None = None, on_ready=None) -> None: + """Serve until /shutdown, should_abort() or `stop` is set. NIS calls run + here, on the calling thread; HTTP requests are handled on others.""" + jobs = queue.Queue() + stop = stop or threading.Event() + srv = http.server.ThreadingHTTPServer(("127.0.0.1", port), make_handler(jobs, stop)) + threading.Thread(target=srv.serve_forever, daemon=True).start() + print(f"NIS bridge v{VERSION} listening on 127.0.0.1:{srv.server_address[1]}, pid {os.getpid()}") + if on_ready: + on_ready(srv.server_address[1]) + try: + while not stop.is_set() and not should_abort(): + try: + fn, body, box = jobs.get(timeout=0.2) + except queue.Empty: + continue + try: + box.put(fn(bridge, body)) + except Exception as e: + box.put({"error": f"{type(e).__name__}: {e}"}) + finally: + srv.shutdown() + srv.server_close() + print("NIS bridge stopped") + + +def run(imgs: "tuple[limjob.Image]", Job: "limjob.JobParam", macro: "limjob.MacroParam", + ctx: "limjob.RunContext"): + """JOBS entry point: serve until Abort or POST /shutdown. The hints are + the JOBS template's own, as strings so this file imports off NIS.""" + serve(Bridge(Macro()), should_abort=getattr(ctx, "shouldAbort", lambda: False)) diff --git a/acquisition/orchestration/stage_positions.py b/acquisition/orchestration/stage_positions.py index 1a9efd6..c62871f 100644 --- a/acquisition/orchestration/stage_positions.py +++ b/acquisition/orchestration/stage_positions.py @@ -7,7 +7,7 @@ # # Saved positions persist to protocols/stage_positions.json, so they can # be reused across sessions (e.g. to build up a protocol's `positions:` -# list - see protocols/example_protocol.yaml). +# list - see load_positions_from_yaml). # # Run directly for a quick sanity check (works on Mac/Linux/Windows dev # machines via MockNIS, no NIS-Elements required), from the repo root: @@ -22,14 +22,14 @@ from acquisition.paths import data_root as _data_root -# Stage travel limits (see nis_mock.py / docs/microscope-notes.md's hardware +# Stage travel limits (see nis_mock.py, from the Nikon Ti2-E hardware # spec) - imported unconditionally since nis_mock.py has no hardware # dependency of its own, so these constants are always available regardless # of whether the real `nis` module or MockNIS ends up being used below. from acquisition.backends.nis_mock import X_LIMIT_UM, Y_LIMIT_UM # ── 1. Connect to the NIS-Elements Python API, or fall back to the mock ───── -# Unlike nis_jobs_connection_test.py / run_protocol.py (which only ever run ON the +# Unlike ConfocalOrchestrator's nis_jobs_connection_test.py / run_protocol.py (which only ever run ON the # microscope PC and hard-fail without the real API), this module is meant # to be usable for offline development too, so it falls back to MockNIS # when the real `nis` module isn't available. @@ -38,7 +38,7 @@ except ImportError: from acquisition.backends.nis_mock import MockNIS nis = MockNIS() - # stderr, not stdout - this module is imported by mcp_server/server.py, + # stderr, not stdout - this module is imported (via loop_tools.py) by mcp_server/server_loop.py, # whose stdout is the MCP stdio JSON-RPC channel; anything else written # there corrupts the protocol stream. print( @@ -72,7 +72,7 @@ def to_plain_float(value) -> float: def validate_position(x: float, y: float) -> None: """Raise ValueError if (x, y) - in microns - is outside the Ti2-E's stage travel limits (X +/-57mm, Y +/-36.5mm - see nis_mock.X_LIMIT_UM / - Y_LIMIT_UM, sourced from docs/microscope-notes.md's hardware spec). + Y_LIMIT_UM, sourced from the Nikon Ti2-E hardware spec). Called before a position is saved (save_current) or moved to (go_to), so a bad reading or a hand-edited positions file can't silently send @@ -149,8 +149,8 @@ def define_position(self, label: str, x: float, y: float, z: float) -> dict: return position def load_positions_from_yaml(self, yaml_path: Path) -> dict: - """Load the `positions:` list from a protocol YAML file (see - protocols/example_protocol.yaml) and define each one by its + """Load the `positions:` list from a protocol YAML file - entries of + `label`, `x`, `y`, `z` (microns) - and define each one by its `label`, validating each against the stage's travel limits. Returns the newly-defined positions as {label: {"x", "y", "z"}}. @@ -206,8 +206,8 @@ def go_to(self, label: str) -> dict: ValueError if the saved position is outside the stage's travel limits (see validate_position) - e.g. from a hand-edited positions file. Callers driving real hardware (not MockNIS) should confirm - with the user before calling this - the same way nis_jobs_connection_test.py - and run_protocol.py confirm before any stage move. + with the user before calling this - the same way ConfocalOrchestrator's + nis_jobs_connection_test.py and run_protocol.py confirm before any stage move. XY moves first, then Z - if Z_Move then fails (e.g. nis_sdk's per-call step-size safety cap, since a saved position's Z commonly @@ -258,9 +258,5 @@ def delete(self, label: str) -> None: print(f"\nPositions saved to: {POSITIONS_FILE}") - protocol_path = Path(__file__).resolve().parent.parent.parent / "protocols" / "example_protocol.yaml" - print(f"\nLoading positions from {protocol_path}...") - print(" ", manager.load_positions_from_yaml(protocol_path)) - print("\nSummary of all saved positions:") manager.print_summary() diff --git a/analysis/physarum_short.py b/analysis/physarum_short.py new file mode 100644 index 0000000..f05f488 --- /dev/null +++ b/analysis/physarum_short.py @@ -0,0 +1,274 @@ +# physarum_short.py +# ------------------------------------------------------------ +# YouTube Short (1080x1920, 30 fps, H.264 + music) from the whole-dish +# Physarum mosaic runs: the time-lapse with a live growth curve drawn +# underneath, then a wipe from the last frame to the green new-growth map. +# +# python -m analysis.physarum_short data/physarum_whole_dish_combined_20260923 +# python -m analysis.physarum_short --preview # 6 stills, seconds +# +# INPUTS, all in (written by its combined_all_script.py): +# combined_all_script.py - run list + alignment into one pixel frame +# oat_approach_all.csv - on-agar area per round (the curve) +# growth_start_to_end_colour.jpg - new growth in green (the reveal) +# Output: /shorts/physarum_short_growth.mp4 (+ .wav). +# +# ALIGNMENT is not redone here: the part of combined_all_script.py before +# "# ---- movie setup" (ECC warps of every run into the overnight run's +# frame, agar/oat masks) is executed as-is, so the short shows exactly the +# frames the analysis measured. The growth map is that same frame resized +# under a caption header, so its bottom H*width/W rows are pixel-aligned +# with the panel and the wipe needs no registration. +# +# THE CURVE is total_mm2 smoothed over 7 rounds (35 min) so the single- +# round segmentation flicker does not read as growth. Gaps between runs +# are drawn as straight lines. +# +# THIS IS THE 23-25 SEP 2026 EXPERIMENT'S SHORT: the phase captions (by +# hour), the oat label position and the default reveal caption describe +# that dish. Another experiment needs those changed, not just new paths. +# ------------------------------------------------------------ + +from __future__ import annotations + +import argparse +import csv +import datetime +from pathlib import Path + +import cv2 +import numpy as np +from PIL import Image, ImageDraw, ImageFilter + +from analysis.make_movie import wb_gains +from analysis.make_soundtrack import render +from analysis.short_video import (FONT_B, FONT_I, FONT_R, FPS, GREY, SR, VH, VW, ShortWriter, canvas, + centred, font, write_wav) + +YELLOW = (255, 214, 64) +GREEN = (80, 230, 90) +S_HOOK, S_INTRO, S_TL, S_REVEAL, S_END = 3, 3, 22, 9, 4 +DURATION = S_HOOK + S_INTRO + S_TL + S_REVEAL + S_END +REVEAL_AT = S_HOOK + S_INTRO + S_TL +PANEL_Y = 400 +A_LO, A_HI = 110, 180 # curve y range, mm^2 + + +def phase(t_h: float) -> str: + if t_h < 9: + return "Exploring in every direction" + if t_h < 20: + return "Spreading toward the food" + if t_h < 24: + return "Pulling back & rebuilding" + return "Holding its network overnight" + + +def load_alignment(combined: Path) -> dict: + """Run the analysis script's setup (runs, warps, frame size) into a namespace.""" + script = combined / "combined_all_script.py" + ns = {"__file__": str(script), "__name__": "combined_setup"} + exec(script.read_text().split("# ---- movie setup")[0], ns) + return ns + + +def soundtrack() -> np.ndarray: + """The Physarum theme, plus a soft tremolo swell into the reveal.""" + mix = render(DURATION) + t = np.arange(len(mix)) / SR + sw = np.clip((t - (REVEAL_AT - 2.0)) / 2.0, 0, 1) * np.exp(-np.clip(t - REVEAL_AT, 0, None) / 1.5) + for f, pan in ((523.25, 0.3), (659.25, 0.7), (783.99, 0.5), (1046.5, 0.5)): + s = np.sin(2 * np.pi * f * t) * (0.5 + 0.5 * np.sin(2 * np.pi * 5.5 * t)) * sw * 0.06 + mix[:, 0] += s * (1 - pan); mix[:, 1] += s * pan + return mix * 10 ** (-1 / 20) / np.abs(mix).max() + + +def make(combined: Path, out: Path, caption: str, preview: bool) -> None: + ns = load_alignment(combined) + RUNS, REF, W, H, UM_PX, to_ref = ns["RUNS"], ns["REF"], ns["W"], ns["H"], ns["UM_PX"], ns["to_ref"] + + ref0 = cv2.imread(str(REF / "mosaic_000.png")) + gains = wb_gains(ref0) + panel_h = int(H * VW / W) + um_panel = UM_PX * W / VW + + def dish_frame(run, fname): + img = cv2.imread(str(run / fname)) + img = np.clip(img.astype(np.float32) * gains, 0, 255).astype(np.uint8) + img = cv2.resize(img, (img.shape[1] * W // ref0.shape[1], img.shape[0] * W // ref0.shape[1]), + interpolation=cv2.INTER_AREA) + img = to_ref(img, run) + return cv2.cvtColor(cv2.resize(img, (VW, panel_h), interpolation=cv2.INTER_AREA), cv2.COLOR_BGR2RGB) + + def oats_label(d): + tip = (40, PANEL_Y + int(0.50 * panel_h)) + base = (95, PANEL_Y + int(0.66 * panel_h)) + d.line([base, tip], fill=(0, 0, 0), width=10) + d.line([base, tip], fill=YELLOW, width=5) + d.polygon([tip, (tip[0] - 14, tip[1] + 30), (tip[0] + 18, tip[1] + 24)], fill=YELLOW) + d.text((12, PANEL_Y + int(0.68 * panel_h)), "OATS", font=font(FONT_B, 46), fill=YELLOW, + stroke_width=4, stroke_fill=(0, 0, 0)) + + def scale_bar(d): + px = int(round(5000 / um_panel)) + x1, y = VW - 40, PANEL_Y + panel_h - 30 + d.rectangle([x1 - px, y - 10, x1, y], fill=(255, 255, 255), outline=(0, 0, 0), width=2) + f = font(FONT_B, 30) + d.text((x1 - px / 2 - d.textlength("5 mm", font=f) / 2, y - 48), "5 mm", font=f, + fill=(255, 255, 255), stroke_width=3, stroke_fill=(0, 0, 0)) + + # ---- growth curve ---- + rows = list(csv.DictReader((combined / "oat_approach_all.csv").open())) + CT = np.array([float(r["t_h"]) for r in rows]) + CA = np.array([float(r["total_mm2"]) for r in rows]) + CA_S = np.convolve(np.pad(CA, 3, mode="edge"), np.ones(7) / 7, mode="valid") + peak = int(np.argmax(CA_S)) + x0, x1 = 120, VW - 60 + y0, y1 = PANEL_Y + panel_h + 150, VH - 110 + + def cx(t): + return x0 + (x1 - x0) * t / CT[-1] + + def cy(a): + return y1 - (y1 - y0) * (a - A_LO) / (A_HI - A_LO) + + def curve(img, t_h, show_peak): + d = ImageDraw.Draw(img) + f_ax, f_lb = font(FONT_R, 30), font(FONT_B, 38) + d.text((x0 - 60, y0 - 112), "Slime mould on the agar (mm²)", font=f_lb, fill=(235, 235, 235)) + for a in (120, 140, 160, 180): + y = cy(a) + d.line([(x0, y), (x1, y)], fill=(48, 48, 54), width=2) + d.text((x0 - 14 - d.textlength(str(a), font=f_ax), y - 20), str(a), font=f_ax, fill=GREY) + for h in (0, 10, 20, 30): + d.text((cx(h) - 12, y1 + 8), f"{h}h", font=f_ax, fill=GREY) + n = int(np.searchsorted(CT, t_h, side="right")) + if n >= 2: + pts = [(cx(t), cy(a)) for t, a in zip(CT[:n], CA_S[:n])] + glow = Image.new("RGBA", img.size, (0, 0, 0, 0)) + ImageDraw.Draw(glow).line(pts, fill=YELLOW + (150,), width=16, joint="curve") + glow = glow.filter(ImageFilter.GaussianBlur(8)) + img.paste(glow, (0, 0), glow) + d = ImageDraw.Draw(img) + d.line(pts, fill=YELLOW, width=6, joint="curve") + x, y = pts[-1] + d.ellipse([x - 12, y - 12, x + 12, y + 12], fill=(255, 255, 255), outline=YELLOW, width=4) + lab = f"{CA_S[n - 1]:.0f} mm²" + d.text((min(x + 18, x1 - d.textlength(lab, font=f_lb)), y - 58), lab, font=f_lb, + fill=(255, 255, 255), stroke_width=3, stroke_fill=(0, 0, 0)) + if show_peak and n > peak: + x, y = cx(CT[peak]), cy(CA_S[peak]) + d.line([(x, y - 16), (x, y - 50)], fill=GREEN, width=3) + t = f"peak {CA_S[peak]:.0f} mm² at {CT[peak]:.0f} h" + d.text((x - d.textlength(t, font=f_ax) / 2, y - 92), t, font=f_ax, fill=GREEN) + + # ---- rounds, in time order across runs ---- + rounds, t0 = [], None + for run in RUNS: + starts = {} + for r in csv.DictReader((run / "tiles.csv").open()): + starts.setdefault(int(r["round"]), datetime.datetime.fromisoformat(r["wall_clock"])) + for rnd in sorted(starts): + fn = f"mosaic_{rnd:03d}.png" + if (run / fn).exists(): + t0 = t0 or starts[rnd] + rounds.append((run, fn, (starts[rnd] - t0).total_seconds() / 3600)) + total_h = rounds[-1][2] + print(f"{len(rounds)} rounds, {total_h:.1f} h; area peak {CA_S[peak]:.0f} mm² at {CT[peak]:.1f} h") + + gm = Image.open(combined / "growth_start_to_end_colour.jpg").convert("RGB") + gm = gm.crop((0, gm.height - round(H * gm.width / W), gm.width, gm.height)) + growth = np.asarray(gm.resize((VW, panel_h), Image.LANCZOS)) + + wav = None + if not preview: + wav = out.with_suffix(".wav") + write_wav(soundtrack(), wav) + stills = {2 * FPS, 5 * FPS, (S_HOOK + S_INTRO + 15) * FPS, (REVEAL_AT + 1) * FPS, + (REVEAL_AT + 6) * FPS, (DURATION - 2) * FPS} + with ShortWriter(out, wav, preview, stills) as w: + # 1. hook + 2. intro over the first frame + first = dish_frame(*rounds[0][:2]) + for i in range((S_HOOK + S_INTRO) * FPS): + if not w.wants(i): + continue + c = canvas(); c.paste(Image.fromarray(first), (0, PANEL_Y)); d = ImageDraw.Draw(c) + if i < S_HOOK * FPS: + centred(d, 110, "This is alive.", font(FONT_B, 104)) + centred(d, 255, "A single cell. No brain.", font(FONT_R, 62), fill=(220, 220, 220)) + else: + centred(d, 90, "Slime mould", font(FONT_B, 92)) + centred(d, 205, "Physarum polycephalum", font(FONT_I, 60), fill=YELLOW) + centred(d, 290, f"{total_h:.0f} hours under a microscope", font(FONT_R, 54), + fill=(220, 220, 220)) + oats_label(d); scale_bar(d); curve(c, 0, False) + w.emit(c, i) + + # 3. time-lapse with the live curve + base = (S_HOOK + S_INTRO) * FPS + n_tl = S_TL * FPS + cache: dict = {} + for i in range(n_tl): + if not w.wants(base + i) and i != n_tl - 1: + continue + k = min(len(rounds) - 1, int(i * len(rounds) / n_tl)) + if k not in cache: + cache = {k: dish_frame(*rounds[k][:2])} + t_h = rounds[k][2] + c = canvas(); c.paste(Image.fromarray(cache[k]), (0, PANEL_Y)); d = ImageDraw.Draw(c) + centred(d, 110, phase(t_h), font(FONT_B, 70)) + centred(d, 215, f"hour {t_h:4.1f}", font(FONT_R, 60), fill=YELLOW) + oats_label(d); scale_bar(d); curve(c, t_h, True) + w.emit(c, base + i) + if not preview and i % 60 == 0: + print(f"timelapse {i}/{n_tl}", flush=True) + last = cache[k] + + # 4. wipe from the last frame to the new-growth map + base = REVEAL_AT * FPS + for i in range(S_REVEAL * FPS): + if not w.wants(base + i): + continue + p = min(1.0, i / (2.0 * FPS)); p = 0.5 - 0.5 * np.cos(np.pi * p) + xw = int(p * VW) + panel = last.copy(); panel[:, :xw] = growth[:, :xw] + c = canvas(); c.paste(Image.fromarray(panel), (0, PANEL_Y)); d = ImageDraw.Draw(c) + if 0 < xw < VW: + d.line([(xw, PANEL_Y), (xw, PANEL_Y + panel_h)], fill=(255, 255, 255), width=5) + centred(d, 100, "So what actually grew?", font(FONT_B, 76)) + if i > 1.5 * FPS: + centred(d, 215, "Green = brand-new slime mould", font(FONT_B, 54), fill=GREEN) + if i > 3.5 * FPS: + centred(d, 290, caption, font(FONT_R, 48), fill=(220, 220, 220)) + curve(c, total_h, True) + w.emit(c, base + i) + + # 5. end card + base = (REVEAL_AT + S_REVEAL) * FPS + for i in range(S_END * FPS): + if not w.wants(base + i): + continue + c = canvas(); c.paste(Image.fromarray(growth), (0, PANEL_Y)); d = ImageDraw.Draw(c) + centred(d, 90, f"{total_h:.0f} hours in {DURATION} seconds", font(FONT_B, 76)) + centred(d, 205, "It never stopped rebuilding itself.", font(FONT_R, 52), fill=(220, 220, 220)) + centred(d, 275, "Nikon Ti2 · 4x · 1 photo-mosaic every 5 min", font(FONT_R, 42), fill=GREY) + curve(c, total_h, True) + w.emit(c, base + i) + print("wrote", out if not preview else f"preview stills in {out.parent}") + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("combined_dir", type=Path) + ap.add_argument("--out", type=Path, help="default: /shorts/physarum_short_growth.mp4") + ap.add_argument("--caption", default="+20 mm² grew · 17 mm² pulled back", + help="reveal caption (the default is the 23-25 Sep figures)") + ap.add_argument("--preview", action="store_true", help="save a few stills instead of rendering") + a = ap.parse_args() + out = a.out or a.combined_dir / "shorts" / "physarum_short_growth.mp4" + out.parent.mkdir(parents=True, exist_ok=True) + make(a.combined_dir.resolve(), out.resolve(), a.caption, a.preview) + + +if __name__ == "__main__": + main() diff --git a/analysis/short_video.py b/analysis/short_video.py new file mode 100644 index 0000000..17e416d --- /dev/null +++ b/analysis/short_video.py @@ -0,0 +1,115 @@ +# short_video.py +# ------------------------------------------------------------ +# Shared pieces for the vertical (1080x1920) YouTube Shorts made by +# physarum_short.py and worm_short.py: canvas, fonts, centred text, and a +# writer that pipes RGB frames straight into ffmpeg (x264) with a WAV +# soundtrack muxed on as AAC in the same pass. +# +# WHY PIPE INTO FFMPEG rather than cv2.VideoWriter: cv2's Windows build has +# no reliable H.264 (see make_movie.py), and the MSMF H.264 it sometimes +# offers writes ~40 Mbit/s - the first Physarum Short was 200 MB. libx264 +# at CRF 19 looks the same at a few MB, and taking the audio in the same +# ffmpeg run avoids a second mux step. The ffmpeg used is the one +# imageio-ffmpeg bundles (in the `analysis` dependency group). +# +# PREVIEW: with preview=True the writer saves the frames whose index is in +# `stills` as PNGs instead of encoding anything, so a layout can be checked +# in seconds before a render that may take most of an hour. +# ------------------------------------------------------------ + +from __future__ import annotations + +import subprocess +import wave +from pathlib import Path + +import numpy as np +from PIL import Image, ImageDraw, ImageFont + +VW, VH, FPS = 1080, 1920, 30 +SR = 44100 +BG = (12, 12, 14) +GREY = (170, 170, 170) + +# Segoe UI ships with Windows; elsewhere fall back to PIL's scalable default. +FONT_B = "C:/Windows/Fonts/segoeuib.ttf" +FONT_R = "C:/Windows/Fonts/segoeui.ttf" +FONT_I = "C:/Windows/Fonts/segoeuii.ttf" +_fonts: dict = {} + + +def font(path: str, size: int) -> ImageFont.FreeTypeFont: + key = (path, size) + if key not in _fonts: + try: + _fonts[key] = ImageFont.truetype(path, size) + except OSError: + _fonts[key] = ImageFont.load_default(size) + return _fonts[key] + + +def canvas(bg=BG) -> Image.Image: + return Image.new("RGB", (VW, VH), bg) + + +def centred(draw: ImageDraw.ImageDraw, y: float, text: str, f, fill=(255, 255, 255)) -> None: + draw.text(((VW - draw.textlength(text, font=f)) / 2, y), text, font=f, fill=fill) + + +def write_wav(mix: np.ndarray, path: Path) -> None: + """Stereo float mix in [-1, 1] -> 16-bit WAV.""" + with wave.open(str(path), "wb") as w: + w.setnchannels(2); w.setsampwidth(2); w.setframerate(SR) + w.writeframes((np.clip(mix, -1, 1) * 32767).astype(np.int16).tobytes()) + + +class ShortWriter: + """Frames in, finished MP4 (H.264 + AAC) out. + + with ShortWriter(out, wav) as w: + w.emit(img) # PIL RGB image, VW x VH + """ + + def __init__(self, out: Path, wav: Path | None, preview: bool = False, + stills: set[int] = frozenset(), still_dir: Path | None = None): + self.out, self.preview, self.stills = Path(out), preview, set(stills) + self.still_dir = Path(still_dir or self.out.parent) + self.index = 0 + self._ff = None + if preview: + return + import imageio_ffmpeg + cmd = [imageio_ffmpeg.get_ffmpeg_exe(), "-y", "-loglevel", "error", + "-f", "rawvideo", "-pix_fmt", "rgb24", "-s", f"{VW}x{VH}", "-r", str(FPS), "-i", "-"] + if wav: + cmd += ["-i", str(wav), "-map", "0:v", "-map", "1:a", "-c:a", "aac", "-b:a", "192k", "-shortest"] + cmd += ["-c:v", "libx264", "-preset", "slow", "-crf", "19", "-pix_fmt", "yuv420p", + "-movflags", "+faststart", str(self.out)] + self._ff = subprocess.Popen(cmd, stdin=subprocess.PIPE) + + def wants(self, index: int) -> bool: + """False for frames a preview would skip, so callers can avoid drawing them.""" + return not self.preview or index in self.stills + + def emit(self, img: Image.Image, index: int | None = None) -> None: + if index is not None: + self.index = index + if self.preview: + if self.index in self.stills: + img.save(self.still_dir / f"_preview_{self.index:04d}.png") + else: + self._ff.stdin.write(np.asarray(img.convert("RGB")).tobytes()) + self.index += 1 + + def close(self) -> None: + if self._ff: + self._ff.stdin.close() + if self._ff.wait() != 0: + raise RuntimeError(f"ffmpeg failed writing {self.out}") + self._ff = None + + def __enter__(self): + return self + + def __exit__(self, *exc): + self.close() diff --git a/analysis/worm_short.py b/analysis/worm_short.py new file mode 100644 index 0000000..fca0587 --- /dev/null +++ b/analysis/worm_short.py @@ -0,0 +1,424 @@ +# worm_short.py +# ------------------------------------------------------------ +# YouTube Short (1080x1920, 30 fps, H.264 + original music) from an AML18 +# C. elegans time-lapse that aml18_survey has already been run on. +# +# python -m analysis.worm_short D:\...\CelegansAML-18\Timelapse_1h.nd2 +# python -m analysis.worm_short file.nd2 --stars 2,43,5 # pick the worms +# python -m analysis.worm_short file.nd2 --preview # 7 stills, seconds +# +# Reads the .nd2 (TD + neuron channel) and _survey/worms.csv; writes +# /shorts/_short.mp4 (+ .wav). +# +# SEQUENCE: "lights off" (neurons only, on black) -> brightfield fades in +# -> the whole run in 17 s with trails for up to three "star" worms and a +# live distance-crawled chart -> the speed comparison -> every track at +# once -> end card. Music is synthesised here (no samples), with a soft +# tick for every frame of the time-lapse. +# +# PICTURE = TD brightfield screened with the neuron channel as a pink glow. +# Default RFP, because on the AX it is read out with TD and lines up with +# the worms; a sequentially scanned GFP lands offset (see aml18_survey.py). +# +# TRACKS are re-linked from worms.csv with aml18_track.link (default 0.5 mm +# max step, stricter than aml18_track's 1 mm, so crowded L1s are not chained +# into long false tracks). Each track's position then gets a 3-frame median, +# which removes one-frame centroid jumps (a curled or touching worm +# segmented oddly) before distances are summed. +# +# STARS: by default the three adult tracks followed for the most frames, +# named by speed - fastest "Explorer", slowest "Homebody", middle +# "Wanderer". --stars picks the track ids instead (same naming rule). +# +# THE SPEED CLAIM. Positions are sampled once per frame (1 min here), so +# "distance crawled" undercounts the real path and the Explorer/Homebody +# ratio depends on the clean-up: 6.5x from raw steps, 13.9x after the +# median, for Timelapse_1h. The Short states the smaller one as "at least +# Nx". The chart compares worms fairly; it is not an absolute speed. +# ------------------------------------------------------------ + +from __future__ import annotations + +import argparse +import csv +from pathlib import Path + +import cv2 +import numpy as np +from PIL import Image, ImageDraw, ImageFilter + +from analysis.aml18_track import link, summarise +from analysis.short_video import (FONT_B, FONT_I, FONT_R, FPS, GREY, SR, VH, VW, ShortWriter, canvas, + centred, font, write_wav) + +BG = (10, 10, 14) +PINK = (255, 90, 210) +GLOW_RGB = np.array([1.0, 0.25, 0.85]) +ROLES = {"Explorer": (255, 205, 60), "Wanderer": (90, 220, 255), "Homebody": (120, 255, 140)} +PANEL_Y = 380 +P = VW # square panel + +S_DARK, S_FADE, S_TL, S_HOLD, S_MAP, S_END = 3.5, 3.0, 17.0, 3.5, 5.0, 3.5 +T_FADE = S_DARK +T_TL = T_FADE + S_FADE +T_HOLD = T_TL + S_TL +T_MAP = T_HOLD + S_HOLD +T_END = T_MAP + S_MAP +DURATION = T_END + S_END + + +# ---- picture ------------------------------------------------------------------- +def load(nd2_path: Path, channel: str): + import nd2 + with nd2.ND2File(nd2_path) as f: + um_px = float(f.voxel_size().x) + names = [c.channel.name for c in f.metadata.channels] + stack = f.asarray() # T, C, Y, X + t_s = np.array([e["Time [s]"] for e in f.events()]) + return stack[:, names.index("TD")], stack[:, names.index(channel)], um_px, t_s + + +def make_composites(td, neuron): + def glow(t): + r = np.clip((neuron[t].astype(np.float32) - 6) / 250, 0, 1) ** 0.6 + return np.clip(cv2.GaussianBlur(r, (0, 0), 1.2) + 0.6 * cv2.GaussianBlur(r, (0, 0), 6), 0, 1) + + def bright(t): + x = td[t].astype(np.float32) + lo, hi = np.percentile(x, [0.5, 99.8]) + g = np.clip((x - lo) / (hi - lo), 0, 1) ** 1.3 * 0.85 + return np.repeat(g[..., None], 3, -1) + + def comp(t, light=1.0): + rgb = 1 - (1 - bright(t) * light) * (1 - np.clip(glow(t)[..., None] * GLOW_RGB * (1.6 - 0.6 * light), 0, 1)) + return cv2.resize((np.clip(rgb, 0, 1) * 255).astype(np.uint8), (P, P), interpolation=cv2.INTER_AREA) + + return [comp(t) for t in range(len(td))], comp(0, light=0.0) + + +# ---- tracks ---------------------------------------------------------------------- +def build_tracks(worms_csv: Path, um_px: float, k: float, dt_min: float, max_step_mm: float): + """{track: [(minute, x, y, step_mm)]} smoothed, plus raw mm/min per track and the summary.""" + rows = link(list(csv.DictReader(worms_csv.open(encoding="utf-8"))), um_px, max_step_mm) + summary = {s["track"]: s for s in summarise(rows)} + raw: dict = {} + for r in rows: + raw.setdefault(r["track"], []).append((int(r["t_index"]) * dt_min, float(r["x_px"]) * k, + float(r["y_px"]) * k, + float(r["step_mm"]) if r["step_mm"] != "" else 0.0)) + tracks, raw_rate = {}, {} + for tid, v in raw.items(): + v.sort() + if len(v) > 1: + raw_rate[tid] = sum(p[3] for p in v) / (v[-1][0] - v[0][0]) + xs = np.array([p[1] for p in v]); ys = np.array([p[2] for p in v]) + if len(v) >= 3: + xs = np.array([np.median(xs[max(0, i - 1):i + 2]) for i in range(len(v))]) + ys = np.array([np.median(ys[max(0, i - 1):i + 2]) for i in range(len(v))]) + st = np.r_[0, np.hypot(np.diff(xs), np.diff(ys)) / k * um_px / 1000] + tracks[tid] = [(p[0], x, y, s) for p, x, y, s in zip(v, xs, ys, st)] + return tracks, raw_rate, summary + + +def pick_stars(tracks, summary, ids): + if not ids: + adults = [t for t, s in summary.items() if s["stage"] == "adult" and s["frames"] >= 10] + ids = sorted(adults, key=lambda t: -summary[t]["frames"])[:3] + if len(ids) < 2: + raise SystemExit("need at least two tracks to compare (pass --stars)") + rate = {t: sum(p[3] for p in tracks[t]) / (tracks[t][-1][0] - tracks[t][0][0]) for t in ids} + order = sorted(ids, key=lambda t: -rate[t]) + names = ["Explorer", "Wanderer", "Homebody"] if len(order) == 3 else ["Explorer", "Homebody"] + return {t: (n, ROLES[n]) for t, n in zip(order, names)}, rate + + +def pos_at(tr, tm): + ts = [p[0] for p in tr] + if tm < ts[0] or tm > ts[-1]: + return None + i = int(np.searchsorted(ts, tm, side="right")) - 1 + if i >= len(tr) - 1: + return tr[-1][1], tr[-1][2] + a, b = tr[i], tr[i + 1] + u = (tm - a[0]) / (b[0] - a[0]) + return a[1] + u * (b[1] - a[1]), a[2] + u * (b[2] - a[2]) + + +# ---- soundtrack -------------------------------------------------------------------- +def soundtrack(n_ticks: int) -> np.ndarray: + n = int(SR * DURATION); t = np.arange(n) / SR + rng = np.random.default_rng(3) + mix = np.zeros((n, 2)) + + def note(m): + return 440.0 * 2 ** ((m - 69) / 12) + + def add(sig, at, pan=0.5): + i0 = int(at * SR) + if i0 >= n: + return + L = min(len(sig), n - i0) + mix[i0:i0 + L, 0] += sig[:L] * (1 - pan) + mix[i0:i0 + L, 1] += sig[:L] * pan + + def marimba(m, lvl, decay=0.45): + tt = np.arange(int(SR * decay * 5)) / SR + f = note(m) + s = np.sin(2 * np.pi * f * tt) + 0.25 * np.sin(2 * np.pi * 4 * f * tt) * np.exp(-tt / 0.05) + return s * np.exp(-tt / decay) * np.clip(tt / 0.003, 0, 1) * lvl + + def pad(ms, a, b, lvl): + e = np.clip(np.minimum((t - a) / 1.5, (b - t) / 1.5), 0, 1) + e = 0.5 - 0.5 * np.cos(np.pi * e) + for m in ms: + for det, pan in ((-0.1, 0.25), (0.1, 0.75)): + s = np.sin(2 * np.pi * note(m) * 2 ** (det / 12) * t + rng.uniform(0, 6.3)) * e * lvl / len(ms) + mix[:, 0] += s * (1 - pan); mix[:, 1] += s * pan + + # dark opening: low drone + glassy shimmer, then the "lights on" swell + pad([38, 45, 50], 0, T_TL + 1, 0.45) + sh = np.clip(t / 2.5, 0, 1) * np.clip((T_TL - t) / 1.5, 0, 1) + for f_, pan in ((1174.7, 0.2), (1396.9, 0.8), (1760.0, 0.5)): + s = np.sin(2 * np.pi * f_ * t) * (0.5 + 0.5 * np.sin(2 * np.pi * 0.7 * t + pan * 4)) * sh * 0.025 + mix[:, 0] += s * (1 - pan); mix[:, 1] += s * pan + pad([50, 57, 62, 66], T_FADE, T_TL + 0.8, 0.5) + # time-lapse: D-major pentatonic marimba at 112 bpm over a I-vi-IV-V loop + beat = 60 / 112 + melody = [74, 76, 78, 81, 78, 76, 74, 71, 74, 78, 81, 83, 81, 78, 76, 74] + chords = [[50, 57, 62, 66], [47, 54, 59, 62], [43, 50, 55, 59], [45, 52, 57, 61]] + t_, i = T_TL, 0 + while t_ < T_MAP: + bar = int((t_ - T_TL) / (4 * beat)) % 4 + if i % 2 == 0 or rng.random() < 0.6: + add(marimba(melody[i % 16] - (12 if i % 8 == 7 else 0), 0.16), t_, pan=0.3 + 0.4 * (i % 2)) + if i % 4 == 0: + add(marimba(chords[bar][0] - 12, 0.30, decay=0.9), t_, pan=0.5) + if i % 8 == 0: + pad(chords[bar][1:], t_, t_ + 8 * beat * 0.5 + 0.3, 0.30) + t_ += beat / 2; i += 1 + for k in range(n_ticks): # a soft tick per time-lapse frame + tt = np.arange(int(SR * 0.04)) / SR + add(rng.standard_normal(len(tt)) * np.exp(-tt / 0.006) * 0.05, T_TL + S_TL * k / (n_ticks - 1)) + # every-path map + end: rising arpeggio into a warm resolve + for k, m in enumerate([62, 66, 69, 74, 78, 81, 86]): + add(marimba(m, 0.13, decay=0.8), T_MAP + 0.25 * k, pan=0.2 + 0.1 * k) + pad([50, 57, 62, 66, 69, 74], T_MAP + 1.5, DURATION + 2, 0.75) + add(marimba(86, 0.12, decay=2.0), T_END + 0.2, pan=0.6) + # reverb (FFT convolution with decaying noise), fades, -1 dBFS + ir_len = int(SR * 1.6) + ir = rng.standard_normal((ir_len, 2)) * np.exp(-np.arange(ir_len) / (SR * 0.4))[:, None] + ir[0] = 0 + nfft = 1 << int(np.ceil(np.log2(n + ir_len))) + wet = np.stack([np.fft.irfft(np.fft.rfft(mix[:, c], nfft) * np.fft.rfft(ir[:, c], nfft), nfft)[:n] + for c in range(2)], axis=1) + wet *= np.abs(mix).max() / (np.abs(wet).max() + 1e-9) + out = 0.75 * mix + 0.35 * wet + out *= (np.clip(t / 0.2, 0, 1) * np.clip((DURATION - t) / 2.0, 0, 1))[:, None] + return out * 10 ** (-1 / 20) / np.abs(out).max() + + +# ---- the Short ----------------------------------------------------------------------- +def make(nd2_path: Path, survey: Path, out: Path, channel: str, max_step_mm: float, + star_ids: list[int], preview: bool) -> None: + td, neuron, um_px, t_s = load(nd2_path, channel) + nt, k = len(td), P / td.shape[-1] + dt_min = (t_s[-1] - t_s[0]) / (nt - 1) / 60 + run_min = dt_min * (nt - 1) + run_txt = "1 hour" if round(run_min) == 60 else f"{run_min:.0f} min" + every_txt = "1 photo every minute" if round(dt_min, 1) == 1 else f"1 photo every {dt_min:.1f} min" + print(f"{nt} frames, {dt_min:.2f} min apart, {um_px:.2f} um/px; compositing", flush=True) + frames, dark0 = make_composites(td, neuron) + + tracks, raw_rate, summary = build_tracks(survey / "worms.csv", um_px, k, dt_min, max_step_mm) + stars, rate = pick_stars(tracks, summary, star_ids) + by_name = {n: t for t, (n, _) in stars.items()} + fast, slow = by_name["Explorer"], by_name["Homebody"] + faster = min(rate[fast] / rate[slow], raw_rate[fast] / raw_rate[slow]) + for t, (n, _) in stars.items(): + print(f"{n}: track {t}, {tracks[t][0][0]:.0f}-{tracks[t][-1][0]:.0f} min, " + f"{rate[t]:.3f} mm/min (raw {raw_rate[t]:.3f})") + print(f"Explorer/Homebody: {raw_rate[fast] / raw_rate[slow]:.1f}x raw, " + f"{rate[fast] / rate[slow]:.1f}x filtered -> 'at least {int(faster)}x'") + others = [v for t, v in tracks.items() if t not in stars and len(v) >= 5] + crawl = {t: (np.array([p[0] for p in tracks[t]]), np.cumsum([p[3] for p in tracks[t]])) for t in stars} + y_max = max(1.0, np.ceil(max(c[-1] for _, c in crawl.values()) * 2) / 2) + + def trails(img, tm, labels=True, show_others=False): + lay = Image.new("RGBA", img.size, (0, 0, 0, 0)) + d = ImageDraw.Draw(lay) + if show_others: + for tr in others: + pts = [(x, y) for m, x, y, _ in tr if m <= tm] + if len(pts) > 1: + d.line(pts, fill=(255, 255, 255, 110), width=3, joint="curve") + for tid, (_, col) in stars.items(): + pts = [(x, y) for m, x, y, _ in tracks[tid] if m <= tm] + cur = pos_at(tracks[tid], tm) + if cur is not None: + pts.append(cur) + if len(pts) > 1: + d.line(pts, fill=col + (255,), width=7, joint="curve") + blur = lay.filter(ImageFilter.GaussianBlur(6)) + img.paste(blur, (0, 0), blur) + img.paste(lay, (0, 0), lay) + d = ImageDraw.Draw(img) + for tid, (name, col) in stars.items(): + tr = tracks[tid] + cur = pos_at(tr, tm) + if cur is None: + if tm > tr[-1][0]: # track ended: hollow marker + x, y = tr[-1][1], tr[-1][2] + d.ellipse([x - 10, y - 10, x + 10, y + 10], outline=col, width=4) + continue + x, y = cur + rad = 13 + 3 * np.sin(tm * 2.0) + d.ellipse([x - rad, y - rad, x + rad, y + rad], fill=col, outline=(0, 0, 0), width=3) + if labels: + f = font(FONT_B, 36) + w = d.textlength(name, font=f) + lx = min(max(x - w / 2, 10), P - w - 10) + ly = y - 72 if y > 90 else y + 26 + d.rounded_rectangle([lx - 12, ly - 4, lx + w + 12, ly + 46], radius=14, fill=(0, 0, 0)) + d.text((lx, ly), name, font=f, fill=col) + + def scale_bar(d): + px = 500 / um_px * k + x1, y = P - 36, PANEL_Y + P - 34 + d.rectangle([x1 - px, y - 9, x1, y], fill=(255, 255, 255), outline=(0, 0, 0), width=2) + f = font(FONT_B, 30) + d.text((x1 - px / 2 - d.textlength("0.5 mm", font=f) / 2, y - 46), "0.5 mm", font=f, + fill=(255, 255, 255), stroke_width=3, stroke_fill=(0, 0, 0)) + + cx0, cx1 = 120, VW - 70 + cy0, cy1 = PANEL_Y + P + 175, VH - 100 + + def chart(img, tm): + d = ImageDraw.Draw(img) + f_ax, f_lb = font(FONT_R, 30), font(FONT_B, 38) + d.text((cx0 - 60, cy0 - 120), "Distance crawled (mm)", font=f_lb, fill=(235, 235, 235)) + for v in np.arange(0, y_max + 0.01, 0.5): + y = cy1 - (cy1 - cy0) * v / y_max + d.line([(cx0, y), (cx1, y)], fill=(46, 46, 54), width=2) + t = f"{v:g}" + d.text((cx0 - 18 - d.textlength(t, font=f_ax), y - 20), t, font=f_ax, fill=GREY) + for m in np.linspace(0, run_min, 5): + x = cx0 + (cx1 - cx0) * m / run_min + t = f"{m:.0f} min" + d.text((x - d.textlength(t, font=f_ax) / 2, cy1 + 8), t, font=f_ax, fill=GREY) + for tid, (_, col) in stars.items(): + ms, cs = crawl[tid] + keep = ms <= tm + if not keep.any(): + continue + pts = [(cx0 + (cx1 - cx0) * m / run_min, cy1 - (cy1 - cy0) * c / y_max) + for m, c in zip(ms[keep], cs[keep])] + if len(pts) > 1: + d.line(pts, fill=col, width=6, joint="curve") + x, y = pts[-1] + d.ellipse([x - 9, y - 9, x + 9, y + 9], fill=col) + d.text((x + 14, y - 22), f"{cs[keep][-1]:.1f}", font=font(FONT_B, 32), fill=col, + stroke_width=3, stroke_fill=BG) + + def panel_at(tm): + u = tm / dt_min + i = min(int(u), nt - 1); j = min(i + 1, nt - 1); u -= i + return frames[i] if u <= 0 or i == j else cv2.addWeighted(frames[i], 1 - u, frames[j], u, 0) + + def with_panel(arr): + c = canvas(BG) + c.paste(arr if isinstance(arr, Image.Image) else Image.fromarray(arr), (0, PANEL_Y)) + return c, ImageDraw.Draw(c) + + def span(a, b): + return range(int(round(a * FPS)), int(round(b * FPS))) + + wav = None + if not preview: + wav = out.with_suffix(".wav") + write_wav(soundtrack(nt), wav) + stills = {int(s * FPS) for s in (1.5, T_FADE + 2.0, T_TL + 6, T_TL + 14, T_HOLD + 2, T_MAP + 3, T_END + 2)} + dim = (frames[0].astype(np.float32) * 0.45).astype(np.uint8) + with ShortWriter(out, wav, preview, stills) as w: + for i in span(0, T_FADE): # 1. lights off + if not w.wants(i): + continue + c, d = with_panel((dark0 * min(1, i / FPS)).astype(np.uint8)) + centred(d, 95, "Lights off.", font(FONT_B, 100)) + if i / FPS > 1.2: + centred(d, 235, "Every glowing dot is a neuron.", font(FONT_R, 58), fill=PINK) + w.emit(c, i) + for i in span(T_FADE, T_TL): # 2. lights on + if not w.wants(i): + continue + u = np.clip((i / FPS - T_FADE) / 1.5, 0, 1); u = 0.5 - 0.5 * np.cos(np.pi * u) + c, d = with_panel(cv2.addWeighted(dark0, 1 - u, frames[0], u, 0)) + centred(d, 80, "C. elegans", font(FONT_I, 100)) + centred(d, 215, "1 mm worms with 302 neurons —", font(FONT_R, 50), fill=(225, 225, 225)) + centred(d, 280, "engineered so each one glows", font(FONT_R, 50), fill=PINK) + scale_bar(d); chart(c, -1) + w.emit(c, i) + for i in span(T_TL, T_HOLD): # 3. the run + if not w.wants(i): + continue + tm = min(run_min, (i / FPS - T_TL) / S_TL * run_min) + pan = Image.fromarray(panel_at(tm)); trails(pan, tm) + c, d = with_panel(pan) + centred(d, 80, f"Same worms, {run_txt}", font(FONT_B, 84)) + centred(d, 205, f"minute {tm:4.1f}", font(FONT_R, 60), fill=ROLES["Explorer"]) + centred(d, 290, every_txt, font(FONT_R, 40), fill=GREY) + scale_bar(d); chart(c, tm) + w.emit(c, i) + if not preview and i % 60 == 0: + print("timelapse", i, flush=True) + for i in span(T_HOLD, T_MAP): # 4. the comparison + if not w.wants(i): + continue + pan = Image.fromarray(frames[-1]); trails(pan, run_min) + c, d = with_panel(pan) + centred(d, 70, "The Explorer moved", font(FONT_B, 76), fill=ROLES["Explorer"]) + centred(d, 180, f"at least {int(faster)}× faster than the Homebody", font(FONT_B, 50)) + if i / FPS - T_HOLD > 1.2: + centred(d, 285, "Same species. Same plate. Different personalities?", font(FONT_I, 40), + fill=GREY) + scale_bar(d); chart(c, run_min) + w.emit(c, i) + for i in span(T_MAP, DURATION): # 5. every path + 6. end card + if not w.wants(i): + continue + u = np.clip((i / FPS - T_MAP) / 3.0, 0, 1) + pan = Image.fromarray(dim); trails(pan, run_min * u, labels=False, show_others=True) + c, d = with_panel(pan) + if i / FPS < T_END: + centred(d, 80, f"Every path in {run_txt}", font(FONT_B, 84)) + centred(d, 205, f"{len(others) + len(stars)} worms followed for 5+ frames", font(FONT_R, 50), + fill=(225, 225, 225)) + else: + centred(d, 80, f"{run_txt} in {S_TL:.0f} seconds", font(FONT_B, 84)) + centred(d, 205, "Nikon AX confocal · brightfield + RFP neurons", font(FONT_R, 44), + fill=(225, 225, 225)) + centred(d, 275, f"AML18 C. elegans · {every_txt.replace('1 photo', '1 frame')}", + font(FONT_R, 44), fill=GREY) + scale_bar(d); chart(c, run_min) + w.emit(c, i) + print("wrote", out if not preview else f"preview stills in {out.parent}") + + +def main() -> None: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("nd2", type=Path) + ap.add_argument("--survey", type=Path, help="survey folder (default _survey)") + ap.add_argument("--out", type=Path, help="default: /shorts/_short.mp4") + ap.add_argument("--channel", default="RFP", help="neuron channel for the glow") + ap.add_argument("--max-step-mm", type=float, default=0.5) + ap.add_argument("--stars", help="comma-separated track ids to feature (default: auto, see header)") + ap.add_argument("--preview", action="store_true", help="save a few stills instead of rendering") + a = ap.parse_args() + survey = a.survey or a.nd2.with_name(a.nd2.stem + "_survey") + out = a.out or a.nd2.parent / "shorts" / f"{a.nd2.stem}_short.mp4" + out.parent.mkdir(parents=True, exist_ok=True) + stars = [int(s) for s in a.stars.split(",")] if a.stars else [] + make(a.nd2, survey, out.resolve(), a.channel, a.max_step_mm, stars, a.preview) + + +if __name__ == "__main__": + main() diff --git a/docs/nis-bridge.md b/docs/nis-bridge.md new file mode 100644 index 0000000..5d8cbfc --- /dev/null +++ b/docs/nis-bridge.md @@ -0,0 +1,98 @@ +# NIS bridge: driving the microscope through NIS-Elements + +The rest of this repo reaches the hardware directly: the Ti2 stand through +its ActiveX SDK and the Baumer camera through GenICam. Some things only +NIS-Elements can do, above all the AX confocal lasers and the ND +experiments set up in its ND Acquisition window. The NIS bridge makes +those available to our code and to an MCP client. + +**Status: built and tested off the microscope, not yet run inside NIS.** +The test plan below is what still has to pass on the scope. Driving +capture through NIS reverses an earlier team decision (see the header of +`mcp_server/loop_tools.py`), so it also needs sign-off before it is used +for real work. + +## How it fits together + +``` +MCP client (Claude) + │ stdio + ▼ +mcp_server/server_nis.py confirm gate, capture-folder rule, estop tool + │ +acquisition/backends/nis_bridge.py e-stop checked before sending + │ HTTP, 127.0.0.1:8766 + ▼ +acquisition/nis_bridge/bridge_job.py JOBS Python task, runs inside nis_ar.exe + │ ctypes + ▼ +g5_regprocs.dll (NIS macro functions: StgMoveXY, ChangeObjective, Capture, ND_RunExperiment, ...) +``` + +The MCP server stays in this repo's environment. Only `bridge_job.py` +runs inside NIS, and it needs nothing beyond the Python standard library, +so nothing has to be installed into NIS's own Python. If our code fails, +NIS keeps running. + +`server_loop.py` and its tools are unchanged. The NIS tools are a +separate server, `confocal-mcp-nis`. + +## What was already verified (2026-09-29) + +- A JOBS Python task runs inside `nis_ar.exe` itself and can bind a port + that another process on the PC reaches (`nis_port_probe.py`). +- `g5_regprocs.dll` exports the macro functions by name. NIS's own + `limpy.macro` calls `WaitText` this way. Signatures come from the macro + reference in `C:\Program Files\NIS-Elements\Docs\nis\eng_ar`. + +## Setting up the bridge in NIS + +1. **Start conditions.** The Ti2 controller is on, NIS-Elements was started + after it, and the e-stop is clear (`python -m acquisition.estop status`). +2. In NIS, open **JOBS** and create a job with one **Python** task. +3. Replace the task's code with the whole of + `acquisition/nis_bridge/bridge_job.py`, then save the job, for example + as `NIS bridge`. +4. Run the job. Its log shows + `NIS bridge v1 listening on 127.0.0.1:8766, pid `. +5. The bridge runs until you **Abort** the job, or until + `python -m acquisition.backends.nis_bridge shutdown`. + +## Test plan for the microscope + +Run each step from a terminal in the repo (`.venv\Scripts\python -m +acquisition.backends.nis_bridge ...`). Write down the output. Stop at the +first surprise. Press the STOP panel if anything moves unexpectedly. + +| # | Step | Command | Pass if | +|---|------|---------|---------| +| 0 | NIS usable while the job runs | click around NIS for 30 s | NIS does not freeze. **If it does, stop here**: JOBS runs Python on the main thread, and the bridge needs a different host. | +| 1 | Read-only status | `status` | `xy_um` and `z_um` match the Ti2 Pad (Z ≈ 4600, not 1.0). `nosepiece_objectives` lists the turret. **Note whether position 1 is the 4×**, i.e. whether the index is 0- or 1-based. | +| 2 | Small move | `move 100 0`, then `move -100 0` | `after_um` changes by +100 then −100 in X, and the Ti2 Pad shows the same. The sign of X on the image is recorded. | +| 3 | Move cap | `move 1500 0` | Refused, and nothing moves. | +| 4 | E-stop | `python -m acquisition.estop engage`, then `move 10 0`, then release | Refused by the client. Then send the same move with `curl -X POST -d "{\"dx\":10}" http://127.0.0.1:8766/move` and check the bridge refuses it too. | +| 5 | Objective | `objective ""`, then back to the 4× | **Does the nosepiece physically turn?** `nosepiece_before/after` go 1 → 2 → 1, and NIS's objective display follows. If only the calibration changes, the bridge needs a nosepiece call instead. | +| 6 | Capture | `capture D:\QurratulAin_ConfocalOrchestratorProject\CelegansAML-18\bridge_check_01.nd2` | The file exists and opens with GFP/RFP/TD (check with `python -m analysis.aml18_survey `). No stray image window is left open in NIS. | +| 7 | Saved ND experiment | In ND Acquisition, set a **2-loop, 1-minute** time-lapse with Save to File and **Save** it as `bridge test`. Then `nd-run "bridge test"`. | The run starts and the file appears. **Record `call_returned_after_s`**: ~0 means the call returns at once, ~60 means it blocks until the run ends. | +| 8 | Status during a run | during step 7, from a second terminal: `status` | Answers with `nd_running: true`, or hangs until the run ends (the blocking case from step 7). | +| 9 | Finish early | run `bridge test` with more loops, then `nd-finish` | The run ends after the current loop, and the file keeps the frames so far. | +| 10 | Shutdown | `shutdown` | The job ends by itself, and `status` then reports the bridge unreachable. | +| 11 | MCP | Add `confocal-mcp-nis` to the MCP client config and ask for `nis_status` | Same answer as step 1. Motion tools refuse without `confirm=True`. | + +If steps 7–8 show that `ND_RunExperiment` blocks, the bridge can still run +experiments, but nothing else can be asked while one is in progress. The +fix is to start runs in a way that returns at once. Record the result +before changing anything. + +## Safety rules (enforced in code) + +- Motion and acquisition answer **POST** only, so a browser cannot trigger them. +- The **e-stop** is checked in the client and again in the bridge. Ending an + ND run is always allowed. +- XY moves are **relative, at most 1000 µm per axis** per call. **Z is never moved.** +- Objective changes are limited to **4× and 10×**, the long working distances. +- MCP tools that move or acquire need **`confirm=True`**. +- `nis_capture` saves only into the capture folder + (`CONFOCAL_NIS_CAPTURE_DIR`, else `data/nis_captures`), by plain file name, + and never overwrites. +- The bridge listens on **127.0.0.1** only. diff --git a/mcp_server/nis_tools.py b/mcp_server/nis_tools.py new file mode 100644 index 0000000..2e3d35c --- /dev/null +++ b/mcp_server/nis_tools.py @@ -0,0 +1,109 @@ +# nis_tools.py +# ------------------------------------------------------------ +# MCP tools that drive the microscope THROUGH NIS-Elements, via the NIS +# bridge (acquisition/nis_bridge/bridge_job.py, running as a JOBS task; +# client in acquisition/backends/nis_bridge.py). Served by their own +# server, mcp_server/server_nis.py - kept apart from loop_tools.py, which +# by team direction has exactly its agreed tools and no NIS-Elements. +# +# WHAT THIS ADDS over loop_tools: things only NIS can do - the AX laser +# acquisition set up in NIS (capture, saved ND experiments) and NIS's own +# objective change. +# +# SAFETY, same shape as loop_tools: +# * every tool that moves or acquires requires confirm=True +# (PermissionError otherwise), exactly like backend="sdk" there +# * the e-stop is checked in the client before a request is sent, and +# again inside NIS by the bridge +# * XY moves are relative and capped at 1 mm; there is no Z tool +# * objective changes are limited to 4x/10x by the bridge +# * captures are written only under the data folder, by file name - a +# chat prompt cannot choose an arbitrary path +# * nis_finish_experiment and estop never need confirm: they only stop +# ------------------------------------------------------------ + +from __future__ import annotations + +import os +import re +from pathlib import Path + +from acquisition.backends.nis_bridge import NISBridge +from acquisition.paths import data_root + + +def _require_confirm(confirm: bool, what: str) -> None: + if not confirm: + raise PermissionError( + f"{what} drives real microscope hardware through NIS-Elements and " + "requires confirm=True. Refusing to proceed without explicit confirmation.") + + +def _bridge() -> NISBridge: + return NISBridge() + + +def capture_dir() -> Path: + """Where nis_capture writes: CONFOCAL_NIS_CAPTURE_DIR, else /data/nis_captures.""" + d = Path(os.environ.get("CONFOCAL_NIS_CAPTURE_DIR") or data_root() / "data" / "nis_captures") + d.mkdir(parents=True, exist_ok=True) + return d + + +def nis_status() -> dict: + """Read-only snapshot from NIS-Elements: XY and Z position (um), current + objective, the objective at each nosepiece position, whether an ND + experiment is running, and whether the e-stop is engaged. Safe to call + at any time; needs the NIS bridge job to be running.""" + return _bridge().status() + + +def nis_move_relative(dx_um: float, dy_um: float, confirm: bool = False) -> dict: + """Move the XY stage by (dx_um, dy_um) microns RELATIVE to where it is, + through NIS-Elements. At most 1000 um per axis per call. Returns the + position before and after - treat 'after_um' as ground truth. + Requires confirm=True. Z cannot be moved from here.""" + _require_confirm(confirm, "nis_move_relative") + return _bridge().move_relative(dx_um, dy_um) + + +def nis_change_objective(name: str, confirm: bool = False) -> dict: + """Switch objective in NIS-Elements. `name` must be the exact name + nis_status lists under nosepiece_objectives, and only 4x or 10x + objectives are accepted (long working distance - they cannot reach + the sample). Returns the objective and nosepiece position before and + after. Requires confirm=True.""" + _require_confirm(confirm, "nis_change_objective") + return _bridge().change_objective(name) + + +_SAFE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 _.\-]{0,80}$") + + +def nis_capture(filename: str, confirm: bool = False) -> dict: + """Capture one image with the current NIS-Elements acquisition settings + (lasers, channels, scan) and save it as an ND2 file in the NIS capture + folder. `filename` is a plain name such as "aml18_check" - no folders; + ".nd2" is added if missing and an existing file is never overwritten. + Returns the saved path. Requires confirm=True.""" + _require_confirm(confirm, "nis_capture") + stem = filename[:-4] if filename.lower().endswith(".nd2") else filename + if not _SAFE_NAME.match(stem) or ".." in stem: + raise ValueError("filename must be a plain name (letters, digits, space, _ . -), no folders") + return _bridge().capture(str(capture_dir() / f"{stem}.nd2")) + + +def nis_run_experiment(experiment: str, confirm: bool = False) -> dict: + """Start an ND experiment that was set up and SAVED in NIS-Elements' + ND Acquisition window, by its saved name. It runs with exactly the + saved settings (time loop, channels, save-to-file path). The call may + not return until the experiment ends. Refused if one is already + running. Requires confirm=True.""" + _require_confirm(confirm, "nis_run_experiment") + return _bridge().nd_run(experiment) + + +def nis_finish_experiment() -> dict: + """End the running ND experiment after its current time loop; NIS saves + the frames captured so far. Never needs confirm - it only stops.""" + return _bridge().nd_finish() diff --git a/mcp_server/server_nis.py b/mcp_server/server_nis.py new file mode 100644 index 0000000..ba1b98d --- /dev/null +++ b/mcp_server/server_nis.py @@ -0,0 +1,38 @@ +# server_nis.py +# ------------------------------------------------------------ +# MCP server for driving the microscope through NIS-Elements (see +# nis_tools.py). Separate from server_loop.py on purpose: that server's +# tool set is fixed by team direction and has no NIS-Elements in it, and +# this one needs the NIS bridge job running inside NIS. +# +# Run, either way: +# confocal-mcp-nis (installed console script) +# python -m mcp_server.server_nis (from a source checkout) +# +# { "mcpServers": { "confocal-nis": { "command": "confocal-mcp-nis" } } } +# ------------------------------------------------------------ + +from mcp.server.mcpserver import MCPServer + +from acquisition import estop +from mcp_server import loop_tools, nis_tools + +mcp = MCPServer("ConfocalOrchestrator-NIS") + +mcp.add_tool(nis_tools.nis_status) +mcp.add_tool(nis_tools.nis_move_relative) +mcp.add_tool(nis_tools.nis_change_objective) +mcp.add_tool(nis_tools.nis_capture) +mcp.add_tool(nis_tools.nis_run_experiment) +mcp.add_tool(nis_tools.nis_finish_experiment) +mcp.add_tool(loop_tools.estop) # the same e-stop, reachable from this server too + + +def main() -> None: + """Console entry point (``confocal-mcp-nis``): serve the NIS tools over stdio.""" + estop.launch_panel() + mcp.run(transport="stdio") + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index abb28dc..a0d7d60 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,13 +45,23 @@ harness = [ "anthropic==1.0.0", "python-dotenv==1.2.3", ] +# The offline scripts in analysis/ (ND2 reading, worm/Physarum analysis, +# movie soundtracks). Run from a source checkout - analysis/ is not in the wheel. +analysis = [ + "scipy==1.18.1", + "scikit-image==0.26.0", + "nd2==0.12.0", + "imageio-ffmpeg==0.6.0", + "opencv-python==5.0.0.93", +] # Everything, for a full workstation install. all = [ - "confocal-mcp[camera,sdk,harness,test]", + "confocal-mcp[camera,sdk,harness,analysis,test]", ] [project.scripts] confocal-mcp = "mcp_server.server_loop:main" +confocal-mcp-nis = "mcp_server.server_nis:main" [project.urls] Homepage = "https://github.com/BioNanomics/microscope-agent" diff --git a/tests/test_nis_bridge.py b/tests/test_nis_bridge.py new file mode 100644 index 0000000..0dde225 --- /dev/null +++ b/tests/test_nis_bridge.py @@ -0,0 +1,267 @@ +# The NIS bridge end to end, without NIS: the real bridge_job HTTP server +# and request queue, the real client and the real MCP tools, with a +# stand-in for g5_regprocs.dll that records every macro call. Covers the +# safety rules (POST-only motion, 1 mm cap, 4x/10x only, e-stop in both +# client and bridge, confirm gate, capture paths); what only NIS can +# answer - that the macro calls do the right thing on the Ti2 - is in +# docs/nis-bridge.md's test plan. +import asyncio +import json +import os +import sys +import threading +import urllib.request + +import pytest + +from acquisition import estop +from acquisition.backends import nis_bridge as client_mod +from acquisition.nis_bridge import bridge_job as bj +from mcp_server import nis_tools + + +class FakeNIS: + """Stands in for g5_regprocs.dll: same names, records calls.""" + + def __init__(self): + self.x, self.y, self.z = 100.0, 200.0, 4619.0 + self.objectives = {1: "PLAN APO \u03bbD 4x OFN25", 2: "Plan Fluor 10x Ph1 DLL", + 4: "Plan Apo 60x Oil"} + self.position = 1 + self.nd_running = False + self.load_rc = 1 + self.calls = [] + self.last_saved = None + f = self._fn + self.StgGetPosXY = f("StgGetPosXY", lambda px, py: self._set(px, self.x, py, self.y)) + self.StgGetPosZ = f("StgGetPosZ", lambda pz, dev: self._set(pz, self.z)) + self.StgMoveXY = f("StgMoveXY", self._move) + self.Stg_GetNosepiecePosition = f("Stg_GetNosepiecePosition", lambda: self.position) + self.Stg_GetNosepiecePositions = f("Stg_GetNosepiecePositions", lambda: 6) + self.Stg_GetNosepieceObjectiveName = f("Stg_GetNosepieceObjectiveName", self._name) + self.GetCurrentObjName = f("GetCurrentObjName", self._current) + self.ChangeObjective = f("ChangeObjective", self._change) + self.Capture = f("Capture", lambda: 1) + self.ImageSaveAs = f("ImageSaveAs", self._save) + self.CloseCurrentDocument = f("CloseCurrentDocument", lambda save: 1) + self.ND_LoadExperiment = f("ND_LoadExperiment", lambda name: self.load_rc) + self.ND_RunExperiment = f("ND_RunExperiment", self._run) + self.ND_IsInExperimentCapture = f("ND_IsInExperimentCapture", lambda: int(self.nd_running)) + self.ND_FinishExperiment = f("ND_FinishExperiment", self._finish) + + def _fn(self, name, impl): + def call(*args): + self.calls.append(name) + return impl(*args) + return call + + @staticmethod + def _set(*pairs): + for ptr, value in zip(pairs[::2], pairs[1::2]): + ptr._obj.value = value + return 1 + + def _move(self, dx, dy, relative): + assert relative == bj.MOVE_RELATIVE + self.x += dx; self.y += dy + return 1 + + def _name(self, i, buf, n): + if i in self.objectives: + buf.value = self.objectives[i] + return 1 + return -2 + + def _current(self, buf): + buf.value = self.objectives[self.position] + return 1 + + def _change(self, name): + self.position = next(i for i, v in self.objectives.items() if v == name) + return 1 + + def _save(self, path, kind, compression): + assert kind == bj.ND2_ALL_LAYERS + open(path, "wb").close() + self.last_saved = path + return 1 + + def _run(self, open_after): + self.nd_running = True + return 1 + + def _finish(self): + self.nd_running = False + return 1 + + def moved(self): + return [c for c in self.calls if c in ("StgMoveXY", "ChangeObjective", "Capture", + "ND_RunExperiment")] + + +@pytest.fixture +def estop_file(tmp_path, monkeypatch): + path = tmp_path / "ESTOP" + monkeypatch.setattr(estop, "ESTOP_PATH", path) + monkeypatch.setattr(bj, "ESTOP", path) + return path + + +@pytest.fixture +def bridge(estop_file, monkeypatch): + nis = FakeNIS() + stop, ready, port = threading.Event(), threading.Event(), [] + t = threading.Thread(target=bj.serve, args=(bj.Bridge(bj.Macro(nis)),), + kwargs=dict(port=0, stop=stop, + on_ready=lambda p: (port.append(p), ready.set())), + daemon=True) + t.start() + assert ready.wait(5) + url = f"http://127.0.0.1:{port[0]}" + monkeypatch.setenv("CONFOCAL_NIS_BRIDGE_URL", url) + yield nis, url, client_mod.NISBridge(url) + stop.set() + t.join(5) + + +def raw(url, method, path, body=None): + req = urllib.request.Request(url + path, method=method, + data=json.dumps(body).encode() if body is not None else None) + try: + with urllib.request.urlopen(req, timeout=5) as r: + return r.status, json.loads(r.read()) + except urllib.error.HTTPError as e: + return e.code, json.loads(e.read()) + + +# ---- bridge behaviour ------------------------------------------------- + +def test_status_reports_positions_objectives_and_state(bridge): + nis, url, c = bridge + s = c.status() + assert s["xy_um"] == [100.0, 200.0] and s["z_um"] == 4619.0 + assert s["current_objective"] == nis.objectives[1] + assert s["nosepiece_objectives"] == {str(k): v for k, v in nis.objectives.items()} + assert s["nd_running"] is False and s["estop_engaged"] is False + + +def test_relative_move_and_cap(bridge): + nis, url, c = bridge + r = c.move_relative(150, -50) + assert r["before_um"] == [100.0, 200.0] and r["after_um"] == [250.0, 150.0] + assert "refused" in c.move_relative(1000.1, 0)["error"] + assert (nis.x, nis.y) == (250.0, 150.0) + + +def test_get_cannot_move_or_acquire(bridge): + nis, url, c = bridge + for path in ("/move", "/objective", "/capture", "/nd/run", "/shutdown"): + code, out = raw(url, "GET", path) + assert code == 404 + assert nis.moved() == [] + + +@pytest.mark.parametrize("name,ok", [ + ("Plan Fluor 10x Ph1 DLL", True), + ("Plan Apo 60x Oil", False), + ("CFI 14x", False), + ("Plan Apo 100x Oil", False), +]) +def test_objective_allow_list(bridge, name, ok): + nis, url, c = bridge + if name not in nis.objectives.values(): + nis.objectives[5] = name + r = c.change_objective(name) + assert ("error" not in r) is ok + assert ("ChangeObjective" in nis.calls) is ok + + +def test_objective_change_reports_nosepiece(bridge): + nis, url, c = bridge + r = c.change_objective("Plan Fluor 10x Ph1 DLL") + assert (r["nosepiece_before"], r["nosepiece_after"]) == (1, 2) + + +def test_estop_blocks_in_bridge_and_client(bridge, estop_file): + nis, url, c = bridge + estop_file.write_text("{}") + # client refuses before sending anything + with pytest.raises(estop.EStopEngaged): + c.move_relative(10, 0) + # a request that bypasses the client is refused inside the bridge + for path, body in (("/move", {"dx": 10}), ("/objective", {"name": "Plan Fluor 10x Ph1 DLL"}), + ("/capture", {"path": str(estop_file.parent / "a.nd2")}), + ("/nd/run", {"experiment": "x"})): + code, out = raw(url, "POST", path, body) + assert out.get("estop_engaged") is True, path + assert nis.moved() == [] + # ending a run is always allowed + nis.nd_running = True + assert c.nd_finish()["nd_running"] is False + + +def test_capture_saves_nd2_and_never_overwrites(bridge, tmp_path): + nis, url, c = bridge + target = tmp_path / "check.nd2" + r = c.capture(str(target)) + assert r["saved"] is True and nis.last_saved == str(target) + assert "already exists" in c.capture(str(target))["error"] + assert "nd2" in c.capture(str(tmp_path / "x.tif"))["error"] + + +def test_nd_run_refuses_second_run_and_reports_load_failure(bridge): + nis, url, c = bridge + nis.load_rc = -2 + assert "ND_LoadExperiment" in c.nd_run("missing")["error"] + nis.load_rc = 1 + assert c.nd_run("C. elegans 3h")["nd_running"] is True + assert "already running" in c.nd_run("C. elegans 3h")["error"] + + +def test_shutdown_stops_the_bridge(bridge): + nis, url, c = bridge + assert c.shutdown() == {"stopping": True} + + +def test_unreachable_bridge_is_a_clear_error(): + with pytest.raises(client_mod.BridgeUnavailable, match="bridge job"): + client_mod.NISBridge("http://127.0.0.1:9", timeout_s=2).status() + + +# ---- MCP tools ---------------------------------------------------------- + +def test_tools_need_confirm(bridge): + nis, url, c = bridge + for call in (lambda: nis_tools.nis_move_relative(10, 0), + lambda: nis_tools.nis_change_objective("Plan Fluor 10x Ph1 DLL"), + lambda: nis_tools.nis_capture("a"), + lambda: nis_tools.nis_run_experiment("x")): + with pytest.raises(PermissionError): + call() + assert nis.moved() == [] + assert nis_tools.nis_move_relative(10, 0, confirm=True)["after_um"] == [110.0, 200.0] + + +def test_capture_tool_writes_only_into_the_capture_folder(bridge, tmp_path, monkeypatch): + nis, url, c = bridge + monkeypatch.setenv("CONFOCAL_NIS_CAPTURE_DIR", str(tmp_path / "caps")) + out = nis_tools.nis_capture("aml18_check", confirm=True) + assert out["path"] == str(tmp_path / "caps" / "aml18_check.nd2") + for bad in ("../evil", "C:\\Windows\\x", "a/b", ""): + with pytest.raises(ValueError): + nis_tools.nis_capture(bad, confirm=True) + + +def test_nis_server_exposes_exactly_the_nis_tools(tmp_path): + from mcp.client import Client + from mcp.client.stdio import StdioServerParameters, stdio_client + + async def names(): + env = dict(os.environ, CONFOCAL_NO_ESTOP_PANEL="1", CONFOCAL_MCP_DATA_DIR=str(tmp_path)) + params = StdioServerParameters(command=sys.executable, args=["-m", "mcp_server.server_nis"], env=env) + async with Client(stdio_client(params)) as client: + return {t.name for t in (await client.list_tools()).tools} + + assert asyncio.run(names()) == {"nis_status", "nis_move_relative", "nis_change_objective", + "nis_capture", "nis_run_experiment", "nis_finish_experiment", + "estop"}