test_integration.py

"""Integration test: cross-validate tcscopeview against pytcs CSV parsing.

This is the test equivalent of what ``parse_svb.py`` used to do manually.
It requires the real recording files in ``../data/`` and also a working
``pytcs`` install.  Both are checked via fixture skip conditions so the
regular unit-test suite (``test_helpers.py``, ``test_svbfile.py``) is
never affected.

Run only the integration tests with::

    pytest -m integration

Skip them explicitly with::

    pytest -m "not integration"
"""

from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pytest

from tcscopeview import SvbFile


# ── pytcs import (optional dep — not declared in pyproject.toml) ──────────────


def _import_pytcs():
    """Return ``pytcs.ScopeFile``, inserting the sibling source tree if needed."""
    try:
        from pytcs import ScopeFile
    except ModuleNotFoundError:
        _src = Path(__file__).parent.parent.parent / "pytcs"
        sys.path.insert(0, str(_src))
        from pytcs import ScopeFile  # type: ignore[import]
    return ScopeFile


# ── session-scoped loaded pair ────────────────────────────────────────────────


@pytest.fixture(scope="session")
def loaded_pair(svb_path: Path, csv_path: Path):
    """Returns ``(svb, csv_sf)`` with all channels fully loaded.

    Skips the test if pytcs is unavailable.
    """
    try:
        ScopeFile = _import_pytcs()
    except ModuleNotFoundError:
        pytest.skip("pytcs not available — skipping cross-validation tests")

    svb = SvbFile(svb_path)
    svb.load()

    csv_sf = ScopeFile(csv_path)
    csv_sf.load()

    return svb, csv_sf


# ── helpers ───────────────────────────────────────────────────────────────────


def _aligned_arrays(sc, cc, start_ft: int):
    """Return ``(t_svb, t_csv, v_svb, v_csv)`` aligned on shared timestamps.

    When the CSV export drops NaN rows the sample counts differ.  In that case
    we align by rounding timestamps to 0.1 ms and intersecting.
    """
    t_csv_rel = (cc.time - start_ft) / 10_000.0

    if len(sc.values) == len(cc.values):
        return sc.time_ms, t_csv_rel, sc.values, cc.values.astype(np.float64)

    # Align on rounded timestamps
    svb_ms_r = np.round(sc.time_ms, 1)
    csv_ms_r = np.round(t_csv_rel, 1)
    svb_idx  = {float(t): i for i, t in enumerate(svb_ms_r)}
    csv_rows = np.array([i for i, t in enumerate(csv_ms_r) if float(t) in svb_idx])
    svb_rows = np.array([svb_idx[float(csv_ms_r[i])] for i in csv_rows])

    return (
        sc.time_ms[svb_rows],
        t_csv_rel[csv_rows],
        sc.values[svb_rows],
        cc.values[csv_rows].astype(np.float64),
    )


# ── test class ────────────────────────────────────────────────────────────────


@pytest.mark.integration
class TestSvbMatchesCsv:
    """Every channel in the SVB must produce identical data to the CSV export."""

    def test_shared_channels_non_empty(self, loaded_pair) -> None:
        """There must be at least one channel in common between SVB and CSV."""
        svb, csv_sf = loaded_pair
        shared = set(svb) & set(csv_sf)
        assert shared, "No channels shared between SVB and CSV"

    def test_all_channels_timestamps_match(self, loaded_pair) -> None:
        """Timestamps for every shared channel must agree within 0.1 ms."""
        svb, csv_sf = loaded_pair
        start_ft: int = csv_sf._meta["StartTime"]
        shared = sorted(set(svb) & set(csv_sf))

        failures: list[str] = []
        for name in shared:
            t_svb, t_csv, _, _ = _aligned_arrays(svb[name], csv_sf[name], start_ft)
            if not np.allclose(t_svb, t_csv, atol=0.1):
                max_dt = float(np.max(np.abs(t_svb - t_csv)))
                failures.append(f"{name} (max Δt={max_dt:.4g} ms)")

        assert not failures, "Timestamp mismatch in channels:\n  " + "\n  ".join(failures)

    def test_all_channels_values_match(self, loaded_pair) -> None:
        """Sample values for every shared channel must agree within 1e-4 (rel+abs)."""
        svb, csv_sf = loaded_pair
        start_ft: int = csv_sf._meta["StartTime"]
        shared = sorted(set(svb) & set(csv_sf))

        failures: list[str] = []
        for name in shared:
            _, _, v_svb, v_csv = _aligned_arrays(svb[name], csv_sf[name], start_ft)
            if not np.allclose(v_svb, v_csv, rtol=1e-4, atol=1e-4):
                max_dv = float(np.max(np.abs(v_svb - v_csv)))
                failures.append(f"{name} (max |Δval|={max_dv:.4g})")

        assert not failures, "Value mismatch in channels:\n  " + "\n  ".join(failures)

    def test_sample_counts_reasonable(self, loaded_pair) -> None:
        """SVB and CSV sample counts for each shared channel must be within 1 % of each other."""
        svb, csv_sf = loaded_pair
        shared = sorted(set(svb) & set(csv_sf))

        failures: list[str] = []
        for name in shared:
            n_svb = len(svb[name].values)
            n_csv = len(csv_sf[name].values)
            ratio = abs(n_svb - n_csv) / max(n_svb, n_csv)
            if ratio > 0.01:
                failures.append(f"{name}: SVB={n_svb}, CSV={n_csv} ({ratio:.1%} apart)")

        assert not failures, "Sample-count mismatch:\n  " + "\n  ".join(failures)