fix(stage): tolerate a corrupted cache/last_coords.txt at startup

software/control/widgets.py

@@ -4361,16 +4361,14 @@ def add_components(self, show_trigger_options, show_display_options, show_autole
 
         # Single-frame capture, for light sensitive samples where a free-running
         # live stream would bleach or damage the sample while settings are dialed in.
+        # Illumination is on for that one exposure only in software/hardware trigger
+        # mode; in continuous mode it stays on for up to two exposures.
         self.btn_snap = QPushButton("Snap")
         self.btn_snap.setCheckable(False)
         self.btn_snap.setDefault(False)
         self.btn_snap.setStyleSheet("background-color: #C2C2FF")
         self.btn_snap.setSizePolicy(sizePolicy)
-        self.btn_snap.setToolTip(
-            "Acquire a single frame with the current Live Configuration. "
-            "In software/hardware trigger mode the illumination is on for that one "
-            "exposure only; in continuous mode it stays on for up to two exposures."
-        )
+        self.btn_snap.setToolTip("View a single frame with the current Live Configuration, to reduce photobleaching.")
 
         # line 3: exposure time and analog gain associated with the current mode
         self.entry_exposureTime = QDoubleSpinBox()

software/squid/stage/utils.py

@@ -1,4 +1,5 @@
 from typing import Optional, Callable
+import math
 import os
 
 import squid.logging
@@ -9,36 +10,52 @@
 
 _log = squid.logging.get_logger(__package__)
 _DEFAULT_CACHE_PATH = "cache/last_coords.txt"
-
-"""
-Attempts to load a cached stage position and return it.
-"""
+_MAX_CACHE_BYTES = 4096
 
 
 def get_cached_position(cache_path=_DEFAULT_CACHE_PATH) -> Optional[Pos]:
+    """Load the stage position written by cache_position(), or None if there is no usable one.
+
+    A cache file that cannot be read or parsed (truncated by a crash mid-write, edited by hand,
+    filled with garbage, ...) is treated exactly like a missing one: a warning is logged and None
+    is returned, so an unreadable cache cannot stop the software from starting.  A well-formed but
+    stale position is returned as-is; it is not range checked against the stage's axis limits.
+    """
     if not os.path.isfile(cache_path):
         _log.debug(f"Cache file '{cache_path}' not found, no cached pos found.")
         return None
-    with open(cache_path, "r") as f:
-        for line in f:
-            try:
-                x, y, z = line.strip("\n").strip().split(",")
-                x = float(x)
-                y = float(y)
-                z = float(z)
-                return Pos(x_mm=x, y_mm=y, z_mm=z, theta_rad=None)
-            except RuntimeError as e:
-                raise e
-                pass
-    return None
 
+    try:
+        with open(cache_path, "r") as f:
+            # A valid cache is one short line, so cap the read: a stray large file at this path
+            # must not pull hundreds of MB into memory during startup.  Read one byte past the cap
+            # so an oversized file is rejected below instead of parsed from its truncated prefix.
+            contents = f.read(_MAX_CACHE_BYTES + 1)
+    except (OSError, UnicodeDecodeError) as e:
+        _log.warning(
+            f"Cached position file '{cache_path}' could not be read ({e}). "
+            "Continuing as if there is no cached position."
+        )
+        return None
+
+    try:
+        if len(contents) > _MAX_CACHE_BYTES:
+            raise ValueError(f"file is larger than {_MAX_CACHE_BYTES} bytes")
+        x, y, z = (float(v) for v in contents.strip().split(","))
+        if not all(math.isfinite(v) for v in (x, y, z)):
+            raise ValueError("non-finite coordinate")
+    except ValueError as e:
+        _log.warning(
+            f"Cached position file '{cache_path}' is corrupted ({e}), contents={contents[:100].strip()!r}. "
+            "Continuing as if there is no cached position."
+        )
+        return None
 
-"""
-Write out the current x, y, z position, in mm, so we can use it later as a cached position.
-"""
+    return Pos(x_mm=x, y_mm=y, z_mm=z, theta_rad=None)
 
 
 def cache_position(pos: Pos, stage_config: StageConfig, cache_path=_DEFAULT_CACHE_PATH):
+    """Write out the current x, y, z position, in mm, so we can use it later as a cached position."""
     if stage_config is not None:  # StageConfig not implemented for Prior stage
         x_min = stage_config.X_AXIS.MIN_POSITION
         x_max = stage_config.X_AXIS.MAX_POSITION
@@ -50,9 +67,13 @@ def cache_position(pos: Pos, stage_config: StageConfig, cache_path=_DEFAULT_CACH
             raise ValueError(
                 f"Position {pos} is not cacheable because it is outside of the min/max of at least one axis. x_range=({x_min}, {x_max}), y_range=({y_min}, {y_max}), z_range=({z_min}, {z_max})"
             )
-    with open(cache_path, "w") as f:
-        _log.debug(f"Writing position={pos} to cache path='{cache_path}'")
+    _log.debug(f"Writing position={pos} to cache path='{cache_path}'")
+    # Atomic write: a crash between truncating and writing would leave a partial file that
+    # get_cached_position() can only discard, losing the position it was meant to preserve.
+    tmp_path = f"{cache_path}.tmp"
+    with open(tmp_path, "w") as f:
         f.write(",".join([str(pos.x_mm), str(pos.y_mm), str(pos.z_mm)]))
+    os.replace(tmp_path, cache_path)
 
 
 def _move_to_loading_position_impl(stage: AbstractStage, is_wellplate: bool):

software/tests/squid/test_stage.py

@@ -1,3 +1,5 @@
+import builtins
+import logging
 import pytest
 import tempfile
 
@@ -36,6 +38,107 @@ def test_position_caching():
     assert p_read == p
 
 
+def test_get_cached_position_returns_none_when_cache_file_is_missing(tmp_path):
+    assert squid.stage.utils.get_cached_position(cache_path=str(tmp_path / "missing.txt")) is None
+
+
+@pytest.mark.parametrize(
+    "contents",
+    [
+        pytest.param(b"garbage", id="garbage"),
+        pytest.param(b"1.0,2.0", id="too-few-fields"),
+        pytest.param(b"1.0,2.0,3.0,4.0", id="too-many-fields"),
+        pytest.param(b"1.0,abc,3.0", id="bad-field"),
+        pytest.param(b"", id="empty"),
+        pytest.param(b"nan,2.0,3.0", id="nan"),
+        pytest.param(b"1.0,inf,3.0", id="inf"),
+        pytest.param(b"\xff\xfe\x00\x80 not text", id="undecodable"),
+    ],
+)
+def test_get_cached_position_treats_corrupted_file_as_no_cache(tmp_path, caplog, contents):
+    """A corrupted cache must not raise (it would abort startup); it warns and behaves like a missing cache."""
+    cache_path = tmp_path / "last_coords.txt"
+    cache_path.write_bytes(contents)
+
+    with caplog.at_level(logging.WARNING, logger="squid"):
+        assert squid.stage.utils.get_cached_position(cache_path=str(cache_path)) is None
+
+    assert any(
+        r.levelno == logging.WARNING and str(cache_path) in r.getMessage() for r in caplog.records
+    ), "expected a warning naming the corrupted cache file"
+
+
+def test_get_cached_position_rejects_a_file_larger_than_the_read_cap(tmp_path, caplog):
+    """A valid prefix padded past the read cap must be rejected, not parsed from the truncated prefix."""
+    cache_path = tmp_path / "last_coords.txt"
+    valid_prefix = "11.0,22.0,1.0"
+    padding = " " * (squid.stage.utils._MAX_CACHE_BYTES - len(valid_prefix))
+    cache_path.write_text(valid_prefix + padding + "garbage past the cap")
+
+    with caplog.at_level(logging.WARNING, logger="squid"):
+        assert squid.stage.utils.get_cached_position(cache_path=str(cache_path)) is None
+
+    assert any(
+        r.levelno == logging.WARNING and str(cache_path) in r.getMessage() for r in caplog.records
+    ), "expected a warning naming the oversized cache file"
+
+
+def test_get_cached_position_rejects_a_second_line(tmp_path):
+    """Only a single coordinate triple is a valid cache; extra lines mean the file is not ours."""
+    cache_path = tmp_path / "last_coords.txt"
+    cache_path.write_text("11.0,22.0,1.0\n44.0,55.0,2.0\n")
+
+    assert squid.stage.utils.get_cached_position(cache_path=str(cache_path)) is None
+
+
+def test_get_cached_position_warning_shows_the_corrupt_contents(tmp_path, caplog):
+    """The warning must quote what was in the file, so a bad cache can be diagnosed from a log alone."""
+    cache_path = tmp_path / "last_coords.txt"
+    cache_path.write_text("1.0,not-a-number,3.0")
+
+    with caplog.at_level(logging.WARNING, logger="squid"):
+        squid.stage.utils.get_cached_position(cache_path=str(cache_path))
+
+    assert any("contents='1.0,not-a-number,3.0'" in r.getMessage() for r in caplog.records)
+
+
+class _WriteFailsFile:
+    """Wraps a real file object so opening (and truncating) succeeds but writing fails."""
+
+    def __init__(self, wrapped):
+        self._wrapped = wrapped
+
+    def __enter__(self):
+        self._wrapped.__enter__()
+        return self
+
+    def __exit__(self, *exc_info):
+        return self._wrapped.__exit__(*exc_info)
+
+    def write(self, _data):
+        raise OSError("simulated disk full")
+
+
+def test_cache_position_keeps_the_previous_position_when_the_write_fails(tmp_path, monkeypatch):
+    """An interrupted write must not leave a truncated cache behind - that is what corrupts the file."""
+    cache_path = str(tmp_path / "last_coords.txt")
+    stage_config = squid.config.get_stage_config()
+    already_cached = squid.abc.Pos(x_mm=11.0, y_mm=22.0, z_mm=1.0, theta_rad=None)
+    squid.stage.utils.cache_position(pos=already_cached, stage_config=stage_config, cache_path=cache_path)
+
+    real_open = builtins.open
+    monkeypatch.setattr(builtins, "open", lambda *args, **kwargs: _WriteFailsFile(real_open(*args, **kwargs)))
+    with pytest.raises(OSError):
+        squid.stage.utils.cache_position(
+            pos=squid.abc.Pos(x_mm=33.0, y_mm=44.0, z_mm=2.0, theta_rad=None),
+            stage_config=stage_config,
+            cache_path=cache_path,
+        )
+    monkeypatch.undo()
+
+    assert squid.stage.utils.get_cached_position(cache_path=cache_path) == already_cached
+
+
 # --- PI V-308 / C-414 focus stage --------------------------------------------