feat: acquisition method schema v2 — FOV patterns, per-well Z, tilted-plate pre-AF (GUI + API)

software/control/acquisition_yaml_loader.py

@@ -2,6 +2,7 @@
 Utilities for parsing and validating acquisition YAML files.
 """
 
+import math
 import yaml
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional, Tuple
@@ -47,6 +48,17 @@ class AcquisitionYAMLData:
     # ("A1:B3" or "A1,B2,C3"); None when the method uses explicit regions instead.
     wells: Optional[str] = None
 
+    # Schema v2 (wellplate): per-well FOV pattern. None = legacy coverage behavior
+    # driven by flat scan_size_mm/overlap_percent. When present, always normalized
+    # to a dict with a "type" key; see _normalize_fov_pattern for per-type keys.
+    fov_pattern: Optional[Dict] = None
+    # Schema v2: per-well laser-AF target offsets (µm from the AF reference plane).
+    # Optional "default" key applies to wells not listed. Requires laser_af.
+    well_z_offsets_um: Optional[Dict[str, float]] = None
+    # Schema v2: pre-AF Z plan for tilted plates: {"type": "focus_map",
+    # "generate": bool, "points": [[x,y,z]*3] | None} (generate XOR points).
+    z_plan: Optional[Dict] = None
+
     # Flexible-specific
     nx: int = 1
     ny: int = 1
@@ -55,6 +67,105 @@ class AcquisitionYAMLData:
     flexible_positions: Optional[List[Dict]] = None  # [{name, center_mm}, ...]
 
 
+_FOV_PATTERN_TYPES = ("coverage", "centered_grid", "grid_subset", "random")
+
+
+def _normalize_fov_pattern(raw: Optional[dict], overlap_default: float) -> Optional[dict]:
+    if raw is None:
+        return None
+    if not isinstance(raw, dict) or "type" not in raw:
+        raise ValueError("fov_pattern must be a mapping with a 'type' key")
+    ptype = raw["type"]
+    if ptype not in _FOV_PATTERN_TYPES:
+        raise ValueError(f"fov_pattern type {ptype!r} not one of {_FOV_PATTERN_TYPES}")
+    if ptype == "coverage":
+        return {
+            "type": "coverage",
+            "scan_size_mm": raw.get("scan_size_mm"),
+            "overlap_percent": float(raw.get("overlap_percent", overlap_default)),
+            "shape": raw.get("shape", "Square"),
+        }
+    if ptype in ("centered_grid", "grid_subset"):
+        nx, ny = raw.get("nx"), raw.get("ny")
+        if not (isinstance(nx, int) and isinstance(ny, int) and nx >= 1 and ny >= 1):
+            raise ValueError(f"fov_pattern {ptype}: nx and ny must be integers >= 1")
+        out = {"type": ptype, "nx": nx, "ny": ny, "overlap_percent": float(raw.get("overlap_percent", overlap_default))}
+        if ptype == "grid_subset":
+            tiles = raw.get("tiles")
+            if not isinstance(tiles, list) or not tiles:
+                raise ValueError("fov_pattern grid_subset: 'tiles' must be a non-empty list of [row, col]")
+            norm_tiles = []
+            for t in tiles:
+                if not (isinstance(t, (list, tuple)) and len(t) == 2):
+                    raise ValueError(f"fov_pattern grid_subset: bad tile entry {t!r} (expected [row, col])")
+                row, col = int(t[0]), int(t[1])
+                if not (0 <= row < ny and 0 <= col < nx):
+                    raise ValueError(
+                        f"fov_pattern grid_subset: tile [{row}, {col}] outside {ny}x{nx} grid (rows 0..{ny-1}, cols 0..{nx-1})"
+                    )
+                norm_tiles.append([row, col])
+            if len({tuple(t) for t in norm_tiles}) != len(norm_tiles):
+                raise ValueError("fov_pattern grid_subset: duplicate tiles")
+            out["tiles"] = norm_tiles
+        return out
+    # random
+    n_fovs = raw.get("n_fovs")
+    if not (isinstance(n_fovs, int) and n_fovs >= 1):
+        raise ValueError("fov_pattern random: n_fovs must be an integer >= 1")
+    seed = raw.get("seed")
+    if seed is not None and not isinstance(seed, int):
+        raise ValueError("fov_pattern random: seed must be an integer")
+    return {"type": "random", "n_fovs": n_fovs, "seed": seed}
+
+
+def _validate_well_z_offsets(raw: Optional[dict]) -> Optional[Dict[str, float]]:
+    if raw is None:
+        return None
+    if not isinstance(raw, dict) or not raw:
+        raise ValueError("well_z_offsets_um must be a non-empty mapping of well name -> µm")
+    out = {}
+    for name, value in raw.items():
+        try:
+            f = float(value)
+        except (TypeError, ValueError):
+            raise ValueError(f"well_z_offsets_um[{name!r}]: not a number: {value!r}")
+        if not math.isfinite(f):
+            raise ValueError(f"well_z_offsets_um[{name!r}]: must be finite, got {value!r}")
+        out[str(name)] = f
+    return out
+
+
+def _validate_z_plan(raw: Optional[dict]) -> Optional[dict]:
+    if raw is None:
+        return None
+    if not isinstance(raw, dict) or raw.get("type") != "focus_map":
+        raise ValueError("z_plan: only {'type': 'focus_map', ...} is supported")
+    generate = bool(raw.get("generate", False))
+    points = raw.get("points")
+    if generate == bool(points):
+        raise ValueError("z_plan: specify exactly one of 'generate: true' or 'points'")
+    norm_points = None
+    if points is not None:
+        if not (isinstance(points, list) and len(points) == 3):
+            raise ValueError("z_plan: 'points' must be exactly 3 [x_mm, y_mm, z_mm] entries (a plane)")
+        norm_points = []
+        for p in points:
+            if not (isinstance(p, (list, tuple)) and len(p) == 3):
+                raise ValueError(f"z_plan: bad point {p!r} (expected [x_mm, y_mm, z_mm])")
+            fx, fy, fz = (float(v) for v in p)
+            if not all(math.isfinite(v) for v in (fx, fy, fz)):
+                raise ValueError(f"z_plan: non-finite point {p!r}")
+            norm_points.append([fx, fy, fz])
+        # The 3 points must span a plane: reject XY-colinear sets here (a clean
+        # loader ValueError -> INVALID_PARAM at preflight) instead of letting
+        # interpolate_plane raise deep inside start_acquisition (a misleading 500).
+        (x1, y1, _), (x2, y2, _), (x3, y3, _) = norm_points
+        det = (x2 - x1) * (y3 - y1) - (x3 - x1) * (y2 - y1)
+        if abs(det) < 1e-9:
+            raise ValueError("z_plan: the 3 points are colinear in XY and cannot define a plane")
+    return {"type": "focus_map", "generate": generate, "points": norm_points}
+
+
 def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
     """Parse acquisition YAML file and return structured data.
 
@@ -127,6 +238,12 @@ def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
     if wells and wellplate_regions:
         raise ValueError("wellplate_scan: specify either 'wells' or 'regions', not both")
 
+    fov_pattern = _normalize_fov_pattern(wellplate_scan.get("fov_pattern"), overlap)
+    if fov_pattern and fov_pattern["type"] != "coverage" and not wells:
+        raise ValueError(f"fov_pattern {fov_pattern['type']!r} requires wellplate_scan 'wells' (per-well patterns)")
+    well_z_offsets_um = _validate_well_z_offsets(wellplate_scan.get("well_z_offsets_um"))
+    z_plan = _validate_z_plan(wellplate_scan.get("z_plan"))
+
     return AcquisitionYAMLData(
         widget_type=widget_type,
         xy_mode=acq.get("xy_mode", "Select Wells"),
@@ -154,6 +271,9 @@ def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
         scan_shape=scan_shape,
         wellplate_regions=wellplate_regions,
         wells=wells,
+        fov_pattern=fov_pattern,
+        well_z_offsets_um=well_z_offsets_um,
+        z_plan=z_plan,
         # Flexible-specific
         nx=flexible_scan.get("nx", 1),
         ny=flexible_scan.get("ny", 1),

software/control/core/laser_auto_focus_controller.py

@@ -392,6 +392,19 @@ def _move_z(self, um_to_move: float) -> None:
         else:
             self.stage.move_z(um_to_move / 1000)
 
+    def apply_relative_offset_um(self, offset_um: float) -> None:
+        """Open-loop relative Z move of ``offset_um`` (displacement µm, 1:1 with Z), with NO
+        spot re-verification.
+
+        Intended to run right after ``move_to_target(0.0)`` has anchored — and verified — the
+        spot at the reference plane, to then place the sample at a target displacement from
+        that reference. Spot-alignment verification (``_verify_spot_alignment``) always crops
+        at ``x_reference`` and would fail for a deliberately-displaced spot, so it must NOT be
+        used to reach a nonzero target; this method deliberately skips it. No-op for offset 0.
+        """
+        if offset_um:
+            self._move_z(offset_um)
+
     def set_reference(self) -> bool:
         """Set the current spot position as the reference position.
 

software/control/core/multi_point_controller.py

@@ -103,21 +103,37 @@ def _save_acquisition_yaml(
 
     # Add widget-specific scan section
     if widget_type == "wellplate":
-        yaml_dict["wellplate_scan"] = {
-            "scan_size_mm": scan_size_mm,
-            "overlap_percent": overlap_percent,
-            "regions": [
-                {
-                    "name": name,
-                    "center_mm": _serialize_for_yaml(center),
-                    "shape": region_shapes.get(name) if region_shapes else None,
-                }
-                for name, center in zip(
-                    params.scan_position_information.scan_region_names,
-                    params.scan_position_information.scan_region_coords_mm,
-                )
-            ],
-        }
+        region_names = params.scan_position_information.scan_region_names
+        if params.xy_mode == "Select Wells":
+            # Schema v2: names + coverage pattern; X/Y are re-derived from the plate
+            # definition at load time (see acquisition_yaml_loader.parse_acquisition_yaml).
+            yaml_dict["wellplate_scan"] = {
+                "wells": list(region_names),
+                "fov_pattern": params.fov_pattern
+                or {
+                    "type": "coverage",
+                    "scan_size_mm": scan_size_mm,
+                    "overlap_percent": overlap_percent,
+                    "shape": (region_shapes or {}).get(region_names[0]) if region_names else "Square",
+                },
+            }
+        else:
+            # Manual / Current Position / Load Coordinates: inherently coordinate-based.
+            yaml_dict["wellplate_scan"] = {
+                "scan_size_mm": scan_size_mm,
+                "overlap_percent": overlap_percent,
+                "regions": [
+                    {
+                        "name": name,
+                        "center_mm": _serialize_for_yaml(center),
+                        "shape": region_shapes.get(name) if region_shapes else None,
+                    }
+                    for name, center in zip(
+                        region_names,
+                        params.scan_position_information.scan_region_coords_mm,
+                    )
+                ],
+            }
     else:  # flexible
         yaml_dict["flexible_scan"] = {
             "nx": params.NX,
@@ -227,6 +243,8 @@ def __init__(
         self.overlap_percent = 10.0  # FOV overlap percentage
 
         self.focus_map = None
+        self.region_laser_af_offsets = {}
+        self.fov_pattern = None
         self.gen_focus_map = False
         self.focus_map_storage = []
         self.already_using_fmap = False
@@ -416,6 +434,16 @@ def set_gen_focus_map_flag(self, flag):
     def set_focus_map(self, focusMap):
         self.focus_map = focusMap  # None if dont use focusMap
 
+    def set_region_laser_af_offsets(self, offsets):
+        # region_id -> µm offset from the global laser-AF reference plane. Empty dict means
+        # every FOV targets the reference (displacement 0), i.e. current behavior.
+        self.region_laser_af_offsets = dict(offsets or {})
+
+    def set_fov_pattern(self, pattern):
+        # Normalized fov_pattern dict (Task-1 loader shape) or None for coverage-from-scan-size.
+        # Consumed-and-cleared by run_acquisition so it never leaks to a later run.
+        self.fov_pattern = pattern
+
     def set_base_path(self, path):
         self.base_path = path
 
@@ -675,6 +703,17 @@ def get_estimated_mosaic_ram_bytes(self) -> int:
         return mosaic_width * mosaic_height * bytes_per_pixel * num_channels
 
     def run_acquisition(self, acquire_current_fov=False):
+        # Consume the per-region laser-AF offsets for THIS run and clear the sticky controller
+        # copy up-front. Any early return below — or a prior GUI abort that pushed offsets but
+        # never reached run_acquisition (e.g. aborted on the disk/RAM dialog) — then cannot leak
+        # them into a later acquisition from an entry point that never sets them (fluidics
+        # widget, TCP control server).
+        run_region_laser_af_offsets = self.region_laser_af_offsets
+        self.region_laser_af_offsets = {}
+        # Consume-and-clear the fov_pattern for THIS run so a stale pattern from an API run
+        # can never leak into a later GUI run (which never sets it).
+        run_fov_pattern = self.fov_pattern
+        self.fov_pattern = None
         if not self.validate_acquisition_settings():
             # emit acquisition finished signal to re-enable the UI
             self.callbacks.signal_acquisition_finished()
@@ -766,10 +805,15 @@ def run_acquisition(self, acquire_current_fov=False):
                         region_fov_coords[i] = (x, y, z)
                         self.scanCoordinates.update_fov_z_level(region_id, i, z)
 
-            elif self.gen_focus_map and not self.do_reflection_af:
+            elif self.gen_focus_map:
                 self._log.info("Generating autofocus plane for multipoint grid")
                 bounds = self.scanCoordinates.get_scan_bounds()
                 if not bounds:
+                    # No scannable regions: the worker never starts, so emit finished here
+                    # (same as the validate_acquisition_settings early return above) to
+                    # re-enable the GUI and let the service mark the job finished. Without
+                    # this the run hangs in ACQUIRING forever.
+                    self.callbacks.signal_acquisition_finished()
                     return
                 x_min, x_max = bounds["x"]
                 y_min, y_max = bounds["y"]
@@ -823,14 +867,35 @@ def run_acquisition(self, acquire_current_fov=False):
 
                     # Generate and enable the AF map
                     self.autofocusController.gen_focus_map(coord1, coord2, coord3)
-                    self.autofocusController.set_focus_map_use(True)
+                    if self.do_reflection_af:
+                        # Laser-AF run: contrast focus-map interpolation never executes in the
+                        # worker's laser branch. Bake the generated plane into the coordinates
+                        # instead (same mechanism as self.focus_map above), so each FOV is
+                        # pre-positioned within laser-AF capture range on a tilted plate.
+                        from control.utils import interpolate_plane
+
+                        pts = [tuple(p) for p in self.autofocusController.focus_map_coords[:3]]
+                        for region_id in scan_position_information.scan_region_names:
+                            region_fov_coords = scan_position_information.scan_region_fov_coords_mm[region_id]
+                            for i, coords in enumerate(region_fov_coords):
+                                x, y = coords[:2]
+                                z = interpolate_plane(pts[0], pts[1], pts[2], (x, y))
+                                region_fov_coords[i] = (x, y, z)
+                                self.scanCoordinates.update_fov_z_level(region_id, i, z)
+                    else:
+                        self.autofocusController.set_focus_map_use(True)
 
                     # Return to center position
                     self.stage.move_x_to(x_center)
                     self.stage.move_y_to(y_center)
 
                 except ValueError:
                     self._log.exception("Invalid coordinates for autofocus plane, aborting.")
+                    # The plane could not be generated; the worker never starts. Emit finished
+                    # so the GUI re-enables and the service marks the job finished/failed
+                    # (job.started_at is None -> reported as a validation failure) rather than
+                    # leaving it stuck in ACQUIRING.
+                    self.callbacks.signal_acquisition_finished()
                     return
 
             def finish_fn():
@@ -842,7 +907,11 @@ def finish_fn():
 
             updated_callbacks = dataclasses.replace(self.callbacks, signal_acquisition_finished=finish_fn)
 
-            acquisition_params = self.build_params(scan_position_information=scan_position_information)
+            acquisition_params = self.build_params(
+                scan_position_information=scan_position_information,
+                region_laser_af_offsets=run_region_laser_af_offsets,
+                fov_pattern=run_fov_pattern,
+            )
 
             # Gather objective and camera info for YAML
             current_objective = self.objectiveStore.current_objective
@@ -946,7 +1015,12 @@ def finish_fn():
                     self._memory_monitor.stop()
                     self._memory_monitor = None
 
-    def build_params(self, scan_position_information: ScanPositionInformation) -> AcquisitionParameters:
+    def build_params(
+        self,
+        scan_position_information: ScanPositionInformation,
+        region_laser_af_offsets: Optional[dict] = None,
+        fov_pattern: Optional[dict] = None,
+    ) -> AcquisitionParameters:
         # Determine plate dimensions from wellplate format if available
         plate_num_rows = 8  # Default for 96-well
         plate_num_cols = 12
@@ -986,6 +1060,8 @@ def build_params(self, scan_position_information: ScanPositionInformation) -> Ac
             plate_num_rows=plate_num_rows,
             plate_num_cols=plate_num_cols,
             xy_mode=self.xy_mode,
+            region_laser_af_offsets=region_laser_af_offsets if region_laser_af_offsets is not None else {},
+            fov_pattern=fov_pattern,
         )
 
     def _on_acquisition_completed(self):

software/control/core/multi_point_utils.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import List, Tuple, Dict, Optional, Callable, TYPE_CHECKING
 
 from control.core.job_processing import CaptureInfo
@@ -64,6 +64,16 @@ class AcquisitionParameters:
     # XY mode for determining scan type
     xy_mode: str = "Current Position"  # "Current Position", "Select Wells", "Manual", "Load Coordinates"
 
+    # Per-region laser-AF target offsets (µm from the global laser-AF reference plane),
+    # keyed by region_id. Empty unless the focus-map + laser-AF combined mode is active,
+    # in which case each FOV in a region targets that region's offset instead of 0.
+    region_laser_af_offsets: Dict[str, float] = field(default_factory=dict)
+
+    # Schema v2: the FOV pattern that generated this run's regions, carried to the
+    # acquisition-record writer so an API/service run round-trips faithfully. None =>
+    # coverage-from-scan-size (the GUI Select-Wells default).
+    fov_pattern: Optional[dict] = None
+
 
 @dataclass
 class OverallProgressUpdate: