Generating AWWA G400 Compliance Reports with Python

An AWWA G400 asset-management summary is only as defensible as the path between the water asset register and the figure on the page, and a report assembled by hand has no defensible path at all. When an analyst exports the water-main feature class, filters it in a spreadsheet, and types the cast-iron mileage into a submission template, no one can later reproduce that mileage, prove which assets were counted, or show that abandoned segments were handled consistently. The stakes are concrete: a G400 condition-and-inventory picture drives capital-renewal prioritization and rate-case justification, so an inventory that quietly omits a material class or double-counts a lifecycle state distorts a multi-year investment plan. Building the report as a Python pipeline that reads the register from a pinned version, aggregates deterministically, validates before it renders, and stamps the result with a hash turns the submission from an assertion into evidence. This page is a runnable implementation of the reporting discipline established across regulatory compliance reporting automation, and it assumes the asset-register foundations laid out under asset lifecycle and maintenance automation.

Environment Prerequisites

A misconfigured environment yields the most dangerous outcome — a report that renders cleanly while its figures are wrong. Lock the following before running anything:

ArcGIS Pro 3.2+ with a Standard or Advanced license, so utility-network coded-value domains and reconciled versioning behave as documented and arcpy is importable from the target interpreter.
Python 3.11 in an isolated conda environment created with conda create -n un-g400 python=3.11, with geopandas>=1.0 and pandas>=2.0 pinned via conda install -c conda-forge geopandas pandas so aggregation semantics stay reproducible across engineers.
A reconciled, read-only snapshot of the enterprise geodatabase — a DEFAULT replica or a file-geodatabase export versioned to a moment — never an actively edited version, so the extract reflects committed geometry and attributes only.
Coded-value domains loaded for MATERIAL, CONDITIONGRADE, and LIFECYCLESTATUS on the water-main class, with no free-text substitutes in the reporting attributes.
A pressure-zone attribute (or a zone feature class to spatial-join against) so inventory and condition roll up to the reporting unit a G400 review expects.
A field-binding map available at runtime — the attribute-by-attribute crosswalk documented in AWWA G400 audit field mapping for the UN schema — so each reported figure resolves to exactly one attribute and domain.

Schema-Aware Validation Protocol — Run Before the Report

Most G400 reporting defects originate in the register, not the aggregation. Work this ordered checklist first; the earliest item is the most frequent culprit.

Confirm the lifecycle filter matches the report. Decide explicitly which lifecycle states the submission counts. In-service and abandoned mains are reported differently and retired segments usually drop out; inferring state from install age or inspection recency instead of the LIFECYCLESTATUS domain is the classic source of an inflated inventory.
Verify domain conformance on every reporting attribute. A MATERIAL value of CI with a trailing space, or a condition grade typed outside its 1–5 domain, is aggregated as a distinct category and silently shifts a subtotal. Compare the distinct values against the loaded domain, never against a hard-coded list in the script.
Check non-null coverage before aggregating. A material field that is a few percent null produces a length total that is a few percent low and entirely plausible. Quantify nulls per reporting field and treat any null in a non-nullable field as a blocking failure.
Confirm the unit of measure. Main length must be aggregated in the unit the template expects; a length stored in feet against a metre-based template is the classic order-of-magnitude error. Read the unit from the geodatabase, do not assume it.
Pin one version moment across all classes. Extract the water-main and service-line classes from the same version moment. A report stitched from extracts taken minutes apart across an active edit is internally inconsistent even when each figure is individually right.

Minimal Reproducible Implementation

The following pipeline reads the water asset register through arcpy, aggregates the G400 inventory-and-condition metrics per pressure zone with pandas, validates completeness and domains, writes the report tables, and appends a hash-stamped audit record. It handles the empty-frame and missing-attribute cases explicitly and returns a structured result rather than printing.

import arcpy
import pandas as pd
import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

MATERIAL_DOMAIN = {"DI", "CI", "PVC", "HDPE", "STEEL", "AC", "OTHER"}
CONDITION_DOMAIN = {1, 2, 3, 4, 5}
REPORTABLE_STATES = {"IN_SERVICE", "ABANDONED"}


@dataclass
class G400Report:
    """Structured result of a G400 report run."""
    inventory_by_material: pd.DataFrame
    condition_by_zone: pd.DataFrame
    audit_record: dict
    ok: bool = True
    issues: list[str] = field(default_factory=list)


def _read_mains(fc_path: str) -> pd.DataFrame:
    """Read reportable water-main rows into a DataFrame."""
    required = ["MATERIAL", "CONDITIONGRADE", "LIFECYCLESTATUS",
                "PRESSUREZONE", "LENGTH_M"]
    existing = {f.name for f in arcpy.ListFields(fc_path)}
    missing = [c for c in required if c not in existing]
    if missing:
        raise KeyError(f"{fc_path}: missing reporting attributes {missing}")

    rows: list[dict] = []
    with arcpy.da.SearchCursor(fc_path, ["OID@", *required]) as cursor:
        for oid, material, grade, lifecycle, zone, length in cursor:
            if lifecycle not in REPORTABLE_STATES:
                continue
            rows.append({
                "OBJECTID": oid, "MATERIAL": material, "CONDITIONGRADE": grade,
                "LIFECYCLESTATUS": lifecycle, "PRESSUREZONE": zone,
                "LENGTH_M": length,
            })
    return pd.DataFrame.from_records(rows)


def _validate(frame: pd.DataFrame) -> list[str]:
    """Return a list of blocking validation issues; empty means clean."""
    issues: list[str] = []
    if frame.empty:
        return ["no reportable rows extracted"]
    for col in ("MATERIAL", "CONDITIONGRADE", "PRESSUREZONE", "LENGTH_M"):
        nulls = int(frame[col].isna().sum())
        if nulls:
            issues.append(f"{col}: {nulls} null values in non-nullable field")
    bad_material = set(frame["MATERIAL"].dropna().unique()) - MATERIAL_DOMAIN
    if bad_material:
        issues.append(f"MATERIAL out of domain: {sorted(bad_material)}")
    bad_grade = set(frame["CONDITIONGRADE"].dropna().unique()) - CONDITION_DOMAIN
    if bad_grade:
        issues.append(f"CONDITIONGRADE out of domain: {sorted(bad_grade)}")
    return issues


def generate_g400_report(
    mains_fc: str,
    version_moment: str,
    bindings_version: str,
    out_dir: str,
) -> G400Report:
    """Assemble an AWWA G400 inventory-and-condition report.

    Reads reportable mains, validates completeness and domains, aggregates
    metrics by material and pressure zone, writes report tables, and seals
    the run with a hashed audit record. Raises only on unrecoverable I/O or
    schema errors; validation failures are returned in the result.
    """
    frame = _read_mains(mains_fc)
    issues = _validate(frame)

    if issues:
        logging.error("G400 validation failed: %s", "; ".join(issues))
        empty = pd.DataFrame()
        audit = _seal(b"", b"", version_moment, bindings_version)
        return G400Report(empty, empty, audit, ok=False, issues=issues)

    inventory = (
        frame.groupby("MATERIAL")["LENGTH_M"].sum()
        .div(1000.0).round(3).rename("length_km").reset_index()
    )
    condition = (
        frame.groupby("PRESSUREZONE")["CONDITIONGRADE"]
        .mean().round(2).rename("mean_condition").reset_index()
    )

    inv_path = f"{out_dir}/g400_inventory_by_material.csv"
    cond_path = f"{out_dir}/g400_condition_by_zone.csv"
    inventory.to_csv(inv_path, index=False)
    condition.to_csv(cond_path, index=False)

    report_bytes = (
        inventory.to_csv(index=False) + condition.to_csv(index=False)
    ).encode("utf-8")
    snapshot_bytes = frame.to_csv(index=False).encode("utf-8")
    audit = _seal(snapshot_bytes, report_bytes, version_moment, bindings_version)

    logging.info("G400 report written: %d materials, %d zones",
                 len(inventory), len(condition))
    return G400Report(inventory, condition, audit)


def _seal(snapshot: bytes, report: bytes,
          version_moment: str, bindings_version: str) -> dict:
    """Immutable audit record binding inputs to outputs."""
    record = {
        "sealed_utc": datetime.now(timezone.utc).isoformat(),
        "version_moment": version_moment,
        "bindings_version": bindings_version,
        "snapshot_sha256": hashlib.sha256(snapshot).hexdigest(),
        "report_sha256": hashlib.sha256(report).hexdigest(),
        "tool_versions": {"arcpy": arcpy.GetInstallInfo().get("Version", "?"),
                          "pandas": pd.__version__},
    }
    record["record_sha256"] = hashlib.sha256(
        json.dumps(record, sort_keys=True).encode("utf-8")
    ).hexdigest()
    return record


if __name__ == "__main__":
    result = generate_g400_report(
        mains_fc="C:/snapshots/water.gdb/wMain",
        version_moment="2026-07-13T04:00:00Z",
        bindings_version="g400-bindings-v4",
        out_dir="C:/reports/2026Q3",
    )
    print(json.dumps(result.audit_record, indent=2))
    if not result.ok:
        raise SystemExit(f"G400 report blocked: {result.issues}")

The two figures the pipeline produces map directly onto the two halves of a G400 review — an inventory picture by material and a condition picture by zone — and the diagram below shows how those two aggregations share one validated frame and one audit seal.

Two implementation choices carry the reproducibility guarantee. Filtering on LIFECYCLESTATUS inside the cursor means retired assets never enter the frame, so the inventory reflects the reportable system rather than the whole geodatabase. Sealing the exact snapshot bytes the report was built from — not a re-read of the source — means the audit hash proves what was actually aggregated, so a later edit to the register is detectable rather than invisible.

Production Deployment Pattern

A notebook run is a diagnostic; a scheduled gate is an engineering control. Promote the pipeline as follows:

Run against a reconciled, isolated snapshot. Export the water classes from the DEFAULT version or a read-only replica with no active edit session, so the report reflects committed geometry and never contends with editor locks or shifts mid-run.
Schedule it unattended and idempotent. Trigger the job from cron, Windows Task Scheduler, or a CI runner on the reporting cadence. Because a re-run against the same version moment produces byte-identical output and therefore an identical hash, a retried run after a transient failure is provably the same report.
Gate on the validation result. Let a non-empty issues list fail the job so an out-of-domain material or a null-coverage gap blocks the submission instead of degrading it to a plausible-but-wrong report.
Persist the audit record to a write-once store. Append the sealed record — version moment, snapshot and report hashes, bindings version, tool versions — to an append-only table or object-lock bucket whose write path is independent of the geodatabase, so the report can be evidenced even after the source is locked for maintenance.
Push the tables downstream over REST. Once the gate passes, publish the inventory and condition tables to the enterprise asset-management system so the same figures drive capital planning and the regulatory submission, using the standardized field mappings so a main resolves identically in GIS, the report, and the work-management system.

Conclusion

This pipeline replaces a hand-built spreadsheet with a deterministic procedure that reads the water register from a pinned version, filters to reportable lifecycle states, validates domain and completeness before it aggregates, and seals the output with a content hash. Enforcing it keeps the G400 inventory-and-condition figures exact, makes any post-hoc edit detectable, and produces the version-stamped audit trail a rate-case or regulatory review requires. The natural next step is to extend the same binding-driven structure to the EPA service-line inventory, so both submissions regenerate from one validated register. Adopt the shared field crosswalk so every engineer’s report resolves each figure to the same attribute.

Up to the parent topic: Regulatory Compliance Reporting Automation
Up to the section: Asset Lifecycle & Maintenance Automation
AWWA G400 Audit Field Mapping for the UN Schema — the attribute crosswalk this generator reads at runtime.
Lifecycle State Machines for Utility Assets — the in-service and abandoned states the report filters on.

For authoritative reference, consult the AWWA standards program and the Python hashlib documentation.