Formatting SDWIS State Submission Files

Every compliance sample a water system pulls eventually has to leave the laboratory information system and arrive at the state primacy agency in a shape their Safe Drinking Water Information System (SDWIS) can ingest without rejecting the batch. That handoff is the exact task on this page, and it belongs to the DMR Generation & SDWIS Submission Workflows section, alongside the two companion builds it shares a data source with — generating NPDES discharge monitoring reports in Python and building consumer confidence reports from compliance data. The three consume the same validated sample-result records; they diverge only in how those records are packaged, which is exactly why keeping the record model and its validation rules in one shared module pays off across all three deliverables. Here the target is a machine-readable state submission file: sample and result rows keyed by public water system identifier (PWSID), each carrying an analyte code, an analytical method, a numeric result with units, and a detection limit, serialized to a fixed-width or XML transfer format and shipped with a manifest and checksum so the receiving agency can prove nothing was truncated in transit.

Prerequisites & Environment Setup

The build targets Python 3.10+ and leans on pydantic v2 for the record schema, because a state agency will bounce an entire file over a single malformed row and you want that failure caught on your side of the wire. Standard-library xml.etree.ElementTree handles the XML variant and hashlib produces the manifest digest, so the third-party surface stays small. Keeping the dependency footprint minimal is deliberate: a submission tool that a compliance officer runs on a locked-down municipal workstation should not drag in a stack of libraries that a change-control board has to vet before every reporting cycle.

Before writing any records you need two inputs that do not come from a package: your agency’s submission specification (the field-position table and the code lists it accepts) and a crosswalk from your internal analyte identifiers to the SDWIS/STORET parameter codes the state expects. Those codes are not interchangeable between agencies, and guessing them is the fastest way to a rejected batch. Treat both inputs as versioned artifacts checked in alongside the code, because agencies revise their specifications between reporting years and a silently stale field table is indistinguishable from a bug until the file bounces.

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*"
# Optional: pandas if you are staging results out of a lab-system export
pip install "pandas==2.2.*"

Step-by-Step Implementation

Step 1 — Model one sample-result record with pydantic

A SDWIS result row is small but unforgiving: a nine-character PWSID (two-letter state postal code followed by seven digits), a numeric analyte code drawn from the SDWIS/STORET parameter list, a method code, the result value, its unit of measure, a detection limit, and a censoring flag that marks non-detects. Encoding all of that as a typed model means an invalid PWSID or an unknown unit never reaches the serializer. Non-detects deserve special care: agencies commonly want a censored result reported at one-half the method detection limit, a substitution expressed as $r_{\text{rep}} = \tfrac{1}{2}\,\text{MDL}$, and the model is the right place to make that rule explicit rather than scattering it through the emitter.

from __future__ import annotations
import re
from datetime import date
from decimal import Decimal
from enum import Enum
from pydantic import BaseModel, Field, field_validator, model_validator

PWSID_RE = re.compile(r"^[A-Z]{2}\d{7}$")


class Units(str, Enum):
    MG_L = "MG/L"
    UG_L = "UG/L"
    NTU = "NTU"
    PCI_L = "PCI/L"
    COUNT_100ML = "CFU/100ML"


class SampleResult(BaseModel):
    pwsid: str
    analyte_code: str            # SDWIS/STORET parameter code, e.g. "1040" (nitrate)
    method_code: str             # analytical method, e.g. "353.2"
    sample_date: date
    result_value: Decimal
    units: Units
    detection_limit: Decimal
    less_than: bool = False      # True => non-detect (censored at the MDL)
    lab_sample_id: str = Field(min_length=1, max_length=20)

    @field_validator("pwsid")
    @classmethod
    def _check_pwsid(cls, v: str) -> str:
        if not PWSID_RE.match(v):
            raise ValueError(f"PWSID {v!r} is not 2 letters + 7 digits")
        return v

    @field_validator("analyte_code")
    @classmethod
    def _check_analyte(cls, v: str) -> str:
        if not v.isdigit() or not (4 <= len(v) <= 5):
            raise ValueError(f"analyte_code {v!r} must be a 4-5 digit code")
        return v

    @model_validator(mode="after")
    def _check_result(self) -> "SampleResult":
        if self.detection_limit <= 0:
            raise ValueError("detection_limit must be positive")
        if not self.less_than and self.result_value < 0:
            raise ValueError("a detected result cannot be negative")
        return self

    def reported_value(self) -> Decimal:
        """Non-detects report at one-half the detection limit."""
        if self.less_than:
            return (self.detection_limit / Decimal(2)).quantize(Decimal("0.0001"))
        return self.result_value.quantize(Decimal("0.0001"))

Step 2 — Serialize records to a fixed-width transfer file

Many primacy agencies still accept a positional flat file, where every field occupies a fixed column span and the receiver slices the line by offset rather than by delimiter. That format is brittle by design: a value one column too wide silently corrupts the next field, and there is no delimiter to resynchronize on, so a single overflow can cascade through every column that follows. The serializer therefore pads and truncates each field to its declared width from a single position table, so the layout lives in one place and the emitter can never drift from it. Left-justify text, right-justify and zero-fill numerics, and strip the decimal point out of the result into an implied-decimal integer the way the specification dictates. The implied-decimal convention exists because the format has no room for a variable-length decimal representation; both sides agree in advance on the number of fractional places, and the value travels as a scaled integer that the receiver divides back down on ingest.

from dataclasses import dataclass


@dataclass(frozen=True)
class Field_:
    name: str
    start: int      # 1-based start column
    width: int
    numeric: bool = False


LAYOUT = [
    Field_("pwsid", 1, 9),
    Field_("analyte_code", 10, 5, numeric=True),
    Field_("method_code", 15, 8),
    Field_("sample_date", 23, 8, numeric=True),   # YYYYMMDD
    Field_("result_value", 31, 12, numeric=True),  # implied 4 decimals
    Field_("units", 43, 10),
    Field_("detection_limit", 53, 12, numeric=True),
    Field_("less_than", 65, 1),                    # "<" or space
]
RECORD_WIDTH = 65


def _pack(value: str, f: Field_) -> str:
    if len(value) > f.width:
        raise ValueError(f"field {f.name}={value!r} exceeds width {f.width}")
    return value.zfill(f.width) if f.numeric else value.ljust(f.width)


def serialize_fixed_width(rows: list[SampleResult]) -> str:
    lines: list[str] = []
    for r in rows:
        scaled = int((r.reported_value() * 10000).to_integral_value())
        cells = {
            "pwsid": r.pwsid,
            "analyte_code": r.analyte_code,
            "method_code": r.method_code,
            "sample_date": r.sample_date.strftime("%Y%m%d"),
            "result_value": str(scaled),
            "units": r.units.value,
            "detection_limit": str(int((r.detection_limit * 10000).to_integral_value())),
            "less_than": "<" if r.less_than else " ",
        }
        line = "".join(_pack(cells[f.name], f) for f in LAYOUT)
        assert len(line) == RECORD_WIDTH, f"packed width {len(line)} != {RECORD_WIDTH}"
        lines.append(line)
    return "\n".join(lines) + "\n"

Step 3 — Serialize the same records to SDWIS XML

Newer state interfaces prefer an XML envelope, which is self-describing and far kinder to debug than column arithmetic. The element names below are illustrative — map them onto your agency’s schema — but the structure is representative: a submission header carrying the PWSID and a generation timestamp, then one <Result> element per record with the analyte, method, value, units, and detection limit as child elements. Emitting the reported value through the same reported_value() method guarantees the XML and fixed-width files agree to the last digit, which matters when an agency reconciles a resubmission against an earlier batch.

import xml.etree.ElementTree as ET
from datetime import datetime, timezone


def serialize_xml(rows: list[SampleResult], submission_id: str) -> bytes:
    root = ET.Element("SDWISSubmission", attrib={"id": submission_id})
    header = ET.SubElement(root, "Header")
    ET.SubElement(header, "GeneratedUTC").text = (
        datetime.now(timezone.utc).isoformat(timespec="seconds")
    )
    ET.SubElement(header, "RecordCount").text = str(len(rows))

    results = ET.SubElement(root, "Results")
    for r in rows:
        node = ET.SubElement(results, "Result")
        ET.SubElement(node, "PWSID").text = r.pwsid
        ET.SubElement(node, "AnalyteCode").text = r.analyte_code
        ET.SubElement(node, "MethodCode").text = r.method_code
        ET.SubElement(node, "SampleDate").text = r.sample_date.isoformat()
        ET.SubElement(node, "ResultValue").text = str(r.reported_value())
        ET.SubElement(node, "Units").text = r.units.value
        ET.SubElement(node, "DetectionLimit").text = str(r.detection_limit)
        ET.SubElement(node, "LessThanFlag").text = "Y" if r.less_than else "N"

    ET.indent(root)
    return ET.tostring(root, encoding="utf-8", xml_declaration=True)

Step 4 — Validate the batch, then emit a manifest and checksum

Never let a row that failed construction slip silently out of the batch. Run the whole input through the model first, collect every rejection with its row index, and refuse to serialize if anything failed — a partial file is worse than no file, because the agency counts what arrives against what you claim you sent, and a quietly dropped result reads on their side as a monitoring gap you never reported. Surfacing every failure at once, rather than aborting on the first, also spares operators a slow one-error-at-a-time correction loop when a lab export drifts out of spec. Once a clean file exists, compute a SHA-256 digest over its exact bytes and write a small manifest recording the record count and that digest. The receiver recomputes the digest and compares; a mismatch means the transfer was truncated or altered, and the whole submission is repudiated before it touches their database. This manifest is also the first artifact an auditor asks for, which is why it should flow straight into the patterns described in Audit Trail & Data Lineage Storage Patterns.

import hashlib
import json
from pydantic import ValidationError


def build_batch(raw_rows: list[dict]) -> list[SampleResult]:
    valid: list[SampleResult] = []
    errors: list[str] = []
    for i, raw in enumerate(raw_rows):
        try:
            valid.append(SampleResult(**raw))
        except ValidationError as exc:
            errors.append(f"row {i}: {exc.errors()[0]['msg']}")
    if errors:
        raise ValueError("batch rejected:\n" + "\n".join(errors))
    return valid


def write_submission(rows: list[SampleResult], path: str, submission_id: str) -> dict:
    payload = serialize_xml(rows, submission_id)
    with open(path, "wb") as fh:
        fh.write(payload)
    digest = hashlib.sha256(payload).hexdigest()
    manifest = {
        "submission_id": submission_id,
        "file": path,
        "record_count": len(rows),
        "sha256": digest,
        "byte_length": len(payload),
    }
    with open(path + ".manifest.json", "w") as fh:
        json.dump(manifest, fh, indent=2)
    return manifest

Configuration Reference

The positional layout below is the contract for the fixed-width variant; every column span, code list, and flag belongs in versioned configuration, never in an inline literal. When the agency reissues its specification, a change should mean editing this table and rerunning the tests, not hunting for magic numbers scattered through the emitter. The analyte and method codes shown are illustrative SDWIS/STORET values — confirm each against your agency’s current parameter list, and route any code you translate through Violation Code Classification so the same crosswalk governs both submission and alerting.

Verification & Testing

Confirm the serializers agree and that the layout never drifts by width. The test below round-trips a non-detect and asserts both that the fixed-width line is exactly one record wide and that the censored value was substituted at half the detection limit.

from decimal import Decimal
from datetime import date


def _row(**kw):
    base = dict(
        pwsid="TX1234567", analyte_code="01040", method_code="353.2",
        sample_date=date(2026, 5, 1), result_value=Decimal("0"),
        units="MG/L", detection_limit=Decimal("0.05"),
        less_than=True, lab_sample_id="LAB-001",
    )
    base.update(kw)
    return SampleResult(**base)


def test_nondetect_reports_half_mdl():
    assert _row().reported_value() == Decimal("0.0250")


def test_fixed_width_line_length():
    line = serialize_fixed_width([_row()])
    assert len(line.rstrip("\n")) == RECORD_WIDTH


def test_bad_pwsid_is_rejected():
    import pytest
    with pytest.raises(ValueError):
        _row(pwsid="TX12345")   # only 5 digits


def test_manifest_digest_is_stable(tmp_path):
    m = write_submission([_row()], str(tmp_path / "b.xml"), "SUB-1")
    assert len(m["sha256"]) == 64 and m["record_count"] == 1

Acceptance criteria before a batch goes to the state:

• Every PWSID matches the two-letter, seven-digit pattern; malformed identifiers reject the whole batch.
• Analyte and method codes are resolved from the agency’s current parameter list, not hard-coded guesses.
• Non-detects report at one-half the detection limit and carry the < / Y censoring flag.
• Fixed-width lines are exactly RECORD_WIDTH characters; no field overflows its span.
• The fixed-width and XML serializers emit identical reported values for the same records.
• A SHA-256 manifest accompanies the file and its recomputed digest matches on the receiving side.
• The rejection report lists every failed row by index before any file is written.

Troubleshooting & Gotchas

• The whole batch is rejected for one bad row. That is the intended behaviour — build_batch refuses to serialize a partial file. Read the rejection report, fix the source records, and resubmit the complete batch rather than shipping the survivors and reconciling later.
• Column drift corrupts every field after the offending one. A fixed-width value that is one character too wide shifts the rest of the line. The _pack guard and the per-line width assertion catch it at emit time; never bypass them by concatenating raw strings.
• Analyte codes look right but the agency rejects them. SDWIS/STORET codes are not universal across primacy agencies, and a code valid in one state maps to nothing in another. Keep the crosswalk versioned and treat a rejected code as a mapping bug, not a data bug.
• Implied-decimal results are off by orders of magnitude. The fixed-width format strips the decimal point and assumes a fixed number of implied places. If the receiver reads 0.0250 as 250, your scale factor and their layout disagree — reconcile the implied-decimal count in the specification before resubmitting.
• Resubmissions fail checksum reconciliation. Regenerating the same records through both serializers must yield byte-identical files given the same timestamp; a floating timestamp in the XML header changes the digest. Pin the generation time from the batch metadata when you need a reproducible manifest.

Related

• DMR Generation & SDWIS Submission Workflows — the parent section and the shared validated-record contract
• Generating NPDES Discharge Monitoring Reports in Python — the same records packaged for a federal discharge report
• Building Consumer Confidence Reports from Compliance Data — the public-facing report drawn from the same source data
• Violation Code Classification — the crosswalk that keeps analyte and violation codes consistent
• Audit Trail & Data Lineage Storage Patterns — where the manifest and checksum are preserved for audit