From 8d3f96857f62f7cc63cd878c3165ff1f1844d35c Mon Sep 17 00:00:00 2001 From: malice Date: Mon, 6 Jul 2026 20:44:02 -0600 Subject: [PATCH] fix(satpass): clean broadcast format (short names, degrees, compass, friendly observers) (#76) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/central/satpass_handler.py | 168 +++++++++++++++----- work/meshai/env/satpass.py | 7 +- work/meshai/main.py | 10 +- work/tests/test_satpass_broadcast_safety.py | 119 ++++++++++++-- work/tests/test_satpass_compass_fallback.py | 27 ++-- work/tests/test_satpass_handler.py | 16 +- work/tests/test_satpass_native.py | 4 +- work/tests/test_satpass_wire_fields.py | 19 +-- 8 files changed, 275 insertions(+), 95 deletions(-) diff --git a/work/meshai/central/satpass_handler.py b/work/meshai/central/satpass_handler.py index efdf546..6fdfea5 100644 --- a/work/meshai/central/satpass_handler.py +++ b/work/meshai/central/satpass_handler.py @@ -22,13 +22,13 @@ Severity mapping: 3 = priority (>= 45 deg max elevation) <= 2 = routine -Broadcast wire format (two lines, LoRa-tight): - Consolidated (multi-observer): - Line 1: πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} - Line 2: {duration} min window, {rise}–{set} {AM/PM} MDT ({entry_obs}β†’{exit_obs}) - Single observer: - Line 1: πŸ›°οΈ {name} {bucket}, {aos_compass}β†’{los_compass} - Line 2: {duration} min window, {rise}–{set} {AM/PM} MDT +Broadcast wire format (single line, LoRa-tight, absolute local time β€” a +~12h-ahead heads-up): + πŸ›°οΈ {short_name} {rise} {AM/PM} {TZ}[ tomorrow], max {el}Β° {compass} ({dur} min)[ (region)] + - short_name: short ham designation (ISS/AO-27/AO-91), else cleaned catalog + - compass: aosβ†’peakβ†’los with consecutive duplicates collapsed (no Eβ†’Eβ†’E) + - region: appended ONLY for a genuine multi-observer sweep with different + friendly names; dropped for a single observer or the synthetic coverage_center DM wire format (compact, exact degrees): {name} {HH:MM}–{HH:MM} {TZ} max {el}Β° {aos_compass}β†’{los_compass} """ @@ -36,6 +36,7 @@ from __future__ import annotations import json import logging +import re import time from datetime import datetime, timezone from typing import Any, Optional @@ -113,7 +114,11 @@ def _parse_iso_epoch(s) -> Optional[int]: def _elevation_bucket(max_el: float) -> str: - """Map max elevation to human-readable bucket name.""" + """Map max elevation to human-readable bucket name. + + Retained for the DM/other paths; the broadcast wire now shows numeric + degrees (`max NNΒ°`) instead of a bucket word. + """ if max_el >= 60: return "overhead" if max_el >= 30: @@ -121,6 +126,88 @@ def _elevation_bucket(max_el: float) -> str: return "low pass" +# Short ham designations for common broadcast satellites, keyed by NORAD id. +# Listeners recognize "AO-91" far faster than the cluttered catalog name +# "RADFXSAT (FOX-1B)". +_SHORT_SAT_NAMES = { + 25544: "ISS", + 22825: "AO-27", + 43017: "AO-91", +} + +# Name-substring fallback (upper-cased contains) for when the NORAD id isn't +# in the map but the catalog name is recognizable. +_SHORT_NAME_SUBSTR = ( + ("ZARYA", "ISS"), + ("EYESAT", "AO-27"), + ("AO-27", "AO-27"), + ("RADFXSAT", "AO-91"), + ("FOX-1B", "AO-91"), +) + + +def _short_sat_name(norad_id: Optional[int], sat_name: Optional[str]) -> str: + """Resolve a short, listener-friendly satellite name. + + NORAD-id map first, then a name-substring fallback, then a cleaned + catalog name (parenthetical stripped, e.g. "RADFXSAT (FOX-1B)" -> + "RADFXSAT"). Always returns a non-empty string. + """ + nid: Optional[int] = None + if norad_id is not None: + try: + nid = int(norad_id) + except (TypeError, ValueError): + nid = None + if nid is not None and nid in _SHORT_SAT_NAMES: + return _SHORT_SAT_NAMES[nid] + + up = (sat_name or "").upper() + for sub, short in _SHORT_NAME_SUBSTR: + if sub in up: + return short + + cleaned = re.sub(r"\s*\(.*?\)", "", sat_name or "").strip() + return cleaned or (sat_name or "").strip() or "SAT" + + +def _collapse_compass(*points: Optional[str]) -> str: + """Join compass points, dropping empties and consecutive duplicates. + + "E","E","E" -> "E"; "E","SE","SE" -> "Eβ†’SE"; "S","W","NW" -> "Sβ†’Wβ†’NW". + """ + out: list[str] = [] + for p in points: + if not p: + continue + if not out or out[-1] != p: + out.append(p) + return "β†’".join(out) + + +# Synthetic coverage-centroid observer markers β€” these are meaningless to +# listeners, so the region parenthetical is dropped when either endpoint is one. +_SYNTHETIC_OBSERVERS = {"coverage_center", "coverage center"} + + +def _is_synthetic_observer(label: Optional[str]) -> bool: + return bool(label) and label.strip().lower() in _SYNTHETIC_OBSERVERS + + +def _region_paren(entry: Optional[str], exit_: Optional[str]) -> str: + """Region suffix for a genuine multi-observer sweep, else ''. + + Only rendered when both endpoints exist, differ, and neither is the + synthetic coverage-centroid observer. Single observer / synthetic -> + no parenthetical. + """ + if not entry or not exit_ or entry == exit_: + return "" + if _is_synthetic_observer(entry) or _is_synthetic_observer(exit_): + return "" + return f" ({entry}β†’{exit_})" + + def _format_time_12h(epoch: Optional[int]) -> str: """Format epoch to h:mm AM/PM in America/Boise.""" if epoch is None: @@ -204,54 +291,53 @@ def format_pass(*, sat_name: str, max_el: float, broadcast: bool = True, entry_observer: Optional[str] = None, exit_observer: Optional[str] = None, - peak_compass: Optional[str] = None) -> str: + peak_compass: Optional[str] = None, + norad_id: Optional[int] = None) -> str: """Unified pass formatter with mode switch. - broadcast=True: Two-line format with buckets, 12h times, LoRa budget. - πŸ›°οΈ {name} {bucket}, {aos_compass}β†’[peak_compassβ†’]{los_compass} - {duration} min window, {rise}–{set} {AM/PM} {TZ} [tomorrow] [(region)] + broadcast=True: Single clean line, absolute local time (a ~12h-ahead + heads-up), LoRa budget. + πŸ›°οΈ {short_name} {rise} {AM/PM} {TZ}[ tomorrow], max {el}Β° {compass} ({dur} min)[ (region)] broadcast=False: Compact DM format with exact degrees. {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). + peak_compass: compass direction at peak elevation. Threaded through the + compass sweep (aosβ†’peakβ†’los); consecutive duplicate points are + collapsed so a degenerate "Eβ†’Eβ†’E" renders as "E". + + norad_id: used to resolve the short ham name for the broadcast wire. """ - # Compass sweep segment: include the peak point only when supplied. - if peak_compass: - compass_seg = f"{aos_compass}β†’{peak_compass}β†’{los_compass}" + # Compass sweep segment: aos->peak->los with consecutive duplicates dropped. + compass_seg = _collapse_compass(aos_compass, peak_compass, los_compass) + + # Duration in whole minutes + if aos_epoch is not None and los_epoch is not None: + dur_min = max(1, round((los_epoch - aos_epoch) / 60)) else: - compass_seg = f"{aos_compass}β†’{los_compass}" + dur_min = 0 if broadcast: - bucket = _elevation_bucket(max_el) - # Duration in whole minutes - if aos_epoch is not None and los_epoch is not None: - dur_min = max(1, round((los_epoch - aos_epoch) / 60)) - else: - dur_min = 0 + name = _short_sat_name(norad_id, sat_name) rise_str = _format_time_12h(aos_epoch) - set_str = _format_time_12h(los_epoch) - ampm = _format_ampm(los_epoch) + ampm = _format_ampm(aos_epoch) tz = _tz_abbr(aos_epoch) date_lbl = _date_label(aos_epoch) + el = int(round(max_el)) if max_el is not None else 0 + region = _region_paren(entry_observer, exit_observer) - 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}" - - # Region parenthetical: multi-observer sweep or single-observer location - if entry_observer and exit_observer and entry_observer != exit_observer: - line2 = f"{time_part} ({entry_observer}\u2192{exit_observer})" - elif entry_observer: - line2 = f"{time_part} ({entry_observer})" - else: - line2 = time_part + # Elevation + compass; the compass may be empty when no azimuth data + # was available, in which case it is simply omitted (no stray space). + core = f"max {el}\u00B0" + if compass_seg: + core += f" {compass_seg}" + line = ( + f"\U0001F6F0\uFE0F {name} {rise_str} {ampm} {tz}{date_lbl}, " + f"{core} ({dur_min} min){region}" + ) # Safety cap: fit the broadcast string to the mesh packet budget. - return fit_to_budget(f"{line1}\n{line2}", budget_for("satpass")) + return fit_to_budget(line, budget_for("satpass")) else: # DM format: compact with exact degrees aos_str = _format_time_24h(aos_epoch) @@ -500,7 +586,7 @@ def gate_consolidated_pass(consolidated: dict, *, return None # Build consolidated wire β€” always pass observer names for region context - wire = format_pass(sat_name=sat_name, max_el=max_el, + wire = format_pass(sat_name=sat_name, max_el=max_el, norad_id=norad_id, aos_epoch=aos_epoch, los_epoch=los_epoch, aos_compass=aos_compass, los_compass=los_compass, peak_compass=peak_compass, diff --git a/work/meshai/env/satpass.py b/work/meshai/env/satpass.py index 1cdd691..1dd6ceb 100644 --- a/work/meshai/env/satpass.py +++ b/work/meshai/env/satpass.py @@ -159,8 +159,10 @@ class SatpassAdapter: "aos_compass": entry["aos_compass"], "los_compass": exit_["los_compass"], "peak_compass": best["peak_compass"], - "entry_observer": entry["observer"], - "exit_observer": exit_["observer"], + # Entry/exit carry the FRIENDLY observer name for the wire's region + # parenthetical; observer_list stays slugs for the audit column. + "entry_observer": entry.get("observer_label") or entry["observer"], + "exit_observer": exit_.get("observer_label") or exit_["observer"], "observer_list": ",".join(r["observer"] for r in by_aos), } @@ -208,6 +210,7 @@ class SatpassAdapter: "norad_id": norad, "sat_name": name, "observer": obs["slug"], + "observer_label": obs.get("name") or obs["slug"], "max_elevation": p.max_elevation, "aos_epoch": aos_epoch, "los_epoch": los_epoch, diff --git a/work/meshai/main.py b/work/meshai/main.py index a8cf59a..b88e758 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -419,7 +419,15 @@ class MeshAI: _cov_bbox, "native", ) - if _sat_scope is not None: + # Only seed the synthetic coverage-centroid observer when the + # operator has NOT listed explicit satpass observers. Explicit + # observers win β€” the centroid is meaningless to listeners and, if + # seeded alongside real stations, leaks into broadcasts as the + # stale "coverage_center" region. (Existing DB rows are left for an + # ops cleanup; this only stops re-creating it.) + _explicit_observers = list( + getattr(self.config.environmental.satpass, "observers", None) or []) + if _sat_scope is not None and not _explicit_observers: lat, lon = _sat_scope["centroid"] observers_to_seed = [ {"slug": "coverage_center", "name": "Coverage Center", diff --git a/work/tests/test_satpass_broadcast_safety.py b/work/tests/test_satpass_broadcast_safety.py index 2893394..45a0bd7 100644 --- a/work/tests/test_satpass_broadcast_safety.py +++ b/work/tests/test_satpass_broadcast_safety.py @@ -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)" + ) diff --git a/work/tests/test_satpass_compass_fallback.py b/work/tests/test_satpass_compass_fallback.py index 1bd8515..29864ca 100644 --- a/work/tests/test_satpass_compass_fallback.py +++ b/work/tests/test_satpass_compass_fallback.py @@ -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 diff --git a/work/tests/test_satpass_handler.py b/work/tests/test_satpass_handler.py index 3eaf05e..104c8e9 100644 --- a/work/tests/test_satpass_handler.py +++ b/work/tests/test_satpass_handler.py @@ -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).""" diff --git a/work/tests/test_satpass_native.py b/work/tests/test_satpass_native.py index 66dc741..72f6e2d 100644 --- a/work/tests/test_satpass_native.py +++ b/work/tests/test_satpass_native.py @@ -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 diff --git a/work/tests/test_satpass_wire_fields.py b/work/tests/test_satpass_wire_fields.py index feb4916..9467fcb 100644 --- a/work/tests/test_satpass_wire_fields.py +++ b/work/tests/test_satpass_wire_fields.py @@ -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():