"""Deterministic screening logic shared by the report engine and tests. Implements the evidence-precedence, continuity-disclosure, and missing-information rules that must never be delegated to an LLM: 1. Effective-velocity precedence — for links flagged by the worker-vs-.rpt reconciliation, the engine .rpt value governs screening; both values, their differences, and whether the discrepancy changes the screening classification are recorded. Unflagged links use worker values. 2. Continuity disclosure — runoff and routing errors are reported separately, sign preserved, each checked against ABSOLUTE review/warning thresholds; water quality is "Not applicable" when no pollutants are modelled. 3. Missing-information register — deterministic list of evidence the report cannot supply, assembled from metadata, the criteria register, and the checklist. Anything listed here can never be a Pass elsewhere. """ from __future__ import annotations import math from typing import Any, Mapping import pandas as pd # --------------------------------------------------------------------------- # Solver-option and execution-integrity gates # --------------------------------------------------------------------------- def resolve_legacy_solver_options(options: Mapping[str, Any]) -> dict[str, Any]: """Resolve auditable SWMM legacy-zero sentinels for an execution copy. Older/converted INP files can explicitly serialize zero for dynamic-wave options that EPA SWMM displays and executes using unit-aware defaults. This function returns substitutions for an immutable derivative; it never edits the uploaded source model. Negative and non-numeric values remain blocking errors. Omitted values remain omitted for the engine to default. """ opts = {str(k).upper(): str(v).strip() for k, v in options.items()} if opts.get("FLOW_ROUTING", "").upper() != "DYNWAVE": return {"effective_options": dict(opts), "substitutions": [], "errors": []} flow_units = opts.get("FLOW_UNITS", "").upper() si_units = flow_units in {"CMS", "LPS", "MLD"} defaults = { "MAX_TRIALS": (8.0, "count"), "HEAD_TOLERANCE": (0.0015 if si_units else 0.005, "m" if si_units else "ft"), "MIN_SURFAREA": (1.167 if si_units else 12.566, "m2" if si_units else "ft2"), } effective = dict(opts) substitutions: list[dict[str, Any]] = [] errors: list[str] = [] for name, (default, units) in defaults.items(): if name not in opts: continue # omitted means use the engine default try: value = float(opts[name]) except (TypeError, ValueError): errors.append(f"{name} must be numeric for dynamic-wave routing.") continue if value < 0: errors.append( f"{name} cannot be negative for dynamic-wave routing; " f"the uploaded value is {opts[name]!r}." ) elif value == 0: effective[name] = format(default, "g") substitutions.append({ "option": name, "original_value": opts[name], "effective_value": default, "units": units, "reason": "Recognized legacy zero/default sentinel", }) return {"effective_options": effective, "substitutions": substitutions, "errors": errors} def validate_solver_options(options: Mapping[str, Any]) -> list[str]: """Return only blocking errors after legacy-default resolution.""" return list(resolve_legacy_solver_options(options)["errors"]) def execution_integrity_assessment(metadata: Mapping[str, Any]) -> dict[str, Any]: """Classify whether hydraulic results can support screening conclusions.""" def number(key: str, default: float = 0.0) -> float: try: return float(metadata.get(key, default)) except (TypeError, ValueError): return default steps = int(number("routing_steps")) failed = int(number("not_converged_steps")) pct_failed = number("pct_not_converged") flow_error = abs(number("flow_error")) runoff_error = abs(number("runoff_error")) invalid_reasons: list[str] = [] if steps > 0 and failed >= steps: invalid_reasons.append("every routing step failed to converge") elif pct_failed >= 5.0: invalid_reasons.append(f"{pct_failed:.3f}% of routing steps failed to converge") if flow_error >= 10.0: invalid_reasons.append(f"flow-routing continuity error is {flow_error:.3f}%") if invalid_reasons: return { "status": "invalid", "results_usable": False, "hydraulic_conclusions_allowed": False, "reason": "; ".join(invalid_reasons) + ".", } limitations: list[str] = [] if failed > 0: limitations.append(f"{failed} routing step(s) did not converge") if flow_error > 1.0: limitations.append(f"flow-routing continuity error is {flow_error:.3f}%") if runoff_error > 1.0: limitations.append(f"runoff continuity error is {runoff_error:.3f}%") return { "status": "limited" if limitations else "valid", "results_usable": True, "hydraulic_conclusions_allowed": True, "reason": "; ".join(limitations) + ("." if limitations else "Execution-integrity checks passed."), } # --------------------------------------------------------------------------- # Velocity classification and evidence precedence # --------------------------------------------------------------------------- def classify_velocity(velocity: float | None, advisory: float = 3.0, critical: float = 4.0) -> str: """Deterministic dual-threshold screening classification (not a regulatory determination).""" if velocity is None: return "Not assessed" try: v = float(velocity) except (TypeError, ValueError): return "Not assessed" if math.isnan(v): return "Not assessed" if v > critical: return f"Critical screening exceedance (> {critical:g} m/s)" if v > advisory: return f"Advisory screening exceedance (> {advisory:g} m/s)" return f"Below advisory threshold ({advisory:g} m/s)" def effective_velocity_table(link_df: pd.DataFrame, recon_links: pd.DataFrame | None, advisory: float = 3.0, critical: float = 4.0) -> pd.DataFrame: """Per-conduit screening table applying the reconciliation precedence. Columns: Link ID, Worker Peak Velocity, RPT Peak Velocity, Screening Velocity, Evidence Source, Delta (abs), Delta (%), Screening Classification, Classification Changed by Reconciliation. """ if link_df is None or link_df.empty: return pd.DataFrame() vel_col = next((c for c in link_df.columns if c.startswith("Peak Velocity")), None) if vel_col is None: return pd.DataFrame() recon: dict[str, dict[str, Any]] = {} if recon_links is not None and not recon_links.empty: for _, r in recon_links.iterrows(): recon[str(r.get("Link ID"))] = r.to_dict() rows: list[dict[str, Any]] = [] for _, r in link_df.iterrows(): link_id = str(r.get("Link ID")) worker_v = pd.to_numeric(pd.Series([r.get(vel_col)]), errors="coerce").iloc[0] rec = recon.get(link_id, {}) rpt_v = rec.get("RPT Peak Velocity") rpt_v = float(rpt_v) if rpt_v is not None and not (isinstance(rpt_v, float) and math.isnan(rpt_v)) else None flagged = str(rec.get("Overall Status", "OK")) not in ("OK", "Unavailable", "nan", "None") if flagged and rpt_v is not None: eff, source = rpt_v, "engine .rpt (reconciliation-flagged)" else: eff, source = (float(worker_v) if pd.notna(worker_v) else None), "worker time series" worker_class = classify_velocity(float(worker_v) if pd.notna(worker_v) else None, advisory, critical) eff_class = classify_velocity(eff, advisory, critical) delta_abs = (float(worker_v) - rpt_v) if (pd.notna(worker_v) and rpt_v is not None) else None delta_pct = (100.0 * delta_abs / abs(rpt_v)) if (delta_abs is not None and rpt_v not in (None, 0)) else None rows.append({ "Link ID": link_id, "Worker Peak Velocity (m/s)": round(float(worker_v), 3) if pd.notna(worker_v) else None, "RPT Peak Velocity (m/s)": round(rpt_v, 3) if rpt_v is not None else None, "Screening Velocity (m/s)": round(eff, 3) if eff is not None else None, "Evidence Source": source, "Delta (m/s)": round(delta_abs, 3) if delta_abs is not None else None, "Delta (%)": round(delta_pct, 1) if delta_pct is not None else None, "Screening Classification": eff_class, "Classification Changed by Reconciliation": ( "Yes" if (flagged and rpt_v is not None and worker_class != eff_class) else ("No" if flagged else "n/a - not flagged")), }) return pd.DataFrame(rows) # --------------------------------------------------------------------------- # Continuity disclosure # --------------------------------------------------------------------------- def continuity_disclosure(metadata: Mapping[str, Any], review_pct: float = 0.5, warning_pct: float = 1.0, has_pollutants: bool = False) -> list[str]: """Sign-preserving continuity lines with symmetric absolute thresholds.""" lines: list[str] = [] for label, key in (("Surface-runoff continuity error", "runoff_error"), ("Flow-routing continuity error", "flow_error")): val = metadata.get(key) try: v = float(val) except (TypeError, ValueError): lines.append(f"{label}: not reported by the engine.") continue lines.append(f"{label}: {v:+.3f}% (engine-reported sign preserved).") if abs(v) > warning_pct: lines.append(f"⚠️ {label} magnitude |{v:.3f}%| exceeds the {warning_pct:g}% absolute warning threshold and must be reviewed before the results are relied upon.") elif abs(v) > review_pct: lines.append(f"{label} magnitude |{v:.3f}%| exceeds the {review_pct:g}% absolute review threshold.") if has_pollutants: qv = metadata.get("quality_error") try: lines.append(f"Water-quality continuity error: {float(qv):+.3f}%.") except (TypeError, ValueError): lines.append("Water-quality continuity error: pollutants modelled but continuity not reported — review engine output.") else: lines.append("Water-quality continuity: Not applicable — no pollutants modelled.") return lines # --------------------------------------------------------------------------- # Missing-information register # --------------------------------------------------------------------------- _METADATA_LABELS = { "legal_description": "Legal land description", "outline_plan_no": "Outline plan number", "subdivision_no": "Subdivision number", "development_permit_no": "Development permit number", "consultant_file_no": "Consultant file number", "prepared_by": "Prepared by (responsible person)", "checked_by": "Checked by (reviewer)", "client": "Client", "consultant": "Consultant", "construction_drawing_no": "Construction drawing number", "development_agreement_no": "Development agreement number", } def missing_information_register(metadata: Mapping[str, Any], criteria_register: pd.DataFrame | None, checklist: pd.DataFrame | None) -> pd.DataFrame: """Deterministic register of evidence the report cannot supply. Items listed here block any related Pass classification elsewhere. """ rows: list[dict[str, str]] = [] for key, label in _METADATA_LABELS.items(): value = str(metadata.get(key, "") or "").strip() if not value or value.lower() in ("not provided", "none", "-", "—"): rows.append({"Item": label, "Category": "Project information", "Status": "Not provided", "Consequence": "Related administrative checklist items remain incomplete."}) if criteria_register is not None and not criteria_register.empty: status_col = next((c for c in criteria_register.columns if "status" in c.lower()), None) name_col = next((c for c in criteria_register.columns if c.lower() in ("criterion", "requirement", "item", "name")), criteria_register.columns[0]) if status_col: for _, r in criteria_register.iterrows(): if "not established" in str(r.get(status_col, "")).lower(): rows.append({"Item": str(r.get(name_col)), "Category": "Governing criteria", "Status": "Not established", "Consequence": "Related screening cannot be reported as Pass; results remain screening-only."}) if checklist is not None and not checklist.empty and "Status" in checklist.columns: for _, r in checklist.iterrows(): if str(r.get("Status", "")).strip().lower() == "missing": rows.append({"Item": f"{r.get('Item')}: {str(r.get('Requirement'))[:80]}", "Category": "SWMR checklist", "Status": "Missing", "Consequence": "Required for a submission-ready report."}) if not rows: rows.append({"Item": "None identified", "Category": "—", "Status": "—", "Consequence": "All tracked evidence items were supplied."}) return pd.DataFrame(rows)