rpt_reconciliation.py

"""Deterministic reconciliation of app-extracted results against the SWMM .rpt file.

Rationale
---------
The application builds its result tables (and therefore the Calgary velocity
screening and the Table 9A depth-velocity assessment) from time series read
through the OpenSWMM Python API inside the isolated worker. The engine's own
``.rpt`` file independently reports per-timestep maxima in its Link Flow
Summary and Node Depth Summary. On the Kincora Phase 2 reference model these
two sources were found to disagree: pipe peak velocities differed by 5-12%,
and peak velocities in irregular-transect (street) channels were understated
by up to a factor of six in the API-derived tables, while the ``.rpt`` values
matched the original consultant's SWMM 5.0.022 run almost exactly.

This module parses the ``.rpt`` summaries and produces an auditable
reconciliation table plus findings-register entries compatible with the
Preliminary Design Assistant. It performs no simulation and calls no LLM.

Notes and limitations
---------------------
* RPT object names are truncated to 20 characters by the engine. Where a
  worker ID is longer than 20 characters, matching falls back to the
  truncated prefix; ambiguous prefixes are reported as ``Unmatched`` rather
  than guessed.
* Only CONDUIT and CHANNEL rows carry velocity; OUTLET/DUMMY/PUMP/ORIFICE/
  WEIR rows are reconciled on peak flow only.
* The ``.rpt`` is treated as the authoritative cross-check because it is the
  engine's own per-timestep statistic. A discrepancy does not by itself say
  which value is "true"; it says the two extraction paths disagree and the
  responsible engineer must not rely on the affected screening rows until
  the cause is resolved.
"""
from __future__ import annotations

import math
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence

import pandas as pd

_RPT_NAME_WIDTH = 20
_VELOCITY_TYPES = {"CONDUIT", "CHANNEL"}
_FLOW_ONLY_TYPES = {"OUTLET", "DUMMY", "PUMP", "ORIFICE", "WEIR"}


@dataclass
class ReconciliationTolerances:
    """Screening tolerances for worker-vs-RPT deltas.

    Values are relative unless stated. Absolute floors avoid flagging noise
    on near-zero quantities (e.g. a 0.002 vs 0.004 m/s trickle).
    """

    velocity_review_pct: float = 5.0
    velocity_discrepancy_pct: float = 15.0
    velocity_abs_floor: float = 0.05          # m/s or ft/s
    flow_review_pct: float = 2.0
    flow_discrepancy_pct: float = 10.0
    flow_abs_floor: float = 0.005             # model flow units
    depth_review_pct: float = 5.0
    depth_discrepancy_pct: float = 15.0
    depth_abs_floor: float = 0.01             # m or ft
    continuity_abs_review: float = 0.05       # absolute percentage points


# ---------------------------------------------------------------------------
# RPT parsing
# ---------------------------------------------------------------------------

def _section_lines(text: str, header: str) -> list[str]:
    """Return the body lines of a starred RPT section, or an empty list."""
    pattern = re.compile(
        r"^\s*" + re.escape(header) + r"\s*$", re.MULTILINE)
    match = pattern.search(text)
    if not match:
        return []
    lines = text[match.end():].splitlines()
    # The data table starts after the LAST dashed rule that precedes the
    # first data row: title banner (****), blank, dashed rule, column
    # headers, dashed rule, data rows, blank line.
    last_rule = -1
    for i, line in enumerate(lines):
        stripped = line.strip()
        if stripped and set(stripped) <= {"-"}:
            last_rule = i
            continue
        if last_rule >= 0 and stripped and not stripped.startswith("*"):
            # Column-header lines sit between the two rules; a line after a
            # rule that is followed by another rule is a header, so only
            # accept this as the data start if no further rule intervenes
            # before the next blank line.
            remainder = lines[i:]
            if any(set(l.strip()) <= {"-"} and l.strip() for l in remainder[:4]):
                continue  # still inside the header block
            body: list[str] = []
            for data_line in remainder:
                if not data_line.strip():
                    break
                if data_line.strip().startswith("*"):
                    break
                body.append(data_line.rstrip("\n"))
            return body
    return []


def _row_tokens(line: str) -> tuple[str, list[str]]:
    """Split an RPT summary row into (object name, remaining tokens).

    The name field is fixed-width (20 chars) and may itself contain no
    spaces in SWMM inputs, but is sliced positionally to be safe.
    """
    name = line[:2 + _RPT_NAME_WIDTH].strip()
    rest = line[2 + _RPT_NAME_WIDTH:].split()
    return name, rest


def _to_float(token: str) -> float | None:
    try:
        value = float(token)
    except (TypeError, ValueError):
        return None
    return value if math.isfinite(value) else None


def parse_link_flow_summary(rpt_path: str | Path) -> pd.DataFrame:
    """Parse the Link Flow Summary into a DataFrame.

    Columns: ``Link ID``, ``RPT Type``, ``RPT Peak |Flow|``,
    ``RPT Time of Max``, ``RPT Peak |Velocity|``, ``RPT Max/Full Flow``,
    ``RPT Max/Full Depth``. Velocity columns are NaN for flow-only types.
    """
    text = Path(rpt_path).read_text(encoding="utf-8", errors="replace")
    rows: list[dict[str, Any]] = []
    for line in _section_lines(text, "Link Flow Summary"):
        name, tokens = _row_tokens(line)
        if not name or len(tokens) < 2:
            continue
        rtype = tokens[0].upper()
        record: dict[str, Any] = {
            "Link ID": name, "RPT Type": rtype,
            "RPT Peak |Flow|": _to_float(tokens[1]),
            "RPT Time of Max": None, "RPT Peak |Velocity|": None,
            "RPT Max/Full Flow": None, "RPT Max/Full Depth": None,
        }
        if len(tokens) >= 4:
            record["RPT Time of Max"] = f"{tokens[2]} {tokens[3]}"
        if rtype in _VELOCITY_TYPES and len(tokens) >= 7:
            record["RPT Peak |Velocity|"] = _to_float(tokens[4])
            record["RPT Max/Full Flow"] = _to_float(tokens[5])
            record["RPT Max/Full Depth"] = _to_float(tokens[6])
        rows.append(record)
    return pd.DataFrame(rows)


def parse_node_depth_summary(rpt_path: str | Path) -> pd.DataFrame:
    """Parse the Node Depth Summary into a DataFrame."""
    text = Path(rpt_path).read_text(encoding="utf-8", errors="replace")
    rows: list[dict[str, Any]] = []
    for line in _section_lines(text, "Node Depth Summary"):
        name, tokens = _row_tokens(line)
        if not name or len(tokens) < 4:
            continue
        rows.append({
            "Node ID": name, "RPT Type": tokens[0].upper(),
            "RPT Avg Depth": _to_float(tokens[1]),
            "RPT Max Depth": _to_float(tokens[2]),
            "RPT Max HGL": _to_float(tokens[3]),
        })
    return pd.DataFrame(rows)


def parse_continuity_errors(rpt_path: str | Path) -> dict[str, float | None]:
    """Return runoff and flow-routing continuity errors in PERCENT.

    The first ``Continuity Error (%)`` line in the RPT belongs to the runoff
    quantity block and the second to flow routing, matching the engine's
    output order.
    """
    text = Path(rpt_path).read_text(encoding="utf-8", errors="replace")
    values = [
        _to_float(m.group(1))
        for m in re.finditer(r"Continuity Error \(%\) \.+\s+(-?[\d.]+)", text)
    ]
    return {
        "runoff_error_pct": values[0] if len(values) > 0 else None,
        "flow_error_pct": values[1] if len(values) > 1 else None,
    }


# ---------------------------------------------------------------------------
# Reconciliation
# ---------------------------------------------------------------------------

def _match_rpt_row(object_id: str, rpt: pd.DataFrame, id_col: str) -> pd.Series | None:
    exact = rpt[rpt[id_col].astype(str) == str(object_id)]
    if len(exact) == 1:
        return exact.iloc[0]
    prefix = str(object_id)[:_RPT_NAME_WIDTH]
    by_prefix = rpt[rpt[id_col].astype(str) == prefix]
    if len(by_prefix) == 1:
        return by_prefix.iloc[0]
    return None


def _classify(worker: float | None, reference: float | None,
              review_pct: float, discrepancy_pct: float,
              abs_floor: float) -> tuple[str, float | None]:
    """Return (status, delta_pct). Deltas below the absolute floor pass."""
    if worker is None or reference is None:
        return "Unavailable", None
    if abs(worker - reference) <= abs_floor:
        return "OK", 0.0 if reference == 0 else round(
            100.0 * (worker - reference) / abs(reference), 2)
    if reference == 0:
        return "Discrepancy", None
    delta = 100.0 * (worker - reference) / abs(reference)
    if abs(delta) <= review_pct:
        return "OK", round(delta, 2)
    if abs(delta) <= discrepancy_pct:
        return "Review", round(delta, 2)
    return "Discrepancy", round(delta, 2)


def reconcile_links(
    link_summary: pd.DataFrame,
    rpt_path: str | Path,
    tolerances: ReconciliationTolerances | None = None,
) -> pd.DataFrame:
    """Reconcile the app's link table against the RPT Link Flow Summary.

    ``link_summary`` must contain ``Link ID`` plus any of the recognised
    worker columns (``Peak Flow (...)``, ``Peak Velocity (...)``,
    ``Depth Ratio``). Unrecognised columns are ignored so unit-suffix
    variations (m³/s vs cfs, m/s vs ft/s) are handled transparently.
    """
    tol = tolerances or ReconciliationTolerances()
    if link_summary is None or link_summary.empty:
        return pd.DataFrame()
    rpt = parse_link_flow_summary(rpt_path)
    if rpt.empty:
        return pd.DataFrame()

    flow_col = next((c for c in link_summary.columns if c.startswith("Peak Flow (")), None)
    vel_col = next((c for c in link_summary.columns if c.startswith("Peak Velocity (")), None)
    depth_ratio_col = "Depth Ratio" if "Depth Ratio" in link_summary.columns else None

    rows: list[dict[str, Any]] = []
    for _, record in link_summary.iterrows():
        link_id = str(record["Link ID"])
        matched = _match_rpt_row(link_id, rpt, "Link ID")
        if matched is None:
            rows.append({"Link ID": link_id, "RPT Type": None,
                         "Overall Status": "Unmatched"})
            continue
        entry: dict[str, Any] = {"Link ID": link_id, "RPT Type": matched["RPT Type"]}
        statuses: list[str] = []

        worker_flow = _to_float(record.get(flow_col)) if flow_col else None
        entry["Worker Peak Flow"] = worker_flow
        entry["RPT Peak Flow"] = matched["RPT Peak |Flow|"]
        status, delta = _classify(worker_flow, matched["RPT Peak |Flow|"],
                                  tol.flow_review_pct, tol.flow_discrepancy_pct,
                                  tol.flow_abs_floor)
        entry["Flow Delta (%)"], entry["Flow Status"] = delta, status
        statuses.append(status)

        if matched["RPT Type"] in _VELOCITY_TYPES:
            worker_vel = _to_float(record.get(vel_col)) if vel_col else None
            entry["Worker Peak Velocity"] = worker_vel
            entry["RPT Peak Velocity"] = matched["RPT Peak |Velocity|"]
            status, delta = _classify(worker_vel, matched["RPT Peak |Velocity|"],
                                      tol.velocity_review_pct,
                                      tol.velocity_discrepancy_pct,
                                      tol.velocity_abs_floor)
            entry["Velocity Delta (%)"], entry["Velocity Status"] = delta, status
            statuses.append(status)

            if depth_ratio_col is not None:
                worker_ratio = _to_float(record.get(depth_ratio_col))
                entry["Worker Depth Ratio"] = worker_ratio
                entry["RPT Max/Full Depth"] = matched["RPT Max/Full Depth"]
                status, delta = _classify(worker_ratio, matched["RPT Max/Full Depth"],
                                          tol.depth_review_pct,
                                          tol.depth_discrepancy_pct,
                                          tol.depth_abs_floor)
                entry["Depth Ratio Delta (%)"], entry["Depth Ratio Status"] = delta, status
                statuses.append(status)

        order = {"Discrepancy": 3, "Unmatched": 3, "Review": 2, "Unavailable": 1, "OK": 0}
        entry["Overall Status"] = max(statuses, key=lambda s: order.get(s, 0)) if statuses else "Unavailable"
        rows.append(entry)
    return pd.DataFrame(rows)


def reconcile_nodes(
    node_summary: pd.DataFrame,
    rpt_path: str | Path,
    tolerances: ReconciliationTolerances | None = None,
) -> pd.DataFrame:
    """Reconcile app node peak depths against the RPT Node Depth Summary."""
    tol = tolerances or ReconciliationTolerances()
    if node_summary is None or node_summary.empty:
        return pd.DataFrame()
    rpt = parse_node_depth_summary(rpt_path)
    if rpt.empty:
        return pd.DataFrame()
    depth_col = next((c for c in node_summary.columns if c.startswith("Peak Depth (")), None)
    if depth_col is None:
        return pd.DataFrame()
    rows: list[dict[str, Any]] = []
    for _, record in node_summary.iterrows():
        node_id = str(record["Node ID"])
        matched = _match_rpt_row(node_id, rpt, "Node ID")
        if matched is None:
            rows.append({"Node ID": node_id, "Overall Status": "Unmatched"})
            continue
        worker_depth = _to_float(record.get(depth_col))
        status, delta = _classify(worker_depth, matched["RPT Max Depth"],
                                  tol.depth_review_pct, tol.depth_discrepancy_pct,
                                  tol.depth_abs_floor)
        rows.append({
            "Node ID": node_id, "RPT Type": matched["RPT Type"],
            "Worker Peak Depth": worker_depth,
            "RPT Max Depth": matched["RPT Max Depth"],
            "Depth Delta (%)": delta, "Overall Status": status,
        })
    return pd.DataFrame(rows)


def reconcile_continuity(
    simulation_metadata: Mapping[str, Any],
    rpt_path: str | Path,
    tolerances: ReconciliationTolerances | None = None,
) -> pd.DataFrame:
    """Reconcile worker continuity errors against the RPT, in percent.

    Detects the fraction-vs-percent inconsistency: if the worker value is
    approximately the RPT value divided by 100, the row is marked
    ``Unit inconsistency (fraction vs percent)`` rather than a numeric
    discrepancy, since the underlying simulation agrees.
    """
    tol = tolerances or ReconciliationTolerances()
    rpt_values = parse_continuity_errors(rpt_path)
    pairs = [
        ("Runoff continuity", simulation_metadata.get("runoff_error"),
         rpt_values["runoff_error_pct"]),
        ("Flow routing continuity", simulation_metadata.get("flow_error"),
         rpt_values["flow_error_pct"]),
    ]
    rows: list[dict[str, Any]] = []
    for label, worker, reference in pairs:
        worker_f, ref_f = _to_float(worker), _to_float(reference)
        if worker_f is None or ref_f is None:
            status = "Unavailable"
        elif abs(worker_f - ref_f) <= tol.continuity_abs_review:
            status = "OK"
        elif abs(worker_f * 100.0 - ref_f) <= tol.continuity_abs_review:
            status = "Unit inconsistency (fraction vs percent)"
        else:
            status = "Discrepancy"
        rows.append({"Quantity": label, "Worker Value": worker_f,
                     "RPT Value (%)": ref_f, "Status": status})
    return pd.DataFrame(rows)


# ---------------------------------------------------------------------------
# Findings-register integration
# ---------------------------------------------------------------------------

def reconciliation_findings(
    link_recon: pd.DataFrame,
    node_recon: pd.DataFrame | None = None,
    continuity_recon: pd.DataFrame | None = None,
    start_index: int = 1,
) -> list[dict[str, Any]]:
    """Convert reconciliation discrepancies into PDA-style finding dicts.

    Only ``Review``, ``Discrepancy``, ``Unmatched`` and unit-inconsistency
    rows generate findings. Severity: Discrepancy/Unmatched -> High,
    Review -> Medium. These findings carry no proposed model edit; the
    recommended action is to resolve the extraction path before relying on
    the affected screening tables.
    """
    findings: list[dict[str, Any]] = []
    counter = start_index

    def add(severity: str, object_type: str, object_id: str, basis: str) -> None:
        nonlocal counter
        findings.append({
            "finding_id": f"RPT-{counter:03d}",
            "category": "Result reconciliation",
            "severity": severity,
            "finding_type": "Worker/RPT disagreement",
            "object_type": object_type,
            "object_id": object_id,
            "rule_id": "QA-RECON-001",
            "criterion_status": "Deterministic cross-check",
            "deterministic_basis": basis,
            "recommended_action": (
                "Do not rely on the affected screening rows until the API "
                "extraction and the engine report file agree. Verify units, "
                "extraction property, and sampling of the worker time series."
            ),
            "engineer_decision": "Defer",
            "resolution_status": "Open",
        })
        counter += 1

    if link_recon is not None and not link_recon.empty:
        for _, row in link_recon.iterrows():
            status = row.get("Overall Status")
            if status in {"OK", "Unavailable", None}:
                continue
            severity = "High" if status in {"Discrepancy", "Unmatched"} else "Medium"
            parts = []
            for label, w, r, d in [
                ("velocity", row.get("Worker Peak Velocity"), row.get("RPT Peak Velocity"), row.get("Velocity Delta (%)")),
                ("flow", row.get("Worker Peak Flow"), row.get("RPT Peak Flow"), row.get("Flow Delta (%)")),
                ("depth ratio", row.get("Worker Depth Ratio"), row.get("RPT Max/Full Depth"), row.get("Depth Ratio Delta (%)")),
            ]:
                if d is not None and not (isinstance(d, float) and math.isnan(d)) and abs(d) > 5.0:
                    parts.append(f"peak {label} worker={w} vs rpt={r} ({d:+.1f}%)")
            basis = (f"Link '{row['Link ID']}': " + "; ".join(parts)) if parts else (
                f"Link '{row['Link ID']}': status {status}.")
            add(severity, "LINK", str(row["Link ID"]), basis)

    if node_recon is not None and not node_recon.empty:
        for _, row in node_recon.iterrows():
            if row.get("Overall Status") in {"Review", "Discrepancy", "Unmatched"}:
                severity = "Medium" if row["Overall Status"] == "Review" else "High"
                add(severity, "NODE", str(row["Node ID"]),
                    f"Node '{row['Node ID']}': peak depth worker="
                    f"{row.get('Worker Peak Depth')} vs rpt={row.get('RPT Max Depth')}"
                    f" ({row.get('Depth Delta (%)')}%).")

    if continuity_recon is not None and not continuity_recon.empty:
        for _, row in continuity_recon.iterrows():
            if row["Status"] not in {"OK", "Unavailable"}:
                add("Medium", "MODEL", "MODEL",
                    f"{row['Quantity']}: worker={row['Worker Value']} vs "
                    f"rpt={row['RPT Value (%)']}%. {row['Status']}.")
    return findings


def reconciliation_summary(link_recon: pd.DataFrame) -> dict[str, Any]:
    """Compact machine-readable summary for the report package metadata."""
    if link_recon is None or link_recon.empty:
        return {"links_checked": 0, "ok": 0, "review": 0,
                "discrepancy": 0, "unmatched": 0, "verdict": "Not performed"}
    counts = link_recon["Overall Status"].value_counts().to_dict()
    discrepancies = counts.get("Discrepancy", 0) + counts.get("Unmatched", 0)
    verdict = ("Pass - worker tables agree with engine report" if discrepancies == 0
               and counts.get("Review", 0) == 0 else
               "Review required - worker tables disagree with engine report")
    return {
        "links_checked": int(len(link_recon)),
        "ok": int(counts.get("OK", 0)),
        "review": int(counts.get("Review", 0)),
        "discrepancy": int(counts.get("Discrepancy", 0)),
        "unmatched": int(counts.get("Unmatched", 0)),
        "verdict": verdict,
    }