"""Adapter from a deterministic SWMM evidence package to the existing PCSWMM Engineering MCP package schema. This module does not recompute hydraulics. It normalizes the evidence already produced by an approved evidence backend, including PCSWMM SDK or PySWMM so the existing MCP review and SWMR report tools can consume it without a parallel workflow. """ from __future__ import annotations import base64 import hashlib import json from pathlib import Path from typing import Any REQUIRED_FILES = ( "manifest.json", "normalized/calgary_swmr_native_evidence.json", "evidence_selection.json", "missing_information.json", ) def _load(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def _sha256_json(value: Any) -> str: payload = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") return hashlib.sha256(payload).hexdigest() def validate_evidence_directory(evidence_dir: str | Path) -> dict[str, Any]: root = Path(evidence_dir).expanduser().resolve() missing = [name for name in REQUIRED_FILES if not (root / name).is_file()] return { "valid": not missing, "evidence_dir": str(root), "missing": missing, "figure_count": len(list((root / "figures").glob("*.png"))) if (root / "figures").is_dir() else 0, "table_count": len(list((root / "tables").glob("*.csv"))) if (root / "tables").is_dir() else 0, "timeseries_count": len(list((root / "timeseries").glob("*.csv"))) if (root / "timeseries").is_dir() else 0, } def _catalog_to_per_object(catalog: list[dict[str, Any]]) -> dict[str, dict[str, dict[str, Any]]]: out: dict[str, dict[str, dict[str, Any]]] = { "node_depth": {}, "node_head": {}, "node_flooding": {}, "link_flow": {}, "link_depth": {}, "link_depth_ratio": {}, "link_velocity": {}, "subcatchment_runoff": {}, } mapping = { ("nodes", "depth"): "node_depth", ("nodes", "head"): "node_head", ("nodes", "flooding"): "node_flooding", ("links", "flow"): "link_flow", ("links", "depth"): "link_depth", ("links", "capacity"): "link_depth_ratio", ("links", "velocity"): "link_velocity", ("subcatchments", "runoff"): "subcatchment_runoff", } for row in catalog or []: group = str(row.get("group", "")).casefold() variable = str(row.get("variable", "")).casefold() bucket = mapping.get((group, variable)) object_id = str(row.get("object", "")).strip() if not bucket or not object_id: continue record = { "minimum": row.get("minimum"), "maximum": row.get("maximum"), "mean": row.get("mean"), "peak_time": row.get("peak_time"), "units": row.get("units"), "source": row.get("graph_file_path"), } if bucket == "link_depth_ratio": record["depth_ratio"] = row.get("maximum") out[bucket][object_id] = record return out def build_mcp_package_from_evidence(evidence_dir: str | Path) -> dict[str, Any]: check = validate_evidence_directory(evidence_dir) if not check["valid"]: raise ValueError("Incomplete deterministic evidence package: " + "; ".join(check["missing"])) root = Path(check["evidence_dir"]) normalized = _load(root / "normalized/calgary_swmr_native_evidence.json") manifest = _load(root / "manifest.json") selection = _load(root / "evidence_selection.json") missing = _load(root / "missing_information.json") agentic_path = root / "agentic_review/agentic_review.json" agentic = _load(agentic_path) if agentic_path.is_file() else {} project = dict(normalized.get("project") or {}) project_name = project.get("project_name") or project.get("name") or root.name declared_model_file = ( project.get("model_file") or project.get("file_path") or manifest.get("project_identity", {}).get("model_file") ) # Evidence archives are frequently moved between machines. Prefer an INP # physically present in the connected evidence directory over a stale # absolute source path recorded during extraction. inp_candidates = sorted(root.glob("*.inp")) if not inp_candidates: inp_candidates = sorted(root.rglob("*.inp")) actual_model_path = inp_candidates[0].resolve() if inp_candidates else None model_file = str(actual_model_path or declared_model_file or f"{project_name}.inp") project.update({ "project_name": project_name, "model_file": model_file, "declared_model_file": declared_model_file, "model_available": bool(actual_model_path and actual_model_path.is_file()), "swmm_version": project.get("swmm_version") or manifest.get("project_identity", {}).get("swmm_version"), "evidence_directory": str(root), }) catalog = list(normalized.get("result_catalog") or []) tables = dict(normalized.get("tables") or {}) collection_counts = dict(normalized.get("collection_counts") or {}) per_object = _catalog_to_per_object(catalog) source_hashes = manifest.get("source_hashes") or {} source_model_path = str(actual_model_path or next(iter(source_hashes), str(model_file))) source_model_hash = ( source_hashes.get(source_model_path) or source_hashes.get(str(declared_model_file)) ) findings = list(agentic.get("consolidated_findings") or []) if not findings: findings = [ { "severity": "information", "category": "missing_information", "title": f"Missing: {item.get('field')}", "conclusion": item.get("reason") or item.get("status"), "action": "Supply, confirm non-applicability, or formally disposition before issue.", "evidence": ["missing_information.json"], } for item in missing ] package: dict[str, Any] = { "schema_version": "1.1", "sdk_version": "deterministic-evidence-adapter-1.1", "project": project, "source": { "backend": normalized.get("source_engine") or manifest.get("source_engine") or "Unknown SWMM backend", "adapter": normalized.get("evidence_adapter"), "sdk_version": manifest.get("version") or normalized.get("source_engine") or "deterministic SWMM evidence backend", "evidence_directory": str(root), "manifest": str(root / "manifest.json"), }, "source_model": { "filename": Path(str(model_file)).name, "path": str(model_file), "declared_path": str(declared_model_file or ""), "available": bool(actual_model_path and actual_model_path.is_file()), "sha256": source_model_hash, }, "simulation": { "pcswmm_results_available": bool(catalog), "source_engine": normalized.get("source_engine") or manifest.get("source_engine"), "run_status": project.get("run_status"), "active_scenario": project.get("active_scenario"), "result_catalog_count": len(catalog), }, "units": { "flow_units": next((r.get("units") for r in catalog if str(r.get("variable", "")).casefold() == "flow"), None), }, "results": { "included": bool(catalog), "hydraulic_summary": { "result_series_count": len(catalog), "figure_count": check["figure_count"], "table_count": check["table_count"], "timeseries_count": check["timeseries_count"], "collection_counts": collection_counts, }, "per_object_results": per_object, "pond_performance": tables.get("Table 08 Pond Performance", []), "storage_performance": { "freeboard_results": tables.get("Table 13 Storage Trap Low Results", []), }, "selected_tables": selection.get("selected_tables", []), "result_catalog": catalog, }, "engineering_review": { "status": agentic.get("overall_status") or "evidence_available_for_review", "reasoning_narrative": ( "The SWMM model was run through the identified deterministic backend and exported through the controlled " "evidence workflow. The MCP report uses the stored model inputs, result summaries, figures, " "tables, compatibility register, and audit hashes without recalculating hydraulic values." ), "findings": findings, "missing_information": missing, }, "evidence": { "root": str(root), "manifest": manifest, "selection": selection, "normalized_file": str(root / "normalized/calgary_swmr_native_evidence.json"), "workbook": str(root / "tables/City_of_Calgary_SWMR_Native_Evidence.xlsx"), "primary_figures": selection.get("primary_figures", []), "table_inventory": sorted(p.name for p in (root / "tables").glob("*.csv")), "figure_inventory": sorted(p.name for p in (root / "figures").glob("*.png")), }, } package["package_sha256"] = _sha256_json(package) return package def selected_figure_payloads(evidence_dir: str | Path, max_figures: int = 12) -> list[dict[str, str]]: root = Path(evidence_dir).expanduser().resolve() selection = _load(root / "evidence_selection.json") payloads: list[dict[str, str]] = [] for item in selection.get("primary_figures", [])[:max(0, int(max_figures))]: rel = item.get("path") if not rel: continue path = root / rel if not path.is_file() or path.suffix.lower() not in {".png", ".jpg", ".jpeg"}: continue title = item.get("title") or path.stem.replace("_", " ") low = title.casefold() section = ( "Storage and Pond Performance" if any(x in low for x in ("pond", "storage")) else "Minor Drainage System" if any(x in low for x in ("link", "conduit", "node", "head", "depth", "velocity")) else "Hydrology" if any(x in low for x in ("rain", "runoff", "subcatch")) else "Engineering Review" ) payloads.append({ "image_base64": base64.b64encode(path.read_bytes()).decode("ascii"), "caption": title, "section": section, "source_path": str(path), }) return payloads