Configuring spatial thresholds for sensor event triggers

This page solves one concrete problem: turning a high-frequency GNSS coordinate stream into clean ENTER/EXIT events against radial boundaries, entirely on a fanless ARM gateway running embedded Linux, in pure-standard-library Python 3 — no PostGIS, no Shapely, no cloud round trip. Within the Local Spatial Processing Patterns practice, and specifically as a concrete build of threshold-based event mapping, this is the stage that decides — per sample, in well under a millisecond — whether a tracked asset has crossed a geofence and an event payload should be queued for backhaul. Deploying this logic at the network edge eliminates cloud-side evaluation latency, spares metered LTE backhaul, and keeps the geofence verdict alive through the hours-long uplink outages that field nodes routinely face.

The deployment context is unforgiving. The engine shares a single core with a cellular modem and a serial poller, runs against a 4–8 MB application heap, and ingests 10–100 Hz of raw fixes carrying 2–5 m of GNSS drift. A naive distance-vs-radius comparison thrashes wildly whenever a sensor lingers on a boundary, flooding the bandwidth and async sync layer with duplicate alerts. Everything below is built to survive that environment: bounded memory, no per-sample allocation in the steady state, and drift tolerance baked in rather than bolted on.

Why a stateful hysteresis filter fits the constraint envelope #

Spatial threshold evaluation at the gateway is best modelled as a stateful event filter, not a continuous spatial join. Each telemetry packet carries latitude, longitude, and a monotonic timestamp; the gateway holds a small set of active boundaries — circular here, though the same state machine drives the polygonal case handled by on-device geometry filtering. The job is to emit an event when a boundary is crossed, suppress redundant emissions while the sensor sits near that boundary, and queue only state changes for downstream routing.

The dominant field failure is false-positive triggering caused by coordinate jitter exceeding the threshold margin. Receivers operating under canopy or near reflective surfaces routinely show 2–5 m RMS drift, which makes a single-radius test oscillate rapidly as the measured distance crosses back and forth over the boundary. The fix is a dual-threshold (Schmitt-trigger) model: a tight activation radius r_act to fire ENTER, and a wider deactivation radius r_deact = r_act + hysteresis to fire EXIT. As long as the hysteresis band comfortably exceeds the receiver’s drift envelope, the state cannot thrash. This is why the pattern is preferred over a debounce-only timer: hysteresis is a property of the geometry, so it holds regardless of sample rate.

Distance itself uses the haversine great-circle formula, which stays accurate at the small radii typical of asset and zone monitoring without pulling in a projection library. For activation distance d against earth radius R:

$$d = 2R\,\arcsin\!\sqrt{\sin^2\!\left(\tfrac{\Delta\varphi}{2}\right) + \cos\varphi_1\cos\varphi_2\,\sin^2\!\left(\tfrac{\Delta\lambda}{2}\right)}$$

Coordinates are clamped to valid WGS84 ranges before any trig runs, following the spatial data precision standards used across the gateway, so a malformed fix can never propagate a NaN into the comparison. Any projection work — should you swap haversine for a local ENU plane inside a confined zone — is precomputed per coordinate reference systems at the edge, never recomputed per sample.

Dual-threshold hysteresis prevents state thrashing near a boundary.

Self-contained threshold engine #

The module below is a complete, dependency-free spatial threshold evaluator. It is single-threaded by design and intended to run inside one asyncio event loop — the ingestion coroutine awaits the modem/serial source, then calls process_telemetry() synchronously because the evaluation is pure CPU and finishes in microseconds, so it never needs its own task or a lock. Memory is bounded by a fixed-length deque; in the steady state (no crossing) the hot path allocates nothing and creates no garbage, so the CPython cyclic garbage collector never wakes mid-burst. A dict is built only on an actual ENTER/EXIT, which is rare by construction.

import math
import logging
from collections import deque
from typing import Dict, List, Optional, Callable

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

_MAX_HISTORY_LEN = 128          # ~1.5 KB of coordinate ring buffer
_EARTH_RADIUS_M = 6_371_000.0


class SpatialThresholdEngine:
    """Memory-aware radial threshold evaluator for ARM-based IoT gateways.

    Single-threaded; drive it from one asyncio loop. Implements dual-threshold
    hysteresis, drift clamping, and per-threshold debounce. No external GIS deps.
    """

    def __init__(self, max_thresholds: int = 64, cooldown_s: float = 30.0):
        self.thresholds: Dict[str, dict] = {}
        self.history: deque = deque(maxlen=_MAX_HISTORY_LEN)  # bounded => no leak
        self.event_callbacks: List[Callable] = []
        self._max_thresholds = max_thresholds
        self._cooldown_s = cooldown_s

    def add_threshold(self, threshold_id: str, lat: float, lon: float,
                      activation_radius_m: float, hysteresis_m: float) -> bool:
        if len(self.thresholds) >= self._max_thresholds:
            logging.warning("Threshold capacity reached; rejecting %s", threshold_id)
            return False
        # r_deact MUST exceed r_act for monotonic state transitions.
        self.thresholds[threshold_id] = {
            "lat": lat, "lon": lon,
            "r_act": activation_radius_m,
            "r_deact": activation_radius_m + hysteresis_m,
            "state": False,            # False = OUTSIDE, True = INSIDE
            "last_trigger_ts": 0.0,
        }
        return True

    @staticmethod
    def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
        """Great-circle distance in metres using only the stdlib math module."""
        lat1, lon1, lat2, lon2 = map(math.radians, (lat1, lon1, lat2, lon2))
        dlat, dlon = lat2 - lat1, lon2 - lon1
        a = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2
        return _EARTH_RADIUS_M * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

    def process_telemetry(self, lat: float, lon: float, ts: float) -> List[dict]:
        """Evaluate one fix against every active threshold.

        Returns a list of crossing events (possibly empty). Pure CPU; safe to
        call directly from the ingestion coroutine without offloading.
        """
        # Clamp out-of-range / NaN-adjacent values before any trig runs.
        lat = max(-90.0, min(90.0, lat))
        lon = max(-180.0, min(180.0, lon))
        self.history.append((ts, lat, lon))

        events: List[dict] = []
        for tid, cfg in self.thresholds.items():
            dist = self._haversine_m(lat, lon, cfg["lat"], cfg["lon"])

            # Schmitt-trigger transition: enter tight, leave wide.
            if not cfg["state"] and dist <= cfg["r_act"]:
                crossing = "ENTER"
                cfg["state"] = True
            elif cfg["state"] and dist > cfg["r_deact"]:
                crossing = "EXIT"
                cfg["state"] = False
            else:
                continue  # no transition for this threshold

            # Debounce: keep the state change, but suppress a repeat *emission*
            # inside the cooldown window so backhaul never sees duplicate alerts.
            if (ts - cfg["last_trigger_ts"]) < self._cooldown_s:
                continue
            cfg["last_trigger_ts"] = ts
            events.append({
                "type": crossing,
                "threshold_id": tid,
                "distance_m": round(dist, 2),
                "timestamp": ts,
                "coords": (lat, lon),
            })

        for ev in events:                      # fan out only on real crossings
            for cb in self.event_callbacks:
                cb(ev)
        return events

    def register_callback(self, func: Callable) -> None:
        self.event_callbacks.append(func)

Constraint validation #

Each constraint that defines this device class — drawn from the broader device constraints and resource limits profile — maps to a specific mitigation already present in the code above.

Gotchas and edge cases #

Drift tolerance is a config value, not a constant. Size the hysteresis band to the receiver, not a textbook number. For standard L1 receivers, hysteresis_m ≈ 1.5 × expected_drift (15–25 m in urban canyons) keeps the EXIT radius clear of the jitter envelope. Sub-5 m activation radii are only honest with RTK/PPK hardware — civilian single-frequency GPS lands at 3–10 m under open sky per the GPS.gov accuracy specifications, so a 3 m geofence on an L1 receiver is noise, not a boundary.

Pre-filter garbage fixes before the engine sees them. A 2D fix without adequate satellite lock can jump tens of metres in one sample and punch straight through the hysteresis band. Discard telemetry with HDOP > 2.5 or sat_count < 6 upstream; the threshold engine assumes its input is a believable position.

Coordinate-system assumptions. Inputs are WGS84 decimal degrees. Haversine is great-circle distance on a sphere — fine to well under a metre at these radii, but if you swap in a planar approximation, do it on a per-zone projected frame, never on raw lon/lat differences near the poles or the antimeridian.

Timestamp monotonicity. The cooldown math assumes ts never goes backwards. On gateways that take wall-clock time from an NTP step or the GNSS receiver itself, a clock correction can make ts - last_trigger_ts negative and briefly disable debounce. Feed process_telemetry() a monotonic source (time.monotonic()), and carry wall-clock time only inside the payload.

State survives restarts only if you persist it. cfg["state"] lives in RAM. A power cycle re-initialises every threshold to OUTSIDE, which will re-fire ENTER for any asset already inside a zone. Persist the last known states to non-volatile storage (e.g. /var/lib/gateway/threshold_state.json) and reload on boot if duplicate ENTERs on reboot are unacceptable. Validate hot-swapped configurations against a schema and run a dry-run pass that logs distances without mutating cfg["state"] before going live.

Calling it from the event-mapping pipeline #

Wire the engine into the ingestion coroutine and route its output to the sync layer. The callback is where a crossing becomes a queued payload — typically published with a delivery guarantee per MQTT QoS levels for telemetry drops, or folded into a delta sync for GPS coordinate streams when events ride alongside position updates.

import asyncio
import time

engine = SpatialThresholdEngine(max_thresholds=32, cooldown_s=30.0)
engine.add_threshold("bonded-zone-A", 51.5074, -0.1278,
                     activation_radius_m=25.0, hysteresis_m=20.0)

async def ingest(gnss_source, outbound_queue):
    """Pull fixes, evaluate, and hand crossings to the durable sync buffer."""
    engine.register_callback(lambda ev: outbound_queue.put_nowait(ev))
    async for fix in gnss_source:                 # awaits modem / serial I/O
        # Pure-CPU eval: synchronous, sub-millisecond, no offload needed.
        engine.process_telemetry(fix.lat, fix.lon, time.monotonic())
        await asyncio.sleep(0)                     # yield to keep the loop fair

A full bench harness lives one level up in threshold-based event mapping; the outbound_queue is drained by the sync layer, which applies its own retry policy when the link is flaky.

Related #

• Threshold-based event mapping — the parent pattern: debounce, scalar bands, and spatial predicates that this radial engine plugs into.
• Implementing polygon containment checks in C++ — when a circular threshold is too coarse and you need an allocation-free geofence polygon test.
• Optimizing WGS84 vs UTM for low-memory IoT gateways — choosing the coordinate frame that keeps distance math cheap inside a confined zone.
• Setting exponential backoff for cloud sync retries — how the durable buffer behind the callback drains events once the uplink returns.
• Reducing RAM usage for GeoJSON parsing on Raspberry Pi — companion memory-discipline technique for loading the boundary definitions the engine evaluates against.