diff --git a/software/tests/acceptance/__init__.py b/software/tests/acceptance/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/software/tests/acceptance/conftest.py b/software/tests/acceptance/conftest.py new file mode 100644 index 000000000..48861c9e4 --- /dev/null +++ b/software/tests/acceptance/conftest.py @@ -0,0 +1,93 @@ +""" +Fixtures for the simulation-mode acquisition acceptance suite. + +These tests run full acquisitions end-to-end against simulated hardware and +assert on observable artifacts (files on disk, process state), not internals. +They are headless-safe: no QApplication is created (MultiPointController uses +plain threads and plain-callable callbacks), so they run under Xvfb in CI. +""" + +import logging +from unittest.mock import patch + +import pytest + +import control._def +import control.core.multi_point_worker +import control.microcontroller + +logger = logging.getLogger(__name__) + + +def pytest_configure(config): + config.addinivalue_line( + "markers", + "acceptance: end-to-end simulation-mode acquisition acceptance tests", + ) + + +def _make_tracking_init(original_init, instances_list): + """Create a wrapper that tracks Microcontroller instances.""" + + def _tracking_init(self, *args, **kwargs): + original_init(self, *args, **kwargs) + instances_list.append(self) + + return _tracking_init + + +@pytest.fixture(autouse=True) +def cleanup_microcontrollers(): + """ + Automatically close all Microcontroller instances after each test. + + Same rationale as tests/control/conftest.py: the packet-reading background + thread must be stopped or subsequent tests can segfault. + """ + active_microcontrollers = [] + original_init = control.microcontroller.Microcontroller.__init__ + + with patch.object( + control.microcontroller.Microcontroller, + "__init__", + _make_tracking_init(original_init, active_microcontrollers), + ): + yield + + for micro in active_microcontrollers: + try: + if hasattr(micro, "terminate_reading_received_packet_thread"): + if not micro.terminate_reading_received_packet_thread: + micro.close() + except Exception as e: + logger.warning(f"Failed to close Microcontroller in test cleanup: {e}") + + +@pytest.fixture(autouse=True) +def _watchdog_state_to_tmp(tmp_path, monkeypatch): + # Keep acquisition breadcrumbs out of the real user state dir during tests. + monkeypatch.setenv("SQUID_WATCHDOG_STATE_DIR", str(tmp_path / "watchdog")) + + +def set_file_saving_option(monkeypatch, option): + """ + Point acquisition file saving at `option` (a control._def.FileSavingOption). + + multi_point_worker star-imports FILE_SAVING_OPTION, freezing its own + binding at import time, while job_processing reads _def.FILE_SAVING_OPTION + at runtime — so both locations must be patched together. + """ + monkeypatch.setattr(control._def, "FILE_SAVING_OPTION", option) + monkeypatch.setattr(control.core.multi_point_worker, "FILE_SAVING_OPTION", option) + + +@pytest.fixture +def acquisition_defaults(monkeypatch): + """ + Pin the control._def knobs the acceptance scenarios depend on to known + values, restoring them afterwards. Individual tests override per-scenario + (e.g. FILE_SAVING_OPTION, backpressure limits) with the same monkeypatch. + """ + monkeypatch.setattr(control._def, "MERGE_CHANNELS", False) + set_file_saving_option(monkeypatch, control._def.FileSavingOption.INDIVIDUAL_IMAGES) + return monkeypatch diff --git a/software/tests/acceptance/harness.py b/software/tests/acceptance/harness.py new file mode 100644 index 000000000..c7bc80c93 --- /dev/null +++ b/software/tests/acceptance/harness.py @@ -0,0 +1,147 @@ +""" +Shared harness for the simulation-mode acquisition acceptance suite. + +Drives MultiPointController directly (in-process, no GUI, no QApplication), +reusing the wiring helpers from tests/control/test_stubs.py. Assertions in the +scenario files operate on observable artifacts: files on disk and process +state. + +Timing thresholds are deliberately generous — CI runners are slow. +""" + +import csv +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import List, Optional + +import control._def +import control.microscope +from control.core.multi_point_controller import MultiPointController + +import tests.control.test_stubs as ts +from control.core.multi_point_utils import MultiPointControllerFunctions + +# Generous CI-safe waits (seconds). +START_TIMEOUT_S = 60 +FINISH_TIMEOUT_S = 300 + + +class AcquisitionTracker: + """Collects acquisition callbacks so tests can wait on observable events.""" + + def __init__(self): + self.started_event = threading.Event() + self.finished_event = threading.Event() + self.first_image_event = threading.Event() + self.image_count = 0 + self._lock = threading.Lock() + + def get_callbacks(self) -> MultiPointControllerFunctions: + return MultiPointControllerFunctions( + signal_acquisition_start=lambda params: self.started_event.set(), + signal_acquisition_finished=lambda: self.finished_event.set(), + signal_new_image=self._receive_image, + signal_current_configuration=lambda config: None, + signal_current_fov=lambda x, y: None, + signal_overall_progress=lambda progress: None, + signal_region_progress=lambda progress: None, + ) + + def _receive_image(self, frame, info): + with self._lock: + self.image_count += 1 + self.first_image_event.set() + + +@dataclass +class AcquisitionHarness: + """A simulated microscope + MultiPointController pair with cleanup.""" + + scope: "control.microscope.Microscope" + mpc: MultiPointController + tracker: AcquisitionTracker + experiment_dirs: List[Path] = field(default_factory=list) + + def close(self): + try: + self.mpc.close() + finally: + self.scope.close() + + def new_experiment(self, base_path: Path, experiment_id: str) -> None: + """Start a fresh experiment; the timestamped directory it creates is + recorded in self.experiment_dirs. + + start_new_experiment() serializes 'acquisition parameters.json' from + the controller's CURRENT settings — configure NZ/Nt/deltaZ/channels + before calling this if the test asserts on that file's contents.""" + base_path.mkdir(parents=True, exist_ok=True) + before = set(base_path.iterdir()) + self.mpc.set_base_path(str(base_path)) + self.mpc.start_new_experiment(experiment_id) + created = [p for p in base_path.iterdir() if p not in before and p.is_dir()] + assert len(created) == 1, f"expected exactly one new experiment dir, found {created}" + self.experiment_dirs.append(created[0]) + + @property + def experiment_dir(self) -> Path: + assert self.experiment_dirs, "new_experiment() has not been called" + return self.experiment_dirs[-1] + + def add_fov_grid(self, region_id: str = "region0", nx: int = 2, ny: int = 2) -> None: + """Add an nx x ny FOV grid at a stage position guaranteed to be within + limits (coordinates outside stage limits are silently dropped).""" + stage_config = self.mpc.stage.get_config() + x = stage_config.X_AXIS.MIN_POSITION + 1.0 + y = stage_config.Y_AXIS.MIN_POSITION + 1.0 + z = (stage_config.Z_AXIS.MIN_POSITION + stage_config.Z_AXIS.MAX_POSITION) / 2.0 + self.mpc.scanCoordinates.add_flexible_region(region_id, x, y, z, nx, ny, 0) + fovs = self.mpc.scanCoordinates.region_fov_coordinates.get(region_id, []) + assert len(fovs) == nx * ny, f"region {region_id}: expected {nx * ny} FOVs, got {len(fovs)}" + + def select_channels(self, count: int) -> List[str]: + objective = self.scope.objective_store.current_objective + names = [c.name for c in self.mpc.liveController.get_channels(objective)] + assert len(names) >= count, f"need {count} channels, simulation config provides {names}" + selected = names[:count] + self.mpc.set_selected_configurations(selected) + return selected + + def run_and_wait(self, timeout_s: float = FINISH_TIMEOUT_S) -> None: + self.mpc.run_acquisition() + assert self.tracker.started_event.wait(START_TIMEOUT_S), "acquisition did not start" + assert self.tracker.finished_event.wait(timeout_s), f"acquisition did not finish within {timeout_s}s" + + +def make_harness() -> AcquisitionHarness: + """Build a fresh simulated microscope wired to a MultiPointController. + + Callers own cleanup: call harness.close() in a try/finally so JobRunner + subprocesses and semaphores are released. Combine with the + `acquisition_defaults` fixture, which pins MERGE_CHANNELS and the file + saving option with automatic restore. + """ + scope = control.microscope.Microscope.build_from_global_config(True) + tracker = AcquisitionTracker() + mpc = ts.get_test_multi_point_controller(microscope=scope, callbacks=tracker.get_callbacks()) + mpc.scanCoordinates.clear_regions() + return AcquisitionHarness(scope=scope, mpc=mpc, tracker=tracker) + + +def read_coordinates_csv(path: Path) -> List[dict]: + with open(path, newline="") as f: + return list(csv.DictReader(f)) + + +def timepoint_dir(experiment_dir: Path, timepoint: int) -> Path: + # FILE_ID_PADDING is 0 by default, so per-timepoint dirs are "0", "1", ... + return experiment_dir / str(timepoint).zfill(control._def.FILE_ID_PADDING) + + +def list_image_files(directory: Path) -> List[Path]: + """Individual-images mode: per-image files named + {region}_{fov}_{z}_{channel}. directly in the timepoint dir.""" + return sorted(p for p in directory.glob("*.tiff") if p.is_file()) + sorted( + p for p in directory.glob("*.bmp") if p.is_file() + ) diff --git a/software/tests/acceptance/test_abort.py b/software/tests/acceptance/test_abort.py new file mode 100644 index 000000000..a5da63db2 --- /dev/null +++ b/software/tests/acceptance/test_abort.py @@ -0,0 +1,129 @@ +""" +Acceptance scenario 3: abort mid-acquisition, then re-acquire in the same process. + +Drives MultiPointController directly against a simulated microscope (no GUI). A +multi-FOV / multi-channel / multi-Z acquisition is started with enough runway +that a mid-flight abort lands after the first image but before completion. We +then assert: + + * abort stops the run cleanly (finished callback fires, acquisition_in_progress + drops, fewer than the full image count were produced), + * the same process/harness can immediately run a second, smaller acquisition to + completion, and + * once the harness is closed, no orphaned JobRunner subprocesses linger. + +Context (PR #582): the app previously leaked orphaned JobRunner ``spawn_main`` +subprocesses because ``os._exit()`` skipped multiprocessing's atexit hook. +Per-acquisition cleanup now shuts runners down in non-blocking daemon threads; +this test pins that no JobRunner survives ``close()`` at the acceptance level. +""" + +import multiprocessing +import time + +import pytest + +from control.core.job_processing import JobRunner + +from tests.acceptance.harness import make_harness, timepoint_dir, list_image_files + +pytestmark = pytest.mark.acceptance + + +def _wait_until(predicate, timeout_s: float, interval_s: float = 0.2) -> bool: + """Poll ``predicate`` until it is truthy or the deadline passes.""" + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(interval_s) + return predicate() + + +def test_abort_mid_acquisition_then_reacquire(tmp_path, acquisition_defaults): + h = make_harness() + tracker = h.tracker + try: + # ---- Run 1: large enough to guarantee abort lands mid-flight ---------- + # 3x3 FOVs x 2 channels x NZ=2 = 36 images of runway. + h.new_experiment(tmp_path, "abort_run") + h.add_fov_grid("region0", nx=3, ny=3) + h.select_channels(2) + h.mpc.set_NZ(2) + expected_full = 3 * 3 * 2 * 2 + assert expected_full == 36 + + h.mpc.run_acquisition() + + assert tracker.started_event.wait(60), "run 1 did not start" + assert tracker.first_image_event.wait(120), "run 1 produced no image before abort" + + # Abort mid-flight (real API misspelling: request_abort_aquisition). + h.mpc.request_abort_aquisition() + + assert tracker.finished_event.wait(120), "abort did not fire finished callback" + # acquisition_in_progress() reflects the worker thread and may briefly lag + # the finished callback, so poll it down rather than asserting immediately. + assert _wait_until( + lambda: not h.mpc.acquisition_in_progress(), timeout_s=30 + ), "acquisition_in_progress stayed True after abort" + + aborted_count = tracker.image_count + assert aborted_count < expected_full, ( + f"abort produced {aborted_count} images; expected fewer than {expected_full} " + "(abort did not land mid-flight -- increase the grid)" + ) + assert aborted_count >= 1, "expected at least one image before abort" + + # Partial artifacts are allowed; only pin what is stable: the first + # timepoint directory was created for the partial run. + assert timepoint_dir(h.experiment_dir, 0).is_dir(), "run 1 timepoint dir 0/ was not created" + + # ---- Run 2: second experiment in the same process/harness ------------- + # The tracker events are latched from run 1; clear them and remember the + # run-1 image count so we can assert run 2 emits exactly 2 new callbacks. + tracker.started_event.clear() + tracker.finished_event.clear() + tracker.first_image_event.clear() + count_before_run2 = tracker.image_count + + h.mpc.scanCoordinates.clear_regions() + h.new_experiment(tmp_path, "after_abort") + h.add_fov_grid("region0", nx=2, ny=1) + h.select_channels(1) + h.mpc.set_NZ(1) + + h.run_and_wait(timeout_s=300) + + new_callbacks = tracker.image_count - count_before_run2 + assert new_callbacks == 2, ( + f"run 2 emitted {new_callbacks} new-image callbacks, expected 2 " "(2 FOVs x 1 channel x NZ=1)" + ) + + run2_tp0 = timepoint_dir(h.experiment_dir, 0) + run2_images = list_image_files(run2_tp0) + assert len(run2_images) == 2, f"run 2 timepoint 0/ has {run2_images}, expected 2 files" + assert (h.experiment_dir / ".done").exists(), "run 2 .done marker missing" + + finally: + h.close() + + # ---- Orphan assertion (after close) -------------------------------------- + # Formulation: type-based (isinstance(child, JobRunner)) rather than a raw + # active_children() count. JobRunner is a multiprocessing.Process subclass, so + # in the parent process a leaked runner shows up as a JobRunner instance in + # active_children(). This is more robust than a count baseline because: + # (a) the MultiPointController PRE-WARMS a JobRunner at construction and + # after each acquisition start, so a live runner is EXPECTED while the + # harness is open -- a count taken "before the test" would need careful + # timing to exclude it, whereas after close() the invariant is simply + # "zero JobRunners", and + # (b) type-matching ignores unrelated children the test framework or other + # fixtures might spawn, targeting exactly the leak PR #582 is about. + def _no_job_runners() -> bool: + return not [c for c in multiprocessing.active_children() if isinstance(c, JobRunner)] + + assert _wait_until(_no_job_runners, timeout_s=15), ( + "JobRunner subprocess(es) still alive after close(): " + f"{[c for c in multiprocessing.active_children() if isinstance(c, JobRunner)]}" + ) diff --git a/software/tests/acceptance/test_backpressure.py b/software/tests/acceptance/test_backpressure.py new file mode 100644 index 000000000..b8a609e2a --- /dev/null +++ b/software/tests/acceptance/test_backpressure.py @@ -0,0 +1,117 @@ +"""Acceptance test pinning the backpressure byte-release deadlock class. + +This test guards against a historical PERMANENT DEADLOCK in acquisition +throttling (see project CLAUDE.md, "Backpressure" design note): + + Backpressure tracks bytes in the MAIN PROCESS queue. Bytes are incremented + when a job is dispatched and decremented when a job completes. For + DownsampledViewJob, bytes are released *immediately on job completion* + (not when the well/region completes), because the image data moves to + subprocess memory when processed. + + Why this matters: if bytes were only released at well completion, a z-stack + acquisition (FOVs x z-levels x channels) could accumulate more pending bytes + than the byte limit *before any well finishes*. The acquisition would then + block forever in wait_for_capacity() waiting for capacity that only well + completion could free -- a permanent deadlock. + +The scenario below drives a small z-stack under a deliberately tight byte limit +(~1.5 camera frames) so that the region's total pending bytes vastly exceed the +limit before the single region completes. Under correct per-job byte release the +acquisition throttles but drains and finishes; under the regressed +release-at-well-completion behavior it would hang. Completion within the timeout +is therefore the regression assertion. +""" + +import logging + +import pytest + +import control._def +from tests.acceptance.harness import ( + make_harness, + list_image_files, + timepoint_dir, +) + +pytestmark = pytest.mark.acceptance + + +def test_zstack_completes_under_tight_byte_limit(tmp_path, acquisition_defaults, caplog): + """A z-stack whose total bytes far exceed the byte limit still completes. + + Pins the DownsampledViewJob byte-release-on-job-completion design: if bytes + were only released at region completion, the 12-image z-stack (~8x the byte + limit) would deadlock before the single region finished. Completion is the + regression assertion; a hang manifests as the run_and_wait timeout. + """ + monkeypatch = acquisition_defaults + + harness = make_harness() + try: + # Frame size is machine-config dependent (CI's pristine INI crops to a + # quarter of the local default), so measure it instead of hardcoding: + # snap one frame from the simulated camera and size the limit off it. + harness.scope.camera.send_trigger() + frame = harness.scope.camera.read_frame() + assert frame is not None, "simulated camera returned no frame" + image_size_mib = frame.nbytes / (1024 * 1024) + + # Byte limit ~1.5 frames: the second dispatched frame already exceeds + # it, so throttling must engage well before the region's 12 frames are + # all in flight (12 frames = ~8x the limit). + byte_limit_mib = 1.5 * image_size_mib + monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_MB", byte_limit_mib) + monkeypatch.setattr(control._def, "ACQUISITION_MAX_PENDING_JOBS", 3) + monkeypatch.setattr(control._def, "ACQUISITION_THROTTLING_ENABLED", True) + harness.new_experiment(tmp_path / "backpressure", "zstack_tight_limit") + harness.add_fov_grid("region0", nx=2, ny=2) # 4 FOVs + harness.select_channels(1) # 1 channel + + harness.mpc.set_Nt(1) + harness.mpc.set_NZ(3) # 4 FOVs * 1 channel * 3 z = 12 images + harness.mpc.set_deltaZ(1.0) + harness.mpc.set_af_flag(False) + harness.mpc.set_reflection_af_flag(False) + + # 12 frames = ~8x the ~1.5-frame byte limit if none were released -- + # the historical deadlock trigger. + with caplog.at_level(logging.INFO): + # Deadlock (regression) manifests as this timeout. + harness.run_and_wait(timeout_s=300) + + # Completion assertions: all 12 images landed and the region drained. + tp0 = timepoint_dir(harness.experiment_dir, 0) + images = list_image_files(tp0) + assert len(images) == 12, f"expected 12 image files in {tp0}, found {len(images)}: {images}" + + # Guard the premise, not just the outcome: the deadlock path is only + # exercised while total acquisition bytes far exceed the byte limit. + # If the simulated frame ever shrinks, fail loudly instead of going + # green without testing anything. (Uncompressed TIFF size ~= frame + # bytes, so on-disk size is a faithful proxy.) + total_bytes = sum(p.stat().st_size for p in images) + limit_bytes = byte_limit_mib * 1024 * 1024 + assert total_bytes > 5 * limit_bytes, ( + f"acquisition totalled {total_bytes / 1024 / 1024:.1f} MiB, not >5x the " + f"{byte_limit_mib:.1f} MiB byte limit — saved images are much smaller than " + "the snapped frame this test sized the limit from, so the backpressure " + "deadlock path is no longer being exercised" + ) + assert harness.tracker.image_count == 12, f"tracker saw {harness.tracker.image_count} images, expected 12" + + # Root acquisition-complete marker. + done_marker = harness.experiment_dir / ".done" + assert done_marker.exists(), f"expected .done marker at {done_marker}" + + # Informational: confirm throttling actually engaged (not asserted -- + # timing-dependent on CI). Reported by the runner via captured logs. + throttled = any("Backpressure throttling" in r.getMessage() for r in caplog.records) + logging.getLogger(__name__).info( + "backpressure throttling engaged during acquisition: %s (byte limit=%.1f MiB, image=%.3f MiB)", + throttled, + byte_limit_mib, + image_size_mib, + ) + finally: + harness.close() diff --git a/software/tests/acceptance/test_full_acquisition.py b/software/tests/acceptance/test_full_acquisition.py new file mode 100644 index 000000000..3e9ff22fd --- /dev/null +++ b/software/tests/acceptance/test_full_acquisition.py @@ -0,0 +1,214 @@ +""" +End-to-end simulation-mode acquisition acceptance tests. + +These drive a full MultiPointController acquisition against simulated hardware +(no GUI, no QApplication) and assert on observable artifacts only: files on +disk, their counts/shapes, coordinate CSVs, the acquisition-parameters JSON, +``.done`` markers, and the per-acquisition log. They intentionally pin CURRENT +master behavior; where the product writes something surprising the assertion +documents it with a ``# pins current behavior:`` note rather than "fixing" it. + +Ordering note: ``acquisition parameters.json`` is written by +``start_new_experiment()`` (i.e. inside ``harness.new_experiment``), so any NZ / +Nt / channel / region configuration that must appear in that JSON has to be set +*before* ``new_experiment`` is called. The helpers below configure the +controller first, then create the experiment folder, then run. +""" + +import glob +import json + +import pytest + +import tifffile + +from control._def import FileSavingOption +from tests.acceptance.conftest import set_file_saving_option +from tests.acceptance.harness import ( + list_image_files, + make_harness, + read_coordinates_csv, + timepoint_dir, +) + +pytestmark = pytest.mark.acceptance + + +def _open_zarr3(array_path): + """Open a Zarr v3 array written by the product's TensorStore writer. + + The store is true Zarr v3 (zarr.json + ``c/`` chunk keys); zarr-python + 2.15.0 cannot read that layout, so we use TensorStore -- the same library + the product uses to write it. + """ + import tensorstore as ts + + return ts.open({"driver": "zarr3", "kvstore": {"driver": "file", "path": str(array_path)}}).result() + + +def _plane_count_and_shape(full_shape): + """Split an (..., Y, X) shape into (num_planes, (Y, X)).""" + plane_shape = tuple(full_shape[-2:]) + num_planes = 1 + for dim in full_shape[:-2]: + num_planes *= dim + return num_planes, plane_shape + + +def test_full_small_acquisition(tmp_path, acquisition_defaults): + """Scenario 1: 2x2 FOVs x 2 channels x NZ=3, Nt=1, individual images.""" + h = make_harness() + try: + # Configure fully BEFORE new_experiment so the params JSON captures NZ/Nt. + h.add_fov_grid("region0", 2, 2) + h.select_channels(2) + h.mpc.set_NZ(3) + h.mpc.set_deltaZ(1.0) # micrometers + h.mpc.set_Nt(1) + h.mpc.set_deltat(0.0) + h.new_experiment(tmp_path / "exp", "small") + + h.run_and_wait(timeout_s=300) + + ed = h.experiment_dir + + # Image counts: 4 FOVs x 3 z x 2 channels. + assert h.mpc.get_acquisition_image_count() == 24 + assert h.tracker.image_count == 24 + + # Timepoint 0 holds exactly 24 individual image files. + tp0 = timepoint_dir(ed, 0) + assert len(list_image_files(tp0)) == 24 + + # Per-timepoint coordinates.csv: one row per region x fov x z_level = 4x3. + inner = read_coordinates_csv(tp0 / "coordinates.csv") + assert len(inner) == 12 + for col in ("region", "fov", "z_level"): + assert col in inner[0] + + # Top-level coordinates.csv: one row per FOV. + top = read_coordinates_csv(ed / "coordinates.csv") + assert len(top) == 4 + + # acquisition parameters.json (literal space in the filename). + with open(ed / "acquisition parameters.json") as f: + params = json.load(f) + assert params["Nz"] == 3 + assert params["Nt"] == 1 + assert params["dz(um)"] == 1.0 # set_deltaZ(1.0) is micrometers + + # .done markers at experiment root and in the timepoint dir. + assert (ed / ".done").exists() + assert (tp0 / ".done").exists() + + # Per-acquisition log exists and is clean of ERROR-level records. + log_text = (ed / "acquisition.log").read_text() + error_lines = [ln for ln in log_text.splitlines() if " - ERROR - " in ln] + assert error_lines == [], f"unexpected ERROR log lines: {error_lines[:5]}" + finally: + h.close() + + +def test_format_parity_ome_tiff_vs_zarr(tmp_path, acquisition_defaults): + """Scenario 2: same geometry (2x2 FOVs x 2 ch x NZ=2, Nt=1) in OME-TIFF and + Zarr v3 represents identical logical content (16 planes, same per-plane + shape and dtype).""" + # The product's ZARR_V3 writer requires tensorstore (and so does reading + # the store back) — skip where it isn't installed, like test_zarr_writer.py. + pytest.importorskip("tensorstore") + + def _run(option, subdir): + set_file_saving_option(acquisition_defaults, option) + h = make_harness() + try: + h.add_fov_grid("region0", 2, 2) + h.select_channels(2) + h.mpc.set_NZ(2) + h.mpc.set_deltaZ(1.0) + h.mpc.set_Nt(1) + h.mpc.set_deltat(0.0) + h.new_experiment(tmp_path / subdir, subdir) + h.run_and_wait(timeout_s=300) + return h.experiment_dir + finally: + h.close() + + # --- OME-TIFF --- + ome_dir = _run(FileSavingOption.OME_TIFF, "ome") + # pins current behavior: OME-TIFF writes one file per FOV under + # {experiment}/ome_tiff/, NOT under the timepoint dir. + ome_files = sorted(glob.glob(str(ome_dir / "**" / "*.ome.tiff"), recursive=True)) + assert len(ome_files) == 4, f"expected 4 per-FOV OME-TIFF files, found {ome_files}" + ome_total_planes = 0 + ome_plane_shapes = set() + ome_dtypes = set() + for path in ome_files: + with tifffile.TiffFile(path) as tf: + series = tf.series[0] + # pins current behavior: per-FOV series is ZCYX (Z=2, C=2). + planes, plane_shape = _plane_count_and_shape(series.shape) + ome_total_planes += planes + ome_plane_shapes.add(plane_shape) + ome_dtypes.add(series.dtype) + assert ome_total_planes == 16 # 4 FOVs x 2 z x 2 ch + assert len(ome_plane_shapes) == 1 + (ome_plane_shape,) = ome_plane_shapes + assert len(ome_dtypes) == 1 + (ome_dtype,) = ome_dtypes + + # --- Zarr v3 --- + zarr_dir = _run(FileSavingOption.ZARR_V3, "zarr") + # pins current behavior: non-well flexible region writes per-FOV stores at + # {experiment}/zarr/{region_id}/fov_{n}.ome.zarr/0 (experiment ROOT, not + # under the timepoint dir as ZarrWriterInfo's docstring path suggests). + zarr_arrays = sorted(glob.glob(str(zarr_dir / "**" / "fov_*.ome.zarr" / "0"), recursive=True)) + assert len(zarr_arrays) == 4, f"expected 4 per-FOV zarr stores, found {zarr_arrays}" + zarr_total_planes = 0 + zarr_plane_shapes = set() + zarr_dtypes = set() + for array_path in zarr_arrays: + arr = _open_zarr3(array_path) + # pins current behavior: per-FOV array is 5D TCZYX (T=1, C=2, Z=2). + planes, plane_shape = _plane_count_and_shape(tuple(arr.shape)) + zarr_total_planes += planes + zarr_plane_shapes.add(plane_shape) + zarr_dtypes.add(arr.dtype.numpy_dtype) + assert zarr_total_planes == 16 # 4 FOVs x 2 z x 2 ch + assert len(zarr_plane_shapes) == 1 + (zarr_plane_shape,) = zarr_plane_shapes + assert len(zarr_dtypes) == 1 + (zarr_dtype,) = zarr_dtypes + + # --- Parity between formats --- + assert ome_total_planes == zarr_total_planes == 16 + assert ome_plane_shape == zarr_plane_shape + assert ome_dtype == zarr_dtype + + +def test_multi_timepoint_smoke(tmp_path, acquisition_defaults): + """Scenario 5: Nt=2, 2x1 FOVs, 1 channel, NZ=1, individual images.""" + h = make_harness() + try: + h.add_fov_grid("region0", 2, 1) + h.select_channels(1) + h.mpc.set_NZ(1) + h.mpc.set_Nt(2) + h.mpc.set_deltat(0.0) + h.new_experiment(tmp_path / "exp", "multi_t") + + h.run_and_wait(timeout_s=300) + + ed = h.experiment_dir + assert h.tracker.image_count == 4 # 2 FOVs x 1 z x 1 ch x 2 timepoints + + for t in (0, 1): + tp = timepoint_dir(ed, t) + assert tp.is_dir(), f"timepoint dir {t} missing" + assert len(list_image_files(tp)) == 2 + rows = read_coordinates_csv(tp / "coordinates.csv") + assert len(rows) == 2 + assert (tp / ".done").exists() + + assert (ed / ".done").exists() + finally: + h.close()