Editing SWMM Input Files Programmatically with swmmio

import logging
import shutil
from pathlib import Path

import pandas as pd
import swmmio

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")


def model_fingerprint(inp_path: str) -> dict:
    """
    Structural counts used to prove an edit changed only what it meant to.

    Compare before and after: a change in any count that the edit did not
    intend is a corrupted model, whatever SWMM says when it runs.
    """
    m = swmmio.Model(str(inp_path))
    fp = {
        "subcatchments": len(m.inp.subcatchments),
        "junctions": len(m.inp.junctions),
        "conduits": len(m.inp.conduits),
        "outfalls": len(m.inp.outfalls),
    }
    log.info("fingerprint %s: %s", Path(inp_path).name, fp)
    return fp


def make_variant(
    base_inp: str,
    out_inp: str,
    subcatchment_edits: dict | None = None,
    conduit_edits: dict | None = None,
    scale_imperviousness: float | None = None,
) -> str:
    """
    Write a modified copy of a SWMM model.

    Parameters
    ----------
    base_inp             : Authored model — never modified.
    out_inp              : Destination for the variant.
    subcatchment_edits   : {name: {column: value}} for individual edits.
    conduit_edits        : {name: {column: value}}.
    scale_imperviousness : Multiply every subcatchment's PercImperv by this,
                           clipped to 0–100. Used for scenario sweeps.
    """
    base, out = Path(base_inp), Path(out_inp)
    out.parent.mkdir(parents=True, exist_ok=True)
    if out.resolve() == base.resolve():
        raise ValueError("refusing to write a variant over the authored model")

    before = model_fingerprint(str(base))
    shutil.copy(base, out)
    model = swmmio.Model(str(out))

    subs = model.inp.subcatchments
    if scale_imperviousness is not None:
        original = subs["PercImperv"].astype(float)
        subs["PercImperv"] = (original * scale_imperviousness).clip(0.0, 100.0)
        n_clipped = int(((original * scale_imperviousness) > 100.0).sum())
        log.info("Scaled imperviousness by %.2f across %d subcatchments "
                 "(mean %.1f → %.1f %%)", scale_imperviousness, len(subs),
                 float(original.mean()), float(subs["PercImperv"].mean()))
        if n_clipped:
            log.warning("%d subcatchment(s) clipped at 100 %% — the scaling is "
                        "no longer uniform across the model", n_clipped)

    if subcatchment_edits:
        for name, changes in subcatchment_edits.items():
            if name not in subs.index:
                # Editing by name means a typo is an error rather than a silent
                # no-op, which is the whole reason not to use text substitution.
                raise KeyError(f"subcatchment {name!r} not in the model")
            for col, val in changes.items():
                if col not in subs.columns:
                    raise KeyError(f"column {col!r} not in [SUBCATCHMENTS]")
                log.info("  %s.%s: %s → %s", name, col, subs.at[name, col], val)
                subs.at[name, col] = val

    if conduit_edits:
        conduits = model.inp.conduits
        for name, changes in conduit_edits.items():
            if name not in conduits.index:
                raise KeyError(f"conduit {name!r} not in the model")
            for col, val in changes.items():
                log.info("  %s.%s: %s → %s", name, col, conduits.at[name, col], val)
                conduits.at[name, col] = val
        model.inp.conduits = conduits

    model.inp.subcatchments = subs
    model.inp.save()

    after = model_fingerprint(str(out))
    if before != after:
        raise RuntimeError(f"the edit changed the model structure: "
                           f"{before} → {after}")
    log.info("Wrote variant %s with the structure unchanged", out)
    return str(out)


def validate_model(inp_path: str) -> dict:
    """
    Structural checks that catch an edit-induced corruption in under a second.

    SWMM itself only errors on structural problems, so a width column that has
    become a slope runs happily and returns nonsense.
    """
    m = swmmio.Model(str(inp_path))
    problems = []

    subs = m.inp.subcatchments
    for col, floor in (("Area", 0.0), ("Width", 0.0), ("PercSlope", 0.0)):
        if col in subs.columns:
            bad = subs[subs[col].astype(float) <= floor]
            if len(bad):
                problems.append(f"{len(bad)} subcatchment(s) with {col} <= {floor}")
    if "PercImperv" in subs.columns:
        imp = subs["PercImperv"].astype(float)
        bad = subs[(imp < 0) | (imp > 100)]
        if len(bad):
            problems.append(f"{len(bad)} subcatchment(s) with PercImperv outside 0–100")

    nodes = set(m.inp.junctions.index) | set(m.inp.outfalls.index)
    if hasattr(m.inp, "storage"):
        nodes |= set(m.inp.storage.index)
    conduits = m.inp.conduits
    for end in ("InletNode", "OutletNode"):
        if end in conduits.columns:
            dangling = conduits[~conduits[end].isin(nodes)]
            if len(dangling):
                problems.append(f"{len(dangling)} conduit(s) with a missing {end}")

    outlets = subs["Outlet"] if "Outlet" in subs.columns else pd.Series(dtype=object)
    missing = outlets[~outlets.isin(nodes | set(subs.index))]
    if len(missing):
        problems.append(f"{len(missing)} subcatchment(s) draining to a missing outlet")

    for p in problems:
        log.error("validation: %s", p)
    if not problems:
        log.info("validation: %s passed all structural checks", Path(inp_path).name)
    return {"path": inp_path, "problems": problems, "valid": not problems}


# --- Example usage: an imperviousness sweep ---
# for factor in (0.8, 1.0, 1.2, 1.5):
#     path = make_variant("model.inp", f"scratch/imp_{factor:.1f}.inp",
#                         scale_imperviousness=factor)
#     validate_model(path)