fix(pipeline): flag dark-correction intervals that span a missed dark

docs/SciencePipeline.md

@@ -73,7 +73,7 @@ Three things characterise this design and make it auditable:
 | `tcl` | `(N,)` int64 \| None | TelemetryIngestStage (or replay source) | Laser trigger counter |
 | `abs_frame_ids` | `(N,)` int64 | FrameClassificationStage | Monotonic unwrapped frame counter |
 | `frame_type` | `(N,)` `<U8` | FrameClassificationStage | One of `"warmup"`, `"dark"`, `"light"`, `"stale"` |
-| `quality` | `(N,)` `<U14` | TimestampRepairStage | `"ok"`, `"ts_corrected"` (timestamp rewritten), or `"nan_filled"` (synthetic gap placeholder) — see §5.4 |
+| `quality` | `(N,)` `<U14` | TimestampRepairStage | `"ok"`, `"ts_corrected"` (timestamp rewritten), or `"nan_filled"` (synthetic gap placeholder) — see §5.4. Corrected frames can also carry `"wide_interval"`, set by DarkCorrectionStage when their interval spans a missed scheduled dark (§5.8.3) |
 | `mean_raw` | `(N, 2, 8)` float32 | MomentsStage | First moment μ₁ of raw histogram (NaN where count == 0) |
 | `std_raw` | `(N, 2, 8)` float32 | MomentsStage | √(μ₂ − μ₁²) of raw histogram |
 | `contrast_raw` | always `None` | MomentsStage | Reserved; pedestal-subtracted contrast is computed downstream |
@@ -96,6 +96,7 @@ Stages produce events when something doesn't fit cleanly into per-frame arrays.
 | `LiveEmit(channel, payload)` | `Tee` stages, `SideAverageStage` (realtime path, `"live_side"`) | The named channel's sinks |
 | `IntervalClosed(corrected_batch)` | `DarkCorrectionStage` (per-camera; enriched + stencilled downstream), `SideAverageStage` (reduced-mode `cam_id=-1` side averages) | `"final"` channel sinks |
 | `DarkIntegrityWarning(...)` | `DarkIntegrityGuard` (inside DarkCorrectionStage) | `"diagnostics"` |
+| `MissedDarkWarning(side, cam_id, left_abs, right_abs, missed_abs_ids)` | DarkCorrectionStage — an interval closed with one or more scheduled dark positions strictly inside it (§5.8.3) | `"diagnostics"` |
 | `StencilFallback(...)` | `DarkFrameQuadraticStencil` (inside DarkFrameHoldStage) | `"diagnostics"` |
 | `PipelineError(...)` | `ScanRunner` (a stage raised; the batch was dropped, state preserved) | `"diagnostics"` |
 | `TimestampMisalignmentWindow(...)` | `TimestampRepairStage` (per-side coalesced window; the terminal stop-frame artifact — the firmware's laser-off frame fires ~150 ms off-grid at every scan stop — is reclassified at INFO and NOT reported) | `"diagnostics"` |
@@ -480,6 +481,8 @@ On every dark frame: append to history and either set `_left` (first dark) or se
 
 When `_right` is set, `flush()` returns a closed `Interval` and rolls `_left ← _right` for the next pass.
 
+**Missed darks (issue #175).** A scheduled dark that never arrives (dropped frame, camera dropout, stale rejection) leaves no `"dark"` row, so the interval stays open until the next dark that does arrive and is interpolated across the wider span. Before correcting each closed interval (mid-scan and the terminal flush alike), the stage counts the scheduled dark positions strictly inside `[left_abs, right_abs]` — `(abs_id − 1) mod dark_interval == 0`, with `dark_interval` forwarded by the factory. If there are any, the interval is still corrected (keep, not discard), but its light frames get `quality = "wide_interval"` (synthetic `nan_filled` rows keep their own flag), the stencilled dark row inherits that flag from its neighbours, a WARNING is logged, and a `MissedDarkWarning` names the missed abs_ids. `"wide_interval"` ranks worst in the side-average quality, because unlike `nan_filled` it contributes finite but biased values.
+
 #### 5.8.4 LinearInterpolation — batched correction (§8.1, §8.2)
 
 For each light frame `lf` in the closed interval `[D_prev, D_next]`:

omotion/pipeline/__init__.py

@@ -25,6 +25,7 @@
     IntervalClosed,
     LiveEmit,
     DarkIntegrityWarning,
+    MissedDarkWarning,
     StencilFallback,
     TerminalDarkResult,
     PipelineError,
@@ -44,7 +45,8 @@
 
 __all__ = [
     "FrameBatch", "BatchEvent", "IntervalClosed", "LiveEmit",
-    "DarkIntegrityWarning", "StencilFallback", "TerminalDarkResult",
+    "DarkIntegrityWarning", "MissedDarkWarning", "StencilFallback",
+    "TerminalDarkResult",
     "PipelineError", "TimestampMisalignmentWindow",
     "Pipeline", "Stage", "Tee",
     "ScanRunner", "CriticalSinkError",

omotion/pipeline/batch.py

@@ -53,6 +53,20 @@ class DarkIntegrityWarning(BatchEvent):
     threshold: float
 
 
+@dataclass
+class MissedDarkWarning(BatchEvent):
+    """One or more scheduled dark frames never arrived for a camera, so its
+    dark-correction interval spans from the dark before the gap to the dark
+    after it (issue #175). The interval is still corrected, by interpolating
+    the baseline across the wider span; its light frames carry
+    quality == "wide_interval". Diagnostic, not a drop signal."""
+    side: str
+    cam_id: int
+    left_abs: int
+    right_abs: int
+    missed_abs_ids: tuple[int, ...]
+
+
 @dataclass
 class StencilFallback(BatchEvent):
     """The 4-point dark-frame quadratic stencil fell back to a simpler scheme
@@ -443,6 +457,8 @@ class FrameBatch:
     # "ok" = device timestamp passed through unchanged
     # "ts_corrected" = timestamp replaced by re-anchoring interpolation
     # "nan_filled" = synthetic row for a missing frame (zero histogram)
+    # DarkCorrectionStage later sets "wide_interval" on the corrected light
+    # frames of an interval that spans a missed scheduled dark (#175).
     quality:        Optional[np.ndarray] = None
 
     # ── Event queue ──────────────────────────────────────────────────────

omotion/pipeline/factory.py

@@ -86,6 +86,7 @@ def not_warmup_or_stale(frame_type: str) -> bool:
             batch_estimator=LinearInterpolation(),
             pedestals=pedestals,
             realtime_history_size=realtime_dark_history_size,
+            dark_interval=dark_interval,
         ),
 
         ShotNoiseCorrectionStage(pedestals=pedestals, camera_gain_map=CAMERA_GAIN_MAP),

omotion/pipeline/stages/dark.py

@@ -17,7 +17,10 @@
 
 import numpy as np
 
-from ..batch import DarkIntegrityWarning, FrameBatch, IntervalClosed, TerminalDarkResult
+from ..batch import (
+    DarkIntegrityWarning, FrameBatch, IntervalClosed, MissedDarkWarning,
+    TerminalDarkResult,
+)
 from ..pedestal import SensorPedestals
 
 
@@ -33,6 +36,22 @@
 _TERMINAL_FSYNC_ABS_OFFSET = 0
 
 
+def scheduled_darks_between(left_abs: int, right_abs: int,
+                            dark_interval: int) -> list[int]:
+    """Scheduled dark abs_ids strictly between two interval boundaries.
+
+    After the first dark (discard_count + 1), the classifier types a frame
+    dark when (abs_id - 1) % dark_interval == 0 (classify._is_dark). An
+    interval's left edge is always a dark, so it sits at or after the first
+    dark and that rule alone covers everything inside the interval. A
+    non-empty result means those darks never arrived (issue #175).
+    """
+    if dark_interval <= 0:
+        return []
+    first = left_abs + 1 + (-left_abs) % dark_interval
+    return list(range(first, right_abs, dark_interval))
+
+
 @dataclass(frozen=True)
 class DarkObservation:
     """One dark-frame measurement for one camera."""
@@ -402,8 +421,12 @@ def __init__(self, *,
                  batch_estimator: LinearInterpolation,
                  pedestals: Optional[SensorPedestals] = None,
                  realtime_history_size: int = 4,
-                 integrity_max_above_pedestal: float = 5.0):
+                 integrity_max_above_pedestal: float = 5.0,
+                 dark_interval: int = 600):
         self._realtime = realtime_estimator
+        # Same value the classifier schedules darks with; used only to spot
+        # an interval that spans a missed dark (_flag_missed_darks).
+        self._dark_interval = int(dark_interval)
         self._batch = batch_estimator
         self._pedestals = pedestals or SensorPedestals(left=64.0, right=64.0)
         self._history = DarkHistory(max_darks=realtime_history_size)
@@ -438,10 +461,52 @@ def _emit_interval(
         quadratic stencil.
         """
         side, cam_id = key
+        self._flag_missed_darks(key, interval, events)
         corrected = self._batch.correct_interval(interval, side=side, cam_id=cam_id)
         corrected.left_t = interval.left.obs.t
         events.append(IntervalClosed(corrected_batch=corrected))
 
+    def _flag_missed_darks(
+        self,
+        key: "tuple[str, int]",
+        interval: "Interval",
+        events: list,
+    ) -> None:
+        """Detect and flag an interval that spans a missed scheduled dark.
+
+        A dark goes missing when its frame is dropped, the camera drops out,
+        or the frame is rejected as stale. Its row then never reaches this
+        stage typed "dark" (a gap-fill row is typed "light" and carries NaN
+        stats), so the interval stays open until the next dark that does
+        arrive. The baseline is then interpolated across the wider span.
+        The interval is kept, not discarded: its light frames are flagged
+        "wide_interval" and a MissedDarkWarning is emitted. Synthetic
+        nan_filled rows keep their own flag, which consumers use to
+        recognise them as placeholders.
+        """
+        missed = scheduled_darks_between(
+            interval.left_abs, interval.right_abs, self._dark_interval,
+        )
+        if not missed:
+            return
+        side, cam_id = key
+        logger.warning(
+            "missed scheduled dark: side=%s cam_id=%d expected abs_id %s; "
+            "dark correction interpolated across [%d, %d] (%d frames instead "
+            "of <= %d); flagging its frames wide_interval",
+            side, cam_id, ", ".join(str(a) for a in missed),
+            interval.left_abs, interval.right_abs,
+            interval.right_abs - interval.left_abs, self._dark_interval,
+        )
+        events.append(MissedDarkWarning(
+            side=side, cam_id=cam_id,
+            left_abs=interval.left_abs, right_abs=interval.right_abs,
+            missed_abs_ids=tuple(missed),
+        ))
+        for lf in interval.light_frames:
+            if lf.quality != "nan_filled":
+                lf.quality = "wide_interval"
+
     def process(self, batch: FrameBatch) -> FrameBatch:
         n = batch.frame_ids.shape[0]
         baseline_rt = np.full((n, 2, 8), np.nan, dtype=np.float32)

omotion/pipeline/stages/dark_frame_hold.py

@@ -171,6 +171,13 @@ def _interp(attr: str) -> float:
 
         temp_c = _interp("temp_c") if right1.temp_c is not None else None
 
+        # The dark row is built from its neighbours, so a neighbour from an
+        # interval that spans a missed dark (#175) makes it just as suspect.
+        quality = "ok"
+        if any(f is not None and f.quality == "wide_interval"
+               for f in (left2, left1, right1, right2)):
+            quality = "wide_interval"
+
         return EnrichedCorrectedFrame(
             abs_frame_id=d_prev_abs,
             t=d_prev_t,
@@ -181,7 +188,7 @@ def _interp(attr: str) -> float:
             contrast=_interp("contrast"),
             bfi=_interp("bfi"),
             bvi=_interp("bvi"),
-            quality="ok",
+            quality=quality,
             temp_c=temp_c,
         )
 

omotion/pipeline/stages/side_avg.py

@@ -35,8 +35,11 @@
 _SIDE_INT_TO_STR = ("left", "right")
 
 # Higher rank = worse quality; the side average inherits the worst quality
-# of any camera that contributed to it. Mirrors sinks._QUALITY_RANK.
-_QUALITY_RANK = {"ok": 0, "ts_corrected": 1, "nan_filled": 2}
+# of any camera that contributed to it. "wide_interval" (#175) ranks worst:
+# a nan_filled camera is NaN and drops out of the average on its own, but a
+# wide-interval camera contributes finite values built on a stretched dark
+# baseline, so it is the one flag that can bias the average unnoticed.
+_QUALITY_RANK = {"ok": 0, "ts_corrected": 1, "nan_filled": 2, "wide_interval": 3}
 
 
 def _mask_to_cam_indices(mask: int) -> np.ndarray:

tests/test_pipeline/test_corrected_side_avg_stage.py

@@ -227,3 +227,17 @@ def test_reset_clears_pending_window():
     flush = _batch()
     stage.on_scan_stop(flush)
     assert _avg_frames(flush) == []
+
+
+
+def test_wide_interval_outranks_nan_filled():
+    """A wide-interval camera contributes finite but biased values, so its
+    flag must survive alongside a nan_filled camera (issue #175)."""
+    stage = _stage()
+    filled = _ef(12, 5.0, "left", 0, 2.0, 20.0)
+    filled.quality = "nan_filled"
+    wide = _ef(12, 5.0, "left", 1, 6.0, 60.0)
+    wide.quality = "wide_interval"
+    b = _batch([_interval(10, 20, [filled]), _interval(10, 20, [wide])])
+    stage.process(b)
+    assert _avg_frames(b)[0].quality == "wide_interval"

tests/test_pipeline/test_dark_correction_stage.py

@@ -717,3 +717,115 @@ def test_corrected_frame_quality_defaults_to_ok():
         mean=36.0, std=4.8, contrast=0.13, bfi=5.0, bvi=5.0,
     )
     assert ef.quality == "ok"
+
+
+# ── Missed scheduled darks (issue #175) ────────────────────────────────────
+
+from omotion.pipeline.batch import MissedDarkWarning
+from omotion.pipeline.stages.dark import scheduled_darks_between
+
+
+@pytest.mark.parametrize("left, right, expected", [
+    (10, 601, []),            # first interval: discard+1 -> first scheduled dark
+    (601, 1201, []),          # ordinary interval
+    (10, 1201, [601]),        # first scheduled dark after discard missed
+    (601, 1801, [1201]),
+    (601, 2401, [1201, 1801]),
+    (601, 1500, [1201]),      # terminal boundary off the schedule
+    (601, 1150, []),          # terminal boundary before the next dark
+    (10, 605, [601]),         # terminal just after a missed first dark
+])
+def test_scheduled_darks_between(left, right, expected):
+    assert scheduled_darks_between(left, right, 600) == expected
+
+
+def test_scheduled_darks_between_disabled_without_interval():
+    assert scheduled_darks_between(601, 5000, 0) == []
+
+
+def _missed_dark_batch():
+    """dark@601, lights, a nan_filled placeholder where the dark at 1201
+    should have been, more lights, dark@1801."""
+    abs_ids = [601, 602, 1200, 1201, 1202, 1800, 1801]
+    types = ["dark", "light", "light", "light", "light", "light", "dark"]
+    u1 = [65.0, 500.0, 505.0, np.nan, 510.0, 515.0, 66.0]
+    sd = [3.0, 20.0, 20.0, np.nan, 21.0, 21.0, 3.0]
+    n = len(abs_ids)
+    mean = np.array(u1, dtype=np.float32).reshape(n, 1, 1) * np.ones((1, 2, 8), dtype=np.float32)
+    std = np.array(sd, dtype=np.float32).reshape(n, 1, 1) * np.ones((1, 2, 8), dtype=np.float32)
+    batch = _batch(n, types, abs_ids, mean_raw=mean, std_raw=std)
+    batch.quality = np.array(
+        ["ok", "ok", "ok", "nan_filled", "ok", "ts_corrected", "ok"], dtype="<U14")
+    return batch
+
+
+def _stage(**kw):
+    return DarkCorrectionStage(
+        realtime_estimator=HybridRealtimePredictor(),
+        batch_estimator=LinearInterpolation(),
+        dark_interval=600,
+        **kw,
+    )
+
+
+def test_missed_dark_keeps_interval_and_flags_its_frames(caplog):
+    batch = _missed_dark_batch()
+    with caplog.at_level(logging.WARNING, logger="openmotion.sdk.pipeline.stages.dark"):
+        _stage().process(batch)
+
+    warnings = [e for e in batch.events if isinstance(e, MissedDarkWarning)]
+    assert len(warnings) == 1
+    w = warnings[0]
+    assert (w.side, w.cam_id) == ("left", 0)
+    assert (w.left_abs, w.right_abs) == (601, 1801)
+    assert w.missed_abs_ids == (1201,)
+    assert "missed scheduled dark" in caplog.text
+
+    closed = [e for e in batch.events if isinstance(e, IntervalClosed)]
+    assert len(closed) == 1                      # kept, not discarded
+    by_abs = {f.abs_frame_id: f for f in closed[0].corrected_batch.frames}
+    assert sorted(by_abs) == [602, 1200, 1201, 1202, 1800]
+    # Real frames are flagged, ts_corrected included (wide_interval ranks worse)...
+    for a in (602, 1200, 1202, 1800):
+        assert by_abs[a].quality == "wide_interval"
+        assert np.isfinite(by_abs[a].mean)       # still corrected
+    # ...but the synthetic placeholder keeps the flag consumers key on.
+    assert by_abs[1201].quality == "nan_filled"
+
+
+def test_full_interval_is_not_flagged():
+    abs_ids = [601, 602, 1200, 1201]
+    n = len(abs_ids)
+    mean = np.array([65.0, 500.0, 505.0, 66.0], dtype=np.float32).reshape(n, 1, 1) * np.ones((1, 2, 8), dtype=np.float32)
+    std = np.full((n, 2, 8), 5.0, dtype=np.float32)
+    batch = _batch(n, ["dark", "light", "light", "dark"], abs_ids,
+                   mean_raw=mean, std_raw=std)
+    _stage().process(batch)
+
+    assert not [e for e in batch.events if isinstance(e, MissedDarkWarning)]
+    closed = [e for e in batch.events if isinstance(e, IntervalClosed)]
+    assert [f.quality for f in closed[0].corrected_batch.frames] == ["ok", "ok"]
+
+
+def test_missed_dark_in_terminal_interval_is_flagged():
+    """The scan-stop flush closes [last dark, terminal laser-off frame]; a
+    dark missed inside that span is caught the same way."""
+    abs_ids = [601, 602, 1300, 1500]
+    n = len(abs_ids)
+    # 1500 is the firmware's terminal laser-off frame, typed "light".
+    mean = np.array([65.0, 500.0, 505.0, 66.0], dtype=np.float32).reshape(n, 1, 1) * np.ones((1, 2, 8), dtype=np.float32)
+    std = np.full((n, 2, 8), 5.0, dtype=np.float32)
+    batch = _batch(n, ["dark", "light", "light", "light"], abs_ids,
+                   mean_raw=mean, std_raw=std)
+    stage = _stage()
+    stage.process(batch)
+    assert not [e for e in batch.events if isinstance(e, MissedDarkWarning)]
+
+    flush = _batch(0, [], [], mean_raw=np.zeros((0, 2, 8), np.float32),
+                   std_raw=np.zeros((0, 2, 8), np.float32))
+    stage.on_scan_stop(flush)
+    warnings = [e for e in flush.events if isinstance(e, MissedDarkWarning)]
+    assert len(warnings) == 1
+    assert warnings[0].missed_abs_ids == (1201,)
+    closed = [e for e in flush.events if isinstance(e, IntervalClosed)]
+    assert {f.quality for f in closed[0].corrected_batch.frames} == {"wide_interval"}

tests/test_pipeline/test_dark_frame_hold_stage.py

@@ -139,3 +139,19 @@ def test_stencilled_dark_row_temp_none_when_neighbour_unstamped():
     assert dark_row.abs_frame_id == 10
     assert dark_row.temp_c is None
     assert dark_row.bfi == pytest.approx(4.0)  # other metrics still stencil
+
+
+
+def test_stencilled_dark_row_inherits_wide_interval_flag():
+    """The dark row is fabricated from its neighbours, so it is flagged
+    when any neighbour comes from an interval that spans a missed dark
+    (issue #175), and stays "ok" otherwise."""
+    flagged = _enriched_light(12, 0.300, 36.8)
+    flagged.quality = "wide_interval"
+    eci = _stencil_one_interval([_enriched_light(11, 0.275, 36.7), flagged])
+    assert eci.frames[0].abs_frame_id == 10
+    assert eci.frames[0].quality == "wide_interval"
+
+    eci = _stencil_one_interval(
+        [_enriched_light(11, 0.275, 36.7), _enriched_light(12, 0.300, 36.8)])
+    assert eci.frames[0].quality == "ok"

tests/test_pipeline/test_factory.py

@@ -146,3 +146,22 @@ def test_pipeline_order_raw_tee_before_timestamp_repair():
         f"Tee('raw') at index {raw_tee_idx} must come before "
         f"TimestampRepairStage at index {repair_idx}"
     )
+
+
+
+def test_default_pipeline_forwards_dark_interval_to_dark_correction():
+    """Missed-dark detection (issue #175) must use the classifier's
+    schedule; a non-default value proves it is forwarded, not defaulted."""
+    meta = ScanMetadata(
+        scan_id="x", subject_id="y", operator="z",
+        started_at_iso="2026-05-22T00:00:00Z", duration_sec=60,
+        left_camera_mask=0xFF, right_camera_mask=0xFF, reduced_mode=False,
+    )
+    pipeline = default_pipeline(
+        metadata=meta, calibration=_trivial_calibration(),
+        pedestals=SensorPedestals(left=64.0, right=64.0),
+        dark_interval=123,
+    )
+    stages = {stage.name: stage for stage in pipeline.stages}
+    assert stages["frame_classification"].dark_interval == 123
+    assert stages["dark_correction"]._dark_interval == 123