Translating EPA Violation Codes to Internal Alerts

Operationalizing Safe Drinking Water Act (SDWA) compliance requires the deterministic translation of federal violation codes into actionable internal alerts. Water utility operators, environmental compliance teams, and municipal developers routinely face a gap between regulatory reporting cycles and real-time SCADA telemetry: a Maximum Contaminant Level (MCL) is exceeded at a treatment train, but the fact does not surface as an operator-facing alert until a lab report or a monthly reconciliation catches it. This page shows how to build the translation layer that closes that gap — a small, stateful service that maps EPA violation codes (51 for MCL violations, 52 for MRDL violations, 55 for monitoring-and-reporting failures, the 60 series for public-notification lapses) directly onto internal escalation matrices. It sits one level below the Violation Code Classification subsystem, where federal identifiers are normalized into machine-readable schemas, and it consumes threshold definitions resolved from the SDWA MCL Reference Mapping. The translation must be deterministic, auditable, and resilient to SCADA data anomalies — deliberately stateful where it must be, because a small, well-bounded state cache is what prevents alert flapping during transient sensor noise.

Prerequisites & Environment Setup

This translation engine is typically deployed as a scheduled Python worker or a containerized microservice wired into the utility’s data historian. It reads normalized telemetry — ideally after it has passed through the Time-Series Alignment Strategies module — evaluates it against a version-controlled taxonomy, and routes any resulting violation to operations.

You need Python 3.10 or newer (the code uses datetime timezone-aware comparisons and dataclasses), plus requests for webhook dispatch. Everything else is standard library, which keeps the compliance-critical path small and auditable.

python3 -m venv .venv
source .venv/bin/activate
pip install "requests>=2.31,<3"        # only external dependency on the hot path
pip install "pytest>=8.0"              # for the verification suite below

The service also needs read access to the historian (REST or OPC-UA — see OPC-UA Data Extraction for the extraction side), write access to an append-only audit store, and outbound network reach to your alert channels. Run it under a service account with least privilege; it should never need write access to the historian.

Step-by-Step Implementation

The pipeline has four stages: normalize telemetry, load the EPA taxonomy, apply direction-aware state-machine logic, and route with a deterministic fallback chain. Each stage is idempotent so the worker can be restarted or re-run over a replayed window without producing duplicate alerts.

1. Ingest, Normalize, and Stage Telemetry

Raw historian payloads must be normalized before evaluation. Polling over REST or OPC-UA requires strict timestamp alignment to UTC, unit standardization, and drift detection. Persist every raw payload to an immutable staging table to preserve chain of custody for compliance audits. Note that the normalized timestamp remains timezone-aware (UTC), which keeps all downstream window and cooldown comparisons unambiguous.

import datetime
import logging
from dataclasses import dataclass
from typing import Optional

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.FileHandler("sdwa_compliance_audit.log")]
)

@dataclass(frozen=True)
class TelemetryPoint:
    sensor_id: str
    parameter: str
    raw_value: float
    unit: str
    timestamp_utc: datetime.datetime

def normalize_telemetry(raw_payload: dict) -> Optional[TelemetryPoint]:
    """Convert historian payload to standardized TelemetryPoint."""
    try:
        ts = datetime.datetime.fromisoformat(raw_payload["timestamp"].replace("Z", "+00:00"))
        # Normalize to a timezone-aware UTC datetime so every downstream
        # comparison (cooldowns, evaluation windows) is unambiguous.
        ts_utc = ts.astimezone(datetime.timezone.utc)

        # Unit conversion: mg/L -> µg/L (multiply by 1000)
        value = float(raw_payload["value"])
        unit = raw_payload["unit"].lower()
        if unit == "mg/l":
            value *= 1000.0
            unit = "µg/l"

        return TelemetryPoint(
            sensor_id=raw_payload["sensor_id"],
            parameter=raw_payload["parameter"],
            raw_value=value,
            unit=unit,
            timestamp_utc=ts_utc
        )
    except Exception as e:
        logging.error(f"Normalization failed: {e}")
        return None

2. Load the EPA Violation Code Taxonomy

Maintain a version-controlled JSON mapping table that records EPA code definitions, the associated SDWA rules (for example, the Lead and Copper Rule Revisions, Stage 2 DBPR, and the Revised Total Coliform Rule), and the internal alert severity for each code. Cross-reference this table against the parent Violation Code Classification subsystem to keep mappings aligned with regulation and to catch drift introduced by primacy updates. The loader below falls back to an embedded copy whenever the on-disk file is missing or fails to parse, so a corrupt taxonomy never halts evaluation.

import json
import logging
from pathlib import Path

# Representative subset — extend this table to cover your full parameter inventory.
# Note: turbidity violations typically generate monitoring/reporting codes (55 series)
# rather than MCL codes, because turbidity at the distribution tap has no numeric MCL
# under the Surface Water Treatment Rule; the trigger is a treatment technique deviation.
EPA_TAXONOMY = {
    "51": {"name": "MCL Exceedance", "severity": "CRITICAL", "rule": "Various", "direction": "above"},
    "52": {"name": "MRDL Exceedance", "severity": "HIGH", "rule": "Stage 2 DBPR", "direction": "above"},
    "55": {"name": "Monitoring/Reporting Failure", "severity": "MEDIUM", "rule": "RTCR", "direction": "gap"},
    "60": {"name": "Public Notification Lapse", "severity": "HIGH", "rule": "SDWA §1414", "direction": "gap"}
}

def load_taxonomy(path: Path = Path("epa_codes_v2024.json")) -> dict:
    """Load the on-disk taxonomy, falling back to the embedded copy on any
    read or parse failure so evaluation never aborts on a malformed file."""
    if path.exists():
        try:
            with open(path, "r", encoding="utf-8") as f:
                return json.load(f)
        except (OSError, json.JSONDecodeError) as e:
            logging.error("Failed to load taxonomy from %s: %s", path, e)
    logging.warning("Falling back to embedded taxonomy. Verify against primacy agency updates.")
    return EPA_TAXONOMY

3. Apply Direction-Aware Translation & State-Machine Logic

Each parameter maps to an EPA code, a threshold, and a direction: most MCL violations fire when a value is above a maximum, but some parameters are minimum requirements. Free chlorine residual, for example, is a minimum — the SDWA treatment technique requires a detectable residual at the distribution system entry point, so an alert fires when the value is below the operational floor, not above it. The MRDL (code 52) applies to the total disinfectant residual in the distribution system and is evaluated as a running annual average against an upper limit. Confusing these directions produces either missed alerts or constant false positives, which is why the direction-aware MCL Exceedance Logic Implementation treats it as a first-class field rather than an assumption.

A deterministic state machine suppresses duplicate alerts during transient sensor noise: a parameter must breach its threshold on consecutive evaluations before it escalates from WARNING to VIOLATION, and a per-rule cooldown window prevents re-firing while a violation persists. The severity attached to each fired code feeds directly into the Severity Scoring Models, so escalation stays consistent with the rest of the rule engine.

Alert state machine with flapping suppression A per-sensor state machine with three states — NOMINAL, WARNING and VIOLATION. From an initial entry point the sensor starts NOMINAL. The first threshold breach moves it to WARNING but fires no alert; a breach on the next evaluation escalates WARNING to VIOLATION and routes the alert. While the value stays breached a self-loop on VIOLATION suppresses re-firing until the cooldown window elapses. From either WARNING or VIOLATION, a value that recovers within the threshold returns the sensor to NOMINAL. NOMINAL WARNING VIOLATION within threshold first breach · no alert alert routed value breaches threshold (first) breach on next evaluation value recovers within threshold value recovers within threshold within cooldown · suppress re-fire
Alert state machine: a first breach only arms WARNING; a sustained breach escalates to VIOLATION and routes once, while the cooldown self-loop suppresses re-firing until the value recovers.
import datetime
from enum import Enum
from typing import Dict, Optional, Tuple

# TelemetryPoint is defined in the ingestion module (Step 1).
from telemetry import TelemetryPoint

class AlertState(Enum):
    NOMINAL = "nominal"
    WARNING = "warning"
    VIOLATION = "violation"

# Parameter-to-rule mapping.
# "direction" controls which side of the threshold constitutes a breach:
#   "above"  – MCL/MRDL upper limit; alert when value EXCEEDS threshold
#   "below"  – minimum requirement; alert when value FALLS BELOW threshold
PARAMETER_RULES = {
    "free_chlorine_residual_mg_l": {
        "epa_code": "55",          # Monitoring/TT failure when residual is absent
        "threshold_mg_l": 0.2,     # Minimum detectable residual at distribution entry
        "direction": "below",
        "description": "Distribution system free-chlorine residual below minimum"
    },
    "nitrate_mg_l": {
        "epa_code": "51",          # MCL = 10 mg/L as N (acute, single confirmed sample)
        "threshold_mg_l": 10.0,
        "direction": "above",
        "description": "Nitrate MCL exceedance"
    },
    "arsenic_ug_l": {
        "epa_code": "51",          # MCL = 10 µg/L
        "threshold_ug_l": 10.0,
        "direction": "above",
        "description": "Arsenic MCL exceedance"
    },
}

class ComplianceEngine:
    def __init__(self, taxonomy: dict):
        self.taxonomy = taxonomy
        # In-memory state tracker: sensor_id -> (state, last_triggered_utc)
        self._state_cache: Dict[str, Tuple[AlertState, Optional[datetime.datetime]]] = {}

    def evaluate(self, point: TelemetryPoint) -> Optional[dict]:
        """Deterministic threshold evaluation with flapping prevention."""
        state, last_triggered = self._state_cache.get(point.sensor_id, (AlertState.NOMINAL, None))

        rule = PARAMETER_RULES.get(point.parameter)
        if rule is None:
            return None  # Parameter not in compliance inventory

        code = rule["epa_code"]
        threshold = rule.get("threshold_mg_l") or rule.get("threshold_ug_l", 0.0)
        direction = rule["direction"]
        taxonomy_entry = self.taxonomy.get(code, {})

        # Cooldown window derived from the rule (defaulting to 4 hours).
        cooldown_seconds = float(taxonomy_entry.get("threshold_hours", 4.0)) * 3600.0

        # Direction-aware breach detection
        if direction == "above":
            breached = point.raw_value > threshold
        else:  # "below"
            breached = point.raw_value < threshold

        if breached:
            if state == AlertState.NOMINAL:
                state = AlertState.WARNING
                self._state_cache[point.sensor_id] = (state, None)
                return None  # First breach: wait for a sustained exceedance.

            # Sustained breach: suppress duplicates inside the cooldown window.
            if last_triggered is not None and \
                    (point.timestamp_utc - last_triggered).total_seconds() < cooldown_seconds:
                return None  # Within cooldown window

            state = AlertState.VIOLATION
            self._state_cache[point.sensor_id] = (state, point.timestamp_utc)

            return {
                "epa_code": code,
                "severity": taxonomy_entry.get("severity", "UNKNOWN"),
                "rule": rule["description"],
                "sensor_id": point.sensor_id,
                "value": point.raw_value,
                "unit": point.unit,
                "threshold": threshold,
                "direction": direction,
                "timestamp_utc": point.timestamp_utc.isoformat(),
                "state": state.value
            }
        else:
            # Reset to nominal
            self._state_cache[point.sensor_id] = (AlertState.NOMINAL, None)
            return None

4. Route with a Deterministic Fallback Chain

Alert routing must survive network partitions, API rate limits, and gateway failures. Implement a cascading fallback chain — primary webhook, then secondary SMS gateway, then tertiary email, and finally a local immutable log. Each step is idempotent and retry-aware, and the chain applies exponential backoff between channels (skipping the wait after the last one, which would only delay the local-log fallback).

Cascading alert routing with exponential backoff A generated alert enters a cascading fallback chain. It first attempts the primary webhook; on success it reaches the Delivered state. On failure it backs off and falls through to the secondary SMS gateway, then to the tertiary email channel, each of which also short-circuits to Delivered on success. If every channel fails, the alert is written to a local immutable audit log so it is never lost. The delay between channels grows as two to the power of the channel index. Alert generated Webhook SMS gateway Email Local immutable audit log Delivered primary secondary tertiary chain exhausted · never lost routed & sealed fail · backoff 2⁰ s fail · backoff 2¹ s fail · backoff 2² s success
Cascading routing: each channel short-circuits to Delivered on success; on failure the chain backs off exponentially and falls through, guaranteeing capture in the local audit log.
import logging
import time
import requests
from typing import Callable, List

def route_alert(alert: dict, fallback_chain: List[Callable]) -> bool:
    """Execute alert routing with an exponential-backoff fallback chain."""
    last_index = len(fallback_chain) - 1
    for i, dispatcher in enumerate(fallback_chain):
        try:
            if dispatcher(alert):
                logging.info("Alert routed successfully via channel %d: %s", i, dispatcher.__name__)
                return True
            logging.warning("Channel %d (%s) reported failure. Attempting fallback.", i, dispatcher.__name__)
        except Exception as e:
            logging.warning("Channel %d (%s) raised: %s. Attempting fallback.", i, dispatcher.__name__, e)
        # Back off before the next channel; skip the wait after the final one.
        if i < last_index:
            time.sleep(2 ** i)  # Exponential backoff
    logging.critical("ALL ROUTING CHANNELS FAILED. ALERT WRITTEN TO LOCAL AUDIT LOG: %s", alert)
    return False

# Example dispatchers
def webhook_dispatch(alert: dict) -> bool:
    resp = requests.post("https://ops.internal/api/v1/alerts", json=alert, timeout=5)
    resp.raise_for_status()
    return True

def sms_dispatch(alert: dict) -> bool:
    # Placeholder for Twilio/ClickSend integration
    logging.info(f"SMS fallback triggered for {alert['sensor_id']}")
    return True

def email_dispatch(alert: dict) -> bool:
    # Placeholder for SMTP integration
    logging.info(f"Email fallback triggered for {alert['sensor_id']}")
    return True

FALLBACK_PIPELINE = [webhook_dispatch, sms_dispatch, email_dispatch]

Configuration Reference

Two tables govern the behaviour of the engine: the EPA code taxonomy (severity and default routing per code family) and the parameter-rule table (threshold and direction per monitored point). Keep both in version control and diff them on every primacy update.

EPA violation code families

Code Name SDWA rule Internal severity Direction Typical response
51 MCL Exceedance Various (40 CFR 141) CRITICAL above Confirmation sample + public notification workflow
52 MRDL Exceedance Stage 2 DBPR HIGH above Running-annual-average review, disinfection tuning
55 Monitoring/Reporting Failure RTCR MEDIUM gap Technician dispatch, historian gap-fill
60 Public Notification Lapse SDWA §1414 HIGH gap Immediate notice issuance, primacy agency contact

Parameter-rule fields

Field Type Example Purpose
epa_code string "51" Links the parameter to a taxonomy entry
threshold_mg_l / threshold_ug_l float 10.0 Numeric limit in the parameter’s native unit
direction enum "above" | "below" Which side of the threshold constitutes a breach
description string "Nitrate MCL exceedance" Human-readable alert label
threshold_hours (taxonomy) float 4.0 Cooldown window sizing per rule family

For the acute nitrate case, the MCL is expressed as a single-sample limit, so the effective evaluation reduces to a direct comparison:

Cmeasured>MCLNO3=10 mg/L as NC_{\text{measured}} > \text{MCL}_{\text{NO}_3} = 10\ \text{mg/L as N}

For running-annual-average parameters such as the MRDL, the comparison instead uses the mean of the trailing four quarters, which the Monitoring Frequency Scheduling module supplies as an aligned window:

C‾RAA=1n∑i=1nCi>MRDL\overline{C}_{\text{RAA}} = \frac{1}{n}\sum_{i=1}^{n} C_i > \text{MRDL}

Verification & Testing

Confirm the engine’s two most error-prone behaviours before deploying: direction-aware breach detection and flapping suppression. The test below drives a below-direction parameter (chlorine residual) and asserts that a single dip does not fire, but a sustained dip does.

import datetime
import pytest
from telemetry import TelemetryPoint
from engine import ComplianceEngine, AlertState

TAXONOMY = {"55": {"severity": "MEDIUM", "threshold_hours": 4.0}}

def _point(value, minutes):
    base = datetime.datetime(2026, 7, 3, 12, 0, tzinfo=datetime.timezone.utc)
    return TelemetryPoint(
        sensor_id="CL2-ENTRY-01",
        parameter="free_chlorine_residual_mg_l",
        raw_value=value,
        unit="mg/l",
        timestamp_utc=base + datetime.timedelta(minutes=minutes),
    )

def test_single_dip_does_not_fire():
    engine = ComplianceEngine(TAXONOMY)
    assert engine.evaluate(_point(0.1, 0)) is None          # WARNING, not VIOLATION
    assert engine._state_cache["CL2-ENTRY-01"][0] is AlertState.WARNING

def test_sustained_dip_fires_once():
    engine = ComplianceEngine(TAXONOMY)
    engine.evaluate(_point(0.1, 0))                          # WARNING
    alert = engine.evaluate(_point(0.1, 15))                 # VIOLATION
    assert alert is not None
    assert alert["epa_code"] == "55"
    assert alert["direction"] == "below"
    # Within the 4h cooldown, a further breach is suppressed.
    assert engine.evaluate(_point(0.1, 30)) is None

def test_recovery_resets_state():
    engine = ComplianceEngine(TAXONOMY)
    engine.evaluate(_point(0.1, 0))
    engine.evaluate(_point(0.5, 15))                         # back above the floor
    assert engine._state_cache["CL2-ENTRY-01"][0] is AlertState.NOMINAL

Acceptance criteria before this service is allowed to route to production channels:

Troubleshooting & Gotchas

  1. Inverted direction (silent miss or alert storm). If a below-direction parameter like chlorine residual is configured as above, a dangerous low-residual event never fires and a healthy reading alarms constantly. Diagnose by asserting direction is present for every rule in a startup check; never let it default.
  2. Timezone-naive timestamps. A naive datetime subtracted from a timezone-aware one raises TypeError inside the cooldown comparison, and the alert is swallowed by the routing except. Enforce astimezone(timezone.utc) in normalize_telemetry and reject payloads without an offset.
  3. Unit mismatch against the threshold. Arsenic is regulated in µg/L but many historians publish mg/L. If normalization does not run, a real 10 µg/L reading arrives as 0.01 and never crosses the 10.0 threshold. Verify the unit field on the TelemetryPoint matches the threshold_* key in the rule.
  4. Cooldown too short for the rule’s averaging shape. A running-annual-average parameter re-evaluated every 15 minutes will re-fire once the cooldown lapses even though nothing changed. Size threshold_hours to the rule’s reporting cadence (often 24 h or longer), sourced from the taxonomy.
  5. Stale taxonomy after a primacy update. Codes and thresholds shift with rule revisions (LCR revisions, DBPR updates). Schedule a weekly job that diffs your local taxonomy against the EPA SDWA compliance portal and opens a CI/CD update when the two diverge, rather than trusting the embedded fallback indefinitely.

By decoupling telemetry ingestion from regulatory translation, utilities retire manual spreadsheet reconciliation and cut compliance latency from days to seconds. The cascading routing layer ensures that even during infrastructure degradation, violation codes are captured, routed, and logged without data loss.