Acquisition core refactor (Phase 1 of 4): single-loop acquisition, decoupled tracking, design for recording/focus/UI phases

CONTEXT.md

@@ -0,0 +1,39 @@
+# WormsPy — Context & Glossary
+
+A glossary of the domain language used in WormsPy. Implementation details live in code and ADRs, not here.
+
+## Glossary
+
+### Frame-synced recording
+The guarantee that a frame from the **brightfield/behaviour** camera and a frame from the
+**fluorescence/calcium** camera can be reliably paired for offline analysis (e.g. correlating
+RIA compartmental calcium with head-bending).
+
+Two distinct properties are bundled under this term; keep them separate:
+
+- **Sensor-level sync** — the two sensors expose at the same instant. Currently **not** provided
+  (cameras free-run). Achievable via FLIR GPIO master–slave hardware trigger. Considered
+  *desirable but optional*: **tens of ms of offset between the two streams is acceptable** for the
+  current science.
+- **Loop coherence (no drift)** — the pairing between the two streams must **not drift without
+  bound** over a recording. This is a **hard requirement**. The current two-independent-generator
+  design violates it.
+
+### Hold Focus (formerly "FocusLock" / "autofocus")
+Keeping the neuron of interest in focus on the fluorescence/calcium feed. True autofocus is
+**not achievable** on this rig:
+- There is **one shared z motor and one objective**, so both cameras focus at a single plane —
+  you cannot independently focus brightfield and fluorescence.
+- Best focus is defined by the **target neuron**, whose depth inside the worm **changes every
+  session**, so there is no fixed parfocal offset to calibrate once.
+
+Therefore focus is **set by the operator** (manually, via joystick, watching the fluorescence
+feed) and software only **holds or restores** that z. An optional, opt-in **drift compensation**
+mode may actively nudge z using a brightfield reference captured at focus-set time, at the cost of
+a focus/intensity artifact in the recording (see [[ADR 0002]]).
+
+### Loop drift
+The accumulating divergence between the brightfield and fluorescence acquisition loops because
+each runs its own independent `while cap.isOpened()` loop with its own wall-clock timestamps and
+its own (possibly different, possibly dropped-frame) cadence. **Unacceptable** — see
+[[frame-synced-recording]].

WormSpy/backend/code/core/__init__.py

@@ -0,0 +1 @@
+"""WormsPy acquisition core: hardware-independent acquisition, tracking, and stage control."""

WormSpy/backend/code/core/acquisition.py

@@ -0,0 +1,61 @@
+import threading
+import time
+from typing import Callable, Optional
+
+from core.cameras import Camera
+from core.frames import AcquiredPair, FrameHub
+
+Sink = Callable[[AcquiredPair], None]
+
+
+class AcquisitionEngine:
+    """Single loop that owns both cameras. Grabs both per tick, assigns a shared monotonic
+    frame_index, publishes the latest frame of each to the hub for display/tracking, and
+    forwards the paired result to an optional sink (the recorder). This is the only place that
+    reads from the cameras, so the two streams cannot drift relative to each other."""
+
+    LEFT = "left"
+    RIGHT = "right"
+
+    def __init__(self, left: Camera, right: Camera, hub: FrameHub,
+                 sink: Optional[Sink] = None) -> None:
+        self._left = left
+        self._right = right
+        self._hub = hub
+        self._sink = sink
+        self._next_index = 0
+        self._stop = threading.Event()
+        self.ticks_completed = 0
+
+    def run_once(self) -> AcquiredPair:
+        index = self._next_index
+        self._next_index += 1
+        ts = time.monotonic()
+
+        left_frame = self._left.read()
+        right_frame = self._right.read()
+
+        if left_frame is not None:
+            self._hub.publish(self.LEFT, left_frame)
+        if right_frame is not None:
+            self._hub.publish(self.RIGHT, right_frame)
+
+        pair = AcquiredPair(frame_index=index, capture_ts=ts,
+                            left=left_frame, right=right_frame)
+        if self._sink is not None:
+            self._sink(pair)
+        self.ticks_completed += 1
+        return pair
+
+    def run(self, max_ticks: Optional[int] = None) -> None:
+        produced = 0
+        while not self._stop.is_set():
+            if max_ticks is not None and produced >= max_ticks:
+                break
+            if not (self._left.is_open() and self._right.is_open()):
+                break
+            self.run_once()
+            produced += 1
+
+    def stop(self) -> None:
+        self._stop.set()

WormSpy/backend/code/core/cameras.py

@@ -0,0 +1,53 @@
+import time
+from abc import ABC, abstractmethod
+from typing import Iterable, List, Optional, Set
+
+import numpy as np
+
+from core.frames import Frame
+
+
+class Camera(ABC):
+    """Acquisition interface. Adapters wrap real SDKs; FakeCamera is the test double."""
+
+    @abstractmethod
+    def read(self) -> Optional[Frame]:
+        """Grab the next frame, or return None if the grab failed / stream ended."""
+
+    @abstractmethod
+    def is_open(self) -> bool:
+        ...
+
+    @abstractmethod
+    def release(self) -> None:
+        ...
+
+
+class FakeCamera(Camera):
+    """Deterministic camera for tests. Emits a scripted list of images."""
+
+    def __init__(self, images: Iterable[np.ndarray],
+                 frame_ids: Optional[List[int]] = None,
+                 fail_on: Optional[Set[int]] = None) -> None:
+        self._images: List[np.ndarray] = list(images)
+        self._frame_ids = frame_ids if frame_ids is not None else list(range(len(self._images)))
+        self._fail_on = fail_on or set()
+        self._pos = 0
+        self._open = True
+
+    def read(self) -> Optional[Frame]:
+        if not self._open or self._pos >= len(self._images):
+            return None
+        idx = self._pos
+        self._pos += 1
+        if idx in self._fail_on:
+            return None
+        return Frame(image=self._images[idx],
+                     device_frame_id=self._frame_ids[idx],
+                     capture_ts=time.monotonic())
+
+    def is_open(self) -> bool:
+        return self._open
+
+    def release(self) -> None:
+        self._open = False

WormSpy/backend/code/core/cameras_spinnaker.py

@@ -0,0 +1,49 @@
+"""FLIR/Spinnaker camera adapter via EasyPySpin. Imports the SDK lazily so tests don't load it.
+
+Sets the camera as the acquisition clock: a fixed AcquisitionFrameRate and a fixed exposure,
+with auto-exposure/auto-gain off. EasyPySpin exposes these through cv2-style properties; the
+device frame id is read from the grabbed image metadata when available, else -1."""
+
+import time
+from typing import Optional
+
+import cv2
+import numpy as np
+
+from core.cameras import Camera
+from core.frames import Frame
+
+
+class SpinnakerCamera(Camera):
+    def __init__(self, identifier, frame_rate: float, exposure_us: float) -> None:
+        import EasyPySpin    # lazy import
+        self._cap = EasyPySpin.VideoCapture(identifier)
+        if not self._cap.isOpened():
+            print(f"SpinnakerCamera: could not open camera {identifier!r}")
+            return
+        # Camera is the clock: disable auto, pin rate + exposure.
+        self._cap.set(cv2.CAP_PROP_AUTO_EXPOSURE, 0)        # manual exposure
+        self._cap.set(cv2.CAP_PROP_EXPOSURE, exposure_us)
+        self._cap.set(cv2.CAP_PROP_FPS, frame_rate)
+        self.width = self._cap.get(cv2.CAP_PROP_FRAME_WIDTH)
+        self.height = self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
+
+    def read(self) -> Optional[Frame]:
+        ok, image = self._cap.read()
+        ts = time.monotonic()
+        if not ok or image is None:
+            return None
+        if image.dtype != np.uint8:
+            # Preserve raw for recording elsewhere; this path is display-oriented in Phase 1.
+            image = cv2.normalize(image, None, 0, 255, cv2.NORM_MINMAX, dtype=cv2.CV_8U)
+        if len(image.shape) == 3:
+            image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        frame_id = int(self._cap.get(cv2.CAP_PROP_POS_FRAMES))
+        return Frame(image=image, device_frame_id=frame_id, capture_ts=ts)
+
+    def is_open(self) -> bool:
+        return self._cap is not None and self._cap.isOpened()
+
+    def release(self) -> None:
+        if self._cap is not None:
+            self._cap.release()

WormSpy/backend/code/core/frames.py

@@ -0,0 +1,38 @@
+import threading
+from dataclasses import dataclass
+from typing import Dict, Optional
+
+import numpy as np
+
+
+@dataclass
+class Frame:
+    """A single grabbed image plus the metadata needed to align it offline."""
+    image: np.ndarray
+    device_frame_id: int   # camera-provided frame id; -1 if the camera cannot supply one
+    capture_ts: float      # time.monotonic() captured as close to the grab as possible
+
+
+@dataclass
+class AcquiredPair:
+    """One acquisition tick: a brightfield and a fluorescence frame sharing an index."""
+    frame_index: int       # monotonic counter assigned by the engine, contiguous, never reused
+    capture_ts: float      # engine timestamp for the tick (time.monotonic())
+    left: Optional[Frame]  # brightfield/behaviour; None if that grab failed this tick
+    right: Optional[Frame] # fluorescence/calcium; None if that grab failed this tick
+
+
+class FrameHub:
+    """Thread-safe store of the latest frame per named stream, for display and tracking."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._latest: Dict[str, Frame] = {}
+
+    def publish(self, name: str, frame: Frame) -> None:
+        with self._lock:
+            self._latest[name] = frame
+
+    def latest(self, name: str) -> Optional[Frame]:
+        with self._lock:
+            return self._latest.get(name)

WormSpy/backend/code/core/stage.py

@@ -0,0 +1,84 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Tuple
+
+
+def clamp_native(value: float, minimum: float, maximum: float) -> Tuple[float, bool]:
+    """Clamp value into [minimum, maximum]. Returns (clamped_value, limit_was_hit)."""
+    if value > maximum:
+        return maximum, True
+    if value < minimum:
+        return minimum, True
+    return value, False
+
+
+class Stage(ABC):
+    """XYZ stage interface in native (microstep) units. Adapters wrap real hardware."""
+
+    @abstractmethod
+    def is_connected(self) -> bool:
+        ...
+
+    @abstractmethod
+    def get_position(self, axis: str) -> float:
+        ...
+
+    @abstractmethod
+    def limits(self, axis: str) -> Tuple[float, float]:
+        ...
+
+    @abstractmethod
+    def move_relative(self, axis: str, native: float) -> bool:
+        """Move axis by `native` microsteps, clamped to limits. Returns True if a limit was hit."""
+
+    @abstractmethod
+    def move_absolute(self, axis: str, native: float) -> bool:
+        """Move axis to absolute `native` microsteps, clamped to limits. Returns True if hit."""
+
+
+class NullStage(Stage):
+    """Used when no hardware is present so the app still runs. All moves are no-ops."""
+
+    def is_connected(self) -> bool:
+        return False
+
+    def get_position(self, axis: str) -> float:
+        return 0.0
+
+    def limits(self, axis: str) -> Tuple[float, float]:
+        return (0.0, 0.0)
+
+    def move_relative(self, axis: str, native: float) -> bool:
+        return False
+
+    def move_absolute(self, axis: str, native: float) -> bool:
+        return False
+
+
+class FakeStage(Stage):
+    """In-memory stage for tests. Honors limits via clamp_native."""
+
+    def __init__(self, limits: Dict[str, Tuple[float, float]],
+                 positions: Dict[str, float]) -> None:
+        self._limits = dict(limits)
+        self._pos = dict(positions)
+
+    def is_connected(self) -> bool:
+        return True
+
+    def get_position(self, axis: str) -> float:
+        return self._pos[axis]
+
+    def limits(self, axis: str) -> Tuple[float, float]:
+        return self._limits[axis]
+
+    def move_relative(self, axis: str, native: float) -> bool:
+        lo, hi = self._limits[axis]
+        target, hit = clamp_native(self._pos[axis] + native, lo, hi)
+        self._pos[axis] = target
+        return hit
+
+    def move_absolute(self, axis: str, native: float) -> bool:
+        lo, hi = self._limits[axis]
+        target, hit = clamp_native(native, lo, hi)
+        self._pos[axis] = target
+        return hit