Building Consumer Confidence Reports from Compliance Data

Every community water system in the United States owes its customers one plain-language document each year: the Consumer Confidence Report, the annual drinking-water quality report mandated by 40 CFR Part 141 Subpart O. This page sits inside the parent DMR Generation & SDWIS Submission Workflows section, and where the sibling on generating NPDES discharge monitoring reports in Python points a regulated effluent stream back at the state, this page points the other direction — outward, to the people who drink the water. The engineering task is specific and repeatable: take a full calendar year of finished-water compliance results, reduce every detected contaminant to a range and an average, line each one up against its maximum contaminant level, fold in any violations, and render a mandatory-format report that a non-specialist can read. Done by hand each spring it is tedious and error-prone; done as code it becomes a deterministic build step that reuses the same validated dataset your SDWA MCL Reference Mapping already curates.

Prerequisites & Environment Setup

The pipeline targets Python 3.10+ and leans on three well-known libraries: pydantic for a typed sample model, jinja2 for HTML templating, and pandas only if your source results already live in a dataframe (it is optional — the core aggregation runs on the standard library). Everything here consumes already-validated compliance results; this page does not re-derive them. The upstream quality flags, provenance, and immutable timestamps come from the ingestion and Audit Trail & Data Lineage Storage Patterns layers, so the CCR builder can assume each row it receives has already cleared validation.

Two non-code inputs matter as much as the packages. First, the reporting year boundary: Subpart O covers a single calendar year and the finished report is due to customers by July 1 of the following year, so your query must clip to [Jan 1, Dec 31] of the target year in the plant’s local reporting calendar. Second, the canonical contaminant reference — the MCL, the MCLG, the reporting unit, and the EPA-worded “likely source” and health-effects language for each regulated analyte. Treat that reference as versioned data, never as inline constants, because the numbers and the mandated sentences are set by rule and change on EPA’s schedule.

One conceptual point is worth settling before any code runs. A CCR is a summary document, not a raw data dump. The rule does not ask you to reprint every sample; it asks you to distill a year of monitoring into one representative figure and one range per contaminant, alongside a factual account of anything that went wrong. That framing drives every design choice below: the pipeline is a reduction, and the value of the reduction depends entirely on the results that feed it being complete and correctly flagged. If a required sample was never collected, the honest output is a monitoring-and-reporting violation on the report — not a silently empty row — which is why the same dataset that produces the table also has to produce the violation narrative.

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*" "jinja2==3.1.*"
# Optional, only if your compliance results arrive as a dataframe:
pip install "pandas==2.2.*"

Step-by-Step Implementation

Step 1 — Model a year of compliance results

Start with a typed record so the aggregator never has to guess at units or detection status. A single result carries its canonical contaminant key, the measured concentration expressed in the reference unit for that analyte, an explicit detected boolean (True only when the result is at or above the method reporting limit), and the sample date. Reporting a non-detect as a literal zero is the single most common way a CCR average ends up wrong, so the model forces the distinction to be recorded rather than inferred.

from datetime import date
from pydantic import BaseModel, field_validator


class ComplianceResult(BaseModel):
    contaminant: str        # canonical key into the MCL reference
    value: float            # concentration in the analyte's reference unit
    detected: bool          # True only if at or above the reporting limit
    sample_date: date

    @field_validator("value")
    @classmethod
    def non_negative(cls, v: float) -> float:
        if v < 0:
            raise ValueError("concentration cannot be negative")
        return v

Step 2 — Reduce detects to a range and an average

Subpart O asks for two numbers per detected contaminant: the range of what was measured across the year and a representative level. For a set of detected results $c_1, c_2, \dots, c_n$ , the range is simply the closed interval

[\,\min_i c_i,\; \max_i c_i\,]

and the average reported for most contaminants is the arithmetic mean of the detected samples,

\bar{c} = \frac{1}{n}\sum_{i=1}^{n} c_i .

The critical rule is that only detected results enter this calculation. Undetected samples are excluded from the mean entirely — they are not zeros — and a contaminant that was never detected all year does not earn a row in the table at all. This is more than a rounding nicety. A system that collects twelve monthly TTHM samples and detects the by-product in only four of them reports the average of those four, and the range spans just those four measured values; the eight non-detects vanish from the arithmetic even though they were dutifully collected and analyzed. Encoding the detected flag as first-class data, rather than reconstructing it from whether a value happens to be zero, is what keeps that rule enforceable in code.

from collections import defaultdict
from statistics import mean
from typing import Iterable


def summarize_detections(results: Iterable[ComplianceResult]) -> dict[str, dict]:
    """Group by contaminant, keep only detects, and reduce each set to
    its count, range endpoints, and mean."""
    grouped: dict[str, list[float]] = defaultdict(list)
    for r in results:
        if r.detected:
            grouped[r.contaminant].append(r.value)

    summary: dict[str, dict] = {}
    for name, values in grouped.items():
        summary[name] = {
            "n": len(values),
            "low": min(values),
            "high": max(values),
            "average": round(mean(values), 4),
        }
    return summary

Step 3 — Join the MCL reference and decide the compliance figure

Each detected contaminant now needs its regulatory context: the MCL, the health goal (MCLG), the reporting unit, and the EPA “likely source” phrase. This is the same reference the SDWA MCL Reference Mapping section maintains, and the CCR builder simply reads from it. The subtle part is which number to compare against the MCL. For an acute contaminant such as nitrate, compliance turns on the single highest result, because one sample above 10 ppm is a violation. For contaminants governed by a running annual average — the disinfection by-products TTHM and HAA5 — the reported figure is the average, not the peak, because their health basis is chronic exposure rather than a single bad day. The reference therefore carries a basis field, and the row builder honors it rather than hard-coding one behavior for every analyte. Getting this join right is what lets the same curated numbers feed both this customer report and the state-facing encoders, and it keeps the CCR consistent with whatever the formatting SDWIS state submission files step reports for the identical year.

MCL_REFERENCE: dict[str, dict] = {
    "nitrate": {
        "label": "Nitrate (as N)", "unit": "ppm", "mcl": 10.0, "mclg": 10.0,
        "basis": "highest",
        "likely_source": "Runoff from fertilizer use; leaching from septic tanks; "
                         "erosion of natural deposits",
    },
    "tthm": {
        "label": "Total Trihalomethanes (TTHM)", "unit": "ppb", "mcl": 80.0, "mclg": None,
        "basis": "average",
        "likely_source": "By-product of drinking water disinfection",
    },
    "arsenic": {
        "label": "Arsenic", "unit": "ppb", "mcl": 10.0, "mclg": 0.0,
        "basis": "highest",
        "likely_source": "Erosion of natural deposits; runoff from orchards; "
                         "glass and electronics production wastes",
    },
}


def build_ccr_rows(summary: dict[str, dict],
                   reference: dict[str, dict]) -> list[dict]:
    rows = []
    for key, stats in summary.items():
        ref = reference[key]
        level = stats["high"] if ref["basis"] == "highest" else stats["average"]
        violation = ref["mcl"] is not None and level > ref["mcl"]
        rows.append({
            "label": ref["label"],
            "unit": ref["unit"],
            "mclg": ref["mclg"],
            "mcl": ref["mcl"],
            "detected_level": round(level, 4),
            "range_low": stats["low"],
            "range_high": stats["high"],
            "likely_source": ref["likely_source"],
            "violation": violation,
        })
    return rows

Step 4 — Assemble violation summaries and mandatory language

A CCR is not just a table. Subpart O requires a written summary of every violation that occurred during the year — whether an MCL exceedance, a treatment-technique failure, or a monitoring-and-reporting lapse — each with its length, the steps taken, and any EPA-worded health-effects statement the rule attaches to that contaminant. The builder keeps those statements beside the reference numbers so they travel together and cannot be forgotten. A monitoring gap that never breached a limit still surfaces here, which is exactly the kind of event the Audit Trail & Data Lineage Storage Patterns records make defensible after the fact.

HEALTH_EFFECTS: dict[str, str] = {
    "nitrate": (
        "Infants below the age of six months who drink water containing nitrate in "
        "excess of the MCL could become seriously ill and, if untreated, may die. "
        "Symptoms include shortness of breath and blue-baby syndrome."
    ),
    "arsenic": (
        "Some people who drink water containing arsenic in excess of the MCL over many "
        "years could experience skin damage or circulatory problems and may have an "
        "increased risk of getting cancer."
    ),
}


def violation_summaries(rows: list[dict]) -> list[dict]:
    """Turn every flagged row into a plain-language violation entry with the
    mandated health-effects statement attached."""
    out = []
    for row in rows:
        if not row["violation"]:
            continue
        key = next(k for k, v in MCL_REFERENCE.items()
                   if v["label"] == row["label"])
        out.append({
            "contaminant": row["label"],
            "detected_level": row["detected_level"],
            "mcl": row["mcl"],
            "unit": row["unit"],
            "health_effects": HEALTH_EFFECTS.get(key, ""),
        })
    return out

Step 5 — Render the HTML report with jinja2

Finally, hand the rows, the violation summaries, and the system identifiers to a Jinja2 template. Autoescaping is enabled because contaminant labels and system names are text you do not want interpreted as markup. Keeping the presentation in a template rather than string-concatenating HTML means the same reduced dataset can drive a printable mailer, a web page, and a plain-text alternative without recomputing a single figure — the numbers are fixed once and rendered many ways. The template below is deliberately compact; a production template also carries the required definitions (MCL, MCLG, ppm/ppb), the standard statement for immuno-compromised customers directing them to consult their health-care providers, the source-water assessment note, and the utility’s contact block for the public meeting where customers can ask questions.

from jinja2 import Environment, BaseLoader, select_autoescape

CCR_TEMPLATE = """<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<title> —  Water Quality Report</title></head>
<body>
<h1>  Consumer Confidence Report</h1>
<p>PWSID . Covering January 1 &ndash; December 31, .</p>
<table border="1" cellpadding="4">
  <thead><tr><th>Contaminant</th><th>MCLG</th><th>MCL</th>
  <th>Level Detected</th><th>Range</th><th>Unit</th>
  <th>Violation</th><th>Likely Source</th></tr></thead>
  <tbody>
  
  </tbody>
</table>

<p>Your water met all federal drinking water standards this year.</p>

</body></html>"""


def render_ccr(system_name: str, pwsid: str, year: int,
               rows: list[dict], violations: list[dict]) -> str:
    env = Environment(loader=BaseLoader(),
                      autoescape=select_autoescape(["html"]))
    template = env.from_string(CCR_TEMPLATE)
    return template.render(system_name=system_name, pwsid=pwsid,
                           year=year, rows=rows, violations=violations)

Configuration Reference

The table below documents the fields that drive one contaminant entry. Populate them from your versioned MCL reference, not from literals in the rendering code.

Field	Type	Example	Meaning
`contaminant`	str	`nitrate`	Canonical key joining a result to the reference
`label`	str	`Nitrate (as N)`	Public-facing name printed in the table
`unit`	str	`ppm`	Reporting unit; must match the sample values
`mcl`	float / None	`10.0`	Maximum contaminant level; `None` for action-level analytes
`mclg`	float / None	`10.0`	Health goal; `None` printed as “n/a” for by-products
`basis`	str	`highest`	`highest` for acute analytes, `average` for RAA-based ones
`likely_source`	str	`Runoff from fertilizer use…`	EPA-worded probable source of contamination
`detected`	bool	`True`	Per-sample flag; only `True` results enter the average

Verification & Testing

The aggregation and the compliance decision are the two places a wrong number does real harm, so pin both with a deterministic test. Undetected samples must be excluded from the mean, and an acute contaminant above its MCL must flag as a violation while an in-bounds one must not.

from datetime import date


def _r(value, detected):
    return ComplianceResult(contaminant="nitrate", value=value,
                            detected=detected, sample_date=date(2025, 6, 1))


def test_average_excludes_non_detects():
    results = [_r(4.0, True), _r(6.0, True), _r(0.0, False)]
    summary = summarize_detections(results)
    assert summary["nitrate"]["n"] == 2
    assert summary["nitrate"]["low"] == 4.0
    assert summary["nitrate"]["high"] == 6.0
    assert summary["nitrate"]["average"] == 5.0  # not 3.33 — the non-detect is dropped


def test_acute_violation_uses_highest_result():
    results = [_r(9.5, True), _r(11.2, True)]
    rows = build_ccr_rows(summarize_detections(results), MCL_REFERENCE)
    row = rows[0]
    assert row["detected_level"] == 11.2   # highest, not the 10.35 average
    assert row["violation"] is True


def test_in_bounds_is_not_a_violation():
    results = [_r(2.0, True), _r(3.0, True)]
    rows = build_ccr_rows(summarize_detections(results), MCL_REFERENCE)
    assert rows[0]["violation"] is False

Acceptance criteria before a report is published to customers:

Only contaminants actually detected during the reporting year appear in the table.
Non-detects are excluded from every average, never counted as zero.
Acute contaminants are judged on the highest single result; RAA contaminants on the average.
Every flagged violation carries its EPA-worded health-effects statement.
Units in the sample values match the unit field of each contaminant’s reference row.
Lead and copper are reported as the 90th-percentile result against the action level, not as an average against an MCL.
The rendered report names the reporting year, the PWSID, and the utility contact.

Troubleshooting & Gotchas

Non-detects counted as zero drag the average down. If your source query returns 0.0 for a non-detect and the model marks it detected=True, the mean is understated and the report is inaccurate. Carry the reporting-limit flag through ingestion and only let detected results reach summarize_detections.
Using the average where the rule wants the peak. Nitrate, arsenic, and other acute or non-averaged contaminants are judged on the single highest result. Comparing their yearly average to the MCL can silently hide a real exceedance — always drive the comparison from the reference basis field.
Unit mismatches between MCL and sample. Arsenic’s MCL is 0.010 mg/L, which is 10 ppb; feed a value in mg/L into a table keyed in ppb and a genuine violation reads as compliant by a factor of a thousand. Normalize every sample into the analyte’s reference unit before aggregating.
Lead and copper modeled as MCLs. These are action-level contaminants reported as a 90th-percentile household result, not an MCL against a plant average. Give them a None MCL and a separate action-level path so build_ccr_rows never mislabels them.
Mandatory language dropped for by-products. Contaminants like TTHM have no MCLG and their row prints “n/a” — but that must not cascade into skipping their required educational text. Keep the health-effects and definitions blocks independent of whether an MCLG value exists.

DMR Generation & SDWIS Submission Workflows — the parent section this report belongs to
Generating NPDES Discharge Monitoring Reports in Python — the outward-to-the-regulator counterpart to this customer-facing report
Formatting SDWIS State Submission Files — encoding the same compliance results for the state primacy agency
SDWA MCL Reference Mapping — the source of the MCL, MCLG, and likely-source values this builder joins against
Audit Trail & Data Lineage Storage Patterns — the provenance behind every result summarized here

Building Consumer Confidence Reports from Compliance Data

Related pages