9 min read2,018 words

Automating EPA LCRI Service Line Inventory Reporting

A service-line inventory is unlike most regulatory returns in one respect: the category that matters most is the one representing an absence of knowledge. Lead and galvanised-requiring- replacement services drive a replacement obligation; unknown services drive an investigation obligation; and an inventory that quietly omits its unknowns looks complete while understating both. Building the return by hand from a customer database and a stack of tap cards is how most utilities produced their first one, and it does not survive annual repetition. This guide assembles the inventory from the network model and the evidence record, keeps the utility-owned and customer-owned portions separate as the rule requires, reports unknowns as a category, and seals the output so a later reviewer can reproduce it. It extends the reporting machinery in regulatory compliance reporting automation.

Environment Prerequisites

  • Python 3.11 with pandas>=2.0 and geopandas>=1.0, and a read-only connection to a reconciled snapshot of the network.
  • Service lines split into their utility-owned and customer-owned portions, either as separate features or as an attribute that identifies the boundary at the property line or the curb stop.
  • An evidence record per portion — the source of the material classification, its type (records, observation, statistical) and its date — because the rule cares how a classification was reached.
  • Coded-value domains for material that include every category the rule defines, including unknown, with no free-text substitutes.
  • The reporting unit the rule specifies, and a spatial or attribute join that resolves every service to one.
  • A write-once store for the sealed return, its hashes and the version moment.

Schema-Aware Validation Protocol — Run Before Assembling the Return

  1. Confirm both portions exist for every service. A service represented as one feature cannot report a utility-side and customer-side material separately, and the rule requires both.
  2. Check the evidence type is populated wherever a material is claimed. A material with no evidence behind it is functionally an unknown, and reporting it as a classification is the defect a reviewer looks for.
  3. Verify unknowns are represented, not absent. Services missing from the inventory entirely are the failure mode this validation exists to catch; count the services in the customer system and reconcile.
  4. Confirm domain conformance on the material field. One free-text variant of “lead” drops a whole group from a category total.
  5. Pin one version moment across every class read. An inventory stitched from extracts taken hours apart describes two different estates.

Minimal Reproducible Implementation

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, field

import pandas as pd

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
LOG = logging.getLogger("lcri-inventory")

CATEGORIES = ("LEAD", "GRR", "NON_LEAD", "UNKNOWN")
EVIDENCE_RANK = {"OBSERVATION": 3, "RECORDS": 2, "STATISTICAL": 1}


@dataclass
class ServiceClassification:
    service_id: str
    utility_side: str
    customer_side: str
    evidence_utility: str
    evidence_customer: str
    reporting_unit: str


@dataclass
class InventoryReturn:
    rows: list[ServiceClassification] = field(default_factory=list)
    issues: list[str] = field(default_factory=list)

    def totals(self) -> pd.DataFrame:
        frame = pd.DataFrame([{
            "reporting_unit": r.reporting_unit,
            "utility_side": r.utility_side,
            "customer_side": r.customer_side,
        } for r in self.rows])
        return (frame.melt(id_vars="reporting_unit", var_name="portion",
                           value_name="category")
                .groupby(["reporting_unit", "portion", "category"])
                .size().rename("services").reset_index())


def classify(portions: pd.DataFrame) -> InventoryReturn:
    """Build the inventory, taking the strongest evidence per portion.

    Where several records describe one portion — a tap card, a field observation, a
    statistical inference — the strongest evidence wins, and a portion with no record
    at all becomes UNKNOWN rather than being dropped. Dropping it is what makes an
    inventory look complete while understating the obligation behind it.
    """
    result = InventoryReturn()

    for service_id, rows in portions.groupby("service_id"):
        sides: dict[str, tuple[str, str]] = {}
        for side in ("UTILITY", "CUSTOMER"):
            candidates = rows[rows["side"] == side]
            if candidates.empty:
                sides[side] = ("UNKNOWN", "NONE")
                continue
            best = candidates.assign(
                rank=candidates["evidence_type"].map(EVIDENCE_RANK).fillna(0)
            ).sort_values("rank", ascending=False).iloc[0]
            category = str(best["material_category"]).upper()
            if category not in CATEGORIES:
                result.issues.append(
                    f"{service_id}/{side}: {category!r} is not a reportable category")
                category = "UNKNOWN"
            sides[side] = (category, str(best["evidence_type"]))

        unit = str(rows.iloc[0]["reporting_unit"])
        result.rows.append(ServiceClassification(
            service_id=str(service_id),
            utility_side=sides["UTILITY"][0], customer_side=sides["CUSTOMER"][0],
            evidence_utility=sides["UTILITY"][1], evidence_customer=sides["CUSTOMER"][1],
            reporting_unit=unit))

    LOG.info("classified %d service(s), %d issue(s)", len(result.rows),
             len(result.issues))
    return result


def seal(totals: pd.DataFrame, version_moment: str, method: str) -> dict:
    """Hash the return so a reviewer can verify it against the inputs."""
    payload = {
        "version_moment": version_moment, "method": method,
        "totals": totals.to_dict(orient="records"),
    }
    blob = json.dumps(payload, sort_keys=True).encode("utf-8")
    payload["sha256"] = hashlib.sha256(blob).hexdigest()
    return payload
The service-line material categories an inventory must report, and what each requires A service-line inventory classifies both the utility-owned and the customer-owned portion of every service. The categories are defined by the rule rather than by the utility, and the evidence standard differs: a lead classification usually rests on a physical observation, while unknown is a valid category that carries an obligation to resolve it. Category Evidence typically required Obligation Lead observation or records replacement programme Galvanised requiring replacement observation + lead history replacement programme Non-lead records or observation none beyond reporting Unknown none — that is the point investigate and resolve Unknown is a reportable category with an obligation, not a gap in the return.

Taking the strongest evidence per portion, rather than the most recent, is the choice worth defending. A physical observation from five years ago is better evidence than a statistical inference made last month, and the ranking makes that explicit rather than leaving it to whichever record happened to be written last.

Assembling a service-line inventory from the network model and the evidence record Each service line is resolved into its utility-owned and customer-owned portions, because they are classified separately. Material comes from the strongest evidence available and the evidence type travels with it. Anything unresolved is reported as unknown rather than omitted. The output is aggregated by the reporting unit the rule specifies, and the whole run is sealed with its version moment. SPLIT utility side / customer side CLASSIFY strongest evidence per portion UNKNOWN reported, not omitted AGGREGATE by reporting unit the rule names SEAL version moment and hashes the tempting error OMITTING UNKNOWNS makes the inventory look complete and understates the replacement obligation An inventory that omits its unknowns is not conservative; it is wrong.

Reporting the Unknowns Honestly

An inventory’s unknown count is the number that drives the investigation programme, and there are two ways to make it look better than it is. The first is omission: services absent from the inventory entirely, usually because they are missing from the network model rather than because anyone decided to exclude them. The reconciliation against the customer system is what catches this, and it belongs in the validation rather than in a spot check.

The second is over-claiming: classifying a portion from evidence that does not support it, most often a records-based inference from a construction era. Where a jurisdiction permits records-based classification the practice is legitimate — but the evidence type has to travel with the classification, so a reviewer can see how much of a non-lead total rests on inference rather than observation.

Both failures produce a return that is easier to file and harder to defend, and both are visible in the evidence-type breakdown that the classifier already produces.

Inventory composition across three annual returns as investigation proceeds The useful measure of an inventory programme is not the lead count but the unknown count, because unknowns are the part of the estate whose obligation has not yet been established. A programme that is working moves services out of unknown in both directions, and a lead count that rises while unknowns fall is a sign the investigation is finding what it was meant to find. share of service connections by category, one system Year 1 — unknown 42 % baseline Year 2 — unknown 24 % investigation underway Year 3 — unknown 9 % records + potholing Year 3 — lead 11 % found, now programmed Falling unknowns and a rising lead count together mean the programme is working.

Production Deployment Pattern

  1. Run the assembly against a pinned snapshot and seal the output, exactly as the G400 reporting workflow does.
  2. Reconcile against the customer system every run. A service in billing and absent from the inventory is the omission this exercise most needs to catch.
  3. Publish the evidence-type breakdown alongside the category totals, internally at minimum, because it is the honest description of how much the return rests on inference.
  4. Feed unknowns into the investigation programme ranked by exposure — services feeding schools, childcare and healthcare first — so the programme is aimed rather than alphabetical.
  5. Track the unknown share between annual returns. It is the measure of programme progress; the lead count on its own can rise for good reasons.
  6. Keep every filed return and its hashes. Year-on-year comparison is part of the obligation, and a return that cannot be reproduced cannot be compared.

Conclusion

A service-line inventory is an assembly problem with an unusual property: its most consequential category is the absence of knowledge. Splitting each service into its two portions, taking the strongest evidence for each, carrying the evidence type through to the return and reporting unknowns as a category rather than a gap produces a filing that is defensible and an investigation queue that is aimed. Sealing the output makes each annual return reproducible, which is what turns a series of filings into a programme with a measurable trend.

For authoritative reference, consult the EPA drinking water regulations and the pandas reshaping guide.

Related pages