feat: advertise IQR-trimmed mean position derived from heard repeater adverts

This commit is contained in:
swaits committed 2026-04-16 13:47:18 -06:00
1 parent 228432f9e3
commit 54228a6f92
7 files changed
+126 -9

No files matched your search

+4
View File
@@ -54,6 +54,10 @@ events:
events-pretty: events-pretty:
@tail -F ~/.donglora/orac-events.jsonl | jq . @tail -F ~/.donglora/orac-events.jsonl | jq .
# Dump heard repeater positions + IQR-trimmed average we'd advertise
positions: _ensure_tools
@uv run python -c 'from orac.state import load_state, heard_positions, average_heard_position; load_state(); ps = heard_positions(); print(f"heard {len(ps)} unique positions:"); [print(f" {lat:+.3f}, {lon:+.3f}") for lat, lon in sorted(ps)]; avg = average_heard_position(); print(f"IQR-trimmed mean: {avg}" if avg else "no positions yet")'
# Clean cached artifacts # Clean cached artifacts
clean: clean:
@rm -rf .pytest_cache .ruff_cache __pycache__ orac/__pycache__ tests/__pycache__ @rm -rf .pytest_cache .ruff_cache __pycache__ orac/__pycache__ tests/__pycache__
+13
View File
@@ -161,6 +161,16 @@ def _handle_sigusr1(signum: int, _frame: object) -> None: # noqa: ARG001
rt.dump_metrics() rt.dump_metrics()
def _handle_sigint(_signum: int, _frame: object) -> None:
"""SIGINT: flag the IOThread for shutdown *before* KeyboardInterrupt
unwinds the main thread, so the interrupted serial read exits cleanly
instead of looking like a device disconnect."""
rt = _active_runtime
if rt is not None:
rt.io_thread.stop()
raise KeyboardInterrupt
# ── Connect loop ───────────────────────────────────────────────── # ── Connect loop ─────────────────────────────────────────────────
@@ -235,6 +245,9 @@ def main() -> None:
# SIGUSR1 dumps metrics; not available on all platforms (e.g. Windows). # SIGUSR1 dumps metrics; not available on all platforms (e.g. Windows).
with contextlib.suppress(ValueError, AttributeError): with contextlib.suppress(ValueError, AttributeError):
signal.signal(signal.SIGUSR1, _handle_sigusr1) signal.signal(signal.SIGUSR1, _handle_sigusr1)
# SIGINT: pre-set stop_event so IOThread's interrupted read exits cleanly.
with contextlib.suppress(ValueError, AttributeError):
signal.signal(signal.SIGINT, _handle_sigint)
port = sys.argv[1] if len(sys.argv) > 1 else None port = sys.argv[1] if len(sys.argv) > 1 else None
+15 -2
View File
@@ -300,9 +300,22 @@ def compute_mac(secret: bytes, ciphertext: bytes) -> bytes:
def build_advert_payload() -> bytes: def build_advert_payload() -> bytes:
"""Build the inner ADVERT payload: pubkey + timestamp + signature + app_data.""" """Build the inner ADVERT payload: pubkey + timestamp + signature + app_data.
If we've heard at least 5 unique repeater positions, include our averaged
position in the advert's app_data.
"""
from orac.state import average_heard_position, heard_position_count
timestamp = struct.pack("<I", int(time.time())) timestamp = struct.pack("<I", int(time.time()))
app_data = bytes([0x81]) + BOT_NAME.encode("utf-8") name_bytes = BOT_NAME.encode("utf-8")
if heard_position_count() >= 5 and (avg := average_heard_position()) is not None:
lat_i = int(round(avg[0] * 1_000_000))
lon_i = int(round(avg[1] * 1_000_000))
# flags: has_name (0x80) | has_location (0x10) | chat node (0x01)
app_data = bytes([0x91]) + struct.pack("<ii", lat_i, lon_i) + name_bytes
else:
app_data = bytes([0x81]) + name_bytes
sign_msg = _pubkey_bytes + timestamp + app_data sign_msg = _pubkey_bytes + timestamp + app_data
signature = sign(sign_msg) signature = sign(sign_msg)
return _pubkey_bytes + timestamp + signature + app_data return _pubkey_bytes + timestamp + signature + app_data
+4 -2
View File
@@ -55,7 +55,7 @@ from orac.meshcore import (
) )
from orac.reply_state import PendingAckTable, ReplyCache from orac.reply_state import PendingAckTable, ReplyCache
from orac.runtime import Metrics, TxItem, TxPriority, TxQueue from orac.runtime import Metrics, TxItem, TxPriority, TxQueue
from orac.state import get_channel_history, record_channel_msg from orac.state import get_channel_history, record_channel_msg, record_heard_position
from orac.worker import ChannelReplyWork, ChannelScreenWork, DmReplyWork, Worker from orac.worker import ChannelReplyWork, ChannelScreenWork, DmReplyWork, Worker
log = logging.getLogger("orac") log = logging.getLogger("orac")
@@ -220,10 +220,12 @@ class RxRouter:
result = try_decode_advert(raw_payload) result = try_decode_advert(raw_payload)
if result is None: if result is None:
return return
adv_pubkey, name = result adv_pubkey, name, node_type, position = result
if adv_pubkey == pubkey_bytes(): if adv_pubkey == pubkey_bytes():
return # our own advert echo return # our own advert echo
register_node(adv_pubkey, name) register_node(adv_pubkey, name)
if node_type == 2 and position is not None:
record_heard_position(position[0], position[1])
logfmt.net( logfmt.net(
"ADVERT %s (hash=0x%02x pk=%s...)", "ADVERT %s (hash=0x%02x pk=%s...)",
name, name,
+16 -4
View File
@@ -554,8 +554,15 @@ def try_decrypt_anon_req(
return sender_pubkey, peer_name, None return sender_pubkey, peer_name, None
def try_decode_advert(raw_payload: bytes) -> tuple[bytes, str] | None: def try_decode_advert(
"""Decode and verify an ADVERT payload. Returns (pubkey, name) or None.""" raw_payload: bytes,
) -> tuple[bytes, str, int, tuple[float, float] | None] | None:
"""Decode and verify an ADVERT payload.
Returns (pubkey, name, node_type, position_or_None) or None on failure.
node_type: low nibble of app_data flags (1=chat, 2=repeater, 3=room).
position: (lat, lon) in degrees, or None if advert lacks location.
"""
if len(raw_payload) < 100: if len(raw_payload) < 100:
return None return None
@@ -568,10 +575,15 @@ def try_decode_advert(raw_payload: bytes) -> tuple[bytes, str] | None:
return None return None
name = "" name = ""
node_type = 0
position: tuple[float, float] | None = None
if app_data: if app_data:
flags = app_data[0] flags = app_data[0]
node_type = flags & 0x0F
pos = 1 pos = 1
if flags & 0x10: # has location if flags & 0x10 and pos + 8 <= len(app_data): # has location
lat_i, lon_i = struct.unpack("<ii", app_data[pos : pos + 8])
position = (lat_i / 1_000_000.0, lon_i / 1_000_000.0)
pos += 8 pos += 8
if flags & 0x20: # feat1 if flags & 0x20: # feat1
pos += 2 pos += 2
@@ -580,4 +592,4 @@ def try_decode_advert(raw_payload: bytes) -> tuple[bytes, str] | None:
if flags & 0x80 and pos < len(app_data): # has name if flags & 0x80 and pos < len(app_data): # has name
name = app_data[pos:].decode("utf-8", errors="replace") name = app_data[pos:].decode("utf-8", errors="replace")
return pubkey, name if name else pubkey.hex()[:8] return pubkey, name if name else pubkey.hex()[:8], node_type, position
+11 -1
View File
@@ -28,6 +28,8 @@ from enum import IntEnum
from typing import Any from typing import Any
import donglora as dl import donglora as dl
from serial import SerialException
from orac import logfmt from orac import logfmt
from orac.constants import ( from orac.constants import (
ADVERT_INTERVAL, ADVERT_INTERVAL,
@@ -413,7 +415,13 @@ class IOThread(threading.Thread):
for _ in range(RX_DRAIN_PER_TICK): for _ in range(RX_DRAIN_PER_TICK):
if self._stop_event.is_set(): if self._stop_event.is_set():
return return
pkt = dl.recv(self._conn) try:
pkt = dl.recv(self._conn)
except (SerialException, OSError):
# Shutdown: signal interrupted a blocking serial read.
if self._stop_event.is_set():
return
raise
if pkt is None: if pkt is None:
break break
self._metrics.inc("rx_total") self._metrics.inc("rx_total")
@@ -453,6 +461,8 @@ class IOThread(threading.Thread):
log.exception("gauge_collector failed") log.exception("gauge_collector failed")
self._last_gauge_dump = now self._last_gauge_dump = now
except BaseException as e: # surface fatal errors to main thread except BaseException as e: # surface fatal errors to main thread
if self._stop_event.is_set():
return # expected: shutdown race
self.last_error = e self.last_error = e
log.exception("IOThread died: %s", e) log.exception("IOThread died: %s", e)
raise raise
+63
View File
@@ -21,6 +21,7 @@ _state: dict[str, dict[str, object]] = {
"channel_history": {}, # channel_name -> list of messages "channel_history": {}, # channel_name -> list of messages
"dm_history": {}, # peer_pubkey_hex -> list of messages "dm_history": {}, # peer_pubkey_hex -> list of messages
"known_nodes": {}, # pubkey_hex -> {"name": str, "seen": float} "known_nodes": {}, # pubkey_hex -> {"name": str, "seen": float}
"heard_positions": [], # list of [lat, lon] rounded to 3 decimals (dedup'd)
} }
@@ -94,6 +95,68 @@ def known_node_count() -> int:
return len(_state["known_nodes"]) return len(_state["known_nodes"])
# ── Heard repeater positions ─────────────────────────────────────
def record_heard_position(lat: float, lon: float) -> None:
"""Record a repeater position (rounded to 3 decimals, dedup'd as a set)."""
entry = [round(lat, 3), round(lon, 3)]
positions: list[list[float]] = _state["heard_positions"] # type: ignore[assignment]
if entry in positions:
return
positions.append(entry)
save_state()
def heard_position_count() -> int:
return len(_state["heard_positions"]) # type: ignore[arg-type]
def _iqr_bounds(values: list[float]) -> tuple[float, float]:
"""Return (lo, hi) IQR fence: [Q1 - 1.5*IQR, Q3 + 1.5*IQR]."""
s = sorted(values)
n = len(s)
# Linear interpolation quantiles (same as numpy default).
def q(p: float) -> float:
idx = p * (n - 1)
lo_i = int(idx)
hi_i = min(lo_i + 1, n - 1)
frac = idx - lo_i
return s[lo_i] * (1 - frac) + s[hi_i] * frac
q1, q3 = q(0.25), q(0.75)
iqr = q3 - q1
return q1 - 1.5 * iqr, q3 + 1.5 * iqr
def heard_positions() -> list[tuple[float, float]]:
"""All recorded heard positions."""
return [(p[0], p[1]) for p in _state["heard_positions"]] # type: ignore[union-attr]
def average_heard_position() -> tuple[float, float] | None:
"""IQR-trimmed mean of heard positions. Rejects lat/lon outliers
(e.g., null-island 0/0 or bogus points) before averaging. Falls back
to a plain mean when there are too few points for a robust IQR."""
positions = heard_positions()
if not positions:
return None
if len(positions) < 4:
lat = sum(p[0] for p in positions) / len(positions)
lon = sum(p[1] for p in positions) / len(positions)
return lat, lon
lat_lo, lat_hi = _iqr_bounds([p[0] for p in positions])
lon_lo, lon_hi = _iqr_bounds([p[1] for p in positions])
kept = [p for p in positions if lat_lo <= p[0] <= lat_hi and lon_lo <= p[1] <= lon_hi]
if not kept:
kept = positions
lat = sum(p[0] for p in kept) / len(kept)
lon = sum(p[1] for p in kept) / len(kept)
return lat, lon
# ── Channel history ────────────────────────────────────────────── # ── Channel history ──────────────────────────────────────────────