Threshold-Based Event Mapping

Within the Local Spatial Processing Patterns framework, threshold-based event mapping is the stage that converts continuous sensor telemetry into discrete, geotagged events by evaluating each reading against scalar limits and spatial boundaries directly on the gateway. Rather than streaming every coordinate and metric to centralized infrastructure, an edge node evaluates readings against predefined boundaries in real time and emits only the moments that matter — a temperature excursion inside a bonded storage zone, a vibration spike on a monitored asset, a tracked vehicle crossing a geofence. This keeps computational budgets, intermittent connectivity, and strict power envelopes in check while turning raw streams into decisions at the point of capture.

The pattern sits between ingestion and dispatch. Upstream, on-device geometry filtering rejects out-of-area points cheaply; downstream, the bandwidth and async sync layer ships the surviving events home. Event mapping is where a multi-condition rule fires — and where a sloppy implementation will either flood the backhaul with duplicate alerts or silently miss the one reading that mattered.

An event fires only when debounce, scalar, and spatial conditions all pass.

Constraint Mapping #

The hardware envelope dictates every design choice in an event mapper. The pattern almost always runs on ARM Cortex-A or RISC-V SoCs with no GPU, where the CPU shares cycles with a cellular modem and a serial/Modbus poller. The table below maps the dominant constraints — drawn from the broader device constraints and resource limits profile — to the specific pressure they put on this stage.

Two recurring decisions fall out of this table. First, coordinates are carried as integer microdegrees (WGS84 degrees × 1,000,000) rather than floats — this follows the spatial data precision standards used across the gateway, gives ~11 cm resolution, and keeps the AABB reject branch in pure integer arithmetic. Second, any projection work — WGS84 to a local ENU or UTM frame — is precomputed or cached per coordinate reference systems at the edge, never recomputed per sample inside a confined operational zone.

Implementation: Radial + Scalar Threshold Mapper #

The primary technique evaluates each sample against a set of registered thresholds, where a threshold combines a scalar band, a radial spatial boundary, and a debounce interval. The radial check uses a local-tangent metre approximation that is accurate to well under a metre for the small radii typical of asset and zone monitoring, and avoids a full haversine call on every sample. Per microdegree of latitude the ground distance is constant; longitude is scaled by the cosine of the latitude:

$$\Delta_{\text{lat}}^{m} = \Delta_{\text{lat}}^{\mu\text{deg}} \cdot 0.11132, \qquad \Delta_{\text{lon}}^{m} = \Delta_{\text{lon}}^{\mu\text{deg}} \cdot 0.11132 \cdot \cos(\phi)$$

A bounding-box reject runs before the square root, so the expensive branch only executes for points already near the centre. The implementation below targets gateways running MicroPython, CPython on Yocto, or an embedded Linux distribution. It avoids pandas/geopandas, uses collections.deque for bounded buffering, and exposes an FFI hook so a C-compiled distance function can replace the Python path during high-frequency ingestion. CPython’s GIL is irrelevant here because evaluation is single-threaded and CPU-bound; concurrency is handled with asyncio, not threads, so the eval loop never contends for the interpreter lock.

import math
import time
import asyncio
import logging
import ctypes
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Dict

# Structured logging for field diagnostics
logger = logging.getLogger("edge.event_mapper")

@dataclass
class TelemetryPoint:
    epoch: float
    lat_microdeg: int  # WGS84 degrees * 1_000_000
    lon_microdeg: int
    scalar: float
    sensor_id: str

@dataclass
class SpatialThreshold:
    center_lat: int   # microdegrees
    center_lon: int   # microdegrees
    radius_m: float
    scalar_min: float
    scalar_max: float
    debounce_ms: int = 0
    _last_trigger: float = field(default=0.0, repr=False)

class ThresholdEventMapper:
    # 1 microdegree of latitude ≈ 0.11132 metres
    _LAT_M_PER_UDEG = 0.11132

    def __init__(self, max_history: int = 512, c_lib_path: Optional[str] = None):
        # Fixed-size buffer prevents heap fragmentation
        self.buffer: deque = deque(maxlen=max_history)
        self.thresholds: Dict[str, SpatialThreshold] = {}

        # FFI integration hook for C-optimized haversine/ECEF distance
        if c_lib_path:
            self._lib = ctypes.CDLL(c_lib_path)
            self._lib.calc_dist.argtypes = [ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int]
            self._lib.calc_dist.restype = ctypes.c_double
        else:
            self._lib = None

    def register_threshold(self, name: str, cfg: SpatialThreshold):
        self.thresholds[name] = cfg

    def _bbox_check(self, p: TelemetryPoint, t: SpatialThreshold) -> bool:
        """Fast AABB reject using per-axis metre approximations."""
        lat_deg = t.center_lat / 1_000_000.0
        lon_m_per_udeg = self._LAT_M_PER_UDEG * math.cos(math.radians(lat_deg))
        lat_diff_m = abs(p.lat_microdeg - t.center_lat) * self._LAT_M_PER_UDEG
        lon_diff_m = abs(p.lon_microdeg - t.center_lon) * lon_m_per_udeg
        return lat_diff_m <= t.radius_m and lon_diff_m <= t.radius_m

    def _calc_distance(self, p: TelemetryPoint, t: SpatialThreshold) -> float:
        if self._lib:
            return self._lib.calc_dist(p.lat_microdeg, p.lon_microdeg, t.center_lat, t.center_lon)
        if not self._bbox_check(p, t):
            return t.radius_m + 1.0
        lat_deg = t.center_lat / 1_000_000.0
        lon_m_per_udeg = self._LAT_M_PER_UDEG * math.cos(math.radians(lat_deg))
        dlat_m = (p.lat_microdeg - t.center_lat) * self._LAT_M_PER_UDEG
        dlon_m = (p.lon_microdeg - t.center_lon) * lon_m_per_udeg
        return math.sqrt(dlat_m**2 + dlon_m**2)

    def evaluate(self, point: TelemetryPoint) -> Optional[dict]:
        self.buffer.append(point)
        for name, t in self.thresholds.items():
            if point.epoch < t._last_trigger + (t.debounce_ms / 1000.0):
                continue
            dist = self._calc_distance(point, t)
            if dist <= t.radius_m and t.scalar_min <= point.scalar <= t.scalar_max:
                t._last_trigger = point.epoch
                return {
                    "event_id": f"{name}_{int(point.epoch)}",
                    "ts": point.epoch,
                    "lat": point.lat_microdeg,
                    "lon": point.lon_microdeg,
                    "scalar": point.scalar,
                    "dist_m": round(dist, 2),
                    "threshold": name
                }
        return None

async def async_ingest_loop(mapper: ThresholdEventMapper, telemetry_queue: asyncio.Queue):
    """Async consumer pattern for non-blocking gateway pipelines."""
    while True:
        point = await telemetry_queue.get()
        event = mapper.evaluate(point)
        if event:
            logger.info("THRESHOLD_HIT", extra=event)
            # Serialize via msgpack for low-bandwidth backhaul
            # await publish_to_mqtt(msgpack.dumps(event))
        telemetry_queue.task_done()

The evaluate method is deliberately allocation-light: it appends to a bounded deque, walks a small dict of thresholds, and only constructs a payload dict on an actual hit. When the threshold count grows beyond a handful, pre-screen candidates through a grid or geohash index — the same staging structures used for spatial joins in constrained environments — so the per-sample cost stays bounded regardless of how many zones are registered. The decoupled async_ingest_loop keeps serial/Modbus ingestion off the evaluation path; for fan-out across cores see async execution for spatial workloads.

Implementation: Polygon Corridor and Hysteresis Variant #

A radial boundary is the wrong shape for many real zones — a pipeline right-of-way, a haul road, a no-wake channel, a bonded warehouse footprint. The complementary technique replaces the radius test with a ray-casting point-in-polygon check while reusing the bounding box as a cheap pre-filter, and adds scalar hysteresis so a reading hovering on a boundary cannot rapidly toggle the event state. Hysteresis is the single most effective defence against alert storms: the condition must cross enter to arm and fall back past exit to re-arm, so jitter inside the dead band produces no new events.

from dataclasses import dataclass, field
from typing import List, Tuple, Dict

@dataclass
class PolygonZone:
    # Ring of (lat_microdeg, lon_microdeg); first != last, ring auto-closed
    ring: List[Tuple[int, int]]
    scalar_enter: float           # arm threshold
    scalar_exit: float            # re-arm threshold (dead band = enter - exit)
    _bbox: Tuple[int, int, int, int] = field(default=None, repr=False)
    _armed: bool = field(default=False, repr=False)

    def __post_init__(self):
        lats = [p[0] for p in self.ring]
        lons = [p[1] for p in self.ring]
        self._bbox = (min(lats), min(lons), max(lats), max(lons))

def _in_bbox(lat: int, lon: int, bbox) -> bool:
    return bbox[0] <= lat <= bbox[2] and bbox[1] <= lon <= bbox[3]

def _point_in_ring(lat: int, lon: int, ring) -> bool:
    """Integer ray-casting; no float drift, no allocation."""
    inside = False
    n = len(ring)
    j = n - 1
    for i in range(n):
        yi, xi = ring[i]
        yj, xj = ring[j]
        if (yi > lat) != (yj > lat):
            # Cross-multiply to stay in integer space and avoid div-by-zero
            if (lon - xi) * (yj - yi) < (xj - xi) * (lat - yi):
                inside = not inside
        j = i
    return inside

class CorridorEventMapper:
    def __init__(self):
        self.zones: Dict[str, PolygonZone] = {}

    def register_zone(self, name: str, zone: PolygonZone):
        self.zones[name] = zone

    def evaluate(self, lat: int, lon: int, scalar: float, epoch: float):
        events = []
        for name, z in self.zones.items():
            inside = _in_bbox(lat, lon, z._bbox) and _point_in_ring(lat, lon, z.ring)
            if not z._armed:
                if inside and scalar >= z.scalar_enter:
                    z._armed = True
                    events.append({"zone": name, "edge": "enter", "ts": epoch})
            else:
                # Re-arm only after the reading clears the dead band or leaves the zone
                if (not inside) or scalar <= z.scalar_exit:
                    z._armed = False
                    events.append({"zone": name, "edge": "exit", "ts": epoch})
        return events

This variant emits both enter and exit edges, which downstream consumers need to compute dwell time and to clear alarms. The ray-casting loop stays in integer space — the cross-multiplication form sidesteps both division-by-zero on horizontal edges and the float rounding that plagues naive slope tests. For complex corridors, simplify the ring offline (Douglas–Peucker at the device’s effective GPS resolution) so the per-sample loop touches as few vertices as possible. The configuring-spatial-thresholds workflow below documents how to choose scalar_enter/scalar_exit and ring tolerances for a given sensor.

Configuration and Tuning #

Every knob in an event mapper is a trade between sensitivity and chatter. The detailed procedure lives in configuring spatial thresholds for sensor event triggers; the load-bearing parameters are:

• debounce_ms — the floor on time between two events from the same threshold. Set it to at least one polling period above the sensor’s natural jitter. A 1 Hz GNSS feed with occasional double-reports wants debounce_ms ≥ 1500.
• Scalar dead band — keep scalar_enter − scalar_exit at roughly 3–5% of the operating range. Too narrow and boundary noise toggles the state; too wide and you miss genuine recoveries.
• radius_m / ring tolerance — pad the geometry by the receiver’s real-world error, not its nominal spec (see field diagnostics below).
• Buffer depth (max_history) — size the deque to the longest replay window you need for post-event context, then cap it so the buffer plus threshold registry stays inside the 4–8 MB app heap.
• Serialization — emit compact JSON or MessagePack, strip redundant CRS declarations, and carry integer microdegrees. Match the wire format to the compression strategies for geospatial payloads used on the uplink.

For the FFI path, compile the distance helper with -O2 -fno-exceptions -fno-rtti and a fixed-width integer ABI so ctypes argument marshalling matches the C signature exactly. A mismatch there is silent and corrupts every distance — validate it against a known baseline at startup. Pin the evaluation loop to one core with taskset -c 1 or a cgroups cpuset so context switches from the modem driver do not perturb the latency budget.

Verification and Field Diagnostics #

Confirm the mapper works on a deployed device, not just in a unit test. Three checks catch the overwhelming majority of field failures:

1. Replay before commissioning. Capture an hour of real telemetry, then feed it through evaluate offline. Inject synthetic coordinate drift and scalar spikes to prove the debounce and hysteresis logic suppress duplicates without dropping true edges. Diff the event count against a hand-labelled expectation.
2. Structured event logging. Emit each hit through logger.info("THRESHOLD_HIT", extra=event) with the distance and threshold name attached, so a field tech can grep the journal and see exactly which rule fired and how close to the boundary the reading was. Prefer structured binary logging over verbose JSON dumps during initial commissioning.
3. Heap and latency watch. During a sustained-load soak, sample tracemalloc snapshots (sparingly — it is not free) to confirm the deque and threshold registry are flat, and time the evaluate call to confirm it stays under the 15 ms budget at peak rate.

GPS quality is the dominant real-world variable. Field receivers routinely show 2–5 m horizontal drift under canopy or multipath. To avoid false negatives, expand the effective boundary from the reported HDOP and satellite count at registration time and re-register when the GNSS quality report changes. A serviceable heuristic:

effective_radius = base_radius + (hdop * 1.5) + (1.0 / sat_count)

When the boundary itself is the moving part — a tracked asset crossing a fixed zone — the same drift logic applies to the asset position, not the zone.

Failure Modes and Recovery #

Three failure modes dominate this pattern, each with a clear detection signal and a safe recovery path:

• Alert storms. A scalar parked on a boundary with no dead band, or debounce_ms set below the polling jitter, produces a flood of near-identical events. Detect: events from one threshold arriving faster than the configured debounce. Recover: widen the dead band and raise debounce_ms; never compensate downstream by dropping events you have already emitted.
• Silent misses. An under-padded geofence rejects real crossings because the receiver drifted just outside the boundary. Detect: expected enter/exit pairs missing from replay against ground truth. Recover: apply the HDOP-based radius expansion and re-test against the same trace.
• Lost events during backhaul outage. Connectivity disappears for hours and an in-memory-only mapper loses every event it fired. Detect: gap between local event log and what arrived at the centre. Recover: persist serialized payloads to an SQLite database in WAL mode (PRAGMA journal_mode=WAL) or a flat-file ring buffer, prioritising event metadata over raw telemetry, and drain through message queue management at the edge once the link returns. Pair that with retry and backoff for unstable networks so reconnection does not stampede the uplink.

The unifying principle: an event mapper must be durable across power and network loss. Fire on state change, persist before you transmit, and treat the uplink as best-effort rather than guaranteed.

Related #

• On-Device Geometry Filtering — the upstream stage that rejects out-of-area points before they reach threshold evaluation.
• Spatial Joins in Constrained Environments — grid and geohash staging for correlating samples against many zones cheaply.
• Async Execution for Spatial Workloads — keeping ingestion and evaluation non-blocking under burst load.
• Configuring Spatial Thresholds for Sensor Event Triggers — the step-by-step procedure for choosing debounce, dead band, and geofence padding.
• Message Queue Management at the Edge — durable buffering and drain order for events fired during a backhaul outage.