How to Map EPA MCLs to Relational Database Schemas for Automated Compliance

Translating Safe Drinking Water Act (SDWA) Maximum Contaminant Levels (MCLs) into a production relational database is the specific task of turning a moving regulatory target into stable, queryable rows so that every SCADA reading can be judged against the exact limit in force when it was sampled. Hardcoding thresholds into telemetry tables is the anti-pattern that causes schema drift, breaks historical audits, and turns every rulemaking amendment into a destructive migration. This page is for the municipal Python developers and environmental compliance teams who own that translation layer; it sits under SDWA MCL Reference Mapping and implements the concrete database design that the mapping strategy there describes. The goal is a normalized model that isolates regulatory metadata, temporal validity windows, and sensor readings into distinct structures so compliance calculations are deterministic and audit-ready.

Prerequisites & Environment Setup

This implementation targets PostgreSQL 14+ (for native tsrange exclusion constraints) and Python 3.11+. The compliance resolver uses pandas for vectorized evaluation and SQLAlchemy with the psycopg2 driver for range-based temporal joins. Pin versions so that the QA reconciliation job runs against a reproducible stack:

python3 -m venv .venv
source .venv/bin/activate
pip install "pandas==2.2.2" "SQLAlchemy==2.0.30" "psycopg2-binary==2.9.9" "pytest==8.2.0"

You also need read access to EPA’s official parameter codes so the dimension keys never rely on hand-transcribed values. The regulatory baselines validated below come from the EPA SDWA regulations. Confirm the target schema has a role permitted to create EXCLUDE constraints, which require the btree_gist extension:

CREATE EXTENSION IF NOT EXISTS btree_gist;

Step-by-Step Implementation

MCLs are dynamic: they vary by contaminant, averaging period (single sample, running annual average, or 90th percentile), monitoring frequency, and regulatory revision date. The mapping follows a deterministic pipeline that isolates schema design, identifier standardization, temporal validity, unit conversion, and evaluation. The full model aligns with the shared lineage rules in the Core Architecture & SDWA Compliance Taxonomy, so threshold updates trigger dimension inserts rather than rewrites.

1. Model the dimensional schema

A production model splits regulatory metadata from high-frequency readings across five tables:

• dim_contaminant: EPA regulatory IDs, CAS numbers, contaminant names, units of measure, and grouping flags.
• dim_mcl_threshold: MCL values, averaging periods, effective_date, expiration_date, and an is_current flag (Slowly Changing Dimension Type 2).
• dim_monitoring_point: physical sampling locations, SCADA tag IDs, compliance reporting zones, and jurisdictional boundaries.
• fact_scada_telemetry: validated concentration readings with UTC timestamps, quality flags, and foreign keys to monitoring points.
• fact_compliance_events: the derived table storing calculated compliance status, violation codes, aggregation windows, and audit trails.

2. Standardize identifiers

Use EPA’s Contaminant_ID as the primary foreign key across every table. Do not key on CAS numbers alone: grouped parameters such as TTHM and HAA5 have no single CASRN value, so a CAS-only join silently drops them.

3. Preserve regulatory history with SCD Type 2

Store MCLs exclusively in dim_mcl_threshold and never overwrite a value. When the EPA revises a rule, expire the old row and insert a new one so historical compliance decisions remain reproducible. A tsrange exclusion constraint guarantees that no two rows for the same contaminant ever claim overlapping validity:

CREATE TABLE dim_mcl_threshold (
    threshold_id     BIGSERIAL PRIMARY KEY,
    contaminant_id   TEXT NOT NULL REFERENCES dim_contaminant(contaminant_id),
    mcl_value        NUMERIC(12, 6),          -- NULL = advisory-only, no enforceable MCL
    averaging_period TEXT NOT NULL,
    unit_of_measure  TEXT NOT NULL,
    validity         TSRANGE NOT NULL,        -- [effective_date, expiration_date)
    is_current       BOOLEAN NOT NULL DEFAULT TRUE,
    EXCLUDE USING gist (
        contaminant_id WITH =,
        validity       WITH &&
    )
);

4. Bind the averaging period, not just the number

An MCL is a number plus an averaging rule. The reference row declares which rule applies and the evaluator dispatches on it. For running-annual-average contaminants such as total trihalomethanes, the engine must reproduce the quarterly mean exactly:

where $\bar{C}_{i}$ is the arithmetic mean of the valid samples in quarter $i$ and $q$ is the current quarter. Whether each $\bar{C}_{i}$ is even complete depends on the statutory cadence owned by the Monitoring Frequency Scheduling module, so the reference table and the scheduler share the same contaminant keys.

5. Normalize units before evaluation

Convert every reading to the EPA-mandated unit of measure (mg/L, µg/L, NTU) before comparison. Store conversion factors in a dedicated dim_unit_conversion table rather than embedding them in code, and enforce a temporal join so each reading resolves the threshold active at its own timestamp:

import logging
from datetime import datetime, timezone

import pandas as pd
from sqlalchemy import text

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def resolve_compliance_status(engine, telemetry_batch_id: str):
    """
    Join SCADA telemetry with the MCL threshold active at each reading's
    timestamp, calculate compliance, and route anomalies to fallback handlers.
    """
    query = """
    SELECT
        t.timestamp_utc,
        t.monitoring_point_id,
        t.contaminant_id,
        t.concentration_raw,
        t.unit_of_measure,
        m.mcl_value,
        m.averaging_period,
        m.validity
    FROM fact_scada_telemetry t
    JOIN dim_mcl_threshold m
      ON t.contaminant_id = m.contaminant_id
     AND m.validity @> t.timestamp_utc
    WHERE t.batch_id = :batch_id
    """
    try:
        with engine.connect() as conn:
            df = pd.read_sql(text(query), conn, params={"batch_id": telemetry_batch_id})
    except Exception as exc:
        logger.error("Threshold resolution failed for batch %s: %s", telemetry_batch_id, exc)
        return route_fallback(telemetry_batch_id, error_type="DB_JOIN_FAILURE")

    if df.empty:
        logger.warning("No active thresholds matched batch. Routing to manual review.")
        return route_fallback(telemetry_batch_id, error_type="NO_ACTIVE_MCL")

    if df["unit_of_measure"].nunique() > 1:
        logger.warning("Unit mismatch detected. Applying standard conversion matrix.")
        df = _apply_unit_conversion(df)

    # Readings with no enforceable MCL are flagged for explicit manual review
    # rather than silently passing.
    df["is_compliant"] = df.apply(
        lambda row: (row["concentration_raw"] <= row["mcl_value"])
        if pd.notnull(row["mcl_value"]) else False,
        axis=1,
    )
    df["compliance_timestamp"] = datetime.now(timezone.utc)
    df["violation_code"] = df["is_compliant"].map({True: "COMPLIANT", False: "EXCEEDANCE"})

    try:
        df.to_sql("fact_compliance_events", engine, if_exists="append", index=False)
        logger.info("Compliance events persisted for batch %s", telemetry_batch_id)
    except Exception as exc:
        logger.error("Failed to write compliance events: %s", exc)
        return route_fallback(telemetry_batch_id, error_type="WRITE_FAILURE")

    return df


def route_fallback(batch_id: str, error_type: str):
    """Immediate operational resolution for pipeline failures."""
    logger.critical("Routing batch %s to quarantine due to %s", batch_id, error_type)
    # Production: INSERT INTO fact_compliance_quarantine and trigger an
    # on-call webhook for the SCADA admin or compliance officer.
    return {"batch_id": batch_id, "error_type": error_type, "status": "QUARANTINED"}


def _apply_unit_conversion(df: pd.DataFrame) -> pd.DataFrame:
    """Join dim_unit_conversion and rescale concentration_raw. Stubbed here."""
    return df

The structured violation_code this resolver emits is the exact input consumed downstream by Translating EPA Violation Codes to Internal Alerts, which maps each code onto operator-facing notifications.

Configuration Reference

The tables below define the core parameters an implementer sets once and the enumerated codes the resolver emits. Keep the averaging-period and status enums in a shared module so the database CHECK constraints and the Python evaluator never drift apart.

Verification & Testing

Prove the temporal join resolves the correct threshold across a rule revision before wiring the resolver into production. The test below seeds two non-overlapping validity windows and asserts a reading is judged against the window active on its own date:

from datetime import datetime, timezone

import pandas as pd
from sqlalchemy import create_engine, text


def test_temporal_join_selects_active_threshold(tmp_path):
    engine = create_engine(f"sqlite:///{tmp_path}/compliance.db")
    with engine.begin() as conn:
        conn.execute(text(
            "CREATE TABLE dim_mcl_threshold "
            "(contaminant_id TEXT, mcl_value REAL, effective TEXT, expiration TEXT)"
        ))
        conn.execute(text(
            "INSERT INTO dim_mcl_threshold VALUES "
            "('1005', 0.010, '2006-01-01', '2024-01-01'), "  # older arsenic MCL era
            "('1005', 0.010, '2024-01-01', '2999-01-01')"
        ))

    reading_ts = datetime(2024, 6, 1, tzinfo=timezone.utc).isoformat()
    query = text(
        "SELECT mcl_value FROM dim_mcl_threshold "
        "WHERE contaminant_id = '1005' AND :ts >= effective AND :ts < expiration"
    )
    with engine.connect() as conn:
        rows = pd.read_sql(query, conn, params={"ts": reading_ts})

    assert len(rows) == 1              # exactly one window matches
    assert rows.iloc[0]["mcl_value"] == 0.010

Acceptance criteria for the mapping layer:

• Every dim_contaminant row carries a valid EPA Contaminant_ID, not a bare CAS number.
• The EXCLUDE/tsrange constraint rejects any overlapping validity window on insert.
• A reading whose timestamp precedes all thresholds routes to NO_ACTIVE_MCL, never a silent pass.
• Advisory-only contaminants (mcl_value IS NULL) are flagged for review, not marked compliant.
• Threshold updates use idempotent upserts (ON CONFLICT DO UPDATE) so pipeline retries never duplicate events.
• A nightly job reconciles fact_scada_telemetry row counts against fact_compliance_events.

Troubleshooting & Gotchas

• Threshold expiration gaps. When a new EPA rule supersedes an older one, a temporal gap can open between the old expiration_date and the new effective_date, and readings in that window match no row. Diagnosis: NO_ACTIVE_MCL spikes for a single contaminant on a known revision date. Fix: default to the most recent valid threshold and flag the record REGULATORY_GAP_PENDING_REVIEW rather than dropping it.
• SCADA tag drift. Telemetry tags remap to different physical locations after PLC upgrades or sensor swaps, so a valid reading joins to the wrong compliance zone. Fix: keep a dim_tag_mapping_history table with its own validity windows and route unmatched tags to fact_orphan_readings with a GIS/SCADA alert. The same UTC-alignment discipline in Aligning Irregular SCADA Timestamps to UTC prevents drift-induced join misses.
• Missing MCL values. Some contaminants carry a health advisory but no enforceable MCL. Comparing against a NULL limit silently passes every reading. Fix: treat mcl_value IS NULL as a skip-and-log path (ADVISORY_ONLY_PARAMETER), which the code above does by defaulting is_compliant to False and routing for review.
• Local timezone leakage. Storing effective_date as a naive local date makes a BETWEEN join off by hours near midnight and by a full hour across DST. Fix: store all bounds in UTC and use half-open tsrange (@>), never inclusive BETWEEN on both ends, so a boundary timestamp resolves to exactly one window.
• Overlapping validity windows. Without the exclusion constraint, a bad backfill can leave two active rows and the join fans out, doubling fact_compliance_events. Fix: enforce the EXCLUDE USING gist constraint at the schema level so the bad insert fails loudly instead of corrupting the audit trail. Downstream, the real-time MCL exceedance logic assumes exactly one threshold per reading.

By decoupling regulatory metadata from high-frequency telemetry, utilities gain deterministic compliance calculations, audit-ready reporting, and pipelines that absorb EPA rulemaking through new dimension rows rather than schema refactoring.

Related

• SDWA MCL Reference Mapping — parent overview of the governed MCL translation layer
• Automating Monthly vs Quarterly Monitoring Schedules — cadence that completes each averaging window
• Translating EPA Violation Codes to Internal Alerts — consumes the violation_code emitted here
• Python Logic for Detecting MCL Exceedances in Real Time — evaluation engine that reads these thresholds
• Core Architecture & SDWA Compliance Taxonomy — shared vocabulary and data-lineage rules