From c3b6540bbd1769d78958f8986623a58bdfd2566a Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 16:08:04 -0400 Subject: [PATCH 01/10] Add the NIS bridge job: NIS macro calls over localhost HTTP A JOBS Python task that, while running, serves /status, /move (relative, capped at 1 mm), /objective (4x/10x only), /capture (save as ND2), /nd/run (load a saved ND experiment and run it), /nd/finish and /shutdown on 127.0.0.1:8766. Motion and acquisition answer only POST; nothing moves or starts while the e-stop file exists; Z is never moved. NIS calls stay on the job thread, HTTP threads only queue them. Macro signatures are from the NIS 6.20 reference on the microscope PC. Checked off the scope over real HTTP with a stand-in DLL; not yet run inside NIS. --- acquisition/nis_bridge/__init__.py | 0 acquisition/nis_bridge/bridge_job.py | 280 +++++++++++++++++++++++++++ 2 files changed, 280 insertions(+) create mode 100644 acquisition/nis_bridge/__init__.py create mode 100644 acquisition/nis_bridge/bridge_job.py 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)) From 7bc187fdbc287a1672346da134eacfac0aacaaaf Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 16:08:55 -0400 Subject: [PATCH 02/10] Add the NIS bridge client with a command line for manual tests NISBridge wraps the bridge job's HTTP API: status, move_relative, change_objective, capture, nd_run, nd_finish, shutdown. The e-stop is checked here before any motion or acquisition request is sent, so a stopped microscope does not rely on the NIS side alone; ending an ND run and shutting the bridge down are always allowed. An unreachable bridge raises BridgeUnavailable with a pointer to the setup notes. --- acquisition/backends/nis_bridge.py | 128 +++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 acquisition/backends/nis_bridge.py 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() From d60e90daf0ffbb103e6f0420b1f66a7208a811fb Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 16:10:07 -0400 Subject: [PATCH 03/10] Add the NIS MCP server (confocal-mcp-nis), separate from server_loop Tools that go through the NIS bridge: nis_status, nis_move_relative, nis_change_objective, nis_capture, nis_run_experiment and nis_finish_experiment, plus the shared estop. Everything that moves or acquires needs confirm=True, like backend='sdk' in loop_tools; captures go only to the data folder by plain file name; finishing an ND run and the e-stop never need confirm. A server of its own because server_loop's tool set is fixed by team direction and deliberately free of NIS-Elements; loop_tools.py and server_loop.py are unchanged. --- mcp_server/nis_tools.py | 109 +++++++++++++++++++++++++++++++++++++++ mcp_server/server_nis.py | 38 ++++++++++++++ pyproject.toml | 1 + 3 files changed, 148 insertions(+) create mode 100644 mcp_server/nis_tools.py create mode 100644 mcp_server/server_nis.py 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..e6114e9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,6 +52,7 @@ all = [ [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" From 7a770ccfe17c9666fabee9860682194cfafa2066 Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 16:11:36 -0400 Subject: [PATCH 04/10] Test the NIS bridge, client and MCP tools without NIS Runs the real bridge HTTP server and queue, the real client and the real MCP tools against a stand-in for g5_regprocs.dll that records every macro call. Covers status, relative moves and the 1 mm cap, GET never moving anything, the 4x/10x allow-list, the e-stop in both client and bridge (with finishing an ND run still allowed), ND2 captures that never overwrite, ND run guards, the confirm gate, capture file names confined to the capture folder, and the NIS server's exact tool set. --- tests/test_nis_bridge.py | 267 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 267 insertions(+) create mode 100644 tests/test_nis_bridge.py 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"} From efac5dcb4f812c5777dc4f0cf3a55d2af2069d33 Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 16:12:29 -0400 Subject: [PATCH 05/10] docs: NIS bridge setup, safety rules and on-scope test plan docs/nis-bridge.md explains how the bridge, client and NIS MCP server fit together, how to run the bridge as a JOBS task, and the step-by-step test plan for the microscope - including the open questions (does NIS stay usable, does ChangeObjective turn the nosepiece, does ND_RunExperiment block). README and CHANGELOG point at it. --- CHANGELOG.md | 10 +++++ README.md | 15 +++++++ docs/nis-bridge.md | 98 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 123 insertions(+) create mode 100644 docs/nis-bridge.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ac3cec..f9631c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow ## [Unreleased] ### Added +- 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 diff --git a/README.md b/README.md index 9e70adb..7c0b0c8 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 ``` 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. From 18732b3497803d34153062cc504ba5c7ee8c6c0d Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 17:27:28 -0400 Subject: [PATCH 06/10] Declare the analysis/ dependencies as an optional group scipy, scikit-image, nd2, imageio-ffmpeg and opencv-python were imported by the analysis scripts but not declared, so a fresh install could not run them. They are now the 'analysis' group, included in 'all'. --- CHANGELOG.md | 4 ++++ README.md | 3 ++- pyproject.toml | 11 ++++++++++- 3 files changed, 16 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9631c6..fe526ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -87,6 +87,10 @@ 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. ## [0.1.0] - 2026-08-31 diff --git a/README.md b/README.md index 7c0b0c8..8646c9a 100644 --- a/README.md +++ b/README.md @@ -100,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/pyproject.toml b/pyproject.toml index e6114e9..a0d7d60 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,9 +45,18 @@ 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] From 9de3ed71719927c2e5c26067958c9861b341564c Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 17:29:04 -0400 Subject: [PATCH 07/10] Fix references to files that live in ConfocalOrchestrator The stage_positions demo loaded protocols/example_protocol.yaml, which is not in this repo, so it always failed; that step is removed. Comments citing docs/microscope-notes.md, run_protocol.py and the calibration tests now name ConfocalOrchestrator, and old names (mcp_server/server.py, acquisition_tools.py) are replaced with the current ones. --- CHANGELOG.md | 6 ++++++ acquisition/backends/baumer_genicam.py | 2 +- acquisition/backends/nis_mock.py | 16 +++++++------- acquisition/backends/nis_sdk.py | 18 ++++++++-------- acquisition/orchestration/stage_positions.py | 22 ++++++++------------ 5 files changed, 33 insertions(+), 31 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe526ae..46f0ae4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -91,6 +91,12 @@ Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow `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/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_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/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() From 0c494a888b4a4a5c9b32e6593dfd351a3dd318ce Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 21:46:37 -0400 Subject: [PATCH 08/10] Add shared helpers for vertical YouTube Shorts Canvas, fonts, centred text and an ffmpeg writer (x264 + AAC in one pass, or preview stills) used by the Physarum and C. elegans Shorts. --- CHANGELOG.md | 4 ++ analysis/short_video.py | 115 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+) create mode 100644 analysis/short_video.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 46f0ae4..fe1ea66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow ## [Unreleased] ### Added +- `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 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() From 434d4770879b5851ade61beab42c855ca9753205 Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 21:46:37 -0400 Subject: [PATCH 09/10] Add the Physarum Short: time-lapse, live growth curve, growth-map reveal Runs the combined analysis script's alignment, draws the smoothed on-agar area under the time-lapse and wipes to growth_start_to_end_colour.jpg, which is the same frame so needs no registration. --- CHANGELOG.md | 4 + analysis/physarum_short.py | 274 +++++++++++++++++++++++++++++++++++++ 2 files changed, 278 insertions(+) create mode 100644 analysis/physarum_short.py diff --git a/CHANGELOG.md b/CHANGELOG.md index fe1ea66..815acf2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ Format follows [Keep a Changelog](https://keepachangelog.com/); versions follow ## [Unreleased] ### Added +- `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 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() From 3e595b4607e5f7ed916df5e6601ffe130c182c93 Mon Sep 17 00:00:00 2001 From: Qurratul Ain Quais Date: Wed, 30 Sep 2026 21:46:38 -0400 Subject: [PATCH 10/10] Add the C. elegans Short from an AML18 NIS time-lapse Re-links tracks at 0.5 mm with a 3-frame position median, features the longest-followed adults named by speed, and states the Explorer/Homebody ratio as the smaller of the raw and filtered figures. --- CHANGELOG.md | 5 + analysis/worm_short.py | 424 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 429 insertions(+) create mode 100644 analysis/worm_short.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 815acf2..b6b070b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ 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 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()