Harden PAQXOS frame validation

docs/data_contracts.md

@@ -33,7 +33,8 @@ All adapters return `diffractomorph_pipeline.model.Run`, whose primary fields ar
 - `channel_ids`: stable, user-defined channel names;
 - `time_min`: chronological elapsed time;
 - `acquisition`: optional named frame-level variables;
-- `stored_reference`: an optional static channel vector;
+- `stored_reference`: an optional stored-reference channel vector, with its
+  selection and validation defined by the adapter;
 - `provenance`: explicit source, adapter, sample, and independent-unit identity.
 
 No fixed channel count, PAQXOS field, compound name, or preparation-date naming
@@ -47,8 +48,11 @@ Adapters are selected explicitly by ID. Gate 1 includes:
 - `tidy_csv`: a vendor-neutral wide CSV with `time_min`, one or more
   `signal_<channel>` columns, optional `acq_<name>` columns, and optional static
   `ref_<channel>` columns;
-- `paqxos_rtf`: the existing Sympatec PAQXOS export reader mapped into the neutral
-  run model.
+- `paqxos_rtf`: the Sympatec PAQXOS export reader mapped into the neutral run
+  model. When the instrument profile declares `channel_ids`, a frame is retained
+  only if it contains those identifiers exactly once, and output columns follow
+  the declared order. The adapter stores the earliest retained frame's reference
+  vector and flags reference variation across retained frames.
 
 Run `dfm-manifest --example --inspect-runs` to validate and inspect the packaged
 four-channel synthetic example without external data.

docs/file_format.md

@@ -47,20 +47,23 @@ Channel, Ref Value, Measured Value
 | `Channel, Ref Value, Measured Value` | column header |
 | `<ch>, <ref>, <measured>` | per detector ring: index, **reference**, **measured** |
 
-## Three quirks the parser must respect
+## Parser contracts
 
 1. **Two intensity columns.** `Measured` is the per-frame signal (analysis
    target); `Ref` is the instrument's stored reference spectrum. **Capture
    both** — `Ref` feeds optional static-baseline subtraction (`I_bgsub = I − ref`)
-   and the staticness check.
-2. **`Ref` is static within a file** — one fixed vector stamped into every frame
-   (per-channel std ≈ 1e-15). The parser validates this and emits a `ref_static`
-   flag; a varying `Ref` means a malformed/concatenated export.
-3. **Frames are listed newest-first** — raw document order is
-   reverse-chronological. The parser **sorts by timestamp ascending** and records
-   `reverse_order_detected`. (The legacy `diffractomorph_core` bug: it reversed
-   the list on a wrong assumption and computed frame-to-frame metrics on
-   non-adjacent pairs.)
+   and the reference-variation check. The adapter retains the earliest valid
+   frame's stored-reference vector and flags variation across retained frames;
+   variation does not automatically exclude a frame.
+2. **Frame structure is validated against detector identifiers.** A PAQXOS
+   instrument profile may declare numeric `channel_ids` in strictly increasing
+   order. A frame is retained only when its timestamp is parseable and every
+   declared identifier occurs exactly once. Missing, additional, substituted,
+   or duplicate identifiers are counted by exclusion reason. Without declared
+   identifiers, the adapter infers a uniquely most frequent exact set and fails
+   closed when equally supported sets are ambiguous.
+3. **Frames may be listed newest-first.** The parser sorts by timestamp ascending
+   and records whether reverse document order was detected.
 
 `Measured` is **not** background-subtracted: in a blank,
 `Measured ≈ Ref ≈ 0.3–0.6` (not zero). `Measured − Ref` is the drug-attributable
@@ -70,7 +73,9 @@ The current parser treats each new `Measurement Time:` marker as the start of a
 sorts parsed timestamps chronologically. It does not depend on `@PAR(1)` as a structural
 separator. In the PAQXOS 5.1 manual, `@PAR(1)` means the first user-defined measurement
 parameter; it is retained in the archived template body for fidelity but is not required by
-the parser or by intensity extraction.
+the parser or by intensity extraction. Missing Copt, reference variation,
+reverse document order, and interframe gaps are reported as run-level flags,
+not additional automatic frame exclusions.
 
 ## Output contract
 
@@ -83,7 +88,8 @@ compatibility properties. These names are not requirements for other adapters.
 
 `dfm-ingest` / `extract_run(emit_csv=True)` writes one tidy CSV per run —
 `frame, time_iso, t_min, copt, I_ch1 … I_ch31` — plus a `<run>_meta.json`
-sidecar carrying the static `ref` vector and `flags`.
+sidecar carrying the earliest retained frame's stored-reference vector and
+run-level flags.
 
 ## Open questions
 

src/diffractomorph_pipeline/ingest.py

@@ -14,8 +14,8 @@
 1. **Two intensity columns per channel** — ``Ref Value`` and ``Measured Value``.
    ``Measured`` is the per-frame signal; ``Ref`` is the stored reference spectrum.
    Both are captured (``Ref`` feeds optional static-baseline subtraction).
-2. **``Ref`` is static within a file** — validated; a varying ``Ref`` flags a
-   malformed/concatenated export.
+2. **``Ref`` is expected to be static within a file** — variation is retained
+   and flagged for review rather than silently changing frame eligibility.
 3. **Frames are listed newest-first** — the parser sorts by timestamp ascending
    and records ``reverse_order_detected``; document order is never trusted.
 
@@ -26,8 +26,10 @@
 
 import re
 import warnings
+from collections import Counter
 from datetime import datetime
 from pathlib import Path
+from typing import Sequence
 
 import numpy as np
 
@@ -66,14 +68,17 @@ def _parse_frames(text: str) -> list[dict]:
             if cur is not None and cur["channels"]:
                 frames.append(cur)
             cur = {"name": pending_name, "time": line.split(":", 1)[1].strip(),
-                   "copt": float("nan"), "channels": {}}
+                   "copt": float("nan"), "channels": {}, "duplicate_channels": set()}
         elif line.startswith("Optical Concentration:") and cur is not None:
             m = re.search(rf"({_NUM})", line.split(":", 1)[1])
             cur["copt"] = float(m.group(1)) if m else float("nan")
         elif cur is not None:
             m = _CHANNEL_ROW.match(line)
             if m:
-                cur["channels"][int(m.group(1))] = (float(m.group(2)), float(m.group(3)))
+                channel = int(m.group(1))
+                if channel in cur["channels"]:
+                    cur["duplicate_channels"].add(channel)
+                cur["channels"][channel] = (float(m.group(2)), float(m.group(3)))
     if cur is not None and cur["channels"]:
         frames.append(cur)
     return frames
@@ -86,6 +91,7 @@ def extract_run(
     run_kind: str | None = None,
     ref_static_tol: float = 1e-6,
     emit_csv: bool = False,
+    expected_channel_ids: Sequence[int | str] | None = None,
 ) -> RawRun:
     """Parse one PAQXOS RTF into a :class:`RawRun` (spec §4).
 
@@ -99,30 +105,78 @@ def extract_run(
         Max per-channel std of ``ref`` across frames for the static check.
     emit_csv
         If true, also write the CSV mirror + meta sidecar next to the source.
+    expected_channel_ids
+        Detector-channel identifiers declared by the instrument profile. When
+        supplied, output columns follow this declared order. When omitted, the
+        parser infers the unique most frequently observed exact channel set and
+        fails closed if equally supported sets are ambiguous.
     """
     from striprtf.striprtf import rtf_to_text
 
     path = Path(path)
     text = rtf_to_text(path.read_text())
     frames = _parse_frames(text)
 
-    # Canonical channel set = the channel list of the most complete frame.
     if not frames:
         raise ValueError(f"No frames parsed from {path}")
-    canonical = sorted(max(frames, key=lambda fr: len(fr["channels"]))["channels"].keys())
-    n_ch = len(canonical)
-
-    # Drop malformed frames (wrong channel count); fail only if <2 remain (§8).
-    good = [fr for fr in frames if sorted(fr["channels"].keys()) == canonical]
-    dropped = len(frames) - len(good)
-    if len(good) < 2 and len(frames) >= 2:
-        # too many dropped to trust — but a genuine single-frame file is allowed below
-        good = [fr for fr in frames if len(fr["channels"]) == n_ch] or good
-    if len(good) < 1:
-        raise ValueError(f"{path}: no valid frames after channel-count filtering")
+
+    for frame in frames:
+        try:
+            frame["parsed_time"] = datetime.strptime(frame["time"], _TIME_FMT)
+        except ValueError:
+            frame["parsed_time"] = None
+
+    if expected_channel_ids is not None:
+        try:
+            canonical = [int(str(channel)) for channel in expected_channel_ids]
+        except ValueError as exc:
+            raise ValueError("expected_channel_ids must contain integer identifiers") from exc
+        if not canonical:
+            raise ValueError("expected_channel_ids must not be empty")
+        if len(set(canonical)) != len(canonical):
+            raise ValueError("expected_channel_ids must be unique")
+        if canonical != sorted(canonical):
+            raise ValueError("expected_channel_ids must be in strictly increasing order")
+        channel_source = "argument"
+    else:
+        eligible_sets = [
+            tuple(sorted(frame["channels"]))
+            for frame in frames
+            if not frame["duplicate_channels"] and frame["parsed_time"] is not None
+        ]
+        if not eligible_sets:
+            raise ValueError(f"{path}: no structurally valid frames from which to infer channels")
+        counts = Counter(eligible_sets)
+        top_count = max(counts.values())
+        leaders = [channel_set for channel_set, count in counts.items() if count == top_count]
+        if len(leaders) != 1:
+            rendered = ", ".join(str(list(channel_set)) for channel_set in sorted(leaders))
+            raise ValueError(
+                f"{path}: ambiguous detector-channel sets ({rendered}); "
+                "declare expected_channel_ids in the instrument profile"
+            )
+        canonical = list(leaders[0])
+        channel_source = "inferred"
+
+    expected_set = set(canonical)
+    dropped_reasons: Counter[str] = Counter()
+    good = []
+    for frame in frames:
+        if frame["duplicate_channels"]:
+            dropped_reasons["duplicate_channel_ids"] += 1
+        elif frame["parsed_time"] is None:
+            dropped_reasons["unparseable_timestamp"] += 1
+        elif set(frame["channels"]) != expected_set:
+            dropped_reasons["channel_set_mismatch"] += 1
+        else:
+            good.append(frame)
+    dropped = sum(dropped_reasons.values())
+    if not good:
+        detail = ", ".join(f"{key}={value}" for key, value in sorted(dropped_reasons.items()))
+        raise ValueError(f"{path}: no structurally valid frames ({detail})")
 
     # Document-order timestamps → detect reverse ordering, then sort ascending (§2.3).
-    times_doc = [datetime.strptime(fr["time"], _TIME_FMT) for fr in good]
+    times_doc = [fr["parsed_time"] for fr in good]
     reverse_order_detected = len(times_doc) > 1 and times_doc[0] > times_doc[-1]
     order = np.argsort(times_doc)
     good = [good[i] for i in order]
@@ -131,7 +185,7 @@ def extract_run(
     I = np.array([[fr["channels"][c][1] for c in canonical] for fr in good], dtype=float)
     ref_all = np.array([[fr["channels"][c][0] for c in canonical] for fr in good], dtype=float)
     copt = np.array([fr["copt"] for fr in good], dtype=float)
-    times = [datetime.strptime(fr["time"], _TIME_FMT) for fr in good]
+    times = [fr["parsed_time"] for fr in good]
     t0 = times[0]
     t_min = np.array([(t - t0).total_seconds() / 60.0 for t in times])
 
@@ -158,6 +212,8 @@ def extract_run(
         "max_gap_min": max_gap_min,
         "n_frames": int(I.shape[0]),
         "dropped_frames": int(dropped),
+        "dropped_frame_reasons": dict(sorted(dropped_reasons.items())),
+        "expected_channel_ids_source": channel_source,
         "copt_nan": copt_nan,
         "run_kind_inferred": inferred,
     }

src/diffractomorph_pipeline/io/paqxos_rtf.py

@@ -9,7 +9,22 @@ class PaqxosRtfReader:
     adapter_id = "paqxos_rtf"
 
     def read(self, spec) -> Run:
-        run = ingest.extract_run(spec.source, run_kind=spec.run_kind)
+        return self._read(spec)
+
+    def read_with_instrument_profile(self, spec, parameters) -> Run:
+        """Read using structural expectations declared by the instrument profile."""
+        expected = parameters.get("channel_ids")
+        run = self._read(spec, expected_channel_ids=expected)
+        if expected is not None:
+            run.flags["expected_channel_ids_source"] = "instrument_profile"
+        return run
+
+    def _read(self, spec, expected_channel_ids=None) -> Run:
+        run = ingest.extract_run(
+            spec.source,
+            run_kind=spec.run_kind,
+            expected_channel_ids=expected_channel_ids,
+        )
         run.provenance = RunProvenance(
             run_id=spec.run_id,
             source_path=str(spec.source),

src/diffractomorph_pipeline/study/manifest.py

@@ -212,16 +212,26 @@ def _validate_loaded_run(self, spec: RunSpec, run: Run) -> Run:
             )
         return run
 
+    def _read_spec(self, spec: RunSpec) -> Run:
+        reader = get_reader(spec.adapter)
+        instrument = self.require_profile("instrument")
+        profile_aware_read = getattr(reader, "read_with_instrument_profile", None)
+        if profile_aware_read is not None:
+            run = profile_aware_read(spec, instrument.parameters)
+        else:
+            run = reader.read(spec)
+        return self._validate_loaded_run(spec, run)
+
     def read_run(self, run_id: str) -> Run:
         """Read one run through the adapter explicitly declared in the manifest."""
         matches = [spec for spec in self.runs if spec.run_id == run_id]
         if not matches:
             raise KeyError(f"unknown run_id {run_id!r}")
         spec = matches[0]
-        return self._validate_loaded_run(spec, get_reader(spec.adapter).read(spec))
+        return self._read_spec(spec)
 
     def read_all_runs(self) -> tuple[Run, ...]:
-        return tuple(self._validate_loaded_run(spec, get_reader(spec.adapter).read(spec)) for spec in self.runs)
+        return tuple(self._read_spec(spec) for spec in self.runs)
 
 
 def bundled_example_manifest() -> Path: