From c230268daca223e71bf51af1c6f02bc8cbb1649c Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 22:35:31 -0700 Subject: [PATCH 01/19] feat: port per-region laser-AF offset core hook from PR #562 Non-GUI subset of PR #562 (per-region-laser-af-offset @ 9d33d549): AcquisitionParameters.region_laser_af_offsets, the controller setter + build_params threading + consume-and-clear, the worker's anchor-at- reference + apply_relative_offset_um consumption, and the open-loop offset method on LaserAutofocusController. Schema-v2's well_z_offsets_um plumbs into this hook. If #562 merges first, this commit becomes a no-op on rebase and should be dropped. Co-Authored-By: Claude Fable 5 --- .../core/laser_auto_focus_controller.py | 13 +++++++++++ .../control/core/multi_point_controller.py | 23 +++++++++++++++++-- software/control/core/multi_point_utils.py | 7 +++++- software/control/core/multi_point_worker.py | 11 ++++++++- 4 files changed, 50 insertions(+), 4 deletions(-) diff --git a/software/control/core/laser_auto_focus_controller.py b/software/control/core/laser_auto_focus_controller.py index c18dc5a31..88ed56aff 100644 --- a/software/control/core/laser_auto_focus_controller.py +++ b/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. diff --git a/software/control/core/multi_point_controller.py b/software/control/core/multi_point_controller.py index 9f117d989..230fd4d8b 100644 --- a/software/control/core/multi_point_controller.py +++ b/software/control/core/multi_point_controller.py @@ -227,6 +227,7 @@ def __init__( self.overlap_percent = 10.0 # FOV overlap percentage self.focus_map = None + self.region_laser_af_offsets = {} self.gen_focus_map = False self.focus_map_storage = [] self.already_using_fmap = False @@ -416,6 +417,11 @@ 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_base_path(self, path): self.base_path = path @@ -675,6 +681,13 @@ 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 = {} if not self.validate_acquisition_settings(): # emit acquisition finished signal to re-enable the UI self.callbacks.signal_acquisition_finished() @@ -842,7 +855,10 @@ 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, + ) # Gather objective and camera info for YAML current_objective = self.objectiveStore.current_objective @@ -946,7 +962,9 @@ 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 + ) -> AcquisitionParameters: # Determine plate dimensions from wellplate format if available plate_num_rows = 8 # Default for 96-well plate_num_cols = 12 @@ -986,6 +1004,7 @@ 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 {}, ) def _on_acquisition_completed(self): diff --git a/software/control/core/multi_point_utils.py b/software/control/core/multi_point_utils.py index 11157e1f9..79dbbb68c 100644 --- a/software/control/core/multi_point_utils.py +++ b/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,11 @@ 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) + @dataclass class OverallProgressUpdate: diff --git a/software/control/core/multi_point_worker.py b/software/control/core/multi_point_worker.py index 7fd973452..0e6990fd4 100644 --- a/software/control/core/multi_point_worker.py +++ b/software/control/core/multi_point_worker.py @@ -124,6 +124,7 @@ def __init__( self.do_reflection_af = acquisition_parameters.do_reflection_autofocus self.use_piezo = acquisition_parameters.use_piezo self.apply_channel_offset = acquisition_parameters.apply_channel_offset + self.region_laser_af_offsets = acquisition_parameters.region_laser_af_offsets self.display_resolution_scaling = acquisition_parameters.display_resolution_scaling self.experiment_ID = acquisition_parameters.experiment_ID @@ -1204,7 +1205,15 @@ def perform_autofocus(self, region_id, fov): # value, NOT by raising — both paths must mark the FOV's AF as failed or # the per-channel z-offset gate would apply offsets from an unanchored z. try: - af_succeeded = self.laser_auto_focus_controller.move_to_target(0) + target_um = self.region_laser_af_offsets.get(region_id, 0.0) + # Anchor closed-loop at the reference (displacement 0), where spot-alignment + # verification is valid, then apply the per-region offset as an OPEN-LOOP + # relative move. Driving move_to_target() straight to a nonzero displacement + # would always fail verification (its crop is fixed at x_reference) and revert. + af_succeeded = self.laser_auto_focus_controller.move_to_target(0.0) + if af_succeeded and target_um: + self._log.info(f"applying per-region laser AF offset for region '{region_id}': {target_um:+.2f} µm") + self.laser_auto_focus_controller.apply_relative_offset_um(target_um) except Exception as e: file_ID = f"{region_id}_focus_camera.bmp" saving_path = os.path.join(self.base_path, self.experiment_ID, str(self.time_point), file_ID) From 501987a59df07bf5262296dd31f633b2c9e40276 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:01:43 -0700 Subject: [PATCH 02/19] feat: parse fov_pattern, well_z_offsets_um, z_plan in acquisition yaml (schema v2) Co-Authored-By: Claude Fable 5 --- software/control/acquisition_yaml_loader.py | 113 ++++++++++++++++++ .../control/test_yaml_loader_schema_v2.py | 112 +++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 software/tests/control/test_yaml_loader_schema_v2.py diff --git a/software/control/acquisition_yaml_loader.py b/software/control/acquisition_yaml_loader.py index d8b80aebe..3fcd7eaa8 100644 --- a/software/control/acquisition_yaml_loader.py +++ b/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,98 @@ 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]) + 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 +231,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 +264,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), diff --git a/software/tests/control/test_yaml_loader_schema_v2.py b/software/tests/control/test_yaml_loader_schema_v2.py new file mode 100644 index 000000000..ae057ce43 --- /dev/null +++ b/software/tests/control/test_yaml_loader_schema_v2.py @@ -0,0 +1,112 @@ +import pytest +import yaml + +from control.acquisition_yaml_loader import parse_acquisition_yaml + + +def _write(tmp_path, wellplate_scan_extra, autofocus=None): + config = { + "acquisition": {"widget_type": "wellplate"}, + "sample": {"wellplate_format": "96 well plate"}, + "z_stack": {"nz": 1, "delta_z_mm": 0.001}, + "time_series": {"nt": 1, "delta_t_s": 0.0}, + "channels": [{"name": "BF LED matrix full"}], + "autofocus": autofocus or {"contrast_af": False, "laser_af": False}, + "wellplate_scan": {"wells": "A1:A2", "overlap_percent": 10, **wellplate_scan_extra}, + } + p = tmp_path / "acq.yaml" + p.write_text(yaml.safe_dump(config)) + return str(p) + + +def test_no_pattern_is_none(tmp_path): + data = parse_acquisition_yaml(_write(tmp_path, {"scan_size_mm": 0.5})) + assert data.fov_pattern is None + assert data.well_z_offsets_um is None + assert data.z_plan is None + + +def test_centered_grid_normalized(tmp_path): + data = parse_acquisition_yaml(_write(tmp_path, {"fov_pattern": {"type": "centered_grid", "nx": 3, "ny": 2}})) + assert data.fov_pattern == {"type": "centered_grid", "nx": 3, "ny": 2, "overlap_percent": 10.0} + + +def test_grid_subset_tile_bounds(tmp_path): + with pytest.raises(ValueError, match="tile"): + parse_acquisition_yaml( + _write(tmp_path, {"fov_pattern": {"type": "grid_subset", "nx": 2, "ny": 2, "tiles": [[0, 0], [2, 0]]}}) + ) + + +def test_grid_subset_ok(tmp_path): + data = parse_acquisition_yaml( + _write(tmp_path, {"fov_pattern": {"type": "grid_subset", "nx": 2, "ny": 2, "tiles": [[0, 0], [1, 1]]}}) + ) + assert data.fov_pattern["tiles"] == [[0, 0], [1, 1]] + + +def test_random_requires_positive_n(tmp_path): + with pytest.raises(ValueError, match="n_fovs"): + parse_acquisition_yaml(_write(tmp_path, {"fov_pattern": {"type": "random", "n_fovs": 0}})) + + +def test_unknown_pattern_type(tmp_path): + with pytest.raises(ValueError, match="fov_pattern"): + parse_acquisition_yaml(_write(tmp_path, {"fov_pattern": {"type": "spiral"}})) + + +def test_pattern_requires_wells(tmp_path): + config_extra = {"fov_pattern": {"type": "centered_grid", "nx": 2, "ny": 2}} + p = tmp_path / "acq2.yaml" + p.write_text( + yaml.safe_dump( + { + "acquisition": {"widget_type": "wellplate"}, + "z_stack": {"nz": 1, "delta_z_mm": 0.001}, + "channels": [{"name": "BF LED matrix full"}], + "wellplate_scan": { + "regions": [{"name": "A1", "center_mm": [14.3, 11.36, 0.5], "shape": "Square"}], + **config_extra, + }, + } + ) + ) + with pytest.raises(ValueError, match="wells"): + parse_acquisition_yaml(str(p)) + + +def test_well_z_offsets_validation(tmp_path): + data = parse_acquisition_yaml( + _write(tmp_path, {"scan_size_mm": 0.5, "well_z_offsets_um": {"A1": 3.0, "default": -1.5}}) + ) + assert data.well_z_offsets_um == {"A1": 3.0, "default": -1.5} + with pytest.raises(ValueError, match="well_z_offsets_um"): + parse_acquisition_yaml(_write(tmp_path, {"scan_size_mm": 0.5, "well_z_offsets_um": {"A1": float("nan")}})) + + +def test_z_plan_points_exactly_three(tmp_path): + with pytest.raises(ValueError, match="z_plan"): + parse_acquisition_yaml( + _write(tmp_path, {"scan_size_mm": 0.5, "z_plan": {"type": "focus_map", "points": [[0, 0, 1]]}}) + ) + data = parse_acquisition_yaml( + _write( + tmp_path, + {"scan_size_mm": 0.5, "z_plan": {"type": "focus_map", "points": [[0, 0, 1.0], [10, 0, 1.1], [0, 10, 1.2]]}}, + ) + ) + assert data.z_plan["points"] == [[0.0, 0.0, 1.0], [10.0, 0.0, 1.1], [0.0, 10.0, 1.2]] + data = parse_acquisition_yaml( + _write(tmp_path, {"scan_size_mm": 0.5, "z_plan": {"type": "focus_map", "generate": True}}) + ) + assert data.z_plan == {"type": "focus_map", "generate": True, "points": None} + with pytest.raises(ValueError, match="z_plan"): + parse_acquisition_yaml( + _write( + tmp_path, + { + "scan_size_mm": 0.5, + "z_plan": {"type": "focus_map", "generate": True, "points": [[0, 0, 1], [1, 0, 1], [0, 1, 1]]}, + }, + ) + ) From 2f4c57a82a84b5dd12b67462dd59bb139c16efab Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:08:27 -0700 Subject: [PATCH 03/19] feat: centered_grid fov_pattern dispatch in service region configuration Co-Authored-By: Claude Fable 5 --- software/squid_service/service.py | 53 +++++++++--- .../control/test_service_fov_patterns.py | 85 +++++++++++++++++++ 2 files changed, 128 insertions(+), 10 deletions(-) create mode 100644 software/tests/control/test_service_fov_patterns.py diff --git a/software/squid_service/service.py b/software/squid_service/service.py index 3e3c3b9ea..f1794d61e 100644 --- a/software/squid_service/service.py +++ b/software/squid_service/service.py @@ -1304,18 +1304,10 @@ def _configure_regions(self, yaml_data, raw: dict, wells_override, sample_format if effective_wells: fmt = sample_format_override or raw.get("sample", {}).get("wellplate_format", "96 well plate") settings = control._def.get_wellplate_settings(fmt) + pattern = yaml_data.fov_pattern for name in parse_well_names(effective_wells): x, y = well_center_mm(name, settings) - sc.add_region( - well_id=name, - center_x=x, - center_y=y, - scan_size_mm=scan_size, - overlap_percent=yaml_data.overlap_percent, - shape=shape, - ) - if name in sc.region_centers: - sc.region_centers[name][2] = z0 + self._add_pattern_region(sc, pattern, name, x, y, z0, yaml_data) else: for region in yaml_data.wellplate_regions: name = region.get("name", "region") @@ -1332,6 +1324,47 @@ def _configure_regions(self, yaml_data, raw: dict, wells_override, sample_format sc.region_centers[name][2] = center[2] if len(center) > 2 else z0 sc.sort_coordinates() + def _add_pattern_region(self, sc, pattern, name, x, y, z0, yaml_data): + """Add one well's FOVs per the fov_pattern (None/coverage -> legacy add_region).""" + if pattern is None or pattern["type"] == "coverage": + scan_size = (pattern or {}).get("scan_size_mm") or yaml_data.scan_size_mm or 2.0 + shape = (pattern or {}).get("shape") or yaml_data.scan_shape or "Square" + overlap = (pattern or {}).get("overlap_percent", yaml_data.overlap_percent) + sc.add_region( + well_id=name, + center_x=x, + center_y=y, + scan_size_mm=scan_size, + overlap_percent=overlap, + shape=shape, + ) + elif pattern["type"] == "centered_grid": + sc.add_flexible_region( + region_id=name, + center_x=x, + center_y=y, + center_z=z0, + Nx=pattern["nx"], + Ny=pattern["ny"], + overlap_percent=pattern["overlap_percent"], + ) + elif pattern["type"] == "grid_subset": + self._add_grid_subset_region(sc, pattern, name, x, y, z0) # Task 3 + elif pattern["type"] == "random": + self._add_random_region(sc, pattern, name, x, y, z0, yaml_data) # Task 4 + if name in sc.region_centers: + sc.region_centers[name][2] = z0 + + def _add_grid_subset_region(self, sc, pattern, name, x, y, z0): + # Replaced by Task 3; unreachable from coverage/centered_grid paths and the + # loader requires wells for per-well patterns, so no yaml can trigger this yet. + raise NotImplementedError("grid_subset fov_pattern is not implemented yet (Task 3)") + + def _add_random_region(self, sc, pattern, name, x, y, z0, yaml_data): + # Replaced by Task 4; unreachable from coverage/centered_grid paths and the + # loader requires wells for per-well patterns, so no yaml can trigger this yet. + raise NotImplementedError("random fov_pattern is not implemented yet (Task 4)") + def _configure_grid_regions(self, grid, z0: float) -> None: import control._def diff --git a/software/tests/control/test_service_fov_patterns.py b/software/tests/control/test_service_fov_patterns.py new file mode 100644 index 000000000..bc4610574 --- /dev/null +++ b/software/tests/control/test_service_fov_patterns.py @@ -0,0 +1,85 @@ +import pytest +import yaml + +import control.microscope +import tests.control.test_stubs as ts +from squid_service.models import AcquisitionRequest +from squid_service.service import SquidCoreService + + +@pytest.fixture(scope="module") +def sim_scope(): + scope = control.microscope.Microscope.build_from_global_config(True) + yield scope + scope.close() + + +@pytest.fixture() +def service(sim_scope, tmp_path): + mpc = ts.get_test_multi_point_controller(sim_scope) + return SquidCoreService( + microscope=sim_scope, + multipoint_controller=mpc, + scan_coordinates=mpc.scanCoordinates, + simulation=True, + job_persist_path=tmp_path / "last_job.json", + methods_dir=tmp_path / "methods", + ) + + +def _write_yaml(tmp_path, sim_scope, wellplate_scan): + objective = sim_scope.objective_store.current_objective + channel = sim_scope.live_controller.get_channels(objective)[0].name + config = { + "acquisition": {"widget_type": "wellplate"}, + "sample": {"wellplate_format": "96 well plate"}, + "z_stack": {"nz": 1, "delta_z_mm": 0.001}, + "time_series": {"nt": 1, "delta_t_s": 0.0}, + "channels": [{"name": channel}], + "autofocus": {"contrast_af": False, "laser_af": False}, + "wellplate_scan": {"wells": "A1:A2", "overlap_percent": 10, **wellplate_scan}, + } + path = tmp_path / "acquisition.yaml" + path.write_text(yaml.safe_dump(config)) + return str(path) + + +def _preflight_and_configure(service, yaml_path, tmp_path): + req = AcquisitionRequest(yaml_path=yaml_path, overrides={"output_path": str(tmp_path)}) + assert service.preflight(req)["ok"] is True + # drive region configuration exactly as start_acquisition does, without running + from control.acquisition_yaml_loader import parse_acquisition_yaml + import yaml as _y + + yaml_data = parse_acquisition_yaml(yaml_path) + raw = _y.safe_load(open(yaml_path)) + z0 = service._microscope.stage.get_pos().z_mm + service._configure_regions(yaml_data, raw, None, None, z0) + return service._scan_coordinates + + +def test_centered_grid_counts_and_rowmajor_order(service, sim_scope, tmp_path): + sc = _preflight_and_configure( + service, + _write_yaml(tmp_path, sim_scope, {"fov_pattern": {"type": "centered_grid", "nx": 3, "ny": 2}}), + tmp_path, + ) + for well in ("A1", "A2"): + coords = sc.region_fov_coordinates[well] + assert len(coords) == 6 + # row-major: y non-decreasing, x increasing within each row + ys = [c[1] for c in coords] + assert ys == sorted(ys) + row0 = coords[0:3] + assert [c[0] for c in row0] == sorted(c[0] for c in row0) + # same relative offsets in every well + a1 = sc.region_fov_coordinates["A1"] + a2 = sc.region_fov_coordinates["A2"] + rel1 = [(round(x - a1[0][0], 6), round(y - a1[0][1], 6)) for x, y, *_ in a1] + rel2 = [(round(x - a2[0][0], 6), round(y - a2[0][1], 6)) for x, y, *_ in a2] + assert rel1 == rel2 + + +def test_coverage_unchanged(service, sim_scope, tmp_path): + sc = _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"scan_size_mm": 0.5}), tmp_path) + assert set(sc.region_fov_coordinates.keys()) == {"A1", "A2"} From 9b89b32aac1cf516f744e93494bbe237c0e07127 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:14:09 -0700 Subject: [PATCH 04/19] feat: grid_subset fov_pattern with deterministic row-major tile indices Co-Authored-By: Claude Fable 5 --- software/squid_service/service.py | 38 +++++++++++++++++-- .../control/test_service_fov_patterns.py | 29 ++++++++++++++ 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/software/squid_service/service.py b/software/squid_service/service.py index f1794d61e..cf759b6fb 100644 --- a/software/squid_service/service.py +++ b/software/squid_service/service.py @@ -1356,9 +1356,41 @@ def _add_pattern_region(self, sc, pattern, name, x, y, z0, yaml_data): sc.region_centers[name][2] = z0 def _add_grid_subset_region(self, sc, pattern, name, x, y, z0): - # Replaced by Task 3; unreachable from coverage/centered_grid paths and the - # loader requires wells for per-well patterns, so no yaml can trigger this yet. - raise NotImplementedError("grid_subset fov_pattern is not implemented yet (Task 3)") + """Generate the full nx x ny centered grid row-major, then keep only `tiles`. + + Flat index convention i = row*nx + col matches the plate-view mosaic + (widgets_mosaic.py) and requires Unidirectional generation: S-Pattern + reverses odd rows, so force it and restore afterwards. + """ + saved_pattern = sc.fov_pattern + sc.fov_pattern = "Unidirectional" + try: + sc.add_flexible_region( + region_id=name, + center_x=x, + center_y=y, + center_z=z0, + Nx=pattern["nx"], + Ny=pattern["ny"], + overlap_percent=pattern["overlap_percent"], + ) + finally: + sc.fov_pattern = saved_pattern + coords = sc.region_fov_coordinates.get(name, []) + expected = pattern["nx"] * pattern["ny"] + if len(coords) != expected: + # stage-limit clipping dropped tiles; flat indices would be wrong + raise F.FaultError( + F.make_fault( + F.FaultCategory.INVALID_PARAM, + F.INVALID_PARAM_OUT_OF_RANGE, + f"grid_subset for well {name}: grid clipped by stage limits " + f"({len(coords)}/{expected} FOVs reachable); move the region or shrink the grid", + detail={"well": name}, + ) + ) + keep = [row * pattern["nx"] + col for row, col in pattern["tiles"]] + sc.region_fov_coordinates[name] = [coords[i] for i in sorted(keep)] def _add_random_region(self, sc, pattern, name, x, y, z0, yaml_data): # Replaced by Task 4; unreachable from coverage/centered_grid paths and the diff --git a/software/tests/control/test_service_fov_patterns.py b/software/tests/control/test_service_fov_patterns.py index bc4610574..9c6c6ca3f 100644 --- a/software/tests/control/test_service_fov_patterns.py +++ b/software/tests/control/test_service_fov_patterns.py @@ -83,3 +83,32 @@ def test_centered_grid_counts_and_rowmajor_order(service, sim_scope, tmp_path): def test_coverage_unchanged(service, sim_scope, tmp_path): sc = _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"scan_size_mm": 0.5}), tmp_path) assert set(sc.region_fov_coordinates.keys()) == {"A1", "A2"} + + +def test_grid_subset_filters_rowmajor_tiles(service, sim_scope, tmp_path): + pattern = {"type": "grid_subset", "nx": 3, "ny": 2, "tiles": [[0, 0], [1, 2]]} + sc = _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"fov_pattern": pattern}), tmp_path) + for well in ("A1", "A2"): + assert len(sc.region_fov_coordinates[well]) == 2 + # tile [0,0] is the grid's min-x/min-y corner; [1,2] is max-x of row 1 + a1 = sc.region_fov_coordinates["A1"] + assert a1[0][0] < a1[1][0] and a1[0][1] < a1[1][1] + # identical relative geometry across wells + a2 = sc.region_fov_coordinates["A2"] + rel1 = [(round(x - a1[0][0], 6), round(y - a1[0][1], 6)) for x, y, *_ in a1] + rel2 = [(round(x - a2[0][0], 6), round(y - a2[0][1], 6)) for x, y, *_ in a2] + assert rel1 == rel2 + + +def test_grid_subset_forces_unidirectional_and_restores(service, sim_scope, tmp_path): + sc = service._scan_coordinates + original = sc.fov_pattern + sc.fov_pattern = "S-Pattern" + try: + pattern = {"type": "grid_subset", "nx": 2, "ny": 2, "tiles": [[1, 0]]} + _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"fov_pattern": pattern}), tmp_path) + a1 = service._scan_coordinates.region_fov_coordinates["A1"] + assert len(a1) == 1 + assert service._scan_coordinates.fov_pattern == "S-Pattern" # restored + finally: + sc.fov_pattern = original From ef1ce434a0c56a7504b54e46cafa69722f03d6b1 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:20:39 -0700 Subject: [PATCH 05/19] feat: random fov_pattern with reproducible per-well seeding Co-Authored-By: Claude Fable 5 --- software/squid_service/service.py | 41 +++++++++++++++++-- .../control/test_service_fov_patterns.py | 26 ++++++++++++ 2 files changed, 64 insertions(+), 3 deletions(-) diff --git a/software/squid_service/service.py b/software/squid_service/service.py index cf759b6fb..eb4fbe5a5 100644 --- a/software/squid_service/service.py +++ b/software/squid_service/service.py @@ -1304,6 +1304,7 @@ def _configure_regions(self, yaml_data, raw: dict, wells_override, sample_format if effective_wells: fmt = sample_format_override or raw.get("sample", {}).get("wellplate_format", "96 well plate") settings = control._def.get_wellplate_settings(fmt) + self._current_wellplate_settings = settings # consumed by _add_random_region pattern = yaml_data.fov_pattern for name in parse_well_names(effective_wells): x, y = well_center_mm(name, settings) @@ -1393,9 +1394,43 @@ def _add_grid_subset_region(self, sc, pattern, name, x, y, z0): sc.region_fov_coordinates[name] = [coords[i] for i in sorted(keep)] def _add_random_region(self, sc, pattern, name, x, y, z0, yaml_data): - # Replaced by Task 4; unreachable from coverage/centered_grid paths and the - # loader requires wells for per-well patterns, so no yaml can trigger this yet. - raise NotImplementedError("random fov_pattern is not implemented yet (Task 4)") + """Sample n_fovs points uniformly inside the well's usable circle. + + Per-well RNG: sha256(f"{seed}:{well}") so runs are reproducible for a given + seed but differ well-to-well (hash() is process-salted; never use it). + seed None -> os.urandom-backed nondeterministic sampling. + """ + import hashlib + import random as _random + + fov_mm = sc.objectiveStore.get_pixel_size_factor() * sc.camera.get_fov_size_mm() + fmt_settings = self._current_wellplate_settings # set by _configure_regions, see below + usable_radius = fmt_settings["well_size_mm"] / 2.0 - fov_mm / 2.0 + if usable_radius <= 0: + raise F.FaultError( + F.make_fault( + F.FaultCategory.INVALID_PARAM, + F.INVALID_PARAM_OUT_OF_RANGE, + f"random fov_pattern: FOV ({fov_mm:.3f} mm) does not fit in a " + f"{fmt_settings['well_size_mm']} mm well", + detail={"well": name}, + ) + ) + seed = pattern["seed"] + if seed is None: + rng = _random.Random() + else: + digest = hashlib.sha256(f"{seed}:{name}".encode()).digest() + rng = _random.Random(int.from_bytes(digest[:8], "big")) + coords = [] + while len(coords) < pattern["n_fovs"]: + px = rng.uniform(-usable_radius, usable_radius) + py = rng.uniform(-usable_radius, usable_radius) + if px * px + py * py <= usable_radius * usable_radius: + coords.append((x + px, y + py, z0)) + sc.region_centers[name] = [x, y, z0] + sc.region_shapes[name] = "Square" # keep region dicts key-consistent (sort/clear paths) + sc.region_fov_coordinates[name] = coords def _configure_grid_regions(self, grid, z0: float) -> None: import control._def diff --git a/software/tests/control/test_service_fov_patterns.py b/software/tests/control/test_service_fov_patterns.py index 9c6c6ca3f..504e898ee 100644 --- a/software/tests/control/test_service_fov_patterns.py +++ b/software/tests/control/test_service_fov_patterns.py @@ -112,3 +112,29 @@ def test_grid_subset_forces_unidirectional_and_restores(service, sim_scope, tmp_ assert service._scan_coordinates.fov_pattern == "S-Pattern" # restored finally: sc.fov_pattern = original + + +def test_random_seeded_reproducible_and_per_well_different(service, sim_scope, tmp_path): + pattern = {"type": "random", "n_fovs": 5, "seed": 42} + sc1 = _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"fov_pattern": pattern}), tmp_path) + a1_first = list(sc1.region_fov_coordinates["A1"]) + a2_first = list(sc1.region_fov_coordinates["A2"]) + assert len(a1_first) == 5 and len(a2_first) == 5 + assert a1_first != a2_first # independent per well + # reproducible: reconfigure, same coordinates + sc2 = _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"fov_pattern": pattern}), tmp_path) + assert list(sc2.region_fov_coordinates["A1"]) == a1_first + + +def test_random_within_well_bounds(service, sim_scope, tmp_path): + import control._def + + pattern = {"type": "random", "n_fovs": 8, "seed": 1} + sc = _preflight_and_configure(service, _write_yaml(tmp_path, sim_scope, {"fov_pattern": pattern}), tmp_path) + settings = control._def.get_wellplate_settings("96 well plate") + from squid_service.wells import well_center_mm + + cx, cy = well_center_mm("A1", settings) + radius = settings["well_size_mm"] / 2.0 + for x, y, *_ in sc.region_fov_coordinates["A1"]: + assert ((x - cx) ** 2 + (y - cy) ** 2) ** 0.5 <= radius + 1e-9 From 2a73ee4e5853f198427728b21587e89ece3908fa Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:31:30 -0700 Subject: [PATCH 06/19] feat: plumb well_z_offsets_um into the per-region laser-AF offset hook with preflight validation Resolve per-well laser-AF Z offsets (listed value, else "default", zeros omitted) and push them into MultiPointController.set_region_laser_af_offsets on the yaml start path after regions are configured. Add a preflight check that offsets require effective laser AF (INVALID_PARAM) and that every offset (including "default") is within the laser AF range (INVALID_PARAM_OUT_OF_RANGE). Co-Authored-By: Claude Fable 5 --- software/squid_service/service.py | 62 +++++++++++++++ .../control/test_service_well_z_offsets.py | 75 +++++++++++++++++++ 2 files changed, 137 insertions(+) create mode 100644 software/tests/control/test_service_well_z_offsets.py diff --git a/software/squid_service/service.py b/software/squid_service/service.py index eb4fbe5a5..e52a90430 100644 --- a/software/squid_service/service.py +++ b/software/squid_service/service.py @@ -1111,6 +1111,42 @@ def check_regions(): ) ) + def check_well_z_offsets(): + offsets = ctx["yaml_data"].well_z_offsets_um + if not offsets: + return + # Offsets are displacements from the laser-AF reference plane, so laser AF must run. + # Read the effective flag the same way the z_reference check does (yaml base with + # request overrides applied); here "reflection" is the laser-AF flag. + laser_af, _contrast = self._effective_af_flags(req, ctx["yaml_data"].laser_af, ctx["yaml_data"].contrast_af) + if not laser_af: + raise F.FaultError( + F.make_fault( + F.FaultCategory.INVALID_PARAM, + F.INVALID_PARAM_BAD_VALUE, + "well_z_offsets_um requires autofocus.laser_af " + "(offsets are relative to the laser-AF reference plane)", + component="well_z_offsets_um", + ) + ) + # Range-check every entry (including "default") against the laser-AF search range, + # when a laser-AF controller with properties is attached (may be absent in sim). + controller = getattr(self._mpc, "laserAutoFocusController", None) + af_range = getattr(getattr(controller, "laser_af_properties", None), "laser_af_range", None) + if af_range: + for well, um in offsets.items(): + if abs(um) > af_range: + raise F.FaultError( + F.make_fault( + F.FaultCategory.INVALID_PARAM, + F.INVALID_PARAM_OUT_OF_RANGE, + f"well_z_offsets_um[{well!r}] = {um:+.1f} um exceeds " + f"laser AF range +/-{af_range} um", + component="well_z_offsets_um", + detail={"well": well, "offset_um": um, "laser_af_range_um": af_range}, + ) + ) + checks = [ ("yaml", check_yaml), ("widget_type", check_widget_type), @@ -1123,6 +1159,7 @@ def check_regions(): "z_reference", self._z_reference_check(req, lambda: (ctx["yaml_data"].laser_af, ctx["yaml_data"].contrast_af)), ), + ("well_z_offsets", check_well_z_offsets), ("output_path", self._output_path_check(req, ctx)), ] return checks, ctx @@ -1290,6 +1327,23 @@ def _resolve_z_reference(self, req: AcquisitionRequest) -> float: # "current" and "autofocus" both baseline on the current stage z. return self._microscope.stage.get_pos().z_mm + def _resolve_well_z_offsets(self, well_z_offsets_um, well_names): + """Map each configured well to its laser-AF target offset (µm from the reference plane). + + Precedence per well: its listed value, else the "default" value if present. Zero and + absent offsets are omitted so the pushed dict only carries wells that actually deviate + (an empty dict means every FOV targets the reference plane -- current behavior). + """ + if not well_z_offsets_um: + return {} + default = well_z_offsets_um.get("default") + out = {} + for name in well_names: + value = well_z_offsets_um.get(name, default) + if value: + out[name] = float(value) + return out + # -- controller configuration -- def _configure_regions(self, yaml_data, raw: dict, wells_override, sample_format_override, z0: float) -> None: @@ -1578,6 +1632,14 @@ def start_acquisition(self, req: AcquisitionRequest) -> dict: else: yaml_data, raw = ctx["yaml_data"], ctx["raw"] self._configure_regions(yaml_data, raw, req.overrides.wells, req.overrides.sample_format, z0) + # Resolve per-well laser-AF offsets against the regions just configured and push + # them into the controller (consumed-and-cleared by run_acquisition, so no leak + # to later runs). Empty dict => every FOV targets the reference plane. + self._mpc.set_region_laser_af_offsets( + self._resolve_well_z_offsets( + yaml_data.well_z_offsets_um, list(self._scan_coordinates.region_centers.keys()) + ) + ) self._configure_controller(yaml_data, z0) channel_count = len(yaml_data.channel_names) nz, nt = yaml_data.nz, yaml_data.nt diff --git a/software/tests/control/test_service_well_z_offsets.py b/software/tests/control/test_service_well_z_offsets.py new file mode 100644 index 000000000..120db6e19 --- /dev/null +++ b/software/tests/control/test_service_well_z_offsets.py @@ -0,0 +1,75 @@ +import pytest +import yaml + +import control.microscope +import tests.control.test_stubs as ts +from squid_service.models import AcquisitionRequest +from squid_service.service import SquidCoreService + + +@pytest.fixture(scope="module") +def sim_scope(): + scope = control.microscope.Microscope.build_from_global_config(True) + yield scope + scope.close() + + +@pytest.fixture() +def service(sim_scope, tmp_path): + mpc = ts.get_test_multi_point_controller(sim_scope) + return SquidCoreService( + microscope=sim_scope, + multipoint_controller=mpc, + scan_coordinates=mpc.scanCoordinates, + simulation=True, + job_persist_path=tmp_path / "last_job.json", + methods_dir=tmp_path / "methods", + ) + + +def _write_yaml(tmp_path, sim_scope, offsets, laser_af): + objective = sim_scope.objective_store.current_objective + channel = sim_scope.live_controller.get_channels(objective)[0].name + config = { + "acquisition": {"widget_type": "wellplate"}, + "sample": {"wellplate_format": "96 well plate"}, + "z_stack": {"nz": 1, "delta_z_mm": 0.001}, + "time_series": {"nt": 1, "delta_t_s": 0.0}, + "channels": [{"name": channel}], + "autofocus": {"contrast_af": False, "laser_af": laser_af}, + "wellplate_scan": { + "wells": "A1:A2", + "scan_size_mm": 0.5, + "overlap_percent": 10, + "well_z_offsets_um": offsets, + }, + } + p = tmp_path / "acq.yaml" + p.write_text(yaml.safe_dump(config)) + return str(p) + + +def test_offsets_require_laser_af(service, sim_scope, tmp_path): + req = AcquisitionRequest( + yaml_path=_write_yaml(tmp_path, sim_scope, {"A1": 3.0}, laser_af=False), + overrides={"output_path": str(tmp_path)}, + ) + result = service.preflight(req) + assert result["ok"] is False + # Preflight check results carry the human-readable text under "message" + # (see SquidCoreService._run_checks_report), not "error". + assert any("well_z_offsets_um" in (c.get("message") or "") for c in result["checks"] if not c["ok"]) + + +def test_offsets_resolved_with_default(service, sim_scope, tmp_path): + yaml_path = _write_yaml(tmp_path, sim_scope, {"A1": 3.0, "default": -1.5}, laser_af=True) + from control.acquisition_yaml_loader import parse_acquisition_yaml + + yaml_data = parse_acquisition_yaml(yaml_path) + resolved = service._resolve_well_z_offsets(yaml_data.well_z_offsets_um, ["A1", "A2"]) + assert resolved == {"A1": 3.0, "A2": -1.5} + + +def test_offsets_default_zero_omitted(service, sim_scope, tmp_path): + resolved = service._resolve_well_z_offsets({"A1": 3.0}, ["A1", "A2"]) + assert resolved == {"A1": 3.0} From 197a27ae9f8bbd823d1e8bffa16650b52ff714e5 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:39:07 -0700 Subject: [PATCH 07/19] test: pin explicit-zero omission and laser_af_range faults for well_z_offsets_um Co-Authored-By: Claude Fable 5 --- .../control/test_service_well_z_offsets.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/software/tests/control/test_service_well_z_offsets.py b/software/tests/control/test_service_well_z_offsets.py index 120db6e19..d4958cad3 100644 --- a/software/tests/control/test_service_well_z_offsets.py +++ b/software/tests/control/test_service_well_z_offsets.py @@ -73,3 +73,38 @@ def test_offsets_resolved_with_default(service, sim_scope, tmp_path): def test_offsets_default_zero_omitted(service, sim_scope, tmp_path): resolved = service._resolve_well_z_offsets({"A1": 3.0}, ["A1", "A2"]) assert resolved == {"A1": 3.0} + + +def test_offsets_explicit_zero_omitted(service): + # An explicit 0.0 for a listed well is omitted (that well simply targets the + # reference plane), while a genuinely deviating well is kept. + assert service._resolve_well_z_offsets({"A1": 0.0, "A2": 3.0}, ["A1", "A2"]) == {"A2": 3.0} + # A zero "default" is likewise omitted for the wells that fall back to it (A2 here). + assert service._resolve_well_z_offsets({"A1": 3.0, "default": 0}, ["A1", "A2"]) == {"A1": 3.0} + + +def _laser_af_range(service): + """The sim laser-AF search range read the same way check_well_z_offsets reads it.""" + controller = getattr(service._mpc, "laserAutoFocusController", None) + return getattr(getattr(controller, "laser_af_properties", None), "laser_af_range", None) + + +@pytest.mark.parametrize("key", ["A1", "default"]) +def test_offsets_out_of_range_fails(service, sim_scope, tmp_path, key): + # Pin that an offset larger than the laser-AF range is a preflight fault, for both a + # per-well key and the authorized-deviation "default" key (both are range-checked). + af_range = _laser_af_range(service) + if not af_range: + pytest.skip("sim laser AF controller exposes no laser_af_range") + req = AcquisitionRequest( + yaml_path=_write_yaml(tmp_path, sim_scope, {key: af_range * 2}, laser_af=True), + overrides={"output_path": str(tmp_path)}, + ) + result = service.preflight(req) + assert result["ok"] is False + # The failing check names the field and reports the range violation (OUT_OF_RANGE fault). + assert any( + "well_z_offsets_um" in (c.get("message") or "") and "range" in (c.get("message") or "").lower() + for c in result["checks"] + if not c["ok"] + ) From b6036c554d137fed23683038793779865a281059 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Fri, 10 Jul 2026 23:59:17 -0700 Subject: [PATCH 08/19] feat: z_plan tilted-plate pre-AF positioning (plane points + gen-with-laser-AF) Co-Authored-By: Claude Fable 5 --- .../control/core/multi_point_controller.py | 20 ++++- software/squid_service/service.py | 45 ++++++++++- software/tests/control/test_service_z_plan.py | 74 +++++++++++++++++++ 3 files changed, 134 insertions(+), 5 deletions(-) create mode 100644 software/tests/control/test_service_z_plan.py diff --git a/software/control/core/multi_point_controller.py b/software/control/core/multi_point_controller.py index 230fd4d8b..5497508f5 100644 --- a/software/control/core/multi_point_controller.py +++ b/software/control/core/multi_point_controller.py @@ -779,7 +779,7 @@ 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: @@ -836,7 +836,23 @@ 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) diff --git a/software/squid_service/service.py b/software/squid_service/service.py index e52a90430..41005413c 100644 --- a/software/squid_service/service.py +++ b/software/squid_service/service.py @@ -1505,7 +1505,9 @@ def _configure_grid_regions(self, grid, z0: float) -> None: ) sc.sort_coordinates() - def _reset_z_range_and_focus_map(self, z0: float, nz: int, delta_z_um: float) -> None: + def _reset_z_range_and_focus_map( + self, z0: float, nz: int, delta_z_um: float, keep_gen_focus_map: bool = False + ) -> None: """Set z_range from the resolved baseline z0 and clear focus-map state. Mirrors the GUI's pre-run path (widgets.py toggle_acquisition ~6472-6487): the @@ -1523,12 +1525,37 @@ def _reset_z_range_and_focus_map(self, z0: float, nz: int, delta_z_um: float) -> set_focus_map(None) before starting; the API exposes no focus-map option, so we clear focus_map/gen_focus_map/use_manual_focus_map outright to prevent a stale focus map from a GUI session from bleeding into an API acquisition. + + keep_gen_focus_map preserves an already-set gen_focus_map flag (set by the + z_plan `generate: true` path just before controller configuration); every + other caller keeps today's clear-everything behavior. """ self._mpc.set_z_range(z0, z0 + delta_z_um / 1000.0 * (nz - 1)) self._mpc.set_focus_map(None) - self._mpc.gen_focus_map = False + if not keep_gen_focus_map: + self._mpc.gen_focus_map = False self._mpc.use_manual_focus_map = False + def _apply_z_plan_points(self, points): + """Bake plane z into every configured FOV (pre-AF positioning for tilted plates). + + Mirrors the controller's focus_map coordinate-baking loop + (multi_point_controller.py:770-780): the worker then physically moves to each + FOV's z before autofocus runs, so laser AF starts within capture range. + """ + from control.utils import interpolate_plane + + p1, p2, p3 = (tuple(p) for p in points) + sc = self._scan_coordinates + for region_id, coords in sc.region_fov_coordinates.items(): + for i, c in enumerate(coords): + x, y = c[0], c[1] + z = interpolate_plane(p1, p2, p3, (x, y)) + coords[i] = (x, y, z) + if region_id in sc.region_centers: + cx, cy = sc.region_centers[region_id][0], sc.region_centers[region_id][1] + sc.region_centers[region_id][2] = interpolate_plane(p1, p2, p3, (cx, cy)) + def _configure_controller(self, yaml_data, z0: float) -> None: self._mpc.set_NX(1) self._mpc.set_NY(1) @@ -1543,7 +1570,10 @@ def _configure_controller(self, yaml_data, z0: float) -> None: # TOP shifts (the API path previously left this at the controller default). self._mpc.z_stacking_config = yaml_data.z_stacking_config self._mpc.set_selected_configurations(yaml_data.channel_names) - self._reset_z_range_and_focus_map(z0, yaml_data.nz, yaml_data.delta_z_um) + # Preserve a gen_focus_map flag set by the z_plan `generate: true` path (wired + # in start_acquisition just before this call); all other paths clear it. + keep_gen_focus_map = bool(yaml_data.z_plan and yaml_data.z_plan["generate"]) + self._reset_z_range_and_focus_map(z0, yaml_data.nz, yaml_data.delta_z_um, keep_gen_focus_map=keep_gen_focus_map) def _configure_grid_controller(self, grid, z0: float) -> None: import control._def @@ -1640,6 +1670,15 @@ def start_acquisition(self, req: AcquisitionRequest) -> dict: yaml_data.well_z_offsets_um, list(self._scan_coordinates.region_centers.keys()) ) ) + # z_plan tilted-plate pre-AF positioning: `points` bakes the plane z into + # every FOV now (works with any AF mode); `generate` defers to the + # controller's 3-corner contrast-AF plane generation (flag survives the + # focus-map reset inside _configure_controller via keep_gen_focus_map). + if yaml_data.z_plan: + if yaml_data.z_plan["points"]: + self._apply_z_plan_points(yaml_data.z_plan["points"]) + else: # generate: true + self._mpc.set_gen_focus_map_flag(True) self._configure_controller(yaml_data, z0) channel_count = len(yaml_data.channel_names) nz, nt = yaml_data.nz, yaml_data.nt diff --git a/software/tests/control/test_service_z_plan.py b/software/tests/control/test_service_z_plan.py new file mode 100644 index 000000000..faad5a277 --- /dev/null +++ b/software/tests/control/test_service_z_plan.py @@ -0,0 +1,74 @@ +import pytest +import yaml + +import control.microscope +import tests.control.test_stubs as ts +from squid_service.service import SquidCoreService + + +@pytest.fixture(scope="module") +def sim_scope(): + scope = control.microscope.Microscope.build_from_global_config(True) + yield scope + scope.close() + + +@pytest.fixture() +def service(sim_scope, tmp_path): + mpc = ts.get_test_multi_point_controller(sim_scope) + return SquidCoreService( + microscope=sim_scope, + multipoint_controller=mpc, + scan_coordinates=mpc.scanCoordinates, + simulation=True, + job_persist_path=tmp_path / "last_job.json", + methods_dir=tmp_path / "methods", + ) + + +def test_z_plan_points_bakes_plane_into_coordinates(service, sim_scope, tmp_path): + objective = sim_scope.objective_store.current_objective + channel = sim_scope.live_controller.get_channels(objective)[0].name + # plane: z = 1.0 + 0.01*x (pure x tilt) + config = { + "acquisition": {"widget_type": "wellplate"}, + "sample": {"wellplate_format": "96 well plate"}, + "z_stack": {"nz": 1, "delta_z_mm": 0.001}, + "channels": [{"name": channel}], + "autofocus": {"contrast_af": False, "laser_af": False}, + "wellplate_scan": { + "wells": "A1,A3", + "scan_size_mm": 0.5, + "overlap_percent": 10, + "z_plan": { + "type": "focus_map", + "points": [[0.0, 0.0, 1.0], [100.0, 0.0, 2.0], [0.0, 100.0, 1.0]], + }, + }, + } + p = tmp_path / "acq.yaml" + p.write_text(yaml.safe_dump(config)) + from control.acquisition_yaml_loader import parse_acquisition_yaml + + yaml_data = parse_acquisition_yaml(str(p)) + raw = yaml.safe_load(open(str(p))) + z0 = sim_scope.stage.get_pos().z_mm + service._configure_regions(yaml_data, raw, None, None, z0) + service._apply_z_plan_points(yaml_data.z_plan["points"]) + sc = service._scan_coordinates + a1 = sc.region_fov_coordinates["A1"][0] + a3 = sc.region_fov_coordinates["A3"][0] + # A3 is right of A1 (larger x) -> larger z on this plane + assert a3[0] > a1[0] + assert a3[2] > a1[2] + expected_a1_z = 1.0 + (a1[0] / 100.0) * 1.0 + assert a1[2] == pytest.approx(expected_a1_z, abs=1e-6) + + +def test_gen_focus_map_flag_survives_reset_when_requested(service, sim_scope, tmp_path): + # Real signature: _reset_z_range_and_focus_map(z0, nz, delta_z_um, keep_gen_focus_map=False) + service._mpc.set_gen_focus_map_flag(True) + service._reset_z_range_and_focus_map(1.0, 1, 1.0, keep_gen_focus_map=True) + assert service._mpc.gen_focus_map is True + service._reset_z_range_and_focus_map(1.0, 1, 1.0) + assert service._mpc.gen_focus_map is False From de0d353f0346e66604e90c1d8a89464ff2585a21 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sat, 11 Jul 2026 00:14:57 -0700 Subject: [PATCH 09/19] test: pin gen-focus-map plane baking under reflection AF Add direct MultiPointController coverage for run_acquisition's reflection-AF plane-baking branch (gen_focus_map + do_reflection_af): the generated focus plane is baked into every FOV z instead of enabling the contrast focus-map. The 3-corner measurement and laser-AF closed-loop move are stubbed so the plane is known and the run is deterministic (no reliance on real sim contrast/laser AF). Asserts: every FOV becomes (x, y, plane_z(x)) within 1e-6, set_focus_map_use is never called with True in the laser-AF path, and the run completes without abort. Co-Authored-By: Claude Fable 5 --- .../test_gen_focus_map_reflection_af.py | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 software/tests/control/test_gen_focus_map_reflection_af.py diff --git a/software/tests/control/test_gen_focus_map_reflection_af.py b/software/tests/control/test_gen_focus_map_reflection_af.py new file mode 100644 index 000000000..a545e2838 --- /dev/null +++ b/software/tests/control/test_gen_focus_map_reflection_af.py @@ -0,0 +1,107 @@ +import dataclasses +import threading + +import pytest + +import control._def +import control.microscope +import tests.control.test_stubs as ts +from control.core.multi_point_controller import NoOpCallbacks + + +@pytest.fixture(scope="module") +def sim_scope(): + scope = control.microscope.Microscope.build_from_global_config(True) + yield scope + scope.close() + + +def test_gen_focus_map_with_reflection_af_bakes_plane_into_fovs(sim_scope, monkeypatch): + """gen_focus_map + reflection AF: run_acquisition must bake the generated focus + plane into every FOV coordinate (z = plane(x, y)) instead of enabling the contrast + focus-map. The contrast focus-map interpolation never runs in the worker's laser-AF + branch, so set_focus_map_use(True) must NOT be called; each FOV is pre-positioned on + the tilted plane instead. The 3-corner measurement is stubbed so the plane is known + and no real contrast/laser AF math executes (fully deterministic). + """ + control._def.MERGE_CHANNELS = False + control._def.SUPPORT_LASER_AUTOFOCUS = True + + started = threading.Event() + finished = threading.Event() + callbacks = dataclasses.replace( + NoOpCallbacks, + signal_acquisition_start=lambda *a, **kw: started.set(), + signal_acquisition_finished=lambda *a, **kw: finished.set(), + ) + mpc = ts.get_test_multi_point_controller(sim_scope, callbacks=callbacks) + + # One well A1 as a 2x2 grid -> four FOVs at distinct x positions on the tilted plane. + cfg = sim_scope.stage.get_config() + center_x = cfg.X_AXIS.MIN_POSITION + 1.0 + center_y = cfg.Y_AXIS.MIN_POSITION + 1.0 + center_z = (cfg.Z_AXIS.MAX_POSITION - cfg.Z_AXIS.MIN_POSITION) / 2.0 + mpc.scanCoordinates.clear_regions() + mpc.scanCoordinates.add_flexible_region("A1", center_x, center_y, center_z, 2, 2, 0) + + channel_names = [c.name for c in mpc.liveController.get_channels(mpc.objectiveStore.current_objective)] + mpc.set_selected_configurations(channel_names[0:1]) + + mpc.set_reflection_af_flag(True) + mpc.set_gen_focus_map_flag(True) + + # Known tilted plane: z = 2.0 + 0.01*x (pure x tilt). For the A1 FOV x values + # (~5.6-6.4 mm) the baked z (~2.06 mm) stays well within the sim Z limits [0.05, 6.0]. + def plane_z(x): + return 2.0 + 0.01 * x + + # Stub the 3-corner contrast-AF measurement: no stage moves / no real autofocus, + # just install a known non-colinear focus map that defines the plane above. + def fake_gen_focus_map(coord1, coord2, coord3): + mpc.autofocusController.focus_map_coords = [ + (0.0, 0.0, plane_z(0.0)), + (100.0, 0.0, plane_z(100.0)), + (0.0, 100.0, plane_z(0.0)), + ] + + monkeypatch.setattr(mpc.autofocusController, "gen_focus_map", fake_gen_focus_map) + + # Spy on set_focus_map_use: the reflection-AF path must never enable the contrast focus map. + focus_map_use_calls = [] + real_set_focus_map_use = mpc.autofocusController.set_focus_map_use + + def spy_set_focus_map_use(enable): + focus_map_use_calls.append(enable) + return real_set_focus_map_use(enable) + + monkeypatch.setattr(mpc.autofocusController, "set_focus_map_use", spy_set_focus_map_use) + + # Satisfy validate_acquisition_settings (laser AF requires a stored reference)... + sim_scope.addons.camera_focus.send_trigger() + ref_image = sim_scope.addons.camera_focus.read_frame() + assert ref_image is not None + mpc.laserAutoFocusController.laser_af_properties.set_reference_image(ref_image) + + # ...and neutralize the sim's flaky laser-AF closed-loop move so the run can't fail on it. + monkeypatch.setattr(mpc.laserAutoFocusController, "move_to_target", lambda *a, **kw: True) + + mpc.run_acquisition() + + assert started.wait(60), "acquisition never started" + assert finished.wait(60), "acquisition never finished" + + # (1) Every FOV coordinate is a 3-tuple whose z equals the known plane at its (x, y). + fovs = mpc.scanCoordinates.region_fov_coordinates["A1"] + assert len(fovs) == 4 + for coord in fovs: + assert len(coord) == 3, f"FOV was not baked to (x, y, z): {coord}" + x, y, z = coord + assert z == pytest.approx(plane_z(x), abs=1e-6) + # The plane actually tilts (larger x -> larger z); the z's are not a single constant. + assert max(f[2] for f in fovs) > min(f[2] for f in fovs) + + # (2) The contrast focus-map was never enabled in the reflection-AF path. + assert True not in focus_map_use_calls + + # (3) The run completed, not aborted. + assert mpc.abort_acqusition_requested is False From 83b086a52137f021054d1ae83617a1f4af7d9327 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sat, 11 Jul 2026 01:04:35 -0700 Subject: [PATCH 10/19] feat: GUI wellplate saves emit wells + fov_pattern for Select Wells mode (schema v2) Co-Authored-By: Claude Fable 5 --- .../control/core/multi_point_controller.py | 45 ++++++++++++------- .../tests/control/test_save_yaml_schema_v2.py | 31 +++++++++++++ 2 files changed, 61 insertions(+), 15 deletions(-) create mode 100644 software/tests/control/test_save_yaml_schema_v2.py diff --git a/software/control/core/multi_point_controller.py b/software/control/core/multi_point_controller.py index 5497508f5..11f186220 100644 --- a/software/control/core/multi_point_controller.py +++ b/software/control/core/multi_point_controller.py @@ -103,21 +103,36 @@ 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": { + "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, diff --git a/software/tests/control/test_save_yaml_schema_v2.py b/software/tests/control/test_save_yaml_schema_v2.py new file mode 100644 index 000000000..2595c64fd --- /dev/null +++ b/software/tests/control/test_save_yaml_schema_v2.py @@ -0,0 +1,31 @@ +import control.microscope +import tests.control.test_stubs as ts +from control.acquisition_yaml_loader import parse_acquisition_yaml + + +def test_select_wells_save_emits_wells_and_pattern(tmp_path): + scope = control.microscope.Microscope.build_from_global_config(True) + try: + mpc = ts.get_test_multi_point_controller(scope) + mpc.set_base_path(str(tmp_path)) + mpc.start_new_experiment("save_v2") + # The wellplate widget sets the mode via set_xy_mode() before run_acquisition; + # do the same here so _save_acquisition_yaml sees params.xy_mode == "Select Wells". + mpc.set_xy_mode("Select Wells") + mpc.set_scan_size(0.5) + mpc.set_overlap_percent(10) + sc = mpc.scanCoordinates + sc.clear_regions() + sc.add_region(well_id="A1", center_x=14.3, center_y=11.36, scan_size_mm=0.5, overlap_percent=10, shape="Square") + channel = scope.live_controller.get_channels(scope.objective_store.current_objective)[0].name + mpc.set_selected_configurations([channel]) + mpc.run_acquisition() + # find the saved acquisition.yaml under the experiment dir + saved = list((tmp_path).rglob("acquisition.yaml")) + assert saved, "acquisition.yaml not written" + data = parse_acquisition_yaml(str(saved[0])) + assert data.wells == "A1" + assert data.fov_pattern is not None and data.fov_pattern["type"] == "coverage" + assert data.wellplate_regions is None + finally: + scope.close() From 51be15bc67bf63c60d8d07de2d159dd330b761df Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sat, 11 Jul 2026 01:11:37 -0700 Subject: [PATCH 11/19] test: pin legacy regions form for non-Select-Wells saves and coverage pattern keys Co-Authored-By: Claude Fable 5 --- .../tests/control/test_save_yaml_schema_v2.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/software/tests/control/test_save_yaml_schema_v2.py b/software/tests/control/test_save_yaml_schema_v2.py index 2595c64fd..1bb26d4ea 100644 --- a/software/tests/control/test_save_yaml_schema_v2.py +++ b/software/tests/control/test_save_yaml_schema_v2.py @@ -26,6 +26,41 @@ def test_select_wells_save_emits_wells_and_pattern(tmp_path): data = parse_acquisition_yaml(str(saved[0])) assert data.wells == "A1" assert data.fov_pattern is not None and data.fov_pattern["type"] == "coverage" + # Coverage pattern keys mirror the add_region / set_scan_size / set_overlap values above. + assert data.fov_pattern["shape"] == "Square" + assert data.fov_pattern["overlap_percent"] == 10.0 + assert data.fov_pattern["scan_size_mm"] == 0.5 assert data.wellplate_regions is None finally: scope.close() + + +def test_non_select_wells_save_emits_legacy_regions(tmp_path): + scope = control.microscope.Microscope.build_from_global_config(True) + try: + mpc = ts.get_test_multi_point_controller(scope) + mpc.set_base_path(str(tmp_path)) + mpc.start_new_experiment("save_v2_legacy") + # Do NOT call set_xy_mode("Select Wells"): a fresh test-stub controller keeps the + # default xy_mode == "Current Position", which drives the legacy regions writer branch. + assert mpc.xy_mode == "Current Position" + mpc.set_scan_size(0.5) + mpc.set_overlap_percent(10) + sc = mpc.scanCoordinates + sc.clear_regions() + sc.add_region(well_id="A1", center_x=14.3, center_y=11.36, scan_size_mm=0.5, overlap_percent=10, shape="Square") + channel = scope.live_controller.get_channels(scope.objective_store.current_objective)[0].name + mpc.set_selected_configurations([channel]) + mpc.run_acquisition() + # find the saved acquisition.yaml under the experiment dir + saved = list((tmp_path).rglob("acquisition.yaml")) + assert saved, "acquisition.yaml not written" + data = parse_acquisition_yaml(str(saved[0])) + # Legacy coordinate-based form: explicit regions with center_mm, no wells-by-name. + assert isinstance(data.wellplate_regions, list) and data.wellplate_regions + first = data.wellplate_regions[0] + assert "name" in first + assert "center_mm" in first + assert data.wells is None + finally: + scope.close() From 50ac77c8e82d98dbbccb7b4f6346b1559ab14f00 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sat, 11 Jul 2026 01:26:03 -0700 Subject: [PATCH 12/19] feat: wellplate widget loads v2 wells+coverage yamls; clear warning for API-only patterns WellplateMultiPointWidget._apply_yaml_settings now loads schema-v2 acquisitions: - wells-by-name (range-expanded via squid_service.wells.parse_well_names) select the named wells and apply the coverage fov_pattern's scan_size_mm / overlap_percent / shape to the widget controls. - Non-coverage patterns (centered_grid/grid_subset/random) are not representable in the GUI yet, so the load is aborted with a clear QMessageBox.warning directing the user to the API/method registry. The signal-unblock finally: still runs. - Legacy wellplate_regions path is unchanged (now the elif branch). Adds GUI tests (in the CI-skipped test_HighContentScreeningGui.py, run locally with QT_API=pyqt5) covering both the coverage load and the non-coverage rejection. Co-Authored-By: Claude Fable 5 --- software/control/widgets.py | 25 +++++- .../control/test_HighContentScreeningGui.py | 83 +++++++++++++++++++ 2 files changed, 106 insertions(+), 2 deletions(-) diff --git a/software/control/widgets.py b/software/control/widgets.py index b2fc98892..6c3270f33 100644 --- a/software/control/widgets.py +++ b/software/control/widgets.py @@ -9280,8 +9280,29 @@ def _apply_yaml_settings(self, yaml_data): if yaml_data.xy_mode in ["Current Position", "Select Wells", "Manual", "Load Coordinates"]: self.combobox_xy_mode.setCurrentText(yaml_data.xy_mode) - # Load well regions if present and update XY checkbox state - if yaml_data.wellplate_regions: + # Load well selection: v2 wells-by-name first, then legacy regions + if yaml_data.wells: + from squid_service.wells import parse_well_names + + pattern = yaml_data.fov_pattern or {"type": "coverage"} + if pattern["type"] != "coverage": + QMessageBox.warning( + self, + "Pattern Not Supported in GUI", + f"This acquisition uses fov_pattern '{pattern['type']}', which the GUI " + "cannot represent yet. Run it via the API/method registry instead.", + ) + return + if pattern.get("scan_size_mm"): + self.entry_scan_size.setValue(pattern["scan_size_mm"]) + self.entry_overlap.setValue(pattern.get("overlap_percent", yaml_data.overlap_percent)) + if pattern.get("shape"): + index = self.combobox_shape.findText(pattern["shape"]) + if index >= 0: + self.combobox_shape.setCurrentIndex(index) + self._load_well_regions([{"name": n} for n in parse_well_names(yaml_data.wells)]) + self.checkbox_xy.setChecked(True) + elif yaml_data.wellplate_regions: self._load_well_regions(yaml_data.wellplate_regions) self.checkbox_xy.setChecked(True) else: diff --git a/software/tests/control/test_HighContentScreeningGui.py b/software/tests/control/test_HighContentScreeningGui.py index 35a3ff961..42ad9c697 100644 --- a/software/tests/control/test_HighContentScreeningGui.py +++ b/software/tests/control/test_HighContentScreeningGui.py @@ -1,3 +1,5 @@ +import pytest + import control._def import control.gui_hcs @@ -62,3 +64,84 @@ def confirm_exit(parent, title, text, *args, **kwargs): assert len(z_calls) == 1, f"signal_z_um_delta wired {len(z_calls)} times, expected 1" assert len(click_calls) == 1, f"image_click_coordinates wired {len(click_calls)} times, expected 1" + + +def _build_hcs_gui(qtbot, monkeypatch): + """Build a simulated HCS GUI, auto-confirming the exit dialog on teardown.""" + + def confirm_exit(parent, title, text, *args, **kwargs): + if title == "Confirm Exit": + return QMessageBox.Yes + raise RuntimeError(f"Unexpected QMessageBox: {title} - {text}") + + monkeypatch.setattr(QMessageBox, "question", confirm_exit) + + scope = control.microscope.Microscope.build_from_global_config(True) + win = control.gui_hcs.HighContentScreeningGui(microscope=scope, is_simulation=True) + qtbot.add_widget(win) + return win + + +def test_wellplate_widget_loads_v2_wells_coverage_yaml(qtbot, monkeypatch): + """Schema v2: a wells+coverage yaml selects the named wells (range-expanded) + and applies the coverage scan settings to the wellplate widget controls.""" + from control.acquisition_yaml_loader import AcquisitionYAMLData + + win = _build_hcs_gui(qtbot, monkeypatch) + widget = win.wellplateMultiPointWidget + assert widget.well_selection_widget is not None + + yaml_data = AcquisitionYAMLData( + widget_type="wellplate", + wells="A1:A2", + overlap_percent=10.0, + fov_pattern={ + "type": "coverage", + "scan_size_mm": 2.5, + "overlap_percent": 15.0, + "shape": "Circle", + }, + ) + + widget._apply_yaml_settings(yaml_data) + + assert widget.entry_scan_size.value() == pytest.approx(2.5) + # overlap comes from the pattern (15.0), not the flat default (10.0) + assert widget.entry_overlap.value() == pytest.approx(15.0) + assert widget.combobox_shape.currentText() == "Circle" + assert widget.checkbox_xy.isChecked() + # "A1:A2" range-expands to two wells -> two selected cells + assert len(widget.well_selection_widget.selectedItems()) == 2 + + +def test_wellplate_widget_rejects_noncoverage_pattern(qtbot, monkeypatch): + """Schema v2: a non-coverage fov_pattern is not representable in the GUI yet, + so the load is aborted with a clear warning and no wells are selected.""" + from control.acquisition_yaml_loader import AcquisitionYAMLData + + warnings = [] + + def capture_warning(parent, title, text, *args, **kwargs): + warnings.append((title, text)) + return QMessageBox.Ok + + monkeypatch.setattr(QMessageBox, "warning", capture_warning) + + win = _build_hcs_gui(qtbot, monkeypatch) + widget = win.wellplateMultiPointWidget + assert widget.well_selection_widget is not None + widget.well_selection_widget.clearSelection() + + yaml_data = AcquisitionYAMLData( + widget_type="wellplate", + wells="A1", + fov_pattern={"type": "centered_grid", "nx": 3, "ny": 3, "overlap_percent": 10.0}, + ) + + widget._apply_yaml_settings(yaml_data) + + assert len(warnings) == 1 + assert warnings[0][0] == "Pattern Not Supported in GUI" + assert "centered_grid" in warnings[0][1] + # Load aborted before selecting wells + assert len(widget.well_selection_widget.selectedItems()) == 0 From 00f96c1b3c718725e04360ef404e2cf59a94fc06 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sat, 11 Jul 2026 09:59:44 -0700 Subject: [PATCH 13/19] fix: propagate yaml-load abort as failure so callers cannot act on false success Co-Authored-By: Claude Fable 5 --- software/control/widgets.py | 12 +++- .../control/test_HighContentScreeningGui.py | 64 +++++++++++++++++++ 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/software/control/widgets.py b/software/control/widgets.py index 6c3270f33..359355b12 100644 --- a/software/control/widgets.py +++ b/software/control/widgets.py @@ -951,8 +951,14 @@ def _load_acquisition_yaml(self, file_path: str) -> bool: dialog.exec_() return False - # Apply settings with signal blocking - self._apply_yaml_settings(yaml_data) + # Apply settings with signal blocking. An override may abort the load by + # returning False (e.g. unsupported settings for the GUI); propagate that as + # a failure so callers cannot act on a false success. Overrides that return + # None keep today's behavior (treated as success). + result = self._apply_yaml_settings(yaml_data) + if result is False: + self._log.warning("YAML load aborted by widget (unsupported settings for GUI)") + return False self._log.info(f"Loaded acquisition settings from: {file_path}") return True @@ -9292,7 +9298,7 @@ def _apply_yaml_settings(self, yaml_data): f"This acquisition uses fov_pattern '{pattern['type']}', which the GUI " "cannot represent yet. Run it via the API/method registry instead.", ) - return + return False if pattern.get("scan_size_mm"): self.entry_scan_size.setValue(pattern["scan_size_mm"]) self.entry_overlap.setValue(pattern.get("overlap_percent", yaml_data.overlap_percent)) diff --git a/software/tests/control/test_HighContentScreeningGui.py b/software/tests/control/test_HighContentScreeningGui.py index 42ad9c697..6e7a5e001 100644 --- a/software/tests/control/test_HighContentScreeningGui.py +++ b/software/tests/control/test_HighContentScreeningGui.py @@ -145,3 +145,67 @@ def capture_warning(parent, title, text, *args, **kwargs): assert "centered_grid" in warnings[0][1] # Load aborted before selecting wells assert len(widget.well_selection_widget.selectedItems()) == 0 + + +def test_load_acquisition_yaml_noncoverage_returns_false(qtbot, monkeypatch, tmp_path): + """Full-path regression: a non-coverage v2 yaml aborts inside _apply_yaml_settings, + and _load_acquisition_yaml must propagate that abort as False so workflow-runner + callers do not act on a false success.""" + warnings = [] + + def capture_warning(parent, title, text, *args, **kwargs): + warnings.append((title, text)) + return QMessageBox.Ok + + monkeypatch.setattr(QMessageBox, "warning", capture_warning) + + win = _build_hcs_gui(qtbot, monkeypatch) + widget = win.wellplateMultiPointWidget + + yaml_path = tmp_path / "noncoverage.yaml" + yaml_path.write_text( + "acquisition:\n" + " widget_type: wellplate\n" + "wellplate_scan:\n" + ' wells: "A1"\n' + " fov_pattern:\n" + " type: centered_grid\n" + " nx: 3\n" + " ny: 3\n" + " overlap_percent: 10.0\n" + ) + + assert widget._load_acquisition_yaml(str(yaml_path)) is False + assert len(warnings) == 1 + assert warnings[0][0] == "Pattern Not Supported in GUI" + + +def test_load_acquisition_yaml_coverage_returns_true(qtbot, monkeypatch, tmp_path): + """Full-path regression: a wells+coverage v2 yaml loads successfully via the full + _load_acquisition_yaml path and returns True.""" + + def fail_on_warning(parent, title, text, *args, **kwargs): + raise RuntimeError(f"Unexpected QMessageBox.warning: {title} - {text}") + + monkeypatch.setattr(QMessageBox, "warning", fail_on_warning) + + win = _build_hcs_gui(qtbot, monkeypatch) + widget = win.wellplateMultiPointWidget + + yaml_path = tmp_path / "coverage.yaml" + yaml_path.write_text( + "acquisition:\n" + " widget_type: wellplate\n" + "wellplate_scan:\n" + ' wells: "A1:A2"\n' + " overlap_percent: 10.0\n" + " fov_pattern:\n" + " type: coverage\n" + " scan_size_mm: 2.5\n" + " overlap_percent: 15.0\n" + " shape: Circle\n" + ) + + assert widget._load_acquisition_yaml(str(yaml_path)) is True + assert widget.checkbox_xy.isChecked() + assert len(widget.well_selection_widget.selectedItems()) == 2 From 5c698adb5fe23c3827116184fb11c629d9545edc Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sat, 11 Jul 2026 22:34:27 -0700 Subject: [PATCH 14/19] docs+registry: schema v2 fields in method summaries and API docs Co-Authored-By: Claude Fable 5 --- software/docs/core-service-api.md | 33 +++++++++++++++++++- software/docs/quickstart-api.md | 14 +++++++-- software/squid_service/methods.py | 1 + software/tests/squid_service/test_methods.py | 18 +++++++++++ 4 files changed, 63 insertions(+), 3 deletions(-) diff --git a/software/docs/core-service-api.md b/software/docs/core-service-api.md index 021638503..7323acae4 100644 --- a/software/docs/core-service-api.md +++ b/software/docs/core-service-api.md @@ -284,9 +284,40 @@ curl -X POST http://127.0.0.1:8060/v1/acquisitions -d '{"method": "daily_scan"}' ``` `GET /v1/methods` returns a summary per method including `estimated_duration_s`, which is **always `null` -in this version** — it is a documented placeholder for a future duration estimator, not a bug. Deleting a +in this version** — it is a documented placeholder for a future duration estimator, not a bug. The summary +also reports `pattern` (the `fov_pattern` type, or `"coverage"` when none is set) and `wells`. Deleting a method while an acquisition is in progress is rejected with a `PROTOCOL_WRONG_STATE` fault. +### Method schema v2 + +A wellplate method's `wellplate_scan` accepts three optional schema-v2 fields. All are validated when the +method is parsed (`POST`/`PUT /v1/methods`) and again at acquisition preflight; `fov_pattern.type` and `wells` +are surfaced in the `GET /v1/methods` summary. + +**`fov_pattern`** — per-well FOV layout, a mapping with a `type` key. One of: + +| Type | Keys | Meaning | +|------|------|---------| +| `coverage` | `scan_size_mm`, `overlap_percent`, `shape` | Default when omitted: flat tiling that covers a `scan_size_mm` area. | +| `centered_grid` | `nx`, `ny`, `overlap_percent` | An `nx`×`ny` tile grid centered on each well. | +| `grid_subset` | `nx`, `ny`, `tiles`, `overlap_percent` | A chosen subset of an `nx`×`ny` grid; `tiles` is a non-empty list of `[row, col]` (0-indexed, in-range, no duplicates). | +| `random` | `n_fovs`, `seed` | `n_fovs` randomly placed FOVs; `seed` (optional int) makes the layout reproducible. | + +Every type except `coverage` is a per-well pattern and requires `wellplate_scan.wells`. + +**`well_z_offsets_um`** — per-well laser-AF **target** offsets, a mapping of well name → µm from the AF +reference plane (an optional `default` key applies to unlisted wells). Requires `laser_af`; each entry is +range-checked against the laser-AF search range at preflight (`INVALID_PARAM`, component `well_z_offsets_um`, +if `|offset| >` the range). Consumed via the per-region AF offset hook (`set_region_laser_af_offsets`), +cleared after the run. This is a **TYPE-1** correction — it shifts *where autofocus focuses*, not the coarse +stage Z. + +**`z_plan`** — pre-AF stage positioning for tilted plates: `{"type": "focus_map", ...}` with **exactly one** +of `generate: true` (measure a 3-corner focus-map plane at run start) or `points` (exactly three +`[x_mm, y_mm, z_mm]` entries defining a plane, which is baked into every FOV's coarse Z). This is a **TYPE-2** +correction — it sets the coarse stage Z *before* autofocus runs, distinct from the `well_z_offsets_um` AF +target offsets, and the two may be combined. + ## Polling guidance (URS API-POLL-005) `GET /v1/system/status` and `GET /v1/system/heartbeat` are served from in-process state — they never make a diff --git a/software/docs/quickstart-api.md b/software/docs/quickstart-api.md index d8579b7f3..444fe37b6 100644 --- a/software/docs/quickstart-api.md +++ b/software/docs/quickstart-api.md @@ -86,8 +86,11 @@ curl -X POST http://127.0.0.1:8060/v1/methods \ "channels": [{"name": "BF LED matrix full"}, {"name": "Fluorescence 488 nm Ex"}], "autofocus": {"contrast_af": false, "laser_af": true}, "wellplate_scan": { - "scan_size_mm": 1.5, "overlap_percent": 10, - "wells": "A1:B3" + "overlap_percent": 10, + "wells": "A1:B3", + "fov_pattern": {"type": "centered_grid", "nx": 3, "ny": 3}, + "well_z_offsets_um": {"A1": 0, "default": -5}, + "z_plan": {"type": "focus_map", "generate": true} } } }' @@ -98,6 +101,13 @@ curl -X POST http://127.0.0.1:8060/v1/acquisitions \ "overrides": {"wells": "A1:D6", "output_path": "/data/exp42"}}' ``` +The three schema-v2 fields above are optional (all `wellplate_scan`): +- `fov_pattern` sets the per-well FOV layout — here a 3×3 grid centered on each well (types: `coverage`, `centered_grid`, `grid_subset`, `random`; all but `coverage` require `wells`). +- `well_z_offsets_um` shifts the laser-AF focus target per well in µm from the AF reference plane (needs `laser_af: true`; a `default` key covers unlisted wells). +- `z_plan` positions the stage in Z before AF for tilted plates — `generate: true` measures a focus map at run start, or give three `points` (`[x_mm, y_mm, z_mm]`) defining a plane. + +See the [Method schema v2](core-service-api.md#method-schema-v2) reference for the full field spec. + Name wells with `wellplate_scan.wells` (`"A1:B3"` range or `"A1,A2,B1"` list) — X/Y are derived from the plate definition, so no coordinates are stored in the method. **Z is not stored in a wells-by-name method**: by default the run baselines on the current stage diff --git a/software/squid_service/methods.py b/software/squid_service/methods.py index 9ac8f6261..6b7231661 100644 --- a/software/squid_service/methods.py +++ b/software/squid_service/methods.py @@ -124,6 +124,7 @@ def _summarize(self, name: str, path: Path) -> dict: "objective": data.objective_name, "wellplate_format": raw.get("sample", {}).get("wellplate_format"), "wells": data.wells, + "pattern": (data.fov_pattern or {}).get("type", "coverage"), "nz": data.nz, "nt": data.nt, "estimated_duration_s": None, # best-effort placeholder, documented diff --git a/software/tests/squid_service/test_methods.py b/software/tests/squid_service/test_methods.py index e6391513c..afa5c1c80 100644 --- a/software/tests/squid_service/test_methods.py +++ b/software/tests/squid_service/test_methods.py @@ -38,12 +38,30 @@ def test_save_list_get_roundtrip(tmp_path): assert summary["nz"] == 1 assert summary["nt"] == 1 assert summary["wellplate_format"] == "96 well plate" + # No fov_pattern -> legacy coverage behavior. + assert summary["pattern"] == "coverage" got = reg.get("scan_a") assert got["name"] == "scan_a" assert got["config"]["acquisition"]["widget_type"] == "wellplate" +def test_list_summary_reports_fov_pattern(tmp_path): + reg = MethodRegistry(tmp_path) + config = _valid_config() + # A per-well pattern requires wells-by-name (not explicit regions). + config["wellplate_scan"] = { + "overlap_percent": 10, + "wells": "A1:B2", + "fov_pattern": {"type": "centered_grid", "nx": 2, "ny": 2}, + } + reg.save("grid_method", config, overwrite=False) + + summary = reg.list()[0] + assert summary["pattern"] == "centered_grid" + assert summary["wells"] == "A1:B2" + + def test_save_existing_without_overwrite_faults(tmp_path): reg = MethodRegistry(tmp_path) reg.save("dup", _valid_config(), overwrite=False) From 5d70d33b7cb0f77b7e91d03b879cc79ae54d10a4 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 00:43:38 -0700 Subject: [PATCH 15/19] fix: carry #562 test-stub update for the ported perform_autofocus hook The Task-0 port of the per-region laser-AF hook (c230268d) changed perform_autofocus to read self.region_laser_af_offsets but did not carry #562's matching stub fix (10b24757), so two _af_stub-based tests regressed vs the pre-hook base. Real worker sets this in __init__. Co-Authored-By: Claude Fable 5 --- software/tests/control/test_MultiPointWorker_offsets.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/software/tests/control/test_MultiPointWorker_offsets.py b/software/tests/control/test_MultiPointWorker_offsets.py index fa487e205..f0102299d 100644 --- a/software/tests/control/test_MultiPointWorker_offsets.py +++ b/software/tests/control/test_MultiPointWorker_offsets.py @@ -363,6 +363,9 @@ def _af_stub(*, move_to_target_result=True, move_to_target_raises=False): w.laser_auto_focus_controller.move_to_target.return_value = move_to_target_result w._laser_af_successes = 0 w._laser_af_failures = 0 + # perform_autofocus reads this per-region target map (empty -> target 0.0, the + # pre-existing behavior); the real worker sets it in __init__. + w.region_laser_af_offsets = {} w.perform_autofocus = MultiPointWorker.perform_autofocus.__get__(w) return w From 42b402759dfcc3742cd390e12bd0a523ea6386bd Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 00:49:10 -0700 Subject: [PATCH 16/19] docs: correct well_z_offsets_um lifecycle wording (consumed-and-cleared at run start) Co-Authored-By: Claude Fable 5 --- software/docs/core-service-api.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/software/docs/core-service-api.md b/software/docs/core-service-api.md index 7323acae4..a98389eab 100644 --- a/software/docs/core-service-api.md +++ b/software/docs/core-service-api.md @@ -309,8 +309,8 @@ Every type except `coverage` is a per-well pattern and requires `wellplate_scan. reference plane (an optional `default` key applies to unlisted wells). Requires `laser_af`; each entry is range-checked against the laser-AF search range at preflight (`INVALID_PARAM`, component `well_z_offsets_um`, if `|offset| >` the range). Consumed via the per-region AF offset hook (`set_region_laser_af_offsets`), -cleared after the run. This is a **TYPE-1** correction — it shifts *where autofocus focuses*, not the coarse -stage Z. +consumed-and-cleared at run start (one-shot; re-provide per run — never leaks to a later acquisition). This is +a **TYPE-1** correction — it shifts *where autofocus focuses*, not the coarse stage Z. **`z_plan`** — pre-AF stage positioning for tilted plates: `{"type": "focus_map", ...}` with **exactly one** of `generate: true` (measure a 3-corner focus-map plane at run start) or `points` (exactly three From 30a8e5359f4b6c11a5c5155f8845c32ecf07fd29 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 01:07:58 -0700 Subject: [PATCH 17/19] fix: validate z_plan colinearity (B1) and signal finish on gen-map early returns (B2) B1: _validate_z_plan now rejects XY-colinear point sets with a clear ValueError, so a colinear z_plan fails cleanly as INVALID_PARAM at preflight (via check_yaml) instead of letting interpolate_plane raise a raw ValueError inside start_acquisition._apply_z_plan_points -> misleading HARDWARE_FAULT 500 after preflight already passed. B2: run_acquisition's gen_focus_map branch had two early returns (`if not bounds` and `except ValueError`) that skipped acquisition-finished signaling. After the Task 6 gate loosening, reflection-AF runs reach them, so finish never fired: the service job stuck in ACQUIRING forever and the GUI stayed disabled. Each early return now emits self.callbacks.signal_acquisition_finished() (same as the validate_acquisition_settings early return); the service classifies the finish-without-start as a validation failure and completes the job as FAILURE, returning the instrument to INITIALIZED and re-enabling the GUI. Tests: colinear z_plan rejected at loader + service preflight (INVALID_PARAM, no 500); non-colinear plane still parses; no-bounds gen_focus_map early return still fires the finished signal (no hang). Co-Authored-By: Claude Fable 5 --- software/control/acquisition_yaml_loader.py | 7 +++ .../control/core/multi_point_controller.py | 10 +++++ .../test_gen_focus_map_reflection_af.py | 45 +++++++++++++++++++ software/tests/control/test_service_z_plan.py | 39 ++++++++++++++++ .../control/test_yaml_loader_schema_v2.py | 23 ++++++++++ 5 files changed, 124 insertions(+) diff --git a/software/control/acquisition_yaml_loader.py b/software/control/acquisition_yaml_loader.py index 3fcd7eaa8..6f2e7b020 100644 --- a/software/control/acquisition_yaml_loader.py +++ b/software/control/acquisition_yaml_loader.py @@ -156,6 +156,13 @@ def _validate_z_plan(raw: Optional[dict]) -> Optional[dict]: 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} diff --git a/software/control/core/multi_point_controller.py b/software/control/core/multi_point_controller.py index 11f186220..f395a26d5 100644 --- a/software/control/core/multi_point_controller.py +++ b/software/control/core/multi_point_controller.py @@ -798,6 +798,11 @@ def run_acquisition(self, acquire_current_fov=False): 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"] @@ -875,6 +880,11 @@ def run_acquisition(self, acquire_current_fov=False): 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(): diff --git a/software/tests/control/test_gen_focus_map_reflection_af.py b/software/tests/control/test_gen_focus_map_reflection_af.py index a545e2838..03d36d091 100644 --- a/software/tests/control/test_gen_focus_map_reflection_af.py +++ b/software/tests/control/test_gen_focus_map_reflection_af.py @@ -105,3 +105,48 @@ def spy_set_focus_map_use(enable): # (3) The run completed, not aborted. assert mpc.abort_acqusition_requested is False + + +def test_gen_focus_map_no_bounds_still_signals_finished(sim_scope, monkeypatch): + """B2: gen_focus_map + reflection AF that hits the `if not bounds: return` early + return must STILL emit signal_acquisition_finished. The worker never starts on that + path, so without the emit the service's job stays ACQUIRING forever and the GUI stays + disabled. get_scan_bounds() is forced to None to drive the early return deterministically. + """ + control._def.MERGE_CHANNELS = False + control._def.SUPPORT_LASER_AUTOFOCUS = True + + finished = threading.Event() + callbacks = dataclasses.replace( + NoOpCallbacks, + signal_acquisition_finished=lambda *a, **kw: finished.set(), + ) + mpc = ts.get_test_multi_point_controller(sim_scope, callbacks=callbacks) + + cfg = sim_scope.stage.get_config() + center_x = cfg.X_AXIS.MIN_POSITION + 1.0 + center_y = cfg.Y_AXIS.MIN_POSITION + 1.0 + center_z = (cfg.Z_AXIS.MAX_POSITION - cfg.Z_AXIS.MIN_POSITION) / 2.0 + mpc.scanCoordinates.clear_regions() + mpc.scanCoordinates.add_flexible_region("A1", center_x, center_y, center_z, 2, 2, 0) + + channel_names = [c.name for c in mpc.liveController.get_channels(mpc.objectiveStore.current_objective)] + mpc.set_selected_configurations(channel_names[0:1]) + + mpc.set_reflection_af_flag(True) + mpc.set_gen_focus_map_flag(True) + + # Store a laser-AF reference so validate_acquisition_settings passes and we actually + # reach the gen_focus_map branch (not the earlier validate-failure return). + sim_scope.addons.camera_focus.send_trigger() + ref_image = sim_scope.addons.camera_focus.read_frame() + assert ref_image is not None + mpc.laserAutoFocusController.laser_af_properties.set_reference_image(ref_image) + + # Force the `if not bounds: return` early return in the gen_focus_map branch. + monkeypatch.setattr(mpc.scanCoordinates, "get_scan_bounds", lambda: None) + + mpc.run_acquisition() + + # The finished signal fires synchronously on the early-return path (worker never starts). + assert finished.wait(30), "acquisition-finished signal never fired on the no-bounds early return (run would hang)" diff --git a/software/tests/control/test_service_z_plan.py b/software/tests/control/test_service_z_plan.py index faad5a277..23f27a985 100644 --- a/software/tests/control/test_service_z_plan.py +++ b/software/tests/control/test_service_z_plan.py @@ -3,6 +3,8 @@ import control.microscope import tests.control.test_stubs as ts +from squid_service.faults import FaultCategory, FaultError +from squid_service.models import AcquisitionRequest from squid_service.service import SquidCoreService @@ -65,6 +67,43 @@ def test_z_plan_points_bakes_plane_into_coordinates(service, sim_scope, tmp_path assert a1[2] == pytest.approx(expected_a1_z, abs=1e-6) +def test_preflight_rejects_colinear_z_plan_points(service, sim_scope, tmp_path): + """B1: a z_plan whose 3 points are colinear in XY must fail cleanly at preflight + (INVALID_PARAM on the ``yaml`` check), NOT escape interpolate_plane's raw ValueError + to a HARDWARE_FAULT 500 during start_acquisition after preflight already passed.""" + objective = sim_scope.objective_store.current_objective + channel = sim_scope.live_controller.get_channels(objective)[0].name + config = { + "acquisition": {"widget_type": "wellplate"}, + "sample": {"wellplate_format": "96 well plate"}, + "z_stack": {"nz": 1, "delta_z_mm": 0.001}, + "channels": [{"name": channel}], + "autofocus": {"contrast_af": False, "laser_af": False}, + "wellplate_scan": { + "wells": "A1,A3", + "scan_size_mm": 0.5, + "overlap_percent": 10, + # All y=0 -> colinear in XY -> cannot define a plane. + "z_plan": {"type": "focus_map", "points": [[0.0, 0.0, 1.0], [1.0, 0.0, 2.0], [2.0, 0.0, 3.0]]}, + }, + } + p = tmp_path / "acq_colinear.yaml" + p.write_text(yaml.safe_dump(config)) + req = AcquisitionRequest(yaml_path=str(p), overrides={"output_path": str(tmp_path)}) + + # preflight must not raise and must report ok=False on the yaml check. + result = service.preflight(req) + assert result["ok"] is False + yaml_check = next(c for c in result["checks"] if c["name"] == "yaml") + assert yaml_check["ok"] is False + assert "colinear" in yaml_check["message"].lower() + + # And the underlying fault is INVALID_PARAM (not HARDWARE_FAULT / a 500). + with pytest.raises(FaultError) as ei: + service._load_yaml_or_fault(str(p)) + assert ei.value.fault.category == FaultCategory.INVALID_PARAM + + def test_gen_focus_map_flag_survives_reset_when_requested(service, sim_scope, tmp_path): # Real signature: _reset_z_range_and_focus_map(z0, nz, delta_z_um, keep_gen_focus_map=False) service._mpc.set_gen_focus_map_flag(True) diff --git a/software/tests/control/test_yaml_loader_schema_v2.py b/software/tests/control/test_yaml_loader_schema_v2.py index ae057ce43..a42502494 100644 --- a/software/tests/control/test_yaml_loader_schema_v2.py +++ b/software/tests/control/test_yaml_loader_schema_v2.py @@ -110,3 +110,26 @@ def test_z_plan_points_exactly_three(tmp_path): }, ) ) + + +def test_z_plan_colinear_points_rejected(tmp_path): + # All three points share y=0 -> colinear in XY -> cannot define a plane. + # Must raise a clean loader ValueError (surfaced as INVALID_PARAM at preflight) + # rather than letting interpolate_plane raise deep inside start_acquisition. + with pytest.raises(ValueError, match="colinear"): + parse_acquisition_yaml( + _write( + tmp_path, + {"scan_size_mm": 0.5, "z_plan": {"type": "focus_map", "points": [[0, 0, 1], [1, 0, 2], [2, 0, 3]]}}, + ) + ) + + +def test_z_plan_non_colinear_plane_parses(tmp_path): + data = parse_acquisition_yaml( + _write( + tmp_path, + {"scan_size_mm": 0.5, "z_plan": {"type": "focus_map", "points": [[0, 0, 1.0], [10, 0, 1.1], [0, 10, 1.2]]}}, + ) + ) + assert data.z_plan["points"] == [[0.0, 0.0, 1.0], [10.0, 0.0, 1.1], [0.0, 10.0, 1.2]] From 6a5e6713b49d7301e8240a0179f31b269bf8a24a Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 01:22:03 -0700 Subject: [PATCH 18/19] feat: faithful v2 acquisition-record for API/service runs (fov_pattern threaded to the writer) Thread the real fov_pattern from the service to the acquisition-record writer, mirroring the Task-5 region_laser_af_offsets threading: an AcquisitionParameters field, a controller attribute + set_fov_pattern setter, consume-and-clear at run start, and build_params passthrough. The writer's v2 wellplate branch now emits params.fov_pattern when present, else the existing coverage-from-scan-size default (GUI Select-Wells save stays byte-unchanged). The service sets xy_mode deterministically on every path (wells -> Select Wells + real pattern, legacy regions -> Manual + None, grid -> Select Wells + centered_grid) so the writer's v2-vs-legacy discriminator is never inherited stale, and consume-and-clear guarantees no pattern leak between runs. Co-Authored-By: Claude Fable 5 --- .../control/core/multi_point_controller.py | 20 +++++++- software/control/core/multi_point_utils.py | 5 ++ software/squid_service/service.py | 17 +++++++ .../control/test_service_fov_patterns.py | 46 +++++++++++++++++++ 4 files changed, 86 insertions(+), 2 deletions(-) diff --git a/software/control/core/multi_point_controller.py b/software/control/core/multi_point_controller.py index f395a26d5..e0c0e0c00 100644 --- a/software/control/core/multi_point_controller.py +++ b/software/control/core/multi_point_controller.py @@ -109,7 +109,8 @@ def _save_acquisition_yaml( # definition at load time (see acquisition_yaml_loader.parse_acquisition_yaml). yaml_dict["wellplate_scan"] = { "wells": list(region_names), - "fov_pattern": { + "fov_pattern": params.fov_pattern + or { "type": "coverage", "scan_size_mm": scan_size_mm, "overlap_percent": overlap_percent, @@ -243,6 +244,7 @@ def __init__( 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 @@ -437,6 +439,11 @@ def set_region_laser_af_offsets(self, offsets): # 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 @@ -703,6 +710,10 @@ def run_acquisition(self, acquire_current_fov=False): # 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() @@ -899,6 +910,7 @@ def finish_fn(): 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 @@ -1004,7 +1016,10 @@ def finish_fn(): self._memory_monitor = None def build_params( - self, scan_position_information: ScanPositionInformation, region_laser_af_offsets: Optional[dict] = None + 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 @@ -1046,6 +1061,7 @@ def build_params( 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): diff --git a/software/control/core/multi_point_utils.py b/software/control/core/multi_point_utils.py index 79dbbb68c..8e829b1d9 100644 --- a/software/control/core/multi_point_utils.py +++ b/software/control/core/multi_point_utils.py @@ -69,6 +69,11 @@ class AcquisitionParameters: # 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: diff --git a/software/squid_service/service.py b/software/squid_service/service.py index 41005413c..d29f537cb 100644 --- a/software/squid_service/service.py +++ b/software/squid_service/service.py @@ -1655,6 +1655,12 @@ def start_acquisition(self, req: AcquisitionRequest) -> dict: grid = req.grid self._configure_grid_regions(grid, z0) self._configure_grid_controller(grid, z0) + # A grid request is a wells-driven centered grid — round-trip it as v2 so the + # saved record faithfully carries the wells + pattern. + self._mpc.set_xy_mode("Select Wells") + self._mpc.set_fov_pattern( + {"type": "centered_grid", "nx": grid.nx, "ny": grid.ny, "overlap_percent": grid.overlap_percent} + ) channel_count = len(grid.channels) nz, nt = 1, 1 source = "grid" @@ -1662,6 +1668,17 @@ def start_acquisition(self, req: AcquisitionRequest) -> dict: else: yaml_data, raw = ctx["yaml_data"], ctx["raw"] self._configure_regions(yaml_data, raw, req.overrides.wells, req.overrides.sample_format, z0) + # Set xy_mode deterministically so the writer's v2-vs-legacy discriminator is + # correct on every entry point (never inherited stale). A wells-driven run + # (override or yaml) round-trips as v2 with the real fov_pattern; a legacy + # regions[].center_mm run keeps the coordinate-based regions form. + effective_wells = req.overrides.wells or yaml_data.wells + if effective_wells: + self._mpc.set_xy_mode("Select Wells") + self._mpc.set_fov_pattern(yaml_data.fov_pattern) + else: + self._mpc.set_xy_mode("Manual") # legacy regions -> writer keeps regions form + self._mpc.set_fov_pattern(None) # Resolve per-well laser-AF offsets against the regions just configured and push # them into the controller (consumed-and-cleared by run_acquisition, so no leak # to later runs). Empty dict => every FOV targets the reference plane. diff --git a/software/tests/control/test_service_fov_patterns.py b/software/tests/control/test_service_fov_patterns.py index 504e898ee..64578ae86 100644 --- a/software/tests/control/test_service_fov_patterns.py +++ b/software/tests/control/test_service_fov_patterns.py @@ -126,6 +126,52 @@ def test_random_seeded_reproducible_and_per_well_different(service, sim_scope, t assert list(sc2.region_fov_coordinates["A1"]) == a1_first +def test_api_centered_grid_run_saves_v2_record(service, sim_scope, tmp_path): + import yaml as _yaml + from control.acquisition_yaml_loader import parse_acquisition_yaml + from squid_service.models import AcquisitionRequest + + out = tmp_path / "out" + req = AcquisitionRequest( + yaml_path=_write_yaml(tmp_path, sim_scope, {"fov_pattern": {"type": "centered_grid", "nx": 2, "ny": 2}}), + overrides={"output_path": str(out)}, + ) + assert service.preflight(req)["ok"] is True + handle = service.start_acquisition(req) + assert service.jobs.wait(handle["job_id"], timeout_s=120.0) + saved = list(out.rglob("acquisition.yaml")) + assert saved, "acquisition.yaml record not written" + data = parse_acquisition_yaml(str(saved[0])) + assert data.wells is not None # not the legacy regions form + assert data.wellplate_regions is None + assert data.fov_pattern is not None and data.fov_pattern["type"] == "centered_grid" + assert data.fov_pattern["nx"] == 2 and data.fov_pattern["ny"] == 2 + + +def test_api_legacy_regions_run_saves_regions_form(service, sim_scope, tmp_path): + """An API run of a legacy regions[].center_mm method (no wells) must still save the + coordinate-based `regions` form — proving the service's xy_mode="Manual" branch and + that the writer is byte-unchanged for legacy runs.""" + from control.acquisition_yaml_loader import parse_acquisition_yaml + from squid_service.models import AcquisitionRequest + + regions = [{"name": "R1", "center_mm": [14.3, 11.36, 0.5], "shape": "Square"}] + out = tmp_path / "out" + req = AcquisitionRequest( + yaml_path=_write_yaml(tmp_path, sim_scope, {"wells": None, "scan_size_mm": 0.5, "regions": regions}), + overrides={"output_path": str(out)}, + ) + assert service.preflight(req)["ok"] is True + handle = service.start_acquisition(req) + assert service.jobs.wait(handle["job_id"], timeout_s=120.0) + saved = list(out.rglob("acquisition.yaml")) + assert saved, "acquisition.yaml record not written" + data = parse_acquisition_yaml(str(saved[0])) + assert data.wells is None # legacy regions form preserved (xy_mode="Manual" branch) + assert data.wellplate_regions is not None + assert data.fov_pattern is None + + def test_random_within_well_bounds(service, sim_scope, tmp_path): import control._def From b53c52c0fd243c8c8a8a69e842465e84232405d5 Mon Sep 17 00:00:00 2001 From: Hongquan Li Date: Sun, 12 Jul 2026 01:28:39 -0700 Subject: [PATCH 19/19] test: round-trip the service grid branch as v2 centered_grid Co-Authored-By: Claude Fable 5 --- .../control/test_service_fov_patterns.py | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/software/tests/control/test_service_fov_patterns.py b/software/tests/control/test_service_fov_patterns.py index 64578ae86..3207539d5 100644 --- a/software/tests/control/test_service_fov_patterns.py +++ b/software/tests/control/test_service_fov_patterns.py @@ -148,6 +148,43 @@ def test_api_centered_grid_run_saves_v2_record(service, sim_scope, tmp_path): assert data.fov_pattern["nx"] == 2 and data.fov_pattern["ny"] == 2 +def test_api_grid_branch_run_saves_v2_centered_grid(service, sim_scope, tmp_path): + """A service inline-`grid` request must round-trip its dict construction end-to-end as a + v2 `centered_grid` record. Unlike the yaml/method branch, the grid branch builds the + `fov_pattern` dict itself (service.py: set_xy_mode("Select Wells") + + set_fov_pattern({"type": "centered_grid", nx, ny, overlap_percent})), so this exercises + that inline construction rather than a loader-normalized dict.""" + from control.acquisition_yaml_loader import parse_acquisition_yaml + from squid_service.models import AcquisitionRequest, GridSpec + + objective = sim_scope.objective_store.current_objective + channel = sim_scope.live_controller.get_channels(objective)[0].name + out = tmp_path / "out" + req = AcquisitionRequest( + grid=GridSpec( + wells="A1:A2", + channels=[channel], + nx=2, + ny=3, + overlap_percent=15, + wellplate_format="96 well plate", + ), + overrides={"output_path": str(out)}, + ) + assert service.preflight(req)["ok"] is True + handle = service.start_acquisition(req) + assert service.jobs.wait(handle["job_id"], timeout_s=120.0) + saved = list(out.rglob("acquisition.yaml")) + assert saved, "acquisition.yaml record not written" + data = parse_acquisition_yaml(str(saved[0])) + assert data.wells is not None # not the legacy regions form + assert data.wellplate_regions is None + assert data.fov_pattern is not None and data.fov_pattern["type"] == "centered_grid" + assert data.fov_pattern["nx"] == 2 + assert data.fov_pattern["ny"] == 3 + assert data.fov_pattern["overlap_percent"] == 15 + + def test_api_legacy_regions_run_saves_regions_form(service, sim_scope, tmp_path): """An API run of a legacy regions[].center_mm method (no wells) must still save the coordinate-based `regions` form — proving the service's xy_mode="Manual" branch and