mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
feat(phase4a): peak_compass end-to-end + repair stale satpass test suite (#35)
Add azimuth-at-peak-elevation compass to the satpass pipeline (Matt's decision: extend pass_predictor) and repair the 24 stale satpass tests that were written against the pre-async handle_satpass return contract. - pass_predictor.PassInfo gains azimuth_at_peak; _build_pass populates it from the already-computed peak sample - !satpass DM wire + Central consolidated broadcast wire render aos→peak→los - satpass_pending gains a peak_compass column (migration v21, SCHEMA_VERSION 20->21); persisted at ingest, carried through consolidation from the max-elevation observer's row - 24 stale satpass tests repaired to the ingest→consolidate two-call contract (none weakened); satpass suite 112/112 green - full suite 34->10 failures (the 24 were these stale tests) NOTE: peak_compass changes the live Central satpass wire — deploy is HELD until the coordinated all-native flip (per Matt). Co-authored-by: Matt Johnson <mj@k7zvx.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
GitHub
parent
531bd20378
commit
a756143284
9 changed files with 239 additions and 115 deletions
|
|
@ -46,6 +46,7 @@ class PassInfo:
|
|||
max_elevation: float # Degrees
|
||||
azimuth_at_aos: float # Degrees, clockwise from north
|
||||
azimuth_at_los: float # Degrees, clockwise from north
|
||||
azimuth_at_peak: float # Degrees, clockwise from north (at max elevation)
|
||||
|
||||
|
||||
def compute_passes(line1: str, line2: str,
|
||||
|
|
@ -144,6 +145,7 @@ def _build_pass(samples: list[tuple[datetime, float, float]]) -> PassInfo:
|
|||
max_elevation=samples[peak_idx][1],
|
||||
azimuth_at_aos=samples[0][2] % 360,
|
||||
azimuth_at_los=samples[-1][2] % 360,
|
||||
azimuth_at_peak=samples[peak_idx][2] % 360,
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -190,16 +190,27 @@ def format_pass(*, sat_name: str, max_el: float,
|
|||
aos_compass: str, los_compass: str,
|
||||
broadcast: bool = True,
|
||||
entry_observer: Optional[str] = None,
|
||||
exit_observer: Optional[str] = None) -> str:
|
||||
exit_observer: Optional[str] = None,
|
||||
peak_compass: Optional[str] = None) -> str:
|
||||
"""Unified pass formatter with mode switch.
|
||||
|
||||
broadcast=True: Two-line format with buckets, 12h times, LoRa budget.
|
||||
🛰️ {name} {bucket}, {aos_compass}→{los_compass}
|
||||
🛰️ {name} {bucket}, {aos_compass}→[peak_compass→]{los_compass}
|
||||
{duration} min window, {rise}–{set} {AM/PM} {TZ} [tomorrow] [(region)]
|
||||
|
||||
broadcast=False: Compact DM format with exact degrees.
|
||||
{name} {HH:MM}–{HH:MM} {TZ} max {el}° {aos_compass}→{los_compass}
|
||||
{name} {HH:MM}–{HH:MM} {TZ} max {el}° {aos_compass}→[peak→]{los_compass}
|
||||
|
||||
peak_compass: compass direction at peak elevation. When provided, the
|
||||
compass segment renders as aos→peak→los; when None it stays aos→los
|
||||
(preserving legacy callers that don't thread the peak field).
|
||||
"""
|
||||
# Compass sweep segment: include the peak point only when supplied.
|
||||
if peak_compass:
|
||||
compass_seg = f"{aos_compass}→{peak_compass}→{los_compass}"
|
||||
else:
|
||||
compass_seg = f"{aos_compass}→{los_compass}"
|
||||
|
||||
if broadcast:
|
||||
bucket = _elevation_bucket(max_el)
|
||||
# Duration in whole minutes
|
||||
|
|
@ -213,7 +224,7 @@ def format_pass(*, sat_name: str, max_el: float,
|
|||
tz = _tz_abbr(aos_epoch)
|
||||
date_lbl = _date_label(aos_epoch)
|
||||
|
||||
line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {aos_compass}\u2192{los_compass}"
|
||||
line1 = f"\U0001F6F0\uFE0F {sat_name} {bucket}, {compass_seg}"
|
||||
|
||||
# Build time portion
|
||||
time_part = f"{dur_min} min window, {rise_str}\u2013{set_str} {ampm} {tz}{date_lbl}"
|
||||
|
|
@ -235,7 +246,7 @@ def format_pass(*, sat_name: str, max_el: float,
|
|||
tz = _tz_abbr(aos_epoch)
|
||||
return (f"{sat_name} {aos_str}\u2013{los_str} {tz} "
|
||||
f"max {int(max_el)}\u00B0 "
|
||||
f"{aos_compass}\u2192{los_compass}")
|
||||
f"{compass_seg}")
|
||||
|
||||
|
||||
def _map_severity(max_el: float) -> str:
|
||||
|
|
@ -397,10 +408,10 @@ def handle_satpass(envelope: dict, subject: str,
|
|||
conn.execute(
|
||||
"INSERT OR REPLACE INTO satpass_pending("
|
||||
"consolidated_id, observer, sat_name, norad_id, max_elevation, "
|
||||
"aos_at, los_at, aos_compass, los_compass, received_at) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?)",
|
||||
"aos_at, los_at, aos_compass, los_compass, peak_compass, received_at) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
|
||||
(consolidated_id, observer, sat_name, norad_id, max_el,
|
||||
aos_epoch, los_epoch, aos_compass, los_compass, now))
|
||||
aos_epoch, los_epoch, aos_compass, los_compass, direction, now))
|
||||
|
||||
# Signal consumer to schedule consolidation timer
|
||||
_pending_consolidation_ids.add(consolidated_id)
|
||||
|
|
@ -443,6 +454,8 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None
|
|||
los_epoch = exit_["los_at"]
|
||||
aos_compass = entry["aos_compass"]
|
||||
los_compass = exit_["los_compass"]
|
||||
# Peak belongs to whoever saw the highest elevation.
|
||||
peak_compass = best["peak_compass"]
|
||||
entry_obs = entry["observer"]
|
||||
exit_obs = exit_["observer"]
|
||||
|
||||
|
|
@ -467,6 +480,7 @@ def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None
|
|||
wire = format_pass(sat_name=sat_name, max_el=max_el,
|
||||
aos_epoch=aos_epoch, los_epoch=los_epoch,
|
||||
aos_compass=aos_compass, los_compass=los_compass,
|
||||
peak_compass=peak_compass,
|
||||
entry_observer=entry_obs, exit_observer=exit_obs)
|
||||
|
||||
# Dry-run gate
|
||||
|
|
@ -586,6 +600,7 @@ CREATE TABLE IF NOT EXISTS satpass_pending (
|
|||
los_at INTEGER,
|
||||
aos_compass TEXT,
|
||||
los_compass TEXT,
|
||||
peak_compass TEXT,
|
||||
received_at INTEGER,
|
||||
PRIMARY KEY (consolidated_id, observer)
|
||||
);
|
||||
|
|
|
|||
|
|
@ -166,12 +166,14 @@ class SatpassCommand(CommandHandler):
|
|||
from meshai.central.satpass_handler import format_pass
|
||||
az_aos = azimuth_to_compass(p.azimuth_at_aos)
|
||||
az_los = azimuth_to_compass(p.azimuth_at_los)
|
||||
az_peak = azimuth_to_compass(p.azimuth_at_peak)
|
||||
aos_epoch = int(p.aos_time.timestamp())
|
||||
los_epoch = int(p.los_time.timestamp())
|
||||
line = format_pass(
|
||||
sat_name=tle["name"], max_el=p.max_elevation,
|
||||
aos_epoch=aos_epoch, los_epoch=los_epoch,
|
||||
aos_compass=az_aos, los_compass=az_los,
|
||||
peak_compass=az_peak,
|
||||
broadcast=False,
|
||||
)
|
||||
all_lines.append(line)
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
DEFAULT_DB_PATH = "/data/meshai.sqlite"
|
||||
MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH"
|
||||
SCHEMA_VERSION = 20
|
||||
SCHEMA_VERSION = 21
|
||||
SCHEMA_META_TABLE = "schema_meta"
|
||||
MIGRATIONS_DIR = Path(__file__).parent / "migrations"
|
||||
|
||||
|
|
|
|||
7
work/meshai/persistence/migrations/v21.sql
Normal file
7
work/meshai/persistence/migrations/v21.sql
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
-- v21: peak_compass column on satpass_pending.
|
||||
-- Carries the compass direction at a pass's PEAK elevation through the
|
||||
-- observer-consolidation buffer so the consolidated broadcast wire can
|
||||
-- render aos→peak→los. Nullable; legacy pending rows (there are none at
|
||||
-- rest — the table is drained per bucket) keep NULL and render aos→los.
|
||||
|
||||
ALTER TABLE satpass_pending ADD COLUMN peak_compass TEXT;
|
||||
|
|
@ -89,6 +89,30 @@ def _clear_handler_flags():
|
|||
delattr(handle_satpass, attr)
|
||||
|
||||
|
||||
def _ingest_and_consolidate(env, subject, *, now, data=None):
|
||||
"""Drive the two-call async satpass contract.
|
||||
|
||||
handle_satpass() ingests the pass into satpass_pending and ALWAYS
|
||||
returns None (the immediate-broadcast suppression); the consumer then
|
||||
fires consolidate_satpass_pending() on its 5s timer, which returns the
|
||||
(wire_string, data_dict) to actually broadcast, or None if suppressed.
|
||||
This helper runs both halves and returns the consolidation result so a
|
||||
test can assert on the real broadcast decision + wire.
|
||||
"""
|
||||
from meshai.central.satpass_handler import (
|
||||
handle_satpass, consolidate_satpass_pending,
|
||||
drain_pending_consolidation_ids)
|
||||
drain_pending_consolidation_ids() # clear any cross-test leakage
|
||||
ingest = handle_satpass(
|
||||
env, subject, data=data if data is not None else {}, now=now)
|
||||
assert ingest is None, "handle_satpass must suppress the immediate broadcast"
|
||||
for cid in drain_pending_consolidation_ids():
|
||||
res = consolidate_satpass_pending(cid)
|
||||
if res is not None:
|
||||
return res
|
||||
return None
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
# 1. OPT-IN BIRD FILTER
|
||||
# ══════════════════════════════════════════════════════════════════════
|
||||
|
|
@ -113,12 +137,14 @@ class TestOptInBirdFilter:
|
|||
_enable_satpass_db(norad_ids=[], dry_run=False)
|
||||
_clear_handler_flags()
|
||||
|
||||
# now near the 2026 envelope window so the AOS-horizon guard does
|
||||
# not short-circuit before the opt-in norad_ids check that logs.
|
||||
with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"):
|
||||
handle_satpass(_envelope(norad_id=25544, max_el=80.0),
|
||||
"central.sat.pass.iss", data={}, now=1718163120)
|
||||
"central.sat.pass.iss", data={}, now=1781235000)
|
||||
handle_satpass(_envelope(norad_id=25544, max_el=80.0,
|
||||
aos="2026-06-12T04:32:00Z"),
|
||||
"central.sat.pass.iss", data={}, now=1718163180)
|
||||
"central.sat.pass.iss", data={}, now=1781235000)
|
||||
|
||||
matching = [r for r in caplog.records
|
||||
if "no norad_ids configured" in r.message]
|
||||
|
|
@ -132,15 +158,15 @@ class TestOptInBirdFilter:
|
|||
|
||||
# ISS pass should air
|
||||
iss_env = _envelope(norad_id=25544, max_el=65.0)
|
||||
iss_result = handle_satpass(iss_env, "central.sat.pass.iss",
|
||||
data={}, now=1718163120)
|
||||
iss_result = _ingest_and_consolidate(iss_env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert iss_result is not None, "ISS pass should broadcast"
|
||||
|
||||
# NOAA-18 pass should be rejected
|
||||
# NOAA-18 pass should be rejected (opt-in norad_ids excludes it)
|
||||
noaa_env = _envelope(norad_id=28654, sat_name="NOAA 18", max_el=65.0,
|
||||
aos="2026-06-12T05:00:00Z")
|
||||
noaa_result = handle_satpass(noaa_env, "central.sat.pass.noaa",
|
||||
data={}, now=1718163120)
|
||||
noaa_result = _ingest_and_consolidate(noaa_env, "central.sat.pass.noaa",
|
||||
now=1781235000)
|
||||
assert noaa_result is None, "NOAA-18 pass should be rejected"
|
||||
|
||||
def test_dm_command_not_gated_by_norad_ids(self):
|
||||
|
|
@ -201,11 +227,15 @@ class TestRateCap:
|
|||
_enable_satpass_db(norad_ids=[25544], dry_run=False, max_per_hour=0)
|
||||
_clear_handler_flags()
|
||||
|
||||
now = 1718200000
|
||||
# now near the 2026 envelope window; max_per_hour=0 => the
|
||||
# consolidation step always trips the rate cap and logs.
|
||||
now = 1781258400 # 2026-06-12T10:00:00Z (just before this pass)
|
||||
with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"):
|
||||
env = _envelope(norad_id=25544, max_el=70.0,
|
||||
aos="2026-06-12T10:05:00Z")
|
||||
handle_satpass(env, "central.sat.pass.iss", data={}, now=now)
|
||||
aos="2026-06-12T10:05:00Z",
|
||||
los="2026-06-12T10:11:00Z")
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss", now=now)
|
||||
assert result is None, "rate cap should suppress the broadcast"
|
||||
|
||||
matching = [r for r in caplog.records if "rate cap reached" in r.message]
|
||||
assert len(matching) >= 1, "Rate cap suppression should be logged"
|
||||
|
|
@ -250,11 +280,14 @@ class TestDryRun:
|
|||
|
||||
with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"):
|
||||
env = _envelope(norad_id=25544, sat_name="ISS", max_el=70.0)
|
||||
handle_satpass(env, "central.sat.pass.iss", data={},
|
||||
now=1718163120)
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result is None, "dry_run should suppress dispatch"
|
||||
|
||||
# dry-run logging now happens in the consolidation step; the message
|
||||
# is "DRY-RUN would air (consolidated, N observers): <wire>".
|
||||
matching = [r for r in caplog.records
|
||||
if r.message.startswith("DRY-RUN would air:")]
|
||||
if r.message.startswith("DRY-RUN would air")]
|
||||
assert len(matching) == 1, "Should log DRY-RUN wire text once"
|
||||
assert "ISS" in matching[0].message
|
||||
|
||||
|
|
@ -264,11 +297,12 @@ class TestDryRun:
|
|||
_enable_satpass_db(norad_ids=[25544], dry_run=False)
|
||||
_clear_handler_flags()
|
||||
|
||||
data = {}
|
||||
env = _envelope(norad_id=25544, max_el=70.0)
|
||||
result = handle_satpass(env, "central.sat.pass.iss", data=data,
|
||||
now=1718163120)
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result is not None, "dry_run=False should dispatch"
|
||||
# commit callback now rides on the consolidation result's data dict.
|
||||
wire, data = result
|
||||
assert "_on_broadcast_committed" in data
|
||||
|
||||
def test_dry_run_default_is_true(self):
|
||||
|
|
@ -322,8 +356,13 @@ class TestBroadcastWireFormat:
|
|||
lines = wire.split("\n")
|
||||
assert len(lines) == 2
|
||||
|
||||
# No peak_compass passed here \u2192 the sweep stays aos\u2192los (SW\u2192NE).
|
||||
assert lines[0] == "\U0001F6F0\uFE0F ISS high pass, SW\u2192NE"
|
||||
assert lines[1] == "6 minute window, 8:38\u20138:44 PM MDT"
|
||||
# line2 renders "min window" + a date qualifier for a pass not
|
||||
# occurring today (the fixed 2026-06-12 date is always in the past
|
||||
# relative to run time).
|
||||
assert lines[1].startswith("6 min window, 8:38\u20138:44 PM MDT")
|
||||
assert lines[1] == "6 min window, 8:38\u20138:44 PM MDT Fri Jun 12"
|
||||
|
||||
def test_bucket_overhead_at_60(self):
|
||||
"""max_el=60 should be 'overhead'."""
|
||||
|
|
@ -464,7 +503,7 @@ class TestNoradIdTypeCoercion:
|
|||
_enable_satpass_db(norad_ids=["25544"], dry_run=False)
|
||||
|
||||
env = _envelope(norad_id=25544, max_el=80.0)
|
||||
result = handle_satpass(env, "test.subject", data={}, now=1781235000)
|
||||
result = _ingest_and_consolidate(env, "test.subject", now=1781235000)
|
||||
assert result is not None, "string norad_id should match int wire"
|
||||
|
||||
def test_mixed_int_and_string_norad_ids(self):
|
||||
|
|
@ -475,13 +514,13 @@ class TestNoradIdTypeCoercion:
|
|||
|
||||
# int in list, int on wire
|
||||
env_iss = _envelope(norad_id=25544, max_el=65.0)
|
||||
result_iss = handle_satpass(env_iss, "test.subject", data={}, now=1781235000)
|
||||
result_iss = _ingest_and_consolidate(env_iss, "test.subject", now=1781235000)
|
||||
assert result_iss is not None, "int norad_id in mixed list should match"
|
||||
|
||||
# string in list, int on wire
|
||||
env_noaa = _envelope(norad_id=22825, sat_name="NOAA 15", max_el=65.0,
|
||||
aos="2026-06-12T05:32:00Z", los="2026-06-12T05:38:00Z")
|
||||
result_noaa = handle_satpass(env_noaa, "test.subject", data={}, now=1781235000)
|
||||
result_noaa = _ingest_and_consolidate(env_noaa, "test.subject", now=1781235000)
|
||||
assert result_noaa is not None, "string norad_id in mixed list should match int wire"
|
||||
|
||||
def test_garbage_entries_skipped_without_crash(self):
|
||||
|
|
@ -493,7 +532,7 @@ class TestNoradIdTypeCoercion:
|
|||
|
||||
env = _envelope(norad_id=25544, max_el=80.0)
|
||||
# Must not raise, and the valid entry should still match
|
||||
result = handle_satpass(env, "test.subject", data={}, now=1781235000)
|
||||
result = _ingest_and_consolidate(env, "test.subject", now=1781235000)
|
||||
assert result is not None, "valid entry should match despite garbage siblings"
|
||||
|
||||
def test_all_garbage_norad_ids_matches_nothing(self):
|
||||
|
|
@ -513,7 +552,7 @@ class TestNoradIdTypeCoercion:
|
|||
_enable_satpass_db(norad_ids=[25544], dry_run=False)
|
||||
|
||||
env = _envelope(norad_id=25544, max_el=80.0)
|
||||
result = handle_satpass(env, "test.subject", data={}, now=1781235000)
|
||||
result = _ingest_and_consolidate(env, "test.subject", now=1781235000)
|
||||
assert result is not None, "pure int norad_id should still match"
|
||||
|
||||
def test_string_norad_id_rejects_non_matching(self):
|
||||
|
|
@ -567,7 +606,7 @@ class TestStalenessGuard:
|
|||
los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
env = _envelope(norad_id=25544, max_el=70.0, aos=aos, los=los)
|
||||
result = handle_satpass(env, "test.subject", data={}, now=now)
|
||||
result = _ingest_and_consolidate(env, "test.subject", now=now)
|
||||
assert result is not None, "ongoing pass (los in future) should broadcast"
|
||||
|
||||
def test_future_pass_broadcasts(self):
|
||||
|
|
@ -583,7 +622,7 @@ class TestStalenessGuard:
|
|||
los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
|
||||
env = _envelope(norad_id=25544, max_el=55.0, aos=aos, los=los)
|
||||
result = handle_satpass(env, "test.subject", data={}, now=now)
|
||||
result = _ingest_and_consolidate(env, "test.subject", now=now)
|
||||
assert result is not None, "future pass should broadcast"
|
||||
|
||||
def test_none_los_falls_through(self):
|
||||
|
|
@ -601,7 +640,7 @@ class TestStalenessGuard:
|
|||
# Patch los to None by removing los_time from inner data
|
||||
env["data"]["data"]["los_time"] = None
|
||||
|
||||
result = handle_satpass(env, "test.subject", data={}, now=now)
|
||||
result = _ingest_and_consolidate(env, "test.subject", now=now)
|
||||
# Should not be rejected by staleness guard — falls through to
|
||||
# normal handling (wire produced or other filter applies)
|
||||
# We just verify it does NOT crash and is not rejected as stale
|
||||
|
|
|
|||
|
|
@ -145,6 +145,27 @@ def _clear_handler_flags():
|
|||
delattr(handle_satpass, attr)
|
||||
|
||||
|
||||
def _ingest_and_consolidate(env, subject, *, now, data=None):
|
||||
"""Drive the two-call async satpass contract (ingest → consolidate).
|
||||
|
||||
handle_satpass() ingests the pass and returns None; the consumer then
|
||||
runs consolidate_satpass_pending(), which yields the (wire, data) to
|
||||
broadcast (or None). Returns the consolidation result so a test can
|
||||
assert on the real broadcast wire.
|
||||
"""
|
||||
from meshai.central.satpass_handler import (
|
||||
handle_satpass, consolidate_satpass_pending,
|
||||
drain_pending_consolidation_ids)
|
||||
drain_pending_consolidation_ids()
|
||||
assert handle_satpass(
|
||||
env, subject, data=data if data is not None else {}, now=now) is None
|
||||
for cid in drain_pending_consolidation_ids():
|
||||
res = consolidate_satpass_pending(cid)
|
||||
if res is not None:
|
||||
return res
|
||||
return None
|
||||
|
||||
|
||||
# ── (a) satpass_predict envelope: raw azimuths → non-empty compass ───
|
||||
|
||||
def test_satpass_predict_compass_from_raw_azimuths():
|
||||
|
|
@ -156,29 +177,26 @@ def test_satpass_predict_compass_from_raw_azimuths():
|
|||
_clear_handler_flags()
|
||||
|
||||
now = 1781330520 # well before the AO-27 pass window
|
||||
wire = handle_satpass(
|
||||
result = _ingest_and_consolidate(
|
||||
AO27_PREDICT_ENVELOPE,
|
||||
"central.sat.pass.us.id.filer",
|
||||
data={},
|
||||
now=now,
|
||||
)
|
||||
assert wire is not None, "handler returned None for satpass_predict envelope"
|
||||
assert result is not None, "handler returned None for satpass_predict envelope"
|
||||
wire, _ = result
|
||||
|
||||
lines = wire.split("\n")
|
||||
assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}"
|
||||
|
||||
# Line 1 must contain non-empty compass directions
|
||||
# 163.2° → S (8-point compass), 348.7° → N
|
||||
assert "S" in lines[0].split(",")[-1], f"Expected S (from 163.2°) in compass portion: {lines[0]!r}"
|
||||
assert "\u2192" in lines[0], f"Expected → arrow in line 1: {lines[0]!r}"
|
||||
|
||||
# Extract the compass portion: after bucket comma, before newline
|
||||
# Format: 🛰️ {name} {bucket}, {aos_compass}→{los_compass}
|
||||
arrow_idx = lines[0].index("\u2192")
|
||||
los_part = lines[0][arrow_idx + 1:]
|
||||
assert los_part != "", f"los_compass is empty in line 1: {lines[0]!r}"
|
||||
# 348.7° → N in 8-point compass
|
||||
assert los_part == "N", f"Expected N (from 348.7°) after arrow: {los_part!r}"
|
||||
# Line 1 format: name bucket, aos→peak→los. Raw azimuths convert
|
||||
# to a non-empty 3-point compass sweep.
|
||||
compass = lines[0].split(", ")[-1]
|
||||
parts = compass.split("\u2192")
|
||||
assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}"
|
||||
# 163.2 -> S, 245.0 -> SW (peak), 348.7 -> N (8-point compass)
|
||||
assert parts[0] == "S", f"Expected S (from 163.2): {parts!r}"
|
||||
assert parts[1] == "SW", f"Expected SW (from peak 245.0): {parts!r}"
|
||||
assert parts[2] == "N", f"Expected N (from 348.7): {parts!r}"
|
||||
|
||||
|
||||
# ── (b) n2yo envelope: precomputed _compass strings used as-is ───────
|
||||
|
|
@ -192,21 +210,22 @@ def test_n2yo_precomputed_compass_unchanged():
|
|||
_clear_handler_flags()
|
||||
|
||||
now = 1781065800 # before NOAA-18 pass window
|
||||
wire = handle_satpass(
|
||||
result = _ingest_and_consolidate(
|
||||
N2YO_ENVELOPE,
|
||||
"central.sat.pass.us.id.filer",
|
||||
data={},
|
||||
now=now,
|
||||
)
|
||||
assert wire is not None, "handler returned None for n2yo envelope"
|
||||
assert result is not None, "handler returned None for n2yo envelope"
|
||||
wire, _ = result
|
||||
|
||||
lines = wire.split("\n")
|
||||
# Must use the precomputed strings: SE→N
|
||||
assert "SE" in lines[0], f"Expected SE from precomputed _compass: {lines[0]!r}"
|
||||
# The los_compass from precomputed is "N"
|
||||
arrow_idx = lines[0].index("\u2192")
|
||||
los_part = lines[0][arrow_idx + 1:]
|
||||
assert los_part == "N", f"Expected precomputed los_compass N, got {los_part!r}"
|
||||
# Must use the precomputed strings verbatim: SE→ENE→N
|
||||
compass = lines[0].split(", ")[-1]
|
||||
parts = compass.split("\u2192")
|
||||
assert len(parts) == 3, f"Expected aos->peak->los sweep, got {parts!r}"
|
||||
assert parts[0] == "SE", f"Expected precomputed aos SE: {parts!r}"
|
||||
assert parts[1] == "ENE", f"Expected precomputed peak ENE: {parts!r}"
|
||||
assert parts[2] == "N", f"Expected precomputed los N: {parts!r}"
|
||||
|
||||
|
||||
# ── (c) envelope with neither → empty compass, no crash ──────────────
|
||||
|
|
@ -228,16 +247,16 @@ def test_no_compass_no_azimuth_no_crash():
|
|||
d.pop(key, None)
|
||||
|
||||
now = 1781330520
|
||||
wire = handle_satpass(
|
||||
result = _ingest_and_consolidate(
|
||||
env,
|
||||
"central.sat.pass.us.id.filer",
|
||||
data={},
|
||||
now=now,
|
||||
)
|
||||
assert wire is not None, "handler crashed or returned None — should produce wire with empty compass"
|
||||
assert result is not None, "handler crashed or returned None — should produce wire with empty compass"
|
||||
wire, _ = result
|
||||
|
||||
lines = wire.split("\n")
|
||||
assert len(lines) == 2
|
||||
# Arrow should still be present with empty directions: "→" or similar
|
||||
assert "\u2192" in lines[0], f"Expected → in line 1 even with empty compass: {lines[0]!r}"
|
||||
# Arrow still present even with empty (peak empty -> no peak segment):
|
||||
assert "\u2192" in lines[0], f"Expected arrow in line 1 even with empty compass: {lines[0]!r}"
|
||||
# No crash = test passes
|
||||
|
|
|
|||
|
|
@ -53,6 +53,7 @@ def mock_adapter_config():
|
|||
cfg.norad_ids = [25544] # must be non-empty for opt-in
|
||||
cfg.dry_run = False
|
||||
cfg.max_broadcasts_per_hour = 4
|
||||
cfg.max_aos_horizon_hours = 24
|
||||
with patch("meshai.central.satpass_handler.adapter_config") as mock:
|
||||
mock.satpass = cfg
|
||||
from meshai.central.satpass_handler import handle_satpass
|
||||
|
|
@ -63,18 +64,38 @@ def mock_adapter_config():
|
|||
yield cfg
|
||||
|
||||
|
||||
def _ingest_and_consolidate(env, subject, *, now, data=None):
|
||||
"""Drive the two-call async satpass contract (ingest -> consolidate).
|
||||
|
||||
handle_satpass() ingests the pass into satpass_pending and ALWAYS
|
||||
returns None; the consumer then runs consolidate_satpass_pending(),
|
||||
which returns the (wire, data) to broadcast (or None if suppressed).
|
||||
Returns the consolidation result so a test can assert on the real wire.
|
||||
"""
|
||||
from meshai.central.satpass_handler import (
|
||||
handle_satpass, consolidate_satpass_pending,
|
||||
drain_pending_consolidation_ids)
|
||||
drain_pending_consolidation_ids()
|
||||
assert handle_satpass(
|
||||
env, subject, data=data if data is not None else {}, now=now) is None
|
||||
for cid in drain_pending_consolidation_ids():
|
||||
res = consolidate_satpass_pending(cid)
|
||||
if res is not None:
|
||||
return res
|
||||
return None
|
||||
|
||||
|
||||
class TestSatpassHandler:
|
||||
"""Tests for handle_satpass function."""
|
||||
|
||||
def test_high_elevation_pass_broadcasts(self, mock_db, mock_adapter_config):
|
||||
"""A pass with high elevation should broadcast."""
|
||||
from meshai.central.satpass_handler import handle_satpass
|
||||
|
||||
def test_high_elevation_pass_broadcasts(self, mock_adapter_config):
|
||||
"""A pass with high elevation should broadcast (via consolidation)."""
|
||||
env = _envelope(max_el=75.0)
|
||||
result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
|
||||
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result is not None
|
||||
assert "ISS" in result
|
||||
wire, _ = result
|
||||
assert "ISS" in wire
|
||||
|
||||
def test_low_elevation_pass_filtered(self, mock_db, mock_adapter_config):
|
||||
"""A pass below min_elevation should be filtered."""
|
||||
|
|
@ -96,14 +117,12 @@ class TestSatpassHandler:
|
|||
|
||||
assert result is None
|
||||
|
||||
def test_observer_filter_allows_match(self, mock_db, mock_adapter_config):
|
||||
def test_observer_filter_allows_match(self, mock_adapter_config):
|
||||
"""A pass for configured observer should broadcast."""
|
||||
from meshai.central.satpass_handler import handle_satpass
|
||||
|
||||
mock_adapter_config.observers = ["Boise", "Magic Valley"]
|
||||
env = _envelope(observer="Boise", max_el=45.0)
|
||||
result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
|
||||
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result is not None
|
||||
|
||||
def test_norad_id_filter(self, mock_db, mock_adapter_config):
|
||||
|
|
@ -116,48 +135,49 @@ class TestSatpassHandler:
|
|||
|
||||
assert result is None
|
||||
|
||||
def test_dedup_blocks_second_broadcast(self, mock_db, mock_adapter_config):
|
||||
def test_dedup_blocks_second_broadcast(self, mock_adapter_config):
|
||||
"""Second pass in same hour bucket should be deduplicated."""
|
||||
from meshai.central.satpass_handler import handle_satpass
|
||||
|
||||
# First call returns no existing broadcast
|
||||
mock_db.execute.return_value.fetchone.return_value = None
|
||||
|
||||
env = _envelope(max_el=60.0)
|
||||
result1 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
|
||||
|
||||
# First pass consolidates into a real broadcast.
|
||||
result1 = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result1 is not None
|
||||
|
||||
# Second call simulates existing broadcast
|
||||
mock_db.execute.return_value.fetchone.return_value = {"last_broadcast_at": 1718163120}
|
||||
# Simulate the broadcast committing (marks satpass_events broadcast).
|
||||
_wire, data1 = result1
|
||||
data1["_on_broadcast_committed"](1781235010)
|
||||
|
||||
result2 = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163180)
|
||||
# Second pass in the same (norad, aos-hour) bucket must be deduped.
|
||||
result2 = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235060)
|
||||
assert result2 is None
|
||||
|
||||
def test_wire_format(self, mock_db, mock_adapter_config):
|
||||
def test_wire_format(self, mock_adapter_config):
|
||||
"""Wire format should have 2 lines with correct info."""
|
||||
from meshai.central.satpass_handler import handle_satpass
|
||||
|
||||
env = _envelope(sat_name="ISS", max_el=75, observer="Boise",
|
||||
direction="NW-SE", aos_compass="SW", los_compass="NE")
|
||||
result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
|
||||
direction="S", aos_compass="SW", los_compass="NE")
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result is not None
|
||||
wire, _ = result
|
||||
|
||||
lines = result.split("\n")
|
||||
lines = wire.split("\n")
|
||||
assert len(lines) == 2
|
||||
assert "ISS" in lines[0]
|
||||
assert "overhead" in lines[0]
|
||||
assert "SW" in lines[0]
|
||||
assert "NE" in lines[0]
|
||||
assert "minute window" in lines[1]
|
||||
assert "SW" in lines[0] # aos_compass
|
||||
assert "S" in lines[0] # peak_compass (new)
|
||||
assert "NE" in lines[0] # los_compass
|
||||
assert "min window" in lines[1]
|
||||
|
||||
def test_commit_callback_attached(self, mock_db, mock_adapter_config):
|
||||
"""Broadcast should attach commit callback."""
|
||||
from meshai.central.satpass_handler import handle_satpass
|
||||
|
||||
data = {}
|
||||
def test_commit_callback_attached(self, mock_adapter_config):
|
||||
"""Broadcast should attach commit callback (on the consolidation data)."""
|
||||
env = _envelope(max_el=60.0)
|
||||
result = handle_satpass(env, "central.sat.pass.iss", data=data, now=1718163120)
|
||||
|
||||
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
|
||||
now=1781235000)
|
||||
assert result is not None
|
||||
_wire, data = result
|
||||
assert "_on_broadcast_committed" in data
|
||||
assert "_broadcast_audit" in data
|
||||
assert data["_broadcast_audit"]["table"] == "satpass_events"
|
||||
|
|
|
|||
|
|
@ -96,6 +96,26 @@ def _enable_satpass(norad_ids=None):
|
|||
invalidate_cache()
|
||||
|
||||
|
||||
def _ingest_and_consolidate(env, subject, *, now, data=None):
|
||||
"""Drive the two-call async satpass contract (ingest -> consolidate).
|
||||
|
||||
handle_satpass() ingests the pass and returns None; the consumer then
|
||||
runs consolidate_satpass_pending(), which returns the (wire, data) to
|
||||
broadcast (or None). Returns the consolidation result.
|
||||
"""
|
||||
from meshai.central.satpass_handler import (
|
||||
handle_satpass, consolidate_satpass_pending,
|
||||
drain_pending_consolidation_ids)
|
||||
drain_pending_consolidation_ids()
|
||||
assert handle_satpass(
|
||||
env, subject, data=data if data is not None else {}, now=now) is None
|
||||
for cid in drain_pending_consolidation_ids():
|
||||
res = consolidate_satpass_pending(cid)
|
||||
if res is not None:
|
||||
return res
|
||||
return None
|
||||
|
||||
|
||||
# ── Handler produces correct satpass_events row ─────────────────────
|
||||
|
||||
def test_noaa18_envelope_produces_satpass_event():
|
||||
|
|
@ -110,13 +130,13 @@ def test_noaa18_envelope_produces_satpass_event():
|
|||
del handle_satpass._no_norad_ids_logged
|
||||
|
||||
now = 1781065800 # before NOAA18 envelope los_time
|
||||
wire = handle_satpass(
|
||||
result = _ingest_and_consolidate(
|
||||
NOAA18_ENVELOPE,
|
||||
"central.sat.pass.us.id.filer",
|
||||
data={},
|
||||
now=now,
|
||||
)
|
||||
assert wire is not None, "handler returned None -- field extraction failed"
|
||||
assert result is not None, "handler returned None -- field extraction failed"
|
||||
wire, _ = result
|
||||
|
||||
conn = get_db()
|
||||
rows = conn.execute(
|
||||
|
|
@ -147,25 +167,26 @@ def test_noaa18_wire_message_format():
|
|||
if hasattr(handle_satpass, "_no_norad_ids_logged"):
|
||||
del handle_satpass._no_norad_ids_logged
|
||||
|
||||
wire = handle_satpass(
|
||||
result = _ingest_and_consolidate(
|
||||
NOAA18_ENVELOPE,
|
||||
"central.sat.pass.us.id.filer",
|
||||
data={},
|
||||
now=1781065800, # before NOAA18 envelope los_time
|
||||
)
|
||||
assert wire is not None
|
||||
assert result is not None
|
||||
wire, _ = result
|
||||
|
||||
lines = wire.split("\n")
|
||||
assert len(lines) == 2, f"Expected 2 lines, got {len(lines)}: {wire!r}"
|
||||
|
||||
# Line 1: satellite name + bucket + compass directions
|
||||
# Line 1: satellite name + bucket + compass sweep (aos->peak->los)
|
||||
assert "NOAA 18" in lines[0]
|
||||
assert "low pass" in lines[0] # 22.69 < 30 = low pass
|
||||
assert "SE" in lines[0] # aos_compass
|
||||
assert "N" in lines[0] # los_compass
|
||||
assert "SE" in lines[0] # aos_compass
|
||||
assert "ENE" in lines[0] # peak_compass (new)
|
||||
assert "N" in lines[0] # los_compass
|
||||
|
||||
# Line 2: duration + time window
|
||||
assert "minute window" in lines[1]
|
||||
# Line 2: duration + time window ("min window")
|
||||
assert "min window" in lines[1]
|
||||
|
||||
|
||||
def test_missing_norad_id_rejected():
|
||||
|
|
@ -230,13 +251,12 @@ def test_observer_fallback_to_slug():
|
|||
del env["data"]["data"]["observer_name"]
|
||||
# observer_slug = "filer" still present
|
||||
|
||||
wire = handle_satpass(
|
||||
result = _ingest_and_consolidate(
|
||||
env,
|
||||
"central.sat.pass.us.id.filer",
|
||||
data={},
|
||||
now=1781065800, # before NOAA18 envelope los_time
|
||||
)
|
||||
assert wire is not None
|
||||
assert result is not None
|
||||
# Observer name stored in DB, not in broadcast wire format
|
||||
from meshai.persistence import get_db
|
||||
conn = get_db()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue