Spectrum batches and file checks

Spectrum batches and file checks

These helpers support reading and centroiding timsTOF data. Existing extraction methods continue to return arrays. Processing diagnostics are an internal development tool, and chromatogram extraction is deferred.

To process DIA or PRM windows with bounded reuse, pass windows in their existing order to iter_window_spectra. Adjacent windows of the same frame share one decode. Each result is a (window, peaks) pair. The numerical array matches window.centroid() with the same settings. No diagnostic accounting runs as part of batch extraction.

from itertools import islice
from tdfpy import DIA, MergePeaksCentroider, iter_window_spectra

with DIA(D_PATH) as reader:
    for window, peaks in iter_window_spectra(
        islice(reader.windows, 4),
        centroid=MergePeaksCentroider(max_peaks=10),
    ):
        assert peaks.shape[1] == 3
        assert window.frame_id > 0

For PRM, pass reader.transitions. Unsorted inputs retain caller order and may decode a frame again. Consume the iterator inside the reader's context. It retains the current frame and uses no global spectrum cache or worker pool. Returned numerical arrays remain usable after the reader closes.

For acquisition checks, run tdfpy validate sample.d or python -m tdfpy validate sample.d. Add --full to decode every binary frame. The command writes JSON and exits with status 0 on success or 1 for a failed check. The Python function returns a structured report:

from tdfpy import validate_acquisition

report = validate_acquisition(D_PATH)
assert report.valid
assert report.frames_checked > 0

Metadata mode checks supported metadata and calibration references. It does not validate compressed payloads. Full mode additionally runs the decoder's integrity checks for every frame, collecting frame-specific failures. Neither mode repairs data or proves numerical equivalence to vendor software.

The built-in extraction and gate paths can share an open reader across worker threads. Metadata needed by those paths is snapshotted when the reader opens. Direct access to td.conn retains SQLite's thread rules. User-written filters must be thread-safe themselves. Wait for workers before closing the reader.

All ion_mobility_type="ccs" raw-spectrum conversions assume charge +1. Raw peaks do not identify charge states. Precursor.ccs uses a known precursor charge when present, falling back to +1 when it is absent. Precursor scan coordinates retain the original fractional metadata value.

For AI agents, an optional MCP server exposes acquisition queries, spectrum extraction, conversions, and file checks without changing the core Python installation.

tdfpy.iter_window_spectra

iter_window_spectra(
    windows: Iterable[DiaWindow | PrmTransition],
    *,
    exclude: ChargeStateRegion | None = None,
    smooth: Smooth | None = None,
    noise: NoiseSpec = None,
    centroid: Centroider | None = None,
    ion_mobility_type: Literal[
        "ook0", "ccs", "voltage"
    ] = "ook0"
) -> Iterator[tuple[DiaWindow | PrmTransition, np.ndarray]]

Yield (window, peaks) pairs, decoding adjacent windows' frame once.

Pass reader.windows or reader.transitions in their existing order. Peaks have shape (N, 3), matching window.centroid with the same options. Only the current frame is retained. Unsorted input preserves caller order and may decode a frame again when it reappears. Keep readers open while consuming the iterator. Results contain no diagnostic or provenance wrappers.

Source code in src/tdfpy/processing.py

def iter_window_spectra(
    windows: Iterable[DiaWindow | PrmTransition],
    *,
    exclude: ChargeStateRegion | None = None,
    smooth: Smooth | None = None,
    noise: NoiseSpec = None,
    centroid: Centroider | None = None,
    ion_mobility_type: Literal["ook0", "ccs", "voltage"] = "ook0",
) -> Iterator[tuple[DiaWindow | PrmTransition, np.ndarray]]:
    """Yield (window, peaks) pairs, decoding adjacent windows' frame once.

    Pass reader.windows or reader.transitions in their existing order. Peaks
    have shape (N, 3), matching window.centroid with the same options. Only the
    current frame is retained. Unsorted input preserves caller order and may
    decode a frame again when it reappears. Keep readers open while consuming
    the iterator. Results contain no diagnostic or provenance wrappers.
    """
    cfg = centroid if centroid is not None else MergePeaksCentroider()
    for (td, frame_id), group in groupby(
        windows, key=lambda w: (w.timsdata, w.frame_id)
    ):
        spectrum = read_spectrum(td, frame_id)
        for window in group:
            td._require_open()
            prepared = _prepare_spectrum(
                spectrum,
                td,
                frame_id,
                scan_range=(window.scan_num_begin, window.scan_num_end),
                exclude=exclude,
                smoothing=smooth,
                noise=noise,
                ion_mobility_type=ion_mobility_type,
            )
            peaks = (
                cfg(prepared, td, frame_id, ion_mobility_type=ion_mobility_type)
                if not prepared.empty
                else np.empty((0, 3), dtype=np.float64)
            )
            yield window, peaks

tdfpy.validate_acquisition

validate_acquisition(
    analysis_dir: str | Path, *, full: bool = False
) -> ValidationReport

Check supported metadata, optionally decoding every frame.

Uses the same strict format and calibration guards as extraction. Collects per-frame failures and continues through the remaining frames. This is structural validation, not a comparison with independent vendor spectra.

Source code in src/tdfpy/validation.py

def validate_acquisition(
    analysis_dir: str | Path, *, full: bool = False
) -> ValidationReport:
    """Check supported metadata, optionally decoding every frame.

    Uses the same strict format and calibration guards as extraction. Collects
    per-frame failures and continues through the remaining frames. This is
    structural validation, not a comparison with independent vendor spectra.
    """
    issues = []
    checked = 0
    failures = (
        OSError,
        ValueError,
        KeyError,
        TypeError,
        sqlite3.Error,
        UnsupportedTdfError,
        UnsupportedCalibrationError,
    )
    try:
        with TimsData(analysis_dir) as td:
            for fid in td.frame_ids:
                checked += 1
                try:
                    frame = td.frame_metadata(fid)
                    if (
                        frame.num_scans < 0
                        or frame.num_peaks < 0
                        or not np.isfinite(frame.time)
                    ):
                        raise ValueError("Invalid frame counts or retention time.")
                    td.calibration_key(fid)
                    # Exercise calibration references and finite conversion values.
                    mz = td.indexToMz(fid, [0])
                    mobility = td.scanNumToOneOverK0(fid, [0])
                    if not np.all(np.isfinite(mz)) or not np.all(np.isfinite(mobility)):
                        raise ValueError("Calibration produces non-finite coordinates.")
                    if full:
                        td.read_frame_arrays(fid)
                except failures as exc:
                    issues.append(ValidationIssue(fid, str(exc)))
    except failures as exc:
        issues.append(ValidationIssue(None, str(exc)))
    return ValidationReport(
        str(analysis_dir), "full" if full else "metadata", checked, tuple(issues)
    )