Syncing Smart-Meter AMI Feeds via REST APIs

A REST collector that pulls interval reads from an AMI head-end has exactly one job that matters to billing: never skip a page and never re-count one. This page is the concrete transport implementation behind Step 1 of AMI/AMR Feed Synchronization Protocols, inside Meter Data Ingestion & Validation Pipelines. The failure mode it resolves is subtle: a paginated pull that commits its cursor before the reads are safely handed off, so a mid-batch token expiry, a 429, or a crash silently drops a window of consumption — or a naive retry replays a page and inflates it. The fix is a collector where the durable cursor advances only after the fetched records clear validation and reach the billing boundary, and where every retry is a no-op by construction.

# Prerequisites

Imports and the data contract this collector assumes — nothing else.

import asyncio
import hashlib
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from decimal import Decimal

import httpx  # httpx>=0.27, async client with per-request timeouts

• Python 3.11+ for timezone-aware datetime and modern asyncio; never pytz, never naive timestamps.
• httpx.AsyncClient as the transport so the collector is non-blocking and integrates with async batch processing for high-volume reads.
• decimal.Decimal for every cumulative_kwh register value — binary float drift compounds across millions of intervals into penny-level ledger discrepancies.
• A persisted cursor store. The head-end exposes cursor-based pagination (next_cursor); the last acknowledged cursor must survive a process restart (a row in the staging DB, not in-memory state).
• An OAuth 2.0 client-credentials grant whose access token expires and must be refreshed mid-pull. The head-end returns cumulative registers, not pre-computed deltas.

# Annotated Implementation

The collector below is the whole transport layer. It refreshes the token before expiry, walks pages until the cursor is exhausted or rate limits force a pause, normalizes each raw record into the canonical envelope the rest of the pipeline consumes, and — the load-bearing detail — persists the next cursor only after the page has been handed off downstream.

@dataclass(slots=True, frozen=True)
class SyncEnvelope:
    """Transport-agnostic record handed to validation. source_hash is the
    idempotency basis: two deliveries of the same read collapse to one identity."""
    meter_id: str
    local_ts: str            # meter wall-clock, ISO 8601, still naive
    cumulative_kwh: Decimal
    status_code: int
    source_hash: str


def _source_hash(meter_id: str, local_ts: str, cumulative_kwh: Decimal) -> str:
    # Content address of the physical reading only — never transport metadata —
    # so a page replay produces identical hashes and upserts become no-ops.
    basis = f"{meter_id}|{local_ts}|{cumulative_kwh}"
    return hashlib.sha3_256(basis.encode("utf-8")).hexdigest()


class AMIRestCollector:
    def __init__(self, base_url: str, client: httpx.AsyncClient, cursor_store):
        self._base_url = base_url.rstrip("/")
        self._client = client
        self._cursor_store = cursor_store          # persistent: get()/commit()
        self._token: str | None = None
        self._token_expiry = datetime.min.replace(tzinfo=timezone.utc)

    async def _auth_header(self) -> dict[str, str]:
        # Refresh 60s early: a token that expires mid-page turns a clean pull
        # into a 401 retry storm. Clock skew of a minute is cheap insurance.
        now = datetime.now(timezone.utc)
        if self._token is None or now >= self._token_expiry - timedelta(seconds=60):
            resp = await self._client.post(
                f"{self._base_url}/oauth/token",
                data={"grant_type": "client_credentials"},
            )
            resp.raise_for_status()
            body = resp.json()
            self._token = body["access_token"]
            self._token_expiry = now + timedelta(seconds=int(body["expires_in"]))
        return {"Authorization": f"Bearer {self._token}"}

    @staticmethod
    def _normalize(raw: dict) -> SyncEnvelope:
        # Coerce the register to Decimal at the boundary, before any arithmetic.
        meter_id = raw["meterId"]
        local_ts = raw["readTimeLocal"]
        cumulative = Decimal(str(raw["cumulativeKwh"]))
        return SyncEnvelope(
            meter_id=meter_id,
            local_ts=local_ts,
            cumulative_kwh=cumulative,
            status_code=int(raw.get("statusCode", 0)),
            source_hash=_source_hash(meter_id, local_ts, cumulative),
        )

    async def sync(self, handoff) -> int:
        """Pull every available page; return the count of records handed off.
        `handoff(records)` must durably persist the batch (validate + upsert)
        and RAISE on failure, so a bad handoff never advances the cursor."""
        cursor = await self._cursor_store.get()      # resumes where we stopped
        total = 0

        while True:
            headers = await self._auth_header()
            params = {"cursor": cursor, "pageSize": 1000} if cursor else {"pageSize": 1000}
            resp = await self._client.get(
                f"{self._base_url}/v1/intervals", headers=headers, params=params
            )

            # 429 / 503: obey Retry-After, then re-request the SAME cursor.
            # We have not advanced, so this is a safe wait, not a data gap.
            if resp.status_code in (429, 503):
                await asyncio.sleep(self._retry_after(resp))
                continue
            resp.raise_for_status()

            body = resp.json()
            records = [self._normalize(r) for r in body["data"]]

            # Hand off BEFORE committing the cursor. If handoff raises, we crash
            # with the old cursor intact and simply re-pull this page next run —
            # the source_hash makes that re-pull idempotent downstream.
            if records:
                await handoff(records)
                total += len(records)

            next_cursor = body.get("nextCursor")
            await self._cursor_store.commit(next_cursor)   # durable acknowledge
            cursor = next_cursor

            # Proactively slow down while quota is nearly exhausted.
            if int(resp.headers.get("X-RateLimit-Remaining", "1")) == 0:
                await asyncio.sleep(self._retry_after(resp))
            if next_cursor is None:                         # feed drained
                return total

    @staticmethod
    def _retry_after(resp: httpx.Response) -> float:
        # Prefer the server's explicit hint; fall back to a fixed pause.
        raw = resp.headers.get("Retry-After")
        if raw and raw.isdigit():
            return float(raw)
        reset = resp.headers.get("X-RateLimit-Reset")
        if reset and reset.isdigit():
            delta = int(reset) - int(datetime.now(timezone.utc).timestamp())
            return float(max(delta, 1))
        return 5.0

The ordering in sync() is the entire correctness argument: fetch → normalize → handoff (which validates and upserts) → then commit the cursor. Because the cursor is the only durable “we are done with this page” signal, and it moves last, any failure between fetch and commit is recovered by simply re-pulling — and the source_hash guarantees that re-pull is harmless. The handoff callback is where records enter schema validation and data quality checks and the idempotent upsert; this collector deliberately owns transport only.

# Edge Cases and Billing Gotchas

• Token expiry mid-pagination. A long pull can outlive a short-lived access token, and a 401 arriving on page 40 of 200 strands the run. Refreshing 60 seconds before the recorded expiry (as _auth_header does) absorbs both normal expiry and modest clock skew, so the token is never the reason a window of reads goes missing.
• Rate-limit pause vs. cursor advance. On 429/503 the collector continues against the same cursor after waiting — it has not acknowledged the page, so waiting is safe. The dangerous inversion is advancing the cursor and then discovering the next page was throttled; that gap is a silent revenue leak. For sustained degradation rather than isolated throttling, escalate to the circuit breaker in error handling and retry workflows.
• Duplicate page delivery. A retried request or a head-end that re-emits a boundary page will deliver reads the pipeline has already seen. The source_hash (meter + local timestamp + register) makes each re-delivery collapse to the same identity, so the downstream upsert is a no-op and no interval is double-counted.
• Negative or rolled-over registers. Because the head-end sends cumulative lifetime registers, a value that decreases between reads is either a fixed-width rollover or a genuine fault — never valid consumption. The transport layer must not “fix” it; it passes the raw register through, and disposition happens in reading anomaly detection algorithms. When a read is missing entirely and no schedule can price the gap, control passes to fallback routing for missing rate data.

# Verification Snippet

Prove the cursor never outruns the handoff. This test injects a handoff that fails on the second page and asserts the cursor stays pinned to the last successful acknowledgment, so a re-run resumes from the right place instead of skipping the failed window.

class FakeCursorStore:
    def __init__(self, start=None):
        self.value = start
        self.commits = []
    async def get(self):
        return self.value
    async def commit(self, cursor):
        self.value = cursor
        self.commits.append(cursor)


async def test_cursor_holds_when_handoff_fails():
    store = FakeCursorStore(start="A")
    seen = []

    async def flaky_handoff(records):
        if store.value == "B":            # fail while positioned on page B
            raise RuntimeError("downstream upsert rejected")
        seen.extend(records)

    # ... wire `collector` to a transport returning pages A -> B -> None ...
    with_pytest_raises = False
    try:
        await collector.sync(flaky_handoff)
    except RuntimeError:
        with_pytest_raises = True

    assert with_pytest_raises                     # the failure surfaced
    assert store.value == "B"                      # cursor did NOT advance past B
    assert "C" not in store.commits                # page after B was never acked

The invariant under test is the one that keeps billing whole: on any handoff failure the durable cursor must equal the last page that actually persisted, so the next run re-pulls the failed page rather than jumping past it.

# Frequently Asked Questions

# Related Topics

• AMI/AMR Feed Synchronization Protocols — the parent workflow; this page implements its transport step.
• Schema Validation & Data Quality Checks — where the handed-off records are validated before the ledger.
• Async Batch Processing for High-Volume Reads — scaling the idempotent handoff to hundreds of thousands of reads.
• Reading Anomaly Detection Algorithms — disposition for rolled-over and negative registers this collector passes through.
• Error Handling & Retry Workflows — backoff, dead-letter queues, and the circuit breaker for sustained vendor degradation.

Up one level: AMI/AMR Feed Synchronization Protocols · Parent guide: Meter Data Ingestion & Validation Pipelines · Return to the utilitybilling.org home.