Generating NPDES Discharge Monitoring Reports in Python

Every facility that discharges treated effluent under a National Pollutant Discharge Elimination System permit owes its regulator a periodic Discharge Monitoring Report, and the engineering task on this page is turning a month of bench-sheet and flow-meter data into a defensible DMR record that a NetDMR or EPA CDX intake will accept without rejection. This is a worked implementation inside the parent DMR Generation & SDWIS Submission Workflows section, aimed at the compliance analysts who sign the report and the Python engineers who own the pipeline that assembles it. A DMR is not a spreadsheet dump: each permitted parameter carries a code, a monitoring location, a required statistical basis, and — critically — a No-Data-Indicator convention for the reporting periods when the outfall did not discharge at all. Get the parameter-to-location structure, the monthly-average and daily-maximum math, and the NODI handling right, and serialization to the submission schema becomes mechanical. Get any of them wrong and the whole report bounces.

Prerequisites & Environment Setup

The implementation targets Python 3.10+ with pydantic for the typed DMR model, pandas for grouping raw samples into the reporting statistics, and the standard-library xml.etree.ElementTree for the NetDMR-style serialization — no third-party XML dependency is needed for a record this size. You also need three inputs that do not come from a package: the permit itself, which enumerates the permitted parameter codes, the monitoring locations, and the required statistical bases and limits; a monitoring dataset of dated lab results and flow readings keyed to those same parameters; and the facility’s NPDES permit number and outfall (monitoring point) identifiers, which anchor every record to the correct authority. Treat the permit-derived structure as versioned configuration, exactly as the SDWA MCL Reference Mapping section treats drinking-water limits, so that a permit reissuance is a data change rather than a code change.

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*" "pandas==2.2.*"

Step-by-Step Implementation

The build proceeds in four steps: model the permit-driven structure, aggregate the raw samples into the required statistics, decide between a reported value and a NODI code for each parameter, and serialize the assembled record to a submission-ready tree.

Step 1 — Model the parameter and monitoring-location structure

A DMR is a matrix. Down one axis sit the permitted parameters, each identified by the five-digit EPA parameter code (for example 50050 for flow, 00310 for five-day biochemical oxygen demand, 00530 for total suspended solids). Down the other sit the monitoring locations — the outfall or internal sampling points, each with a one-character location code such as 1 for effluent gross or G for a raw-influent point. Every cell where a parameter meets a location that the permit actually monitors becomes a limit set: a required statistical basis (monthly average, daily maximum, daily minimum), a numeric limit, a unit, and a sample frequency. Modeling this explicitly with pydantic gives every downstream step a validated contract to read from.

from __future__ import annotations
from datetime import date
from decimal import Decimal
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, field_validator


class StatBasis(str, Enum):
    MONTHLY_AVG = "monthly_average"
    DAILY_MAX = "daily_maximum"
    DAILY_MIN = "daily_minimum"


class LimitSet(BaseModel):
    parameter_code: str = Field(pattern=r"^\d{5}$")
    monitoring_location: str = Field(min_length=1, max_length=2)
    stat_basis: StatBasis
    limit: Optional[Decimal] = None      # None => monitor-and-report only
    unit_code: str                       # NetDMR unit code, e.g. "19" mg/L
    sample_frequency: str                # e.g. "01/30" = once per 30 days

    @field_validator("parameter_code")
    @classmethod
    def normalize(cls, v: str) -> str:
        return v.strip()


class DmrHeader(BaseModel):
    npdes_id: str = Field(pattern=r"^[A-Z]{2}\d{7}$")
    monitoring_period_start: date
    monitoring_period_end: date
    permitted_feature: str               # outfall id, e.g. "001"

The LimitSet is the unit of work: one parameter, one location, one statistical basis. A single outfall reporting BOD both as a monthly average and a daily maximum yields two limit sets, and the aggregation step must produce one value per set.

Step 2 — Aggregate monthly average and daily maximum with pandas

Raw monitoring data arrives as individual samples: a date, a parameter code, a location, and a measured result. The reporting statistics are simple reductions over the samples that fall inside the monitoring period. The monthly average is the arithmetic mean of the daily results,

xˉ=1n∑i=1nxi\bar{x} = \frac{1}{n}\sum_{i=1}^{n} x_i

where xix_i is the result for sample day ii and nn is the number of sampled days in the period. The daily maximum is max⁡ixi\max_i x_i and the daily minimum is min⁡ixi\min_i x_i. For a mass-based limit, the concentration must first be converted to a loading. The daily mass load is

L=Q⋅C⋅kL = Q \cdot C \cdot k

where QQ is the flow (MGD), CC is the concentration (mg/L), and k=8.34k = 8.34 is the unit factor that turns million-gallons times mg/L into pounds per day. The function below groups the sample frame by parameter and location and computes each basis in one pass.

import pandas as pd

MASS_FACTOR = Decimal("8.34")  # lb/day per (MGD * mg/L)


def aggregate_period(samples: pd.DataFrame, header: DmrHeader) -> pd.DataFrame:
    """Reduce raw samples to one row per (parameter, location, basis).

    Expected columns: sample_date, parameter_code, monitoring_location, result.
    """
    mask = (
        (samples["sample_date"] >= header.monitoring_period_start)
        & (samples["sample_date"] <= header.monitoring_period_end)
    )
    period = samples.loc[mask].copy()
    if period.empty:
        return pd.DataFrame(columns=["parameter_code", "monitoring_location",
                                     "stat_basis", "value", "n"])

    grouped = period.groupby(["parameter_code", "monitoring_location"])["result"]
    stats = grouped.agg(
        monthly_average="mean",
        daily_maximum="max",
        daily_minimum="min",
        n="count",
    ).reset_index()

    tidy = stats.melt(
        id_vars=["parameter_code", "monitoring_location", "n"],
        value_vars=["monthly_average", "daily_maximum", "daily_minimum"],
        var_name="stat_basis",
        value_name="value",
    )
    return tidy


def mass_load(flow_mgd: Decimal, conc_mg_l: Decimal) -> Decimal:
    """Convert a flow and concentration to a pounds-per-day loading."""
    return (flow_mgd * conc_mg_l * MASS_FACTOR).quantize(Decimal("0.01"))

Rounding is deliberately deferred: keep full precision through the mean, then quantize when the value is placed into the record, because rounding each sample first can shift a borderline monthly average across a permit limit and manufacture a violation that the raw data never supported. That distinction — a real exceedance versus a rounding artifact — is exactly what the Violation Code Classification section exists to adjudicate downstream.

Step 3 — Resolve each limit set to a value or a NODI code

A cell in the DMR is never blank. If the facility monitored and has a result, the cell carries the number. If it did not — the outfall did not discharge, the sample was lost, monitoring was conditionally waived — the cell carries a No-Data-Indicator code that tells the regulator why the number is absent. The most common is C for “no discharge”; a seasonal or wet-weather outfall that stayed dry all month reports C against every parameter rather than reporting a misleading zero. Resolving each limit set means joining the permit’s required cells against the computed statistics and substituting a NODI code wherever a value is missing.

How one DMR cell resolves to a computed value or a No-Data-Indicator code A required limit set flows into a no-discharge test. A dry period yields NODI code C. A discharging period flows into a result-exists test, which yields the computed value when a sample is present and NODI code B when it is not. Required limit set Discharged this period? Result exists? NODI = C (no discharge) value NODI = B (not sampled) no yes yes no 
class DmrCell(BaseModel):
    parameter_code: str
    monitoring_location: str
    stat_basis: StatBasis
    value: Optional[Decimal] = None
    unit_code: str
    nodi_code: Optional[str] = None      # mutually exclusive with value

    @field_validator("nodi_code")
    @classmethod
    def enforce_exclusivity(cls, v, info):
        if v is not None and info.data.get("value") is not None:
            raise ValueError("a cell carries either a value or a NODI code, never both")
        return v


def resolve_cells(limit_sets: list[LimitSet], stats: pd.DataFrame,
                  discharged: bool) -> list[DmrCell]:
    lookup = {
        (r.parameter_code, r.monitoring_location, r.stat_basis): r.value
        for r in stats.itertuples()
    }
    cells: list[DmrCell] = []
    for ls in limit_sets:
        if not discharged:
            cells.append(DmrCell(parameter_code=ls.parameter_code,
                                 monitoring_location=ls.monitoring_location,
                                 stat_basis=ls.stat_basis,
                                 unit_code=ls.unit_code, nodi_code="C"))
            continue
        raw = lookup.get((ls.parameter_code, ls.monitoring_location, ls.stat_basis.value))
        if raw is None or pd.isna(raw):
            cells.append(DmrCell(parameter_code=ls.parameter_code,
                                 monitoring_location=ls.monitoring_location,
                                 stat_basis=ls.stat_basis,
                                 unit_code=ls.unit_code, nodi_code="B"))
        else:
            cells.append(DmrCell(parameter_code=ls.parameter_code,
                                 monitoring_location=ls.monitoring_location,
                                 stat_basis=ls.stat_basis,
                                 unit_code=ls.unit_code,
                                 value=Decimal(str(raw)).quantize(Decimal("0.001"))))
    return cells

Step 4 — Serialize to a NetDMR/CDX-style record

The assembled cells serialize into the nested structure that a NetDMR batch or CDX exchange expects: a header carrying the NPDES identifier and monitoring period, then a PermittedFeature grouping the cells by outfall and parameter. The function below emits an ElementTree, which you can write to disk or hand to the transmission layer. The same provenance discipline that the Audit Trail & Data Lineage Storage Patterns section prescribes applies here: stamp the generated document with the source dataset hash before it leaves the process, so the submitted XML can always be traced back to the exact bench sheets that produced it.

import xml.etree.ElementTree as ET


def serialize_dmr(header: DmrHeader, cells: list[DmrCell]) -> ET.ElementTree:
    root = ET.Element("DischargeMonitoringReport")
    ET.SubElement(root, "PermitIdentifier").text = header.npdes_id
    period = ET.SubElement(root, "MonitoringPeriod")
    ET.SubElement(period, "StartDate").text = header.monitoring_period_start.isoformat()
    ET.SubElement(period, "EndDate").text = header.monitoring_period_end.isoformat()

    feature = ET.SubElement(root, "PermittedFeature",
                            attrib={"id": header.permitted_feature})
    for cell in cells:
        node = ET.SubElement(feature, "Parameter", attrib={
            "code": cell.parameter_code,
            "monitoringLocation": cell.monitoring_location,
            "statisticalBasis": cell.stat_basis.value,
        })
        if cell.nodi_code is not None:
            ET.SubElement(node, "NoDataIndicator").text = cell.nodi_code
        else:
            val = ET.SubElement(node, "ReportedValue")
            val.text = str(cell.value)
            val.set("unitCode", cell.unit_code)
    return ET.ElementTree(root)

Configuration Reference

The tables below capture the permit-driven inputs and the NODI vocabulary the resolver depends on. Pull the parameter codes, location codes, and unit codes directly from the permit and the NetDMR code lists; treat them as versioned data.

Limit-set parameters

Field Type Example Meaning
parameter_code str (5 digits) 00310 EPA parameter code for the analyte
monitoring_location str 1 Sampling point (effluent gross, influent, internal)
stat_basis enum monthly_average Required statistical reduction
limit Decimal or None 30.0 Permit limit; None means monitor-and-report only
unit_code str 19 NetDMR unit code (19 = mg/L, 03 = MGD)
sample_frequency str 01/30 Samples per period, e.g. once per 30 days

Common No-Data-Indicator codes

NODI Meaning Typical use
C No discharge Dry or seasonal outfall for the whole period
B Sampling not performed Missed or lost sample where discharge occurred
9 Conditional monitoring — not required Trigger condition for monitoring was not met
E Analysis lost or not valid Lab result rejected on QA grounds
7 Below detection limit Reported where the method could not quantify

Verification & Testing

Confirm the aggregation and NODI logic with deterministic tests over a small, known sample frame. A three-day BOD dataset of 10, 20, 30 mg/L must yield a monthly average of exactly 20 and a daily maximum of 30, and a period with no discharge must produce a C NODI code on every required cell rather than a numeric zero.

import pandas as pd
from decimal import Decimal


def test_monthly_average_and_daily_max():
    samples = pd.DataFrame({
        "sample_date": pd.to_datetime(["2026-06-05", "2026-06-15", "2026-06-25"]).date,
        "parameter_code": ["00310"] * 3,
        "monitoring_location": ["1"] * 3,
        "result": [10.0, 20.0, 30.0],
    })
    header = DmrHeader(npdes_id="CA1234567", permitted_feature="001",
                       monitoring_period_start=date(2026, 6, 1),
                       monitoring_period_end=date(2026, 6, 30))
    stats = aggregate_period(samples, header)
    avg = stats.query("stat_basis == 'monthly_average'")["value"].iloc[0]
    mx = stats.query("stat_basis == 'daily_maximum'")["value"].iloc[0]
    assert avg == 20.0 and mx == 30.0


def test_no_discharge_yields_nodi_c():
    ls = LimitSet(parameter_code="00310", monitoring_location="1",
                  stat_basis=StatBasis.MONTHLY_AVG, unit_code="19",
                  sample_frequency="01/30")
    empty = pd.DataFrame(columns=["parameter_code", "monitoring_location",
                                  "stat_basis", "value"])
    cells = resolve_cells([ls], empty, discharged=False)
    assert cells[0].nodi_code == "C" and cells[0].value is None


def test_mass_load_uses_834_factor():
    assert mass_load(Decimal("2.0"), Decimal("30.0")) == Decimal("500.40")

Acceptance criteria before a generated DMR is transmitted:

Troubleshooting & Gotchas

  • A dry outfall reports zeros instead of NODI C. A groupby over an empty slice, or a fill of missing values with 0, silently turns “no discharge” into “discharged at zero concentration,” which reads as perfect compliance and is a misreport. Gate the resolver on an explicit discharged flag derived from flow, not on whether the sample frame happens to be empty.
  • Borderline monthly averages flip across the limit. Rounding each daily result before averaging can push a mean across a permit limit. Carry Decimal precision through the mean and quantize once at output; leave the exceedance determination itself to the Violation Code Classification logic.
  • NetDMR rejects the batch on an unknown code. Parameter, unit, and NODI codes are controlled vocabularies. A locally invented unit string or a lowercase NODI letter fails intake. Validate every code against the current NetDMR code list at model construction, the same way the sibling formatting SDWIS state submission files workflow validates state schema codes.
  • Timezone drift splits a sample into the wrong month. A sample stamped near midnight on the last day of the period can fall out of the window if compared as a naive datetime across a timezone boundary. Reduce sample stamps to a plain date in the facility’s operating timezone before masking the period.
  • The same result feeds two reports with different rounding. The DMR sent to the regulator and the public-facing summary produced by the building Consumer Confidence Reports from compliance data workflow must trace to one canonical value. Compute the statistic once, persist it, and let both documents read the stored figure rather than re-aggregating independently.

Related pages