matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

fix(satpass): clean broadcast format (short names, degrees, compass, friendly observers) (#76)

Browse source 
Rewrite the satellite-pass wire to a single clean line: short ham names
(ISS/AO-27/AO-91), numeric max elevation (max 77°) instead of a bucket word,
collapsed compass sweeps (no E→E→E), and friendly observer names — dropping
the meaningless synthetic coverage_center parenthetical (and no longer seeding
that observer when explicit observers are configured). Absolute local time
kept for the 12h-advance heads-up.

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:
malice 2026-07-06 20:44:02 -06:00 • committed by GitHub
commit 8d3f96857f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 276 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

119
work/tests/test_satpass_broadcast_safety.py
View file

@ -333,12 +333,11 @@ class TestElevationDefault:
class TestBroadcastWireFormat:
"""Two-line format, buckets, byte budget."""
def test_exact_two_line_example(self):
"""Formatter produces the exact two-line example from spec."""
def test_exact_single_line_example(self):
"""Formatter produces the exact single-line target format."""
from meshai.central.satpass_handler import format_pass
# ISS high pass, SW→NE, 6 minute window, 8:38–8:44 PM MDT
# We need epoch values that produce 8:38 PM and 8:44 PM MDT
# ISS, SW->NE, 6-minute window, rises 8:38 PM MDT, max 55 deg.
from zoneinfo import ZoneInfo
tz = ZoneInfo("America/Boise")
aos_dt = datetime(2026, 6, 12, 20, 38, 0, tzinfo=tz)
@ -347,22 +346,19 @@ class TestBroadcastWireFormat:
los_epoch = int(los_dt.timestamp())
wire = format_pass(
sat_name="ISS", max_el=55.0,
sat_name="ISS", norad_id=25544, max_el=55.0,
aos_epoch=aos_epoch, los_epoch=los_epoch,
aos_compass="SW", los_compass="NE",
broadcast=True,
)
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"
# 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"
# Single line: absolute local rise time, numeric max elevation, and a
# date qualifier ("Fri Jun 12" \u2014 the fixed date is always in the past).
assert "\n" not in wire
assert wire == (
"\U0001F6F0\uFE0F ISS 8:38 PM MDT Fri Jun 12, "
"max 55\u00B0 SW\u2192NE (6 min)"
)
def test_bucket_overhead_at_60(self):
"""max_el=60 should be 'overhead'."""
@ -648,3 +644,96 @@ class TestStalenessGuard:
# but the staleness guard specifically must not block it.
# Since all other filters pass, wire should be produced.
assert result is not None, "None los_epoch should fall through staleness guard"
# ══════════════════════════════════════════════════════════════════════
# 6. CLEAN FORMAT: short names, degrees, compass collapse, friendly obs
# ══════════════════════════════════════════════════════════════════════
class TestCleanBroadcastFormat:
"""The format-cleanup rules (short names, degrees, compass, observers)."""
@staticmethod
def _wire(**kw):
from meshai.central.satpass_handler import format_pass
from zoneinfo import ZoneInfo
tz = ZoneInfo("America/Boise")
base = dict(
sat_name="X", max_el=50.0,
aos_epoch=int(datetime(2026, 6, 12, 20, 38, 0, tzinfo=tz).timestamp()),
los_epoch=int(datetime(2026, 6, 12, 20, 44, 0, tzinfo=tz).timestamp()),
aos_compass="S", los_compass="N", broadcast=True,
)
base.update(kw)
return format_pass(**base)
def test_short_name_ao91_from_norad(self):
wire = self._wire(norad_id=43017, sat_name="RADFXSAT (FOX-1B)")
assert "AO-91" in wire
assert "RADFXSAT" not in wire
assert "FOX-1B" not in wire
def test_short_name_iss_and_ao27(self):
assert "ISS" in self._wire(norad_id=25544, sat_name="ISS (ZARYA)")
assert "AO-27" in self._wire(norad_id=22825, sat_name="EYESAT-1 (AO-27)")
def test_short_name_substring_fallback(self):
# No NORAD mapping, but the catalog name is recognizable.
wire = self._wire(norad_id=99999, sat_name="RADFXSAT (FOX-1B)")
assert "AO-91" in wire
def test_unmapped_name_is_cleaned(self):
# Parenthetical stripped for an unmapped satellite.
wire = self._wire(norad_id=40000, sat_name="METEOR-M2 (WEATHER)")
assert "METEOR-M2" in wire
assert "(WEATHER)" not in wire
def test_compass_collapse_all_equal(self):
wire = self._wire(aos_compass="E", peak_compass="E", los_compass="E")
assert "E→E→E" not in wire
assert " E " in wire # a lone collapsed "E"
def test_compass_collapse_trailing_dup(self):
wire = self._wire(aos_compass="E", peak_compass="SE", los_compass="SE")
assert "E→SE" in wire
assert "E→SE→SE" not in wire
def test_compass_no_collapse_when_distinct(self):
wire = self._wire(aos_compass="S", peak_compass="W", los_compass="NW")
assert "S→W→NW" in wire
def test_shows_numeric_degrees_not_bucket(self):
wire = self._wire(max_el=77.0)
assert "max 77°" in wire
assert "high pass" not in wire
assert "overhead" not in wire
def test_multi_observer_region_shown(self):
wire = self._wire(entry_observer="Treasure Valley",
exit_observer="Magic Valley")
assert "(Treasure Valley→Magic Valley)" in wire
def test_single_observer_no_region(self):
wire = self._wire(entry_observer="Boise", exit_observer="Boise")
assert "(" not in wire.split("min)")[-1] # nothing after the (N min)
def test_coverage_center_region_dropped(self):
wire = self._wire(entry_observer="coverage_center",
exit_observer="Magic Valley")
assert "coverage_center" not in wire
assert "Coverage Center" not in wire
# Also the friendly synthetic label form.
wire2 = self._wire(entry_observer="Coverage Center",
exit_observer="Magic Valley")
assert "Coverage Center" not in wire2
def test_golden_line(self):
# A full golden line matching the target format for a sample pass.
wire = self._wire(
norad_id=25544, sat_name="ISS (ZARYA)", max_el=77.0,
aos_compass="S", peak_compass=None, los_compass="NW",
)
assert wire == (
"\U0001F6F0️ ISS 8:38 PM MDT Fri Jun 12, "
"max 77° S→NW (6 min)"
)

27
work/tests/test_satpass_compass_fallback.py
View file

@ -185,12 +185,9 @@ def test_satpass_predict_compass_from_raw_azimuths():
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 format: name bucket, aos→peak→los. Raw azimuths convert
# to a non-empty 3-point compass sweep.
compass = lines[0].split(", ")[-1]
# Single-line wire: extract the compass segment between "max NN° " and " (".
assert "\n" not in wire, f"Expected single line: {wire!r}"
compass = wire.split("° ", 1)[1].split(" (", 1)[0]
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)
@ -218,9 +215,9 @@ def test_n2yo_precomputed_compass_unchanged():
assert result is not None, "handler returned None for n2yo envelope"
wire, _ = result
lines = wire.split("\n")
# Must use the precomputed strings verbatim: SE→ENE→N
compass = lines[0].split(", ")[-1]
# Single-line wire: must use the precomputed strings verbatim: SE→ENE→N.
assert "\n" not in wire, f"Expected single line: {wire!r}"
compass = wire.split("° ", 1)[1].split(" (", 1)[0]
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}"
@ -255,8 +252,12 @@ def test_no_compass_no_azimuth_no_crash():
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 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}"
# Single clean line; with no azimuth data the compass segment is simply
# omitted (no arrow, no stray double space) and the wire still renders.
assert "\n" not in wire, f"Expected single line: {wire!r}"
assert wire.startswith("\U0001F6F0")
assert "max" in wire
assert "min)" in wire
assert "\u2192" not in wire # empty compass -> no sweep arrows
assert "\u00b0 (" not in wire # no stray double space where compass would be
# No crash = test passes

16
work/tests/test_satpass_handler.py
View file

@ -154,7 +154,7 @@ class TestSatpassHandler:
assert result2 is None
def test_wire_format(self, mock_adapter_config):
"""Wire format should have 2 lines with correct info."""
"""Wire is a single clean line: short name, degrees, collapsed compass."""
env = _envelope(sat_name="ISS", max_el=75, observer="Boise",
direction="S", aos_compass="SW", los_compass="NE")
result = _ingest_and_consolidate(env, "central.sat.pass.iss",
@ -162,14 +162,12 @@ class TestSatpassHandler:
assert result is not None
wire, _ = result
lines = wire.split("\n")
assert len(lines) == 2
assert "ISS" in lines[0]
assert "overhead" in lines[0]
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]
assert "\n" not in wire # single line now
assert "ISS" in wire
assert "max 75°" in wire # numeric elevation, not a bucket word
assert "overhead" not in wire
assert "min window" not in wire
assert "SW→S→NE" in wire # aos -> peak -> los, collapsed
def test_commit_callback_attached(self, mock_adapter_config):
"""Broadcast should attach commit callback (on the consolidation data)."""

4
work/tests/test_satpass_native.py
View file

@ -136,9 +136,9 @@ def test_two_observers_consolidate_to_one_broadcast(monkeypatch):
assert row["los_at"] == T0 + 400 # twin (latest LOS)
assert "boise" in row["observer"] and "twin" in row["observer"]
# Wire carries entry->exit region + aos->peak->los compass sweep.
# Wire carries entry->exit region (FRIENDLY names) + aos->peak->los sweep.
wire = evt["wire"]
assert "boise→twin" in wire # entry -> exit
assert "(Boise→Twin Falls)" in wire # friendly entry -> exit
assert "SW→S→NE" in wire # aos -> peak(twin) -> los

19
work/tests/test_satpass_wire_fields.py
View file

@ -175,18 +175,13 @@ def test_noaa18_wire_message_format():
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 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 "ENE" in lines[0] # peak_compass (new)
assert "N" in lines[0] # los_compass
# Line 2: duration + time window ("min window")
assert "min window" in lines[1]
# Single clean line: name, numeric elevation, aos->peak->los compass sweep.
assert "\n" not in wire, f"Expected single line: {wire!r}"
assert "NOAA 18" in wire # no mapping -> cleaned catalog name
assert "max 23°" in wire # 22.69 rounds to 23, not "low pass"
assert "low pass" not in wire
assert "min window" not in wire
assert "SE→ENE→N" in wire # aos -> peak -> los
def test_missing_norad_id_rejected():