Debouncing Instrument Alarms with Time Windows

Amplitude noise is only half of why lab alarms flap; the other half is time. A reading that spikes past a limit for a single 10 ms sample and settles is a measurement artifact, not an excursion, yet a naive comparator fires on it every time. Time-window debouncing suppresses these transient spikes by requiring a threshold condition to persist for a dwell time — or across N-of-M consecutive samples — before an alarm is allowed to fire, and by holding it asserted through a separate release delay so it cannot chatter on the way down. This is the time-domain complement to the amplitude-domain fix in Hysteresis Thresholds to Suppress Alert Flapping, and it sits inside the Threshold Tuning & Alerting stage of the Data Capture, Validation & Metadata Sync architecture.

Why Dwell Time and Hysteresis Solve Different Problems

Hysteresis and dwell filtering are frequently conflated, and using one where the other belongs leaves the alarm chattering. Hysteresis separates the trip and clear thresholds along the value axis — the signal must fall measurably back below where it rose before the alarm releases — and it defeats flapping caused by broadband noise sitting right on a single limit. It does nothing for a clean, brief spike that shoots far past the threshold and returns: that excursion clears the trip level and the clear level both, so hysteresis passes it straight through. Dwell filtering separates the events along the time axis instead: it asks not “how far past the limit” but “for how long,” and rejects anything that does not stay past the limit long enough to matter. A robust alarm on a noisy scientific instrument almost always needs both — hysteresis on the amplitude, a dwell window on the duration — because the two failure signatures are independent. This page implements the time-domain half; the amplitude bands come from the parent Threshold Tuning & Alerting evaluator, and for drift that is neither a spike nor a static offset, Rolling Z-Score Anomaly Detection for Sensor Drift tracks the moving baseline the dwell logic then guards.

Two assumptions bound the scope. The incoming samples are already threshold-classified — each carries a boolean “condition true/false” plus its acquisition timestamp — so debouncing operates on the crossing decision, not the raw volts. And the sample stream is roughly periodic at a known poll period, because dwell time is only meaningful relative to how fast you are looking.

On-Delay Integration, Off-Delay Extension, and N-of-M Voting

Two independent timers govern the alarm. The on-delay (integration) timer requires the trip condition to hold continuously for a dwell time t_on before the alarm asserts — it integrates duration and rejects anything shorter. The off-delay (extension) timer holds the alarm asserted for t_off after the condition clears, so a signal hovering at the boundary cannot rapidly de-assert and re-assert. The two are deliberately asymmetric: t_on is tuned to the shortest real excursion you must catch, t_off to how long a genuine event’s tail typically rings.

Because the sampler is discrete, a dwell time maps to a sample count. The alarm fires when the condition has been continuously true for at least t_on, which at poll period T_poll means at least this many consecutive true samples:

 $N_{on} = ⌈ \frac{t _{on}}{T _{poll}} ⌉, fire when Δ t_{true} \geq t_{on}$

Requiring strictly consecutive samples is brittle when a single sample dropout or a lone dip below the limit resets the whole count. Production alarms relax this to N-of-M voting: the alarm arms when at least N of the last M samples satisfy the condition, tolerating isolated gaps while still demanding sustained pressure. The vote over the trailing window W of the most recent M classifications is:

 $arm when i \in W \sum 1 [cond_{i}] \geq N, N \leq M$

Setting N = M recovers the strict consecutive rule; N = M-1 tolerates one bad sample in the window; lower ratios trade sensitivity for noise immunity. A third, orthogonal control governs emission rate rather than assertion: a leaky-bucket limiter caps how many alert messages leave the system per unit time even when the alarm legitimately re-fires, so a channel oscillating just above the dwell threshold cannot flood a pager. The bucket holds a token budget that drains at a fixed leak rate and refuses emission when empty — the alarm state stays correct, only the notification is throttled.

A transient spike shorter than t_on never asserts the alarm; a sustained excursion fires only after the on-delay dwell elapses, then holds through the off-delay release window before clearing.

*A transient spike is rejected on duration alone, while a sustained excursion fires after the integration dwell and releases after the extension delay.*

A DebouncedAlarm with On/Off Timers and an N-of-M Voter

The implementation keeps assertion state, both dwell timers, the sample vote, and the leaky-bucket emission limiter in one object. Every timer is measured against a caller-supplied monotonic clock so wall-clock adjustments (NTP steps, daylight-saving) can never corrupt a dwell interval. The alarm ingests one classified sample at a time and returns a structured event only on a genuine state edge, leaving routing to the caller exactly as the parent evaluator does.

from __future__ import annotations

import logging
from collections import deque
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Deque, Optional

logger = logging.getLogger(__name__)


class AlarmEdge(Enum):
    ASSERTED = "ASSERTED"
    CLEARED = "CLEARED"


@dataclass(frozen=True)
class AlarmEvent:
    edge: AlarmEdge
    at_monotonic: float
    channel: str
    emitted: bool  # False when suppressed by the leaky-bucket limiter


@dataclass
class DebounceConfig:
    t_on: float = 0.75          # on-delay: seconds condition must persist to assert
    t_off: float = 2.0          # off-delay: seconds alarm is held after condition clears
    vote_m: int = 5             # sliding window length for N-of-M voting
    vote_n: int = 4             # samples within the window that must be true to arm
    bucket_capacity: float = 3.0    # max burst of emitted alerts
    bucket_leak_per_s: float = 0.2  # sustained emission rate (tokens refilled / second)

    def __post_init__(self) -> None:
        if not (1 <= self.vote_n <= self.vote_m):
            raise ValueError("require 1 <= vote_n <= vote_m")
        if self.t_on < 0 or self.t_off < 0:
            raise ValueError("dwell times must be non-negative")


class DebouncedAlarm:
    """Time-domain debounce for a single threshold-classified channel.

    Consumes ``(condition, timestamp)`` pairs where ``condition`` is the raw
    threshold-crossing decision and ``timestamp`` comes from a monotonic clock.
    Asserts only after the trip condition survives an on-delay dwell (and an
    N-of-M vote); releases only after an off-delay extension. A leaky bucket
    caps how many asserted edges are actually emitted, decoupling alarm truth
    from notification volume.
    """

    def __init__(
        self,
        channel: str,
        config: DebounceConfig = DebounceConfig(),
        clock: Callable[[], float] = None,
    ) -> None:
        import time
        self.channel = channel
        self.cfg = config
        self._clock = clock or time.monotonic
        self._asserted = False
        self._votes: Deque[bool] = deque(maxlen=config.vote_m)
        self._cond_since: Optional[float] = None   # when the vote first armed
        self._clear_since: Optional[float] = None   # when the vote first disarmed
        self._tokens = config.bucket_capacity
        self._last_leak = self._clock()

    def _armed(self) -> bool:
        """True when at least vote_n of the last vote_m samples satisfied the condition."""
        return sum(self._votes) >= self.cfg.vote_n

    def _take_token(self, now: float) -> bool:
        """Refill by elapsed leak, then consume one token if available."""
        elapsed = now - self._last_leak
        self._last_leak = now
        self._tokens = min(
            self.cfg.bucket_capacity,
            self._tokens + elapsed * self.cfg.bucket_leak_per_s,
        )
        if self._tokens >= 1.0:
            self._tokens -= 1.0
            return True
        return False

    def update(self, condition: bool, timestamp: float) -> Optional[AlarmEvent]:
        """Feed one classified sample; return an AlarmEvent only on a state edge."""
        self._votes.append(condition)
        armed = self._armed()

        if not self._asserted:
            if armed:
                # start (or continue) the on-delay integration window
                if self._cond_since is None:
                    self._cond_since = timestamp
                if timestamp - self._cond_since >= self.cfg.t_on:
                    self._asserted = True
                    self._cond_since = None
                    self._clear_since = None
                    emitted = self._take_token(timestamp)
                    logger.info("%s ASSERTED (emitted=%s)", self.channel, emitted)
                    return AlarmEvent(AlarmEdge.ASSERTED, timestamp, self.channel, emitted)
            else:
                self._cond_since = None  # dip below the vote resets the dwell
            return None

        # already asserted: run the off-delay extension before clearing
        if armed:
            self._clear_since = None  # excursion still active, cancel any release
            return None
        if self._clear_since is None:
            self._clear_since = timestamp
        if timestamp - self._clear_since >= self.cfg.t_off:
            self._asserted = False
            self._clear_since = None
            logger.info("%s CLEARED", self.channel)
            return AlarmEvent(AlarmEdge.CLEARED, timestamp, self.channel, True)
        return None

The vote window and the on-delay timer work together: the vote provides gap tolerance sample-by-sample, and the timer enforces the absolute duration in seconds so the guarantee survives a change in poll rate. Emission suppression is reported, never silent — the emitted flag on every asserted edge lets downstream Metadata Injection Workflows record that an alarm was real but throttled, which is a very different audit fact from an alarm that never occurred.

Verifying Rejection Below Dwell and Assertion Above It

The correctness claim is precise and testable without hardware: a burst shorter than t_on must never assert, and a burst longer than t_on must assert exactly once. Drive the alarm with synthetic timestamped samples at a fixed poll period and assert on the returned edges.

from debounced_alarm import AlarmEdge, DebounceConfig, DebouncedAlarm


def _feed(alarm: DebouncedAlarm, pattern: list[bool], t_poll: float, t0: float = 0.0):
    events = []
    for i, cond in enumerate(pattern):
        ev = alarm.update(cond, t0 + i * t_poll)
        if ev is not None:
            events.append(ev)
    return events


def test_transient_below_dwell_never_fires() -> None:
    cfg = DebounceConfig(t_on=0.75, t_off=2.0, vote_m=5, vote_n=4)
    alarm = DebouncedAlarm("det0", cfg)
    # 3 true samples at 100 ms poll = 0.2 s of dwell, well under t_on=0.75 s
    pattern = [False] * 5 + [True] * 3 + [False] * 10
    assert _feed(alarm, pattern, t_poll=0.1) == []


def test_sustained_above_dwell_fires_once() -> None:
    cfg = DebounceConfig(t_on=0.75, t_off=2.0, vote_m=5, vote_n=4)
    alarm = DebouncedAlarm("det0", cfg)
    # 12 true samples at 100 ms = 1.2 s > t_on, then clear long enough for t_off
    pattern = [False] * 5 + [True] * 12 + [False] * 25
    events = _feed(alarm, pattern, t_poll=0.1)
    edges = [e.edge for e in events]
    assert edges == [AlarmEdge.ASSERTED, AlarmEdge.CLEARED]


def test_one_sample_dropout_tolerated_by_vote() -> None:
    cfg = DebounceConfig(t_on=0.5, t_off=1.0, vote_m=5, vote_n=4)
    alarm = DebouncedAlarm("det0", cfg)
    # a lone False inside the excursion must not reset assertion (4-of-5 holds)
    pattern = [True, True, False, True, True, True, True, True]
    events = _feed(alarm, pattern, t_poll=0.2)
    assert any(e.edge == AlarmEdge.ASSERTED for e in events)

On a live rig, the log signature of a healthy debounce is a quiet channel: transient EMI, lamp flicker, and single-frame dropouts produce no ASSERTED line, while a real excursion logs exactly one assertion followed — after its tail rings out past t_off — by one clear. If the log shows an assertion on every brief spike, t_on is too short for the instrument’s transient profile; if genuine excursions are missing entirely, it is too long (see below). Cross-check that the acquisition feeding the alarm is not itself stalling: a timeout in the poll loop that must be recovered through Timeout Handling & Retry Logic will stretch T_poll and silently change the sample-count arithmetic.

Failure Modes Specific to Time-Window Debouncing

Dwell longer than a real excursion swallows the event. The whole mechanism is a low-pass filter on duration, so any genuine excursion shorter than t_on is discarded as if it were noise — a 300 ms detector saturation vanishes under a 750 ms dwell, exactly the event you needed to catch. The tell is a downstream consequence (a spoiled sample, a tripped hardware interlock) with no corresponding alarm in the log. Fix by sizing t_on to the shortest actionable excursion, not the longest transient; when those overlap, the amplitude domain must disambiguate instead, and a fast excursion that is also large should bypass the dwell via a separate hard limit.

Poll-rate jitter miscounts the sample window. N-of-M voting assumes a stable T_poll; if the acquisition loop jitters — GC pauses, a slow numpy pass, USB scheduling — the same wall-clock excursion spans a different number of samples, and a strict consecutive count can arm early or late. This is why the implementation gates assertion on elapsed time (timestamp - self._cond_since >= t_on) rather than a raw sample tally, and only uses the vote for gap tolerance. Diagnose by logging the sample count inside each dwell and comparing it against ceil(t_on / T_poll); a wandering count confirms jitter and argues for the time-based guard the code already uses.

Wall-clock timestamps corrupt the dwell interval. If samples are timestamped from datetime.now() or a wall clock, an NTP step or DST change mid-dwell can make a 750 ms window appear to elapse in negative time or jump an hour, either firing instantly or hanging asserted forever. Always drive the timers from time.monotonic() (as the clock argument does) and carry that monotonic stamp on every sample from acquisition; reserve wall-clock time for the human-readable record attached later during metadata injection.

Correlated cross-channel crossing becomes an alert storm. When a shared cause — a power glitch, an HVAC swing, a common reference drift — pushes many channels past their limits within the same dwell window, each DebouncedAlarm legitimately asserts and a hundred pagers fire at once. Per-alarm debouncing cannot see the correlation. The leaky bucket caps each channel’s own emission rate, but the real fix is a supervisory layer that aggregates simultaneous asserted edges into one grouped incident. Feeding those edges through an Async Command Queuing System lets a single consumer debounce at the fleet level and collapse a synchronized burst into one actionable event.

Threshold Tuning & Alerting — the parent stage that classifies each reading into the boolean condition this alarm debounces.
Hysteresis Thresholds to Suppress Alert Flapping — the amplitude-domain companion; pair it with dwell filtering for full flap immunity.
Rolling Z-Score Anomaly Detection for Sensor Drift — moving-baseline detection for drift that neither a spike nor a static limit catches.
Metadata Injection Workflows — records the emitted flag so a throttled-but-real alarm is auditable.
Async Command Queuing Systems — fleet-level aggregation that collapses a correlated cross-channel burst into one incident.

← Back to Threshold Tuning & Alerting

References

Python time.monotonic documentation — the monotonic clock guarantee the dwell timers depend on.
Python collections.deque documentation — the bounded sliding window backing the N-of-M vote.