scenario_manager.py

"""Preliminary SWMM scenario generation, execution, and comparison.

Rev22.1 intentionally treats generated storms and parameter changes as preliminary
engineering scenarios.  It does not represent generated rainfall as an approved
City of Calgary design storm unless the user supplies and verifies the source.
"""
from __future__ import annotations

from dataclasses import asdict, dataclass, field
from datetime import datetime, timedelta
from io import BytesIO
from pathlib import Path
from typing import Any, Iterable
import csv
import hashlib
import json
import math
import re
import tempfile
import zipfile

import pandas as pd

from swmm_core import run_swmm


@dataclass
class StormDefinition:
    mode: str = "existing_model"
    name: str = "Existing model rainfall"
    return_period: str = "Model-defined"
    duration_minutes: int = 60
    interval_minutes: int = 5
    total_depth_mm: float | None = None
    peak_position: float = 0.40
    source_status: str = "Existing model input"
    source_reference: str = ""
    selected_timeseries: str = ""


@dataclass
class ScenarioChange:
    object_type: str
    object_id: str
    parameter: str
    old_value: Any = None
    new_value: Any = None
    status: str = "requested"
    note: str = ""


@dataclass
class ScenarioDefinition:
    scenario_id: str
    scenario_name: str
    description: str = ""
    storm: StormDefinition = field(default_factory=StormDefinition)
    imperviousness_overrides: dict[str, float] = field(default_factory=dict)
    conduit_diameter_overrides: dict[str, float] = field(default_factory=dict)
    conduit_roughness_overrides: dict[str, float] = field(default_factory=dict)
    storage_depth_overrides: dict[str, float] = field(default_factory=dict)
    simulation_hours: float | None = None
    review_status: str = "Preliminary"
    created_at: str = field(default_factory=lambda: datetime.now().isoformat(timespec="seconds"))


def safe_scenario_id(value: str) -> str:
    cleaned = re.sub(r"[^A-Za-z0-9_-]+", "_", value.strip()).strip("_")
    return cleaned[:64] or "scenario"


def _split_sections(text: str) -> tuple[list[str], dict[str, list[str]]]:
    preamble: list[str] = []
    sections: dict[str, list[str]] = {}
    current: str | None = None
    for raw in text.splitlines():
        m = re.match(r"^\s*\[([^]]+)\]\s*$", raw)
        if m:
            current = m.group(1).upper()
            sections.setdefault(current, [])
        elif current is None:
            preamble.append(raw)
        else:
            sections[current].append(raw)
    return preamble, sections


def _join_sections(preamble: list[str], sections: dict[str, list[str]]) -> str:
    out = list(preamble)
    if out and out[-1].strip():
        out.append("")
    for name, lines in sections.items():
        out.append(f"[{name}]")
        out.extend(lines)
        out.append("")
    return "\n".join(out).rstrip() + "\n"


def _data_tokens(line: str) -> tuple[list[str], str] | None:
    body, sep, comment = line.partition(";")
    if not body.strip() or body.lstrip().startswith(";"):
        return None
    return body.split(), (sep + comment if sep else "")


def _replace_tokens(line: str, tokens: list[str], comment: str) -> str:
    return "  ".join(str(x) for x in tokens) + ((" " + comment) if comment else "")


def _update_named_row(lines: list[str], object_id: str, token_index: int, new_value: Any) -> tuple[list[str], Any, bool]:
    updated: list[str] = []
    old = None
    changed = False
    for line in lines:
        parsed = _data_tokens(line)
        if parsed and parsed[0] and parsed[0][0].casefold() == object_id.casefold():
            tokens, comment = parsed
            if len(tokens) > token_index:
                old = tokens[token_index]
                tokens[token_index] = f"{new_value:g}" if isinstance(new_value, float) else str(new_value)
                line = _replace_tokens(line, tokens, comment)
                changed = True
        updated.append(line)
    return updated, old, changed


def _set_option(lines: list[str], key: str, value: str) -> list[str]:
    result: list[str] = []
    found = False
    for line in lines:
        parsed = _data_tokens(line)
        if parsed and parsed[0][0].upper() == key.upper():
            tokens, comment = parsed
            tokens = [key, value]
            line = _replace_tokens(line, tokens, comment)
            found = True
        result.append(line)
    if not found:
        result.append(f"{key:<20} {value}")
    return result


def chicago_style_incremental_depths(total_depth_mm: float, duration_minutes: int, interval_minutes: int, peak_position: float = 0.4) -> list[float]:
    """Create a preliminary, mass-conserving Chicago-style alternating-block storm.

    This is deliberately labelled preliminary. It is useful for scenario workflow
    testing, but it is not a replacement for a verified municipal IDF-based storm.
    """
    n = max(1, int(math.ceil(duration_minutes / interval_minutes)))
    # Smooth synthetic intensity pattern with a sharp peak and positive tails.
    ranks = list(range(1, n + 1))
    weights = [1.0 / (r ** 0.72) for r in ranks]
    scale = total_depth_mm / sum(weights)
    blocks = [w * scale for w in weights]
    peak_idx = min(n - 1, max(0, round((n - 1) * peak_position)))
    order = [peak_idx]
    offset = 1
    while len(order) < n:
        right = peak_idx + offset
        left = peak_idx - offset
        if right < n:
            order.append(right)
        if left >= 0 and len(order) < n:
            order.append(left)
        offset += 1
    result = [0.0] * n
    for block, idx in zip(sorted(blocks, reverse=True), order):
        result[idx] = block
    # exact conservation after floating-point arithmetic
    result[-1] += total_depth_mm - sum(result)
    return result


def _inject_preliminary_storm(sections: dict[str, list[str]], storm: StormDefinition, scenario_id: str) -> None:
    if storm.total_depth_mm is None:
        raise ValueError("Total storm depth is required for a generated preliminary storm.")
    interval = max(1, int(storm.interval_minutes))
    duration = max(interval, int(storm.duration_minutes))
    series_id = f"SCN_{safe_scenario_id(scenario_id)}_RAIN"
    depths = chicago_style_incremental_depths(float(storm.total_depth_mm), duration, interval, float(storm.peak_position))

    # Replace/add the time series. Values are interval rainfall depths; the gage is VOLUME.
    ts_lines = [ln for ln in sections.get("TIMESERIES", []) if not (_data_tokens(ln) and _data_tokens(ln)[0][0].casefold() == series_id.casefold())]
    start = datetime(2000, 1, 1, 0, 0)
    ts_lines.append(f"; Preliminary scenario storm: {storm.name}; source status: {storm.source_status}")
    for i, depth in enumerate(depths):
        t = start + timedelta(minutes=i * interval)
        ts_lines.append(f"{series_id:<24} {t.strftime('%m/%d/%Y')} {t.strftime('%H:%M')} {depth:.6f}")
    sections["TIMESERIES"] = ts_lines

    # Point all model rain gages to the generated series while retaining their IDs.
    rg_lines: list[str] = []
    for line in sections.get("RAINGAGES", []):
        parsed = _data_tokens(line)
        if parsed:
            tokens, comment = parsed
            if len(tokens) >= 1:
                gage_id = tokens[0]
                tokens = [gage_id, "VOLUME", f"0:{interval:02d}", "1.0", "TIMESERIES", series_id]
                line = _replace_tokens(line, tokens, comment)
        rg_lines.append(line)
    sections["RAINGAGES"] = rg_lines



def extract_rainfall_event_catalog(text: str) -> list[dict[str, str]]:
    """Return rainfall time-series available inside an INP model.

    SWMM models often retain several design-event series (for example 1:5 and
    1:100) while only one series is referenced by the active rain gage.
    """
    _, sections = _split_sections(text)
    active = set()
    gages = {}
    for line in sections.get("RAINGAGES", []):
        parsed = _data_tokens(line)
        if not parsed:
            continue
        tokens, _ = parsed
        if len(tokens) >= 6 and tokens[4].upper() == "TIMESERIES":
            gages[tokens[0]] = tokens[5]
            active.add(tokens[5].casefold())
    ids=[]
    seen=set()
    for line in sections.get("TIMESERIES", []):
        parsed=_data_tokens(line)
        if not parsed:
            continue
        tokens,_=parsed
        if not tokens:
            continue
        sid=tokens[0]
        if sid.casefold() in seen:
            continue
        seen.add(sid.casefold())
        ids.append({"event_id":sid,"active":"Yes" if sid.casefold() in active else "No","used_by":", ".join(k for k,v in gages.items() if v.casefold()==sid.casefold())})
    return ids


def _select_existing_timeseries(sections: dict[str, list[str]], series_id: str) -> None:
    """Point every TIMESERIES-based rain gage to an existing model series."""
    available={x["event_id"].casefold():x["event_id"] for x in extract_rainfall_event_catalog(_join_sections([], sections))}
    if series_id.casefold() not in available:
        raise ValueError(f"Rainfall time series {series_id!r} was not found in the selected model.")
    canonical=available[series_id.casefold()]
    new=[]
    changed=0
    for line in sections.get("RAINGAGES", []):
        parsed=_data_tokens(line)
        if parsed:
            tokens,comment=parsed
            if len(tokens)>=6 and tokens[4].upper()=="TIMESERIES":
                tokens[5]=canonical
                line=_replace_tokens(line,tokens,comment)
                changed+=1
        new.append(line)
    if changed==0:
        raise ValueError("The selected model has no TIMESERIES-based rain gage to assign.")
    sections["RAINGAGES"]=new

def build_scenario_inp(base_inp: str | Path, scenario: ScenarioDefinition, output_path: str | Path) -> dict[str, Any]:
    base = Path(base_inp)
    text = base.read_text(encoding="utf-8", errors="ignore")
    preamble, sections = _split_sections(text)
    change_log: list[ScenarioChange] = []

    if scenario.storm.mode == "generated_preliminary":
        _inject_preliminary_storm(sections, scenario.storm, scenario.scenario_id)
        change_log.append(ScenarioChange("storm", scenario.storm.name, "rainfall_series", None, scenario.storm.total_depth_mm, "applied", scenario.storm.source_status))
    elif scenario.storm.mode == "existing_timeseries" and scenario.storm.selected_timeseries:
        _select_existing_timeseries(sections, scenario.storm.selected_timeseries)
        change_log.append(ScenarioChange("storm", scenario.storm.selected_timeseries, "active_timeseries", None, scenario.storm.selected_timeseries, "applied", "Existing series selected from model"))

    if scenario.simulation_hours and scenario.simulation_hours > 0:
        start = datetime(2000, 1, 1, 0, 0)
        end = start + timedelta(hours=float(scenario.simulation_hours))
        opts = sections.setdefault("OPTIONS", [])
        for key, val in [
            ("START_DATE", start.strftime("%m/%d/%Y")),
            ("START_TIME", start.strftime("%H:%M:%S")),
            ("REPORT_START_DATE", start.strftime("%m/%d/%Y")),
            ("REPORT_START_TIME", start.strftime("%H:%M:%S")),
            ("END_DATE", end.strftime("%m/%d/%Y")),
            ("END_TIME", end.strftime("%H:%M:%S")),
        ]:
            opts = _set_option(opts, key, val)
        sections["OPTIONS"] = opts

    # SWMM [SUBCATCHMENTS]: Name RainGage Outlet Area %Imperv Width %Slope ...
    for oid, value in scenario.imperviousness_overrides.items():
        lines, old, ok = _update_named_row(sections.get("SUBCATCHMENTS", []), oid, 4, float(value))
        sections["SUBCATCHMENTS"] = lines
        change_log.append(ScenarioChange("subcatchment", oid, "imperviousness_percent", old, value, "applied" if ok else "not_found"))

    # SWMM [XSECTIONS]: Link Shape Geom1 ... ; circular Geom1 is diameter.
    for oid, value in scenario.conduit_diameter_overrides.items():
        lines, old, ok = _update_named_row(sections.get("XSECTIONS", []), oid, 2, float(value))
        sections["XSECTIONS"] = lines
        change_log.append(ScenarioChange("link", oid, "diameter_or_geom1", old, value, "applied" if ok else "not_found"))

    # SWMM [CONDUITS]: Name From To Length Roughness ...
    for oid, value in scenario.conduit_roughness_overrides.items():
        lines, old, ok = _update_named_row(sections.get("CONDUITS", []), oid, 4, float(value))
        sections["CONDUITS"] = lines
        change_log.append(ScenarioChange("conduit", oid, "manning_n", old, value, "applied" if ok else "not_found"))

    # SWMM [STORAGE]: Name Elev MaxDepth InitDepth Shape ...
    for oid, value in scenario.storage_depth_overrides.items():
        lines, old, ok = _update_named_row(sections.get("STORAGE", []), oid, 2, float(value))
        sections["STORAGE"] = lines
        change_log.append(ScenarioChange("storage", oid, "maximum_depth", old, value, "applied" if ok else "not_found"))

    output = Path(output_path)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(_join_sections(preamble, sections), encoding="utf-8")
    return {
        "scenario": asdict(scenario),
        "changes": [asdict(x) for x in change_log],
        "input_sha256": hashlib.sha256(output.read_bytes()).hexdigest(),
        "output_path": str(output),
    }


def summarize_scenario_results(results: dict[str, Any], scenario: ScenarioDefinition) -> dict[str, Any]:
    node_ts = results.get("node_ts", {}) or {}
    link_ts = results.get("link_ts", {}) or {}
    sub_ts = results.get("sub_ts", {}) or {}
    meta = results.get("metadata", {}) or {}

    max_flood = max((max(v.get("flooding", []) or [0.0]) for v in node_ts.values()), default=0.0)
    total_peak_inflow = max((max(v.get("inflow", []) or [0.0]) for v in node_ts.values()), default=0.0)
    max_velocity = 0.0
    max_velocity_link = ""
    max_depth_ratio = 0.0
    max_depth_ratio_link = ""
    peak_link_flow = 0.0
    for lid, values in link_ts.items():
        vel = max((abs(float(x)) for x in values.get("velocity", []) or [0.0]), default=0.0)
        if vel > max_velocity:
            max_velocity, max_velocity_link = vel, lid
        depth = max((float(x) for x in values.get("depth", []) or [0.0]), default=0.0)
        diameter = float(values.get("diameter", 0.0) or 0.0)
        ratio = depth / diameter if diameter > 0 else 0.0
        if ratio > max_depth_ratio:
            max_depth_ratio, max_depth_ratio_link = ratio, lid
        peak_link_flow = max(peak_link_flow, max((abs(float(x)) for x in values.get("flow", []) or [0.0]), default=0.0))
    peak_runoff = max((max(v.get("runoff", []) or [0.0]) for v in sub_ts.values()), default=0.0)
    return {
        "Scenario ID": scenario.scenario_id,
        "Scenario Name": scenario.scenario_name,
        "Storm": scenario.storm.name,
        "Storm Status": scenario.storm.source_status,
        "Simulation Status": "Completed",
        "Runoff Error (%)": float(meta.get("runoff_error", 0.0) or 0.0),
        "Flow Error (%)": float(meta.get("flow_error", 0.0) or 0.0),
        "Peak Subcatchment Runoff": peak_runoff,
        "Peak Link Flow": peak_link_flow,
        "Maximum Link Velocity": max_velocity,
        "Velocity Link": max_velocity_link,
        "Maximum Modelled Depth Ratio": max_depth_ratio,
        "Depth-Ratio Link": max_depth_ratio_link,
        "Maximum Node Flooding": max_flood,
        "Maximum Node Inflow": total_peak_inflow,
        "Review Status": scenario.review_status,
        "Model Role": "Scenario",
    }


def run_scenario(base_inp: str | Path, scenario: ScenarioDefinition, work_dir: str | Path | None = None) -> dict[str, Any]:
    root = Path(work_dir) if work_dir else Path(tempfile.mkdtemp(prefix="swmm_scenario_"))
    root.mkdir(parents=True, exist_ok=True)
    sid = safe_scenario_id(scenario.scenario_id)
    inp = root / f"{sid}.inp"
    manifest = build_scenario_inp(base_inp, scenario, inp)
    rpt = root / f"{sid}.rpt"
    out = root / f"{sid}.out"
    results = run_swmm(inp, rpt, out)
    # Attach final-input geometry to result records so scenario depth-ratio
    # summaries are computed from the scenario model rather than shown as zero.
    final_sections = {}
    try:
        _, final_sections = _split_sections(inp.read_text(encoding="utf-8", errors="ignore"))
        diameters = {}
        for line in final_sections.get("XSECTIONS", []):
            parsed = _data_tokens(line)
            if parsed:
                tokens, _ = parsed
                if len(tokens) >= 3:
                    try:
                        diameters[tokens[0]] = float(tokens[2])
                    except ValueError:
                        pass
        for lid, values in (results.get("link_ts", {}) or {}).items():
            if lid in diameters:
                values["diameter"] = diameters[lid]
    except Exception:
        pass
    summary = summarize_scenario_results(results, scenario)
    summary["Conduit Count"] = len(final_sections.get("CONDUITS", []) or [])
    storage_ids = [(_data_tokens(line)[0][0] if _data_tokens(line) else "") for line in (final_sections.get("STORAGE", []) or [])]
    storage_ids = [x for x in storage_ids if x]
    storage_depth = 0.0
    storage_volume = 0.0
    storage_node = ""
    for node_id in storage_ids:
        values = (results.get("node_ts", {}) or {}).get(node_id, {})
        depth = max((float(x) for x in values.get("depth", []) or [0.0]), default=0.0)
        volume = max((float(x) for x in values.get("volume", []) or [0.0]), default=0.0)
        if depth > storage_depth:
            storage_depth, storage_node = depth, node_id
        storage_volume = max(storage_volume, volume)
    summary["Maximum Storage Depth"] = storage_depth if storage_ids else None
    summary["Maximum Storage Volume"] = storage_volume if storage_ids else None
    summary["Controlling Storage Node"] = storage_node if storage_ids else ""
    manifest.update({
        "report_path": str(rpt),
        "output_path": str(out),
        "report_sha256": hashlib.sha256(rpt.read_bytes()).hexdigest() if rpt.exists() else None,
        "summary": summary,
    })
    return {"definition": asdict(scenario), "manifest": manifest, "results": results, "summary": summary, "files": {"inp": inp.read_bytes(), "rpt": rpt.read_bytes() if rpt.exists() else b"", "out": out.read_bytes() if out.exists() else b""}}



def base_model_record(results: dict[str, Any], *, scenario_name: str = "Base Model", review_status: str = "Reference model", source_name: str = "Base Model") -> dict[str, Any]:
    """Create a comparison record for the currently simulated base model."""
    definition = ScenarioDefinition(
        scenario_id="BASE_MODEL",
        scenario_name=scenario_name,
        description="Current uploaded and simulated reference model.",
        storm=StormDefinition(
            mode="existing_model",
            name="Existing model rainfall",
            return_period="Model-defined",
            source_status="Base model input",
        ),
        review_status=review_status,
    )
    summary = summarize_scenario_results(results or {}, definition)
    summary["Model Role"] = "Base"
    summary["Source Model"] = source_name
    summary["Input Changed"] = "Reference"
    summary["Input SHA256 (short)"] = ""
    summary["Conduit Count"] = sum(1 for values in ((results or {}).get("link_ts", {}) or {}).values() if float(values.get("diameter", 0) or 0) > 0)
    return {
        "definition": asdict(definition),
        "manifest": {"summary": summary, "model_role": "Base", "source": "Current uploaded model"},
        "results": results or {},
        "summary": summary,
        "files": {},
    }


def comparison_with_base(base_results: dict[str, Any] | None, records: Iterable[dict[str, Any]], *, base_name: str = "Base Model") -> pd.DataFrame:
    """Return one comparison table containing the base model and scenario records."""
    combined: list[dict[str, Any]] = []
    if base_results:
        combined.append(base_model_record(base_results, scenario_name=base_name, source_name=base_name))
    combined.extend(list(records))
    df = comparison_dataframe(combined)
    if not df.empty and "Model Role" not in df.columns:
        df.insert(0, "Model Role", ["Base" if str(v) == "BASE_MODEL" else "Scenario" for v in df.get("Scenario ID", [])])
    elif not df.empty:
        df["Model Role"] = df["Model Role"].fillna("Scenario")
    if not df.empty:
        base_rows = df[df["Scenario ID"].astype(str) == "BASE_MODEL"]
        if not base_rows.empty:
            b = base_rows.iloc[0]
            metric_cols = ["Peak Subcatchment Runoff", "Peak Link Flow", "Maximum Link Velocity", "Maximum Modelled Depth Ratio", "Maximum Node Flooding", "Maximum Node Inflow"]
            flags = []
            for _, row in df.iterrows():
                if str(row.get("Scenario ID")) == "BASE_MODEL":
                    flags.append("Reference")
                    continue
                changed = False
                for col in metric_cols:
                    try:
                        if abs(float(row.get(col, 0) or 0) - float(b.get(col, 0) or 0)) > 1e-8:
                            changed = True
                            break
                    except Exception:
                        pass
                flags.append("Different" if changed else "No summary-level difference")
            df["Hydraulic Difference"] = flags
    return df


def deterministic_comparison_analysis(comparison: pd.DataFrame) -> str:
    """Create a conservative, deterministic scenario-comparison narrative."""
    if comparison is None or comparison.empty:
        return "No scenario comparison data are available."
    df = comparison.copy()
    base = df[df["Scenario ID"].astype(str) == "BASE_MODEL"]
    if base.empty:
        base = df.iloc[[0]]
    b = base.iloc[0]
    metrics = [
        ("Peak Subcatchment Runoff", "peak subcatchment runoff"),
        ("Peak Link Flow", "peak link flow"),
        ("Maximum Link Velocity", "maximum link velocity"),
        ("Maximum Modelled Depth Ratio", "maximum modelled depth ratio"),
        ("Maximum Node Flooding", "maximum node flooding"),
        ("Maximum Node Inflow", "maximum node inflow"),
        ("Maximum Storage Depth", "maximum storage depth"),
        ("Maximum Storage Volume", "maximum storage volume"),
    ]
    lines = [
        "The comparison uses the current uploaded model as the reference case. Differences are calculated from deterministic simulation summaries and are intended for preliminary engineering review.",
        "",
    ]
    # Guard: if every base metric is zero while at least one scenario is
    # non-zero, the base record almost certainly did not carry usable
    # results (e.g. it was built from an empty or malformed results dict).
    # Deltas quoted against a zeroed base are misleading, so say so
    # explicitly instead of presenting "+X relative to base" as fact.
    def _metric_value(record, col):
        try:
            return abs(float(record.get(col, 0) or 0))
        except Exception:
            return 0.0
    base_all_zero = all(_metric_value(b, col) <= 1e-12 for col, _ in metrics)
    any_scenario_nonzero = any(
        _metric_value(row, col) > 1e-12
        for _, row in df[df["Scenario ID"].astype(str) != str(b.get("Scenario ID", "BASE_MODEL"))].iterrows()
        for col, _ in metrics
    )
    if base_all_zero and any_scenario_nonzero:
        lines.insert(0, (
            "CAUTION: All reference-case metrics are zero while scenario results are non-zero. "
            "The base record does not appear to contain usable simulation results; differences below "
            "are NOT valid deltas against the uploaded model. Re-simulate or rebuild the base record "
            "before relying on this comparison."
        ))
        lines.insert(1, "")
    scenarios = df[df["Scenario ID"].astype(str) != str(b.get("Scenario ID", "BASE_MODEL"))]
    if scenarios.empty:
        lines.append("No alternative scenario has been completed.")
        return "\n".join(lines)
    for _, row in scenarios.iterrows():
        sid = row.get("Scenario ID", "Scenario")
        name = row.get("Scenario Name", sid)
        lines.append(f"{name} ({sid}):")
        changes=[]
        no_conduits = int(row.get("Conduit Count", 0) or 0) == 0
        for col,label in metrics:
            if no_conduits and col in {"Maximum Link Velocity", "Maximum Modelled Depth Ratio"}:
                changes.append(f"- {label}: Not applicable — no conduits in model")
                continue
            try:
                bv=float(b.get(col,0) or 0); sv=float(row.get(col,0) or 0)
            except Exception:
                continue
            delta=sv-bv
            if abs(bv)>1e-12:
                pct=100*delta/abs(bv)
                changes.append(f"- {label}: {sv:.4g} ({delta:+.4g}; {pct:+.1f}% relative to base)")
            else:
                changes.append(f"- {label}: {sv:.4g} ({delta:+.4g} relative to base)")
        lines.extend(changes)
        if str(row.get("Hydraulic Difference", "")) == "No summary-level difference":
            lines.append("- No difference was detected in the reported summary metrics. Check the input-change status and detailed time series before treating this as a distinct hydraulic alternative.")
        if str(row.get("Input Changed", "")) == "No":
            lines.append("- The final scenario input is identical to the base-model input; identical results are expected.")
        if float(row.get("Maximum Node Flooding",0) or 0)>0:
            lines.append("- Modelled node flooding is present and requires review.")
        if abs(float(row.get("Runoff Error (%)",0) or 0))>1 or abs(float(row.get("Flow Error (%)",0) or 0))>1:
            lines.append("- Numerical continuity exceeds 1% for at least one reported balance and requires review.")
        lines.append("")
    lines.append("The comparison does not establish design adequacy or compliance. The responsible engineer must review model changes, storm sources, controlling elements, physical feasibility, and applicable project criteria.")
    return "\n".join(lines).strip()

def comparison_dataframe(records: Iterable[dict[str, Any]]) -> pd.DataFrame:
    rows = [r.get("summary", {}) for r in records]
    return pd.DataFrame(rows)


def build_scenario_package(records: Iterable[dict[str, Any]]) -> bytes:
    records = list(records)
    buffer = BytesIO()
    with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        comparison = comparison_dataframe(records)
        zf.writestr("scenario_comparison.csv", comparison.to_csv(index=False))
        manifest = []
        for rec in records:
            sid = safe_scenario_id(rec.get("definition", {}).get("scenario_id", "scenario"))
            base = f"scenarios/{sid}"
            editable_clone = rec.get("files", {}).get("editable_clone", b"")
            if editable_clone:
                zf.writestr(f"{base}/{sid}_editable_clone_source.inp", editable_clone)
            zf.writestr(f"{base}/{sid}.inp", rec.get("files", {}).get("inp", b""))
            zf.writestr(f"{base}/{sid}.rpt", rec.get("files", {}).get("rpt", b""))
            out_bytes = rec.get("files", {}).get("out", b"")
            if out_bytes:
                zf.writestr(f"{base}/{sid}.out", out_bytes)
            zf.writestr(f"{base}/scenario_definition.json", json.dumps(rec.get("definition", {}), indent=2, default=str))
            zf.writestr(f"{base}/scenario_manifest.json", json.dumps(rec.get("manifest", {}), indent=2, default=str))
            manifest.append(rec.get("manifest", {}))
        zf.writestr("scenario_register.json", json.dumps(manifest, indent=2, default=str))
        zf.writestr("README.txt", "Rev22.1 preliminary scenario package. Editable clone source files are preserved separately from final scenario input files. Generated storms and parameter changes require professional verification before design use or municipal submission.\n")
    return buffer.getvalue()