Automating Tier 1 Public Notification Triggers

A Tier 1 public notice is the regulatory equivalent of a fire alarm: it covers the acute conditions where drinking water poses an immediate health risk, and the utility has only twenty-four hours from the moment the condition is known to reach the affected population. This page implements the detection half of that obligation inside the parent Public Notification Workflows for SDWA Violations section — the code that watches the stream of violations and validated readings, recognizes an acute condition the instant it appears, computes the hard deadline, and emits a single well-formed event to the dispatcher. Under 40 CFR Part 141 Subpart Q, the acute tier is reserved for conditions such as an E. coli or fecal-coliform maximum contaminant level (MCL) violation, a nitrate or nitrite MCL exceedance, a waterborne disease outbreak, and situations where the state or the utility declares a boil-water or loss-of-disinfection emergency. Getting the predicate right matters more than almost anything else in the Violation Detection & Rule Engine Logic pipeline, because a missed acute trigger is a reportable failure in its own right and a false one erodes the public’s trust in every future notice.

Prerequisites & Environment Setup

The trigger engine is intentionally small and dependency-light so it can run as a step inside the rule engine or as its own service. It targets Python 3.10+ and uses pydantic for the input record contract and a lightweight enum-driven predicate. No network client is required here — the input is a normalized violation or reading record that has already passed through MCL Exceedance Logic Implementation, so this stage never re-parses raw sensor payloads. It consumes a decided record and decides one thing: does this condition mandate a 24-hour notice?

Install the runtime into a fresh virtual environment and pin the versions, since the pydantic v2 validation API differs materially from v1:

python3 -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.7.*"
# Optional: zoneinfo backport only if you are on Python < 3.9
pip install "backports.zoneinfo==0.2.1; python_version < '3.9'"

You also need two non-code inputs. The first is the acute contaminant table — the set of contaminants and conditions that Subpart Q classifies as Tier 1, together with their MCLs; for nitrate this is $10\ \text{mg/L as N}$ and for nitrite $1\ \text{mg/L as N}$. The second is a clock discipline: every timestamp entering this stage must already be timezone-aware UTC, because the deadline arithmetic is unforgiving and a naive local timestamp will silently shift your 24-hour window.

Step-by-Step Implementation

The engine follows four steps: model the incoming record, classify it against the acute condition set, compute the deadline, and emit the event. The decision an operator most needs to trust is the classification branch, so the flow below shows exactly which conditions fall into the acute tier.

Step 1 — Model the incoming record

The trigger stage consumes a decided record, not a raw reading. Model it with pydantic so malformed input is rejected at the boundary rather than corrupting a deadline downstream. The record carries the contaminant, the measured value and its units, the MCL that applied, a boolean for whether an MCL violation was already decided upstream, and any emergency flags a plant operator or the state raised.

from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, field_validator


class Condition(str, Enum):
    MICROBIAL_MCL = "microbial_mcl"          # E. coli / fecal coliform
    NITRATE_MCL = "nitrate_mcl"
    NITRITE_MCL = "nitrite_mcl"
    OUTBREAK = "waterborne_outbreak"
    BOIL_WATER = "boil_water_emergency"
    DISINFECTION_LOSS = "loss_of_disinfection"


class ViolationRecord(BaseModel):
    pwsid: str = Field(..., description="Public water system identifier")
    contaminant: str
    value: Optional[float] = None
    units: Optional[str] = None
    mcl: Optional[float] = None
    mcl_violation: bool = False
    conditions: set[Condition] = Field(default_factory=set)
    detected_at: datetime

    @field_validator("detected_at")
    @classmethod
    def must_be_aware_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("detected_at must be timezone-aware")
        return v.astimezone(timezone.utc)

Step 2 — Write the acute predicate

The predicate is a pure function over the record. It returns True only when at least one acute condition is present. Microbial MCL violations and confirmed outbreaks are acute unconditionally. Nitrate and nitrite are acute when they exceed their MCL, which lets the same predicate double-check the numeric boundary rather than trusting a stale flag. The nitrate ceiling is $10\ \text{mg/L as N}$ and nitrite is $1\ \text{mg/L as N}$; a reading is acute when its running value $v$ satisfies $v > \text{MCL}$. This numeric gate is deliberately redundant with the upstream MCL exceedance logic so that a Tier 1 notice never fires on a formatting artifact alone.

ACUTE_MCL = {"nitrate": 10.0, "nitrite": 1.0}  # mg/L as N

UNCONDITIONAL = {
    Condition.MICROBIAL_MCL,
    Condition.OUTBREAK,
    Condition.BOIL_WATER,
    Condition.DISINFECTION_LOSS,
}


def is_tier1(record: ViolationRecord) -> bool:
    """True when the record mandates a 24-hour acute public notice."""
    if record.conditions & UNCONDITIONAL:
        return True

    key = record.contaminant.strip().lower()
    if key in ACUTE_MCL and record.value is not None:
        # An exceedance of the acute nitrate/nitrite MCL is Tier 1.
        if record.value > ACUTE_MCL[key] and record.mcl_violation:
            return True

    return False

Step 3 — Compute the deadline

Subpart Q sets the acute clock at twenty-four hours from detection: $t_{deadline}=t_{detect}+24\,\text{h}$. Compute it once, in UTC, and carry it with the event so every consumer works from the same instant. The subtlety that trips teams up is the definition of $t_{detect}$: the clock starts when the utility learns of the acute condition, which for an automated pipeline is the moment a validated reading crosses the predicate — not when the lab drew the sample and not when an analyst happens to open the dashboard. Anchoring the deadline to the record’s detected_at makes that instant defensible, because it is the same timestamp the rule engine already recorded upstream. Keeping the deadline immutable at emission time is what lets the deadline-tracking stage prove, after the fact, that the notice went out on time.

NOTICE_WINDOW = timedelta(hours=24)


def deadline_for(record: ViolationRecord) -> datetime:
    """t_deadline = t_detect + 24h, always in UTC."""
    return record.detected_at + NOTICE_WINDOW

Step 4 — Emit the event to the dispatcher

The final step packages the decision into a serializable event and hands it off. This stage does not send SMS, email, or post to the web itself — that is the job of the sibling stage that handles dispatching multi-channel notifications. Here the engine only decides, stamps the deadline, and publishes. The event also carries a severity hint pulled from the severity scoring models so the dispatcher can prioritize the most exposed populations when several notices fire at once.

from typing import Callable


class Tier1Event(BaseModel):
    pwsid: str
    tier: int = 1
    contaminant: str
    conditions: list[Condition]
    detected_at: datetime
    deadline: datetime
    severity_hint: str = "acute"


def evaluate_and_emit(
    record: ViolationRecord,
    publish: Callable[[Tier1Event], None],
) -> Optional[Tier1Event]:
    """Fire a Tier 1 event to the dispatcher when the record is acute."""
    if not is_tier1(record):
        return None

    event = Tier1Event(
        pwsid=record.pwsid,
        contaminant=record.contaminant,
        conditions=sorted(record.conditions, key=lambda c: c.value)
        or [Condition.NITRATE_MCL],
        detected_at=record.detected_at,
        deadline=deadline_for(record),
    )
    publish(event)
    return event

The publish callable is any sink — a message-queue producer, an in-process bus, or a database write that the dispatcher polls. Decoupling emission from delivery keeps the trigger engine testable and lets the delivery layer scale independently. It also means the trigger can be exercised in isolation with an in-memory list as the sink, which is exactly how the verification tests below assert that acute records produce exactly one event and non-acute records produce none. In a live deployment, prefer a durable sink over an in-process call: if the dispatcher is momentarily unavailable, an acute event must survive a restart rather than vanish, because a lost Tier 1 event is indistinguishable from a missed deadline in the eyes of the primacy agency.

Configuration Reference

The engine reads its behavior from a small, versioned configuration. Treat the acute table as regulatory data under change control, never as inline literals scattered through the code.

The acute conditions recognized by the predicate map to Subpart Q as follows:

Verification & Testing

Prove the predicate and the deadline with deterministic unit tests. The critical assertions are that an unconditional condition fires regardless of numbers, that a nitrate reading below its MCL never fires, and that the deadline is exactly detection plus twenty-four hours.

from datetime import datetime, timedelta, timezone


def _rec(**kw):
    base = dict(
        pwsid="MA1234567",
        contaminant="nitrate",
        detected_at=datetime(2026, 7, 17, 9, 0, tzinfo=timezone.utc),
    )
    base.update(kw)
    return ViolationRecord(**base)


def test_microbial_is_unconditionally_acute():
    r = _rec(contaminant="e_coli", conditions={Condition.MICROBIAL_MCL})
    assert is_tier1(r) is True


def test_nitrate_below_mcl_is_not_acute():
    r = _rec(value=8.4, mcl=10.0, mcl_violation=False)
    assert is_tier1(r) is False


def test_nitrate_over_mcl_is_acute():
    r = _rec(value=12.7, mcl=10.0, mcl_violation=True)
    assert is_tier1(r) is True


def test_deadline_is_detection_plus_24h():
    r = _rec(value=12.7, mcl=10.0, mcl_violation=True)
    assert deadline_for(r) - r.detected_at == timedelta(hours=24)


def test_emit_publishes_only_when_acute():
    sink = []
    evaluate_and_emit(_rec(value=8.4, mcl=10.0), sink.append)
    assert sink == []
    evaluate_and_emit(_rec(value=12.7, mcl=10.0, mcl_violation=True), sink.append)
    assert len(sink) == 1 and sink[0].tier == 1

Acceptance criteria before promoting the trigger engine to production:

• Every incoming record carries a timezone-aware UTC detected_at; naive timestamps are rejected at the boundary.
• Microbial MCL violations and confirmed outbreaks fire Tier 1 with no numeric gate.
• Nitrate and nitrite fire only when the value exceeds the acute MCL and the upstream violation flag is set.
• deadline_for returns exactly detected_at + 24h for every acute record.
• The dispatcher sink receives one well-formed Tier1Event per acute record and nothing for non-acute records.
• Each emitted event carries the immutable deadline that the deadline-tracking stage will audit against.

Troubleshooting & Gotchas

• Naive timestamps silently shift the window. A detected_at without a timezone is the single most dangerous input here, because + 24h on local wall-clock time drifts across a daylight-saving boundary and can push the deadline an hour in either direction. The field_validator rejects naive datetimes outright; keep that guard and normalize to UTC at ingestion.
• A stale mcl_violation flag fires a phantom notice. If an upstream retry re-emits an old record with the flag still set, the predicate will happily re-trigger. De-duplicate on (pwsid, contaminant, detected_at) before emission, and let the deadline-tracking stage own idempotency for the notice itself, tracked in state-primacy notification deadlines.
• Nitrate units mismatch. Some labs report nitrate as $\text{NO}_3^-$ rather than as N; $10\ \text{mg/L as N}$ is roughly $44\ \text{mg/L as }\text{NO}_3^-$. Comparing the wrong basis against the MCL either misses a real exceedance or invents one. Convert to as N before the predicate ever sees the value.
• The trigger fires but nothing reaches the public. This stage only publishes an event; delivery lives in the dispatcher. If notices are silently dropped, the fault is almost always in the delivery channels, not the predicate — confirm the publish sink is wired to the live multi-channel dispatcher and not a stub.
• Multiple acute conditions on one record. A boil-water order alongside a microbial violation is one notice, not two. The predicate returns a single boolean, so let the dispatcher merge conditions into one notice per system rather than emitting an event per condition.

Related

• Public Notification Workflows for SDWA Violations — the parent section that sequences detection, dispatch, and deadline tracking
• Dispatching Multi-Channel Notifications: SMS, Email, and Web — the delivery stage that consumes the event emitted here
• Tracking State-Primacy Notification Deadlines — audits the 24-hour deadline this engine stamps
• Severity Scoring Models — supplies the priority hint carried on each acute event
• MCL Exceedance Logic Implementation — decides the exceedance this stage classifies into a tier