Detecting Stuck and Frozen Meter Reads

A meter that never moves is a quieter failure than one that reads negative, and that is exactly what makes it dangerous. When a cumulative register freezes — a jammed dial, a dead communication module, a firmware hang — the interval feed keeps arriving on schedule, every read is structurally valid, and the delta is a perfectly plausible zero. Nothing trips a bounds check, so the account bills at zero or on an estimate month after month while real consumption goes unrecovered. This page is the focused counterpart to the broader Reading Anomaly Detection Algorithms workflow inside the Meter Data Ingestion & Validation Pipelines subsystem, and it solves one specific sub-problem: telling a stuck register apart from a genuinely vacant premise, where a flat line is the honest reading. Its sibling page on detecting negative consumption anomalies handles the register that moves the wrong way; this one handles the register that does not move at all.

# Minimal Prerequisites

• Python 3.11+ — for slots=True on @dataclass, zoneinfo in the standard library, and modern union syntax.
• decimal.Decimal for every register value — cumulative registers become deltas and deltas become charges, so float is never acceptable; exact comparison of two identical readings must be exact, not approximate.
• zoneinfo (not pytz) — the local civil hour of each interval decides whether zero movement is suspicious or expected, so every timestamp must be timezone-aware.
• Data assumptions: for a single meter you have an ordered series of interval reads — each a timezone-aware timestamp and a cumulative register value already validated non-negative by upstream schema validation and data quality checks — plus a baseline_daily, the premise’s historical mean daily consumption.

# Annotated Implementation

The core signal is a run-length: how many consecutive intervals pass with the cumulative register unchanged. A live meter almost never repeats an exact reading for long; a stuck one repeats it indefinitely. But zero movement alone cannot mean “stuck,” because a vacant premise also reads flat, and so does any occupied home in the small hours. Two guards separate the failure from the false alarm. First, the historical baseline: a premise that normally uses nothing is vacant, and its flat line is expected, not broken. Second, the local hour: a flat run is only damning across active hours, when a consuming premise should be advancing the dial. A register that holds steady from midnight to dawn tells you nothing; one that holds steady all afternoon on an account that averages twenty-plus kilowatt-hours a day is stuck.

The detector below computes the longest flat run and, within it, how many of those intervals fall in active hours, then decides between four states with the baseline as the first gate.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from enum import Enum
from zoneinfo import ZoneInfo

UTILITY_TZ = ZoneInfo("America/Chicago")  # the utility's civil time zone


class MeterState(str, Enum):
    LIVE = "live"                    # register advancing normally
    STUCK = "stuck"                  # frozen register on a premise that should consume
    VACANT = "vacant"               # flat line is the honest reading; do not flag
    INDETERMINATE = "indeterminate"  # flat, but only where zero use is plausible


@dataclass(frozen=True, slots=True)
class IntervalRead:
    ts: datetime          # timezone-aware, interval end
    register: Decimal     # cumulative register value, validated >= 0 upstream


@dataclass(frozen=True, slots=True)
class StuckVerdict:
    state: MeterState
    flat_run: int         # longest run of consecutive unchanged registers
    active_flat_run: int  # of that run, intervals falling in active hours
    baseline_daily: Decimal


def _longest_flat_run(ordered: list[IntervalRead]) -> int:
    longest = run = 0
    for prev, curr in zip(ordered, ordered[1:]):
        run = run + 1 if curr.register == prev.register else 0
        longest = max(longest, run)
    return longest


def detect_stuck_meter(
    reads: list[IntervalRead],
    baseline_daily: Decimal,
    *,
    min_flat_intervals: int = 12,             # e.g. 12 hourly reads of no movement
    vacancy_allowance: Decimal = Decimal("0.5"),  # kWh/day below which a premise reads vacant
    active_hours: tuple[int, int] = (7, 22),  # local hours where consumption is expected
) -> StuckVerdict:
    """Flag a frozen cumulative register, separating a stuck meter from a vacant premise.

    A meter is only 'stuck' when it should be moving: the premise historically
    consumes, and the register nonetheless holds a single value across a long run
    of active-hour intervals. Decimal comparison makes the equality exact.
    """
    if len(reads) < 2:
        return StuckVerdict(MeterState.INDETERMINATE, 0, 0, baseline_daily)

    ordered = sorted(reads, key=lambda r: r.ts)

    # Gate 1 — vacancy. A premise that historically uses essentially nothing is
    # vacant, not stuck; a flat register there is the expected read, and flagging
    # it would spawn false work orders for seasonal or unoccupied accounts.
    if baseline_daily <= vacancy_allowance:
        return StuckVerdict(MeterState.VACANT, _longest_flat_run(ordered), 0, baseline_daily)

    # Gate 2 — run-length of zero movement, weighted by active hours.
    longest = longest_active = 0
    run = run_active = 0
    for prev, curr in zip(ordered, ordered[1:]):
        if curr.register == prev.register:            # register did not advance
            run += 1
            hour = curr.ts.astimezone(UTILITY_TZ).hour
            if active_hours[0] <= hour < active_hours[1]:
                run_active += 1                       # zero use here IS suspicious
        else:
            run = run_active = 0                      # any movement resets the run
        longest = max(longest, run)
        longest_active = max(longest_active, run_active)

    if longest_active >= min_flat_intervals:
        return StuckVerdict(MeterState.STUCK, longest, longest_active, baseline_daily)
    if longest >= min_flat_intervals:
        # Flat, but only across hours where zero use is plausible (e.g. overnight):
        # not enough evidence to declare the meter stuck.
        return StuckVerdict(MeterState.INDETERMINATE, longest, longest_active, baseline_daily)
    return StuckVerdict(MeterState.LIVE, longest, longest_active, baseline_daily)

Two design choices carry the weight. The baseline gate runs first so a vacant or seasonal account can never be flagged — its flat line is truth, and the detector must not manufacture a work order from it. The active-hour weighting is what makes the run-length trustworthy: counting only intervals when a consuming premise should be moving the dial keeps a normal overnight lull from accumulating into a false positive. A STUCK verdict does not bill anything; like every other diverted read it should be handed to fallback routing for missing rate data for a clearly flagged provisional estimate and a field work order, so the cycle still closes while the physical meter is inspected.

# Edge Cases and Billing Gotchas

Flat data is where “no consumption” and “no measurement” look identical, and the difference is the whole bill.

• Genuine zero use / vacant premise. A foreclosed home or a shuttered storefront legitimately reads flat for months. The baseline_daily <= vacancy_allowance gate classifies these as VACANT up front so they never reach the run-length test — the opposite failure from the negative-consumption classifier, which never sees a zero delta because a stuck register produces no negative to catch.
• Seasonal low use. A summer-only irrigation account or a vacation cabin can sit near zero for a season without being broken. Deriving baseline_daily from the same season in prior years — not a rolling annual mean — keeps a legitimately dormant meter from tripping the detector when its off-season flat line is entirely normal.
• Partially stuck registers. Some meters freeze one channel (say kWh) while another (demand kW) still updates, or the register advances in coarse steps far apart. A pure equality run misses the coarse-step case, so pair this detector with a low-variance check: an active-hour window whose consumption variance collapses toward zero is suspicious even if the exact value ticks occasionally.
• Night hours and the active-hour window. Setting active_hours too wide drowns the signal in normal overnight flatness; too narrow, and a genuinely stuck meter escapes because its frozen run never overlaps the window. Tune the window to the customer class — a (7, 22) band fits residential, while a 24-hour industrial process may warrant a full-day window with a higher min_flat_intervals.

# Verification Snippet

Pin the four outcomes with a fixture so a change to a threshold fails loudly instead of quietly re-classifying live accounts.

from datetime import timedelta


def _hourly(start: datetime, registers: list[str]) -> list[IntervalRead]:
    return [IntervalRead(start + timedelta(hours=i), Decimal(v))
            for i, v in enumerate(registers)]


def test_stuck_and_vacant_and_live() -> None:
    active_start = datetime(2026, 7, 1, 7, 0, tzinfo=UTILITY_TZ)  # 07:00 local

    # Frozen register across a full active-hour block on a consuming premise.
    stuck = detect_stuck_meter(_hourly(active_start, ["48120.5"] * 16),
                               baseline_daily=Decimal("22.0"))
    assert stuck.state is MeterState.STUCK
    assert stuck.active_flat_run >= 12

    # Same flat line, but the premise historically uses ~nothing: vacant, not stuck.
    vacant = detect_stuck_meter(_hourly(active_start, ["1002.0"] * 16),
                                baseline_daily=Decimal("0.10"))
    assert vacant.state is MeterState.VACANT

    # Register climbing every interval: a healthy, live meter.
    climbing = [str(Decimal("48120.5") + Decimal(i)) for i in range(16)]
    live = detect_stuck_meter(_hourly(active_start, climbing),
                              baseline_daily=Decimal("22.0"))
    assert live.state is MeterState.LIVE

    # Flat, but only across the overnight lull: not enough evidence.
    night_start = datetime(2026, 7, 1, 22, 0, tzinfo=UTILITY_TZ)  # 22:00 local
    overnight = detect_stuck_meter(_hourly(night_start, ["48120.5"] * 16),
                                   baseline_daily=Decimal("22.0"))
    assert overnight.state is MeterState.INDETERMINATE

Replay this against a historical window where field crews confirmed which meters were physically stuck, and assert the detector reproduces their dispositions. Meters that repeatedly land in INDETERMINATE are a signal to widen the active-hour window or lengthen the read history, not to lower the threshold blindly.

# FAQ

# Related Topics

• Reading Anomaly Detection Algorithms — the parent workflow; where this no-movement guardrail fits in the full detection pipeline.
• Detecting Negative Consumption Anomalies in Python — the sibling case for a register that moves the wrong way rather than not at all.
• Meter Data Ingestion & Validation Pipelines — the subsystem this detector belongs to.
• Schema Validation & Data Quality Checks — the upstream contract guaranteeing the well-formed, non-negative registers this detector assumes.
• Fallback Routing for Missing Rate Data — provisional handling for a stuck meter so the cycle still closes while the field work order runs.

Up: Reading Anomaly Detection Algorithms · Meter Data Ingestion & Validation Pipelines