Polling GPS Telematics Without Rate Limiting

This page shows how to pull GPS fixes from a vendor telematics API as fast as your route optimizer needs them — without ever earning a 429 Too Many Requests — by running an async aiohttp poller behind a token-bucket governor that spends the full rate budget and stops exactly at its edge, instead of leaving idle gaps a fixed interval would.

“Without rate limiting” does not mean ignoring the vendor’s limit; it means the client owns the pacing so the server never has to. A fixed 30-second interval wastes most of a 10-requests-per-second budget and still drifts a truck a full block between fixes; a naive asyncio.gather over every vehicle empties the budget in one burst and gets the whole fleet throttled. The token bucket built below keeps the request stream pinned just under the ceiling continuously. This is the pull-side companion to the GPS Polling Strategies cadence design, and it feeds the same ingestion boundary the Telematics & Sensor Data Ingestion framework enforces before any fix reaches a solver.

Environment & Data Prerequisites

This page targets Python 3.10+ — the code uses X | None union syntax and match is avoided for portability. Pin the async HTTP client and its speedups:

python -m pip install aiohttp==3.10.10

The input is a set of vehicle ids you poll and, per response, a single GPS fix with the shape below. Units matter: latitude and longitude are WGS84 decimal degrees, accuracy is the receiver’s reported CEP in meters, and the timestamp is timezone-aware UTC.

Two numbers govern everything: the vendor’s sustained rate (requests per second across your API key) and its burst allowance. A typical municipal telematics contract grants something like 10 req/s sustained with a burst of 20. Poll above the sustained rate for longer than the burst window and the server returns 429 with a Retry-After header. The whole design goal is to spend that budget without crossing it.

Step-by-Step Implementation

Step 1 — Model the fix and the poller config

Fixes are frozen dataclasses so a recorded coordinate is never mutated after ingestion, keeping the audit trail tamper-evident. The config carries the rate budget explicitly rather than hiding it in magic numbers:

from dataclasses import dataclass, field


@dataclass(frozen=True)
class GPSFix:
    vehicle_id: str
    lat: float
    lon: float
    timestamp: str        # ISO-8601, UTC
    gps_accuracy: float


@dataclass
class PollerConfig:
    base_url: str
    fleet_ids: list[str]
    rate_per_sec: float = 10.0      # vendor sustained ceiling
    burst_capacity: float = 20.0    # vendor burst allowance
    max_concurrent: int = 20        # socket pool ceiling
    timeout_sec: float = 2.5
    max_attempts: int = 3

Step 2 — Configure the site-standard JSON audit logger

Every request outcome lands in a machine-readable line. The site uses a custom logging.Formatter subclass — not a third-party logger — so the record shape matches the rest of the compliance stack:

import json
import logging
from datetime import datetime, timezone


class JSONLogFormatter(logging.Formatter):
    """Emit one newline-delimited JSON object per poll event."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "event": record.getMessage(),
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        for key, value in getattr(record, "poll", {}).items():
            entry[key] = value
        return json.dumps(entry)


_handler = logging.StreamHandler()
_handler.setFormatter(JSONLogFormatter())
logger = logging.getLogger("gps_poll")
logger.setLevel(logging.INFO)
logger.handlers = [_handler]

Step 3 — Bind the rate budget into a token bucket

This is the constraint that replaces the fixed interval. The bucket holds up to burst_capacity tokens and refills at rate_per_sec. Every request must acquire() one token; when the bucket is empty the coroutine sleeps only long enough for the next token to accrue. Because the refill is continuous, the request stream sits flush against the ceiling instead of pulsing:

import asyncio
import time


class TokenBucket:
    def __init__(self, rate_per_sec: float, capacity: float):
        self._rate = rate_per_sec
        self._capacity = capacity
        self._tokens = capacity
        self._updated = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            self._tokens = min(
                self._capacity,
                self._tokens + (now - self._updated) * self._rate,
            )
            self._updated = now
            if self._tokens < 1.0:
                deficit = 1.0 - self._tokens
                await asyncio.sleep(deficit / self._rate)
                self._tokens = 0.0
                self._updated = time.monotonic()
            else:
                self._tokens -= 1.0

The token refill count over an elapsed interval is simply the rate times the time delta, clamped to capacity:

$$\text{tokens} = \min\!\left(C,\; \text{tokens}_{0} + R \cdot \Delta t\right)$$

where $C$ is burst_capacity, $R$ is rate_per_sec, and $\Delta t$ is the seconds since the last refill.

Step 4 — Poll concurrently under the budget with backoff

The poller shares one aiohttp.ClientSession with a bounded TCPConnector, so persistent sockets eliminate per-request TLS handshakes. Each fetch first spends a token, then issues the request; a 429 drains the bucket for the server-supplied Retry-After window instead of hammering it, which is the difference between backing off and getting banned:

import aiohttp


class RateAwareGPSPoller:
    def __init__(self, config: PollerConfig):
        self.config = config
        self.bucket = TokenBucket(config.rate_per_sec, config.burst_capacity)
        self.session: aiohttp.ClientSession | None = None

    async def __aenter__(self) -> "RateAwareGPSPoller":
        connector = aiohttp.TCPConnector(
            limit=self.config.max_concurrent,
            ttl_dns_cache=300,
            keepalive_timeout=30,
            enable_cleanup_closed=True,
        )
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=aiohttp.ClientTimeout(total=self.config.timeout_sec),
            headers={"Accept": "application/json"},
        )
        return self

    async def __aexit__(self, *exc: object) -> None:
        if self.session:
            await self.session.close()

    async def _fetch(self, vehicle_id: str) -> dict | None:
        assert self.session is not None
        url = f"{self.config.base_url}/api/v2/telemetry/{vehicle_id}"
        for attempt in range(1, self.config.max_attempts + 1):
            await self.bucket.acquire()
            try:
                async with self.session.get(url) as resp:
                    if resp.status == 429:
                        retry_after = float(resp.headers.get("Retry-After", "1"))
                        # Drain the bucket so no coroutine retries early.
                        self.bucket._tokens = 0.0
                        logger.warning("rate_limited", extra={"poll": {
                            "vehicle_id": vehicle_id,
                            "retry_after_s": retry_after,
                            "attempt": attempt}})
                        await asyncio.sleep(retry_after)
                        continue
                    resp.raise_for_status()
                    payload = await resp.json()
                    logger.info("fix_received", extra={"poll": {
                        "vehicle_id": vehicle_id, "status": resp.status}})
                    return payload
            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                logger.error("fetch_failed", extra={"poll": {
                    "vehicle_id": vehicle_id,
                    "attempt": attempt, "error": str(exc)}})
        logger.critical("fetch_exhausted", extra={"poll": {
            "vehicle_id": vehicle_id}})
        return None

Step 5 — Serialize validated fixes through a plausibility gate

A fast poll is worthless if it forwards a garbage coordinate. Before a fix is yielded it passes a schema check and a displacement plausibility gate: the great-circle distance from the previous fix, divided by the elapsed seconds, must stay under a heavy-vehicle speed cap. Coordinates that violate kinematic reality are quarantined, exactly as Validating GPS Coordinates in Python specifies for the ingestion layer. The Haversine distance between two fixes is:

$$d = 2R \cdot \arcsin\!\sqrt{\sin^{2}\!\frac{\Delta\varphi}{2} + \cos\varphi_1 \cos\varphi_2 \sin^{2}\!\frac{\Delta\lambda}{2}}$$

with $R = 6{,}371{,}000$ m, $\varphi$ latitude and $\lambda$ longitude in radians.

import math
from datetime import datetime

_REQUIRED = {"vehicle_id", "lat", "lon", "timestamp", "gps_accuracy"}
_EARTH_M = 6_371_000.0
_MAX_MPS = 33.3   # ~120 km/h heavy-vehicle cap


def _haversine_m(a: GPSFix, b: GPSFix) -> float:
    p1, p2 = math.radians(a.lat), math.radians(b.lat)
    dphi = math.radians(b.lat - a.lat)
    dlmb = math.radians(b.lon - a.lon)
    h = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2
    return 2 * _EARTH_M * math.asin(math.sqrt(h))


def validate(payload: dict, prev: GPSFix | None) -> GPSFix | None:
    missing = _REQUIRED - payload.keys()
    if missing:
        logger.warning("schema_rejected", extra={"poll": {"missing": sorted(missing)}})
        return None
    fix = GPSFix(
        vehicle_id=payload["vehicle_id"], lat=payload["lat"], lon=payload["lon"],
        timestamp=payload["timestamp"], gps_accuracy=payload["gps_accuracy"],
    )
    if not (-90 <= fix.lat <= 90) or not (-180 <= fix.lon <= 180):
        logger.warning("bounds_rejected", extra={"poll": {
            "vehicle_id": fix.vehicle_id, "lat": fix.lat, "lon": fix.lon}})
        return None
    if prev is not None:
        dt = (datetime.fromisoformat(fix.timestamp)
              - datetime.fromisoformat(prev.timestamp)).total_seconds()
        if dt > 0 and _haversine_m(prev, fix) / dt > _MAX_MPS:
            logger.warning("displacement_rejected", extra={"poll": {
                "vehicle_id": fix.vehicle_id,
                "implied_mps": round(_haversine_m(prev, fix) / dt, 1)}})
            return None
    return fix

The poll loop ties the pieces together: it spends the budget across the fleet with asyncio.as_completed so validated fixes stream out as soon as each vehicle answers, rather than waiting for the slowest:

async def run() -> None:
    config = PollerConfig(
        base_url="https://telemetry.municipal-waste.example",
        fleet_ids=["WST-1042", "WST-1043", "WST-1044"],
        rate_per_sec=10.0, burst_capacity=20.0,
    )
    last: dict[str, GPSFix] = {}
    async with RateAwareGPSPoller(config) as poller:
        tasks = [poller._fetch(vid) for vid in config.fleet_ids]
        for coro in asyncio.as_completed(tasks):
            payload = await coro
            if payload is None:
                continue
            fix = validate(payload, last.get(payload["vehicle_id"]))
            if fix is not None:
                last[fix.vehicle_id] = fix
                logger.info("fix_dispatched", extra={"poll": {
                    "vehicle_id": fix.vehicle_id,
                    "lat": fix.lat, "lon": fix.lon}})


if __name__ == "__main__":
    asyncio.run(run())

Compliance Output

The poller emits one newline-delimited JSON object per event. This is the audit record a municipal contract expects — it proves the client paced itself and never forwarded an implausible fix:

{"event": "fix_received", "timestamp": "2026-03-15T08:42:11.305000+00:00", "vehicle_id": "WST-1042", "status": 200}
{"event": "fix_dispatched", "timestamp": "2026-03-15T08:42:11.306000+00:00", "vehicle_id": "WST-1042", "lat": 40.7128, "lon": -74.006}
{"event": "rate_limited", "timestamp": "2026-03-15T08:42:12.001000+00:00", "vehicle_id": "WST-1043", "retry_after_s": 1.0, "attempt": 1}
{"event": "displacement_rejected", "timestamp": "2026-03-15T08:42:12.140000+00:00", "vehicle_id": "WST-1044", "implied_mps": 210.4}

Each field maps to an operational or regulatory purpose:

• event — the decision class an auditor filters on; a healthy shift shows fix_received/fix_dispatched and no sustained rate_limited runs.
• status — the HTTP status returned, evidence the client stayed inside the vendor’s ceiling rather than being throttled off it.
• retry_after_s — the exact backoff the server asked for and the client honored, so any 429 in the record is provably followed by a compliant pause.
• implied_mps — the ground speed a rejected fix would have required; it is the evidence a quarantined coordinate was a GPS artifact, not a real position that a solver should trust.
• timestamp — a UTC, ISO-8601 instant. Retention and signing of these lines are handled by the Security & Access Boundaries layer so proof-of-service timestamps survive audit.

Verification

Confirm the token bucket actually caps throughput. This drives 40 acquisitions through a 10 req/s bucket sized for a burst of 20 and asserts the elapsed time reflects the sustained rate, with no live endpoint:

import asyncio
import time


def test_token_bucket_caps_sustained_rate():
    async def drive() -> float:
        bucket = TokenBucket(rate_per_sec=10.0, capacity=20.0)
        start = time.monotonic()
        for _ in range(40):
            await bucket.acquire()
        return time.monotonic() - start

    elapsed = asyncio.run(drive())
    # 20 burst tokens are free; the remaining 20 accrue at 10/s -> ~2.0s.
    assert 1.8 <= elapsed <= 2.6

The window is the proof: the first 20 requests drain the burst instantly, the next 20 are metered at ten per second, so 40 requests can never complete faster than roughly two seconds — the client cannot out-run the budget even if the event loop is idle.

Common Errors

RuntimeError: Timeout context manager should be used inside a task — raised when aiohttp.ClientSession is created outside a running event loop, e.g. in __init__ at import time. Root cause: session construction bound to the wrong loop. Fix: build the session inside the async context (__aenter__) as shown, and only ever use it under asyncio.run.

Every request eventually returns 429 no matter how slowly you poll — root cause: two poller processes share one API key, so each sees only half the true request stream and the vendor’s server-side counter sums both. Fix: run a single governor per key, or lower rate_per_sec proportionally to the number of workers so their combined stream stays under the contract ceiling. The same shared-budget idempotency concern is handled push-side by Bin Sensor API Sync.

ValueError: Invalid isoformat string in validate — root cause: the vendor sends a Z-suffixed timestamp (2026-03-15T08:42:11Z) that older datetime.fromisoformat cannot parse. Fix on 3.10 by replacing the suffix (ts.replace("Z", "+00:00")) before parsing, or normalize every timestamp to UTC at ingestion exactly as Parsing IoT Fill-Level Sensor Payloads does for its device feed.

Related

• GPS Polling Strategies for Waste Route Optimization — the adaptive-cadence design this rate governor plugs into.
• Validating GPS Coordinates in Python — the full displacement, bounds, and accuracy gate the plausibility check summarizes.
• Bin Sensor API Sync for Municipal Waste Operations — the push-side idempotency counterpart to this pull-side poller.
• Parsing IoT Fill-Level Sensor Payloads — timestamp normalization and payload parsing patterns reused above.
• VRP Route Optimization Algorithms — the downstream consumer whose distance matrix depends on these paced, validated fixes.

Up: GPS Polling Strategies