From 41178831e4754ab856ef265170e705ce55e6ffd3 Mon Sep 17 00:00:00 2001 From: malice Date: Fri, 17 Jul 2026 15:46:36 -0600 Subject: [PATCH] chore(central-ripout 2b): relocate satellite code to env/satellite/ (#162) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * chore(central-ripout 2b): create env/satellite package, move pass_predictor pass_predictor.py was 100% live (no dead entrypoint) — SGP4 pass computation used by both the native satpass adapter and the on-demand !satpass command. Straight move, no code changes: meshai.central.pass_predictor -> meshai.env.satellite.pass_predictor. Owner directive: satellite code gets its own folder under the feed adapters, separate from env.satpass (the adapter) to avoid colliding with env/satpass.py. Co-Authored-By: Claude Opus 4.8 (1M context) * chore(central-ripout 2b): split satpass_handler.py -> env/satellite/pass_format.py satpass_handler.py was a split file: live wire-formatting/gate logic plus dead Central-envelope ingest machinery whose only caller was the already-deleted central/consumer.py NATS bridge. Moved (live, verified via rg — external callers in env/satpass.py and commands/satpass_cmd.py, or transitively called by them): gate_consolidated_pass, format_pass, _check_rate_cap, _upsert_satpass, _attach_commit, _map_severity, _canonical_id, _azimuth_to_compass, _short_sat_name, _collapse_compass, _region_paren, _is_synthetic_observer, _format_time_12h/24h, _format_ampm, _tz_abbr, _date_label, plus the _SHORT_SAT_NAMES/_SHORT_NAME_SUBSTR/_SYNTHETIC_OBSERVERS tables. Dropped (dead — zero callers outside the already-deleted consumer.py and handle_satpass/consolidate_satpass_pending themselves; verified with rg): handle_satpass, consolidate_satpass_pending, _cleanup_pending, load_pending_schedule, _log_event_returning_id, _coerce_float, _coerce_int, _parse_iso_epoch, _now, CONSOLIDATION_DELAY, _pending_consolidation_ids, drain_pending_consolidation_ids, _elevation_bucket (already-orphaned pre-ripout: superseded by numeric "max NN°" wire format, zero callers anywhere but its own tests), SCHEMA_SATPASS_EVENTS/SCHEMA_SATPASS_PENDING (unused string constants — actual schema lives in persistence/migrations/*.sql, never imported). Also dropped now-unused `json`/`time`/`Any` imports. Straight code move otherwise — no logic changes to any moved function. Two docstrings updated for accuracy (module docstring, and gate_consolidated_pass's docstring which referenced the now-deleted Central consumer path). Co-Authored-By: Claude Opus 4.8 (1M context) * chore(central-ripout 2b): split tle_handler.py -> env/satellite/tle_store.py tle_handler.py was a split file: live storage helpers plus a dead Central-envelope ingest entrypoint whose only caller was the already-deleted central/consumer.py NATS bridge. Moved (live — used by env.tle_fetch, env.satpass, commands.satpass_cmd, verified via rg): upsert_tle, get_fresh_tles, get_tle_by_norad, search_tle_by_name. Dropped (dead — handle_tle's only callers were tests and the deleted consumer.py; verified with rg): handle_tle. Straight code move otherwise — no logic changes. Co-Authored-By: Claude Opus 4.8 (1M context) * chore(central-ripout 2b): repoint satellite consumers at env.satellite Rewire the three production consumers (including their lazy/function-body imports, not just module-top ones) to the new location: - env/satpass.py: central.tle_handler -> env.satellite.tle_store, central.pass_predictor -> env.satellite.pass_predictor, central.satpass_handler -> env.satellite.pass_format - env/tle_fetch.py: central.tle_handler.upsert_tle -> env.satellite.tle_store - commands/satpass_cmd.py: all three, same mapping Also refreshed docstrings that pointed at the old module paths or described the now-fully-deleted Central consolidation path (consolidate_satpass_pending / satpass_pending buffer) as a live alternative, and updated central/__init__.py's module docstring to stop listing the three relocated modules among central's remaining contents. No behavior changes. Co-Authored-By: Claude Opus 4.8 (1M context) * chore(central-ripout 2b): update satpass/tle test suite for the relocation Repoints every remaining test import at the new env.satellite.* modules and removes/adapts coverage for the Central envelope-ingest path deleted in this pass (handle_satpass, consolidate_satpass_pending, handle_tle, and the filtering/coercion/staleness logic that lived only inside them): - test_satpass_native.py, test_tle_fetch.py, test_satpass_command.py: import-path updates only (pass_predictor, tle_store). Also dropped test_satpass_command.py's TestTLEUpsert.test_returns_none_always (handle_tle-specific contract, no longer applicable) and rewrote its two latest-wins tests to call upsert_tle directly — same behavior under test, now exercised through the still-live primitive instead of the dead wrapper. - test_satpass_native.py: deleted test_central_consolidate_feeds_shared_gate_merged (spied on consolidate_satpass_pending, which no longer exists). The merge-across-observers logic it guarded is native-side (_consolidate) and already covered by test_two_observers_consolidate_to_one_broadcast. - test_satpass_handler.py: gutted to the one test that calls format_pass directly (test_format_pass_worst_case_fits_140); the rest exercised handle_satpass's observer/norad/elevation filters, which have no live equivalent (the native adapter filters at the config level, not per-envelope) and is redundant with test_satpass_native.py's dedup/wire coverage via the real SatpassAdapter path. - test_satpass_broadcast_safety.py: kept every test that calls format_pass or gate_consolidated_pass-adjacent REGISTRY checks directly (wire format, clean-format rules, REGISTRY defaults); deleted TestNoradIdTypeCoercion and TestStalenessGuard (handle_satpass-only logic, no live equivalent) and the 6 _elevation_bucket tests (_elevation_bucket itself was dead before this pass too — zero callers anywhere but its own tests, already superseded by the numeric "max NN°" wire format per its own docstring). - test_satpass_persisted_timer.py: dropped test_due_at_persisted_on_normal_ingest (handle_satpass-only); kept the two schema/migration tests, which don't touch satpass_handler. - Deleted outright (tested ONLY the dead Central envelope-ingest path, no live equivalent to port to): test_satpass_event_path.py, test_satpass_compass_fallback.py, test_satpass_wire_fields.py. Full suite: 2059 passed, 0 failed (was 0 failed on main pre-change). Satpass/TLE subset (99 tests across 8 files) verified green in isolation. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/central/__init__.py | 6 +- work/meshai/central/satpass_handler.py | 804 ------------------ work/meshai/commands/satpass_cmd.py | 10 +- work/meshai/env/satellite/__init__.py | 16 + work/meshai/env/satellite/pass_format.py | 425 +++++++++ .../satellite}/pass_predictor.py | 0 .../satellite/tle_store.py} | 82 +- work/meshai/env/satpass.py | 42 +- work/meshai/env/tle_fetch.py | 19 +- work/tests/test_satpass_broadcast_safety.py | 497 +---------- work/tests/test_satpass_command.py | 88 +- work/tests/test_satpass_compass_fallback.py | 263 ------ work/tests/test_satpass_event_path.py | 192 ----- work/tests/test_satpass_handler.py | 207 +---- work/tests/test_satpass_native.py | 72 +- work/tests/test_satpass_persisted_timer.py | 86 +- work/tests/test_satpass_wire_fields.py | 262 ------ work/tests/test_tle_fetch.py | 2 +- 18 files changed, 590 insertions(+), 2483 deletions(-) delete mode 100644 work/meshai/central/satpass_handler.py create mode 100644 work/meshai/env/satellite/__init__.py create mode 100644 work/meshai/env/satellite/pass_format.py rename work/meshai/{central => env/satellite}/pass_predictor.py (100%) rename work/meshai/{central/tle_handler.py => env/satellite/tle_store.py} (56%) delete mode 100644 work/tests/test_satpass_compass_fallback.py delete mode 100644 work/tests/test_satpass_event_path.py delete mode 100644 work/tests/test_satpass_wire_fields.py diff --git a/work/meshai/central/__init__.py b/work/meshai/central/__init__.py index 07eca5c..d4fcda5 100644 --- a/work/meshai/central/__init__.py +++ b/work/meshai/central/__init__.py @@ -2,5 +2,7 @@ JetStream firehose and normalized it into meshai pipeline Events. The NATS consumer is retired (Central is gone); this package now only holds the split-file modules still used by the native adapter paths (firms_handler, -satpass_handler, tle_handler, wfigs_handler _render, budget, -idaho_gauge_sites, pass_predictor).""" +wfigs_handler _render, idaho_gauge_sites). The satellite pieces +(satpass_handler, tle_handler, pass_predictor) have moved to +meshai.env.satellite — they're feed-adapter support code, not consumer +leftovers.""" diff --git a/work/meshai/central/satpass_handler.py b/work/meshai/central/satpass_handler.py deleted file mode 100644 index fd4569e..0000000 --- a/work/meshai/central/satpass_handler.py +++ /dev/null @@ -1,804 +0,0 @@ -"""v0.7 Satellite pass handler. - -Broadcast regional satellite passes from Central's CENTRAL_SAT stream. - -Filter criteria: - (a) Pass must be for an observer in adapter_config.satpass.observers - (empty list = all observers) - (b) Max elevation must meet adapter_config.satpass.min_elevation (default 30) - (c) Opt-in NORAD ID filter via adapter_config.satpass.norad_ids - (empty list = broadcast NOTHING — opt-in only) - -Rate cap: adapter_config.satpass.max_broadcasts_per_hour (default 4). -Dry-run: adapter_config.satpass.dry_run (default True) — logs wire text - at INFO with "DRY-RUN would air:" prefix, does not dispatch. - -Dedup bucketing: canonical event_id = {norad_id}:{aos_bucket} -where aos_bucket = floor(aos_epoch / 3600) -- one broadcast per satellite -per hour window, consolidated across all observers. - -Severity mapping: - 4 = immediate (>= 60 deg max elevation) - 3 = priority (>= 45 deg max elevation) - <= 2 = routine - -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} -""" -from __future__ import annotations - -import json -import logging -import re -import time -from datetime import datetime, timezone -from typing import Any, Optional -from zoneinfo import ZoneInfo - -from meshai.adapter_config import adapter_config -from meshai.notifications.formatters._budget import budget_for, fit_to_budget -from meshai.persistence import get_db - -logger = logging.getLogger(__name__) - -# Mountain time for broadcast display -_TZ = ZoneInfo("America/Boise") - -# Module-level signal: consolidation IDs that need timer scheduling. -# Consumer polls this after each satpass _normalize() call. -_pending_consolidation_ids: set[str] = set() - -# Baseline consolidation delay, in seconds, from a pending row's arrival to -# when its consolidated broadcast should fire. This is the DURABLE fire-time -# basis persisted as satpass_pending.due_at (= received_at + this). -# -# It matches the live consumer's baseline: the consumer schedules the timer -# at `5.0 + N*60` where N is the count of OTHER in-flight timers (a runtime -# anti-thundering-herd stagger). The `+N*60` term depends on transient -# in-memory scheduler state that has no meaning across a restart, so it is -# deliberately NOT persisted; only the N=0 baseline (5s) is durable. The -# live in-memory timer still drives normal operation exactly as before — -# due_at is purely the reboot-recovery backstop the in-memory timer can't be. -CONSOLIDATION_DELAY = 5 - - -def drain_pending_consolidation_ids() -> set[str]: - """Atomically drain and return all pending consolidation IDs.""" - ids = _pending_consolidation_ids.copy() - _pending_consolidation_ids.clear() - return ids - - -def _now() -> int: - return int(time.time()) - - -def _coerce_float(v) -> Optional[float]: - if v is None: - return None - if isinstance(v, (int, float)): - return float(v) - try: - return float(v) - except (TypeError, ValueError): - return None - - -def _coerce_int(v) -> Optional[int]: - if v is None: - return None - if isinstance(v, int): - return v - try: - return int(v) - except (TypeError, ValueError): - return None - - -def _parse_iso_epoch(s) -> Optional[int]: - """Parse ISO-8601 timestamp to epoch seconds.""" - if not s or not isinstance(s, str): - return None - try: - dt = datetime.fromisoformat(s.replace("Z", "+00:00")) - return int(dt.timestamp()) - except Exception: - return None - - -def _elevation_bucket(max_el: float) -> str: - """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: - return "high pass" - 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: - return "?" - try: - dt = datetime.fromtimestamp(epoch, tz=_TZ) - # Use %-I for no-leading-zero hour on Linux, fall back to %I - try: - return dt.strftime("%-I:%M") - except ValueError: - return dt.strftime("%I:%M").lstrip("0") - except Exception: - return "?" - - -def _format_ampm(epoch: Optional[int]) -> str: - """Return AM or PM for an epoch in America/Boise.""" - if epoch is None: - return "" - try: - dt = datetime.fromtimestamp(epoch, tz=_TZ) - return dt.strftime("%p") - except Exception: - return "" - - -def _format_time_24h(epoch: Optional[int]) -> str: - """Format epoch to HH:MM local time string (24h).""" - if epoch is None: - return "?" - try: - dt = datetime.fromtimestamp(epoch, tz=_TZ) - return dt.strftime("%H:%M") - except Exception: - return "?" - - -def _tz_abbr(epoch: Optional[int]) -> str: - """Return timezone abbreviation for an epoch in America/Boise.""" - if epoch is None: - return "MDT" - try: - dt = datetime.fromtimestamp(epoch, tz=_TZ) - return dt.strftime("%Z") - except Exception: - return "MDT" - - -def _azimuth_to_compass(az_deg: float) -> str: - """Convert azimuth in degrees to 8-point compass direction.""" - az = az_deg % 360 - dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"] - idx = int((az + 22.5) / 45) % 8 - return dirs[idx] - - -def _date_label(epoch: Optional[int]) -> str: - """Return a date qualifier for passes not happening today. - - Returns '' for today, 'tomorrow' for tomorrow, or 'Mon Jun 17' - for anything further out. - """ - if epoch is None: - return "" - try: - now_local = datetime.now(tz=_TZ) - pass_local = datetime.fromtimestamp(epoch, tz=_TZ) - delta_days = (pass_local.date() - now_local.date()).days - if delta_days == 0: - return "" - if delta_days == 1: - return " tomorrow" - return pass_local.strftime(" %a %b %-d") - except Exception: - return "" - - -def format_pass(*, sat_name: str, max_el: float, - aos_epoch: Optional[int], los_epoch: Optional[int], - aos_compass: str, los_compass: str, - broadcast: bool = True, - entry_observer: Optional[str] = None, - exit_observer: Optional[str] = None, - peak_compass: Optional[str] = None, - norad_id: Optional[int] = None) -> str: - """Unified pass formatter with mode switch. - - 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. 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: 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: - dur_min = 0 - - if broadcast: - name = _short_sat_name(norad_id, sat_name) - rise_str = _format_time_12h(aos_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) - - # 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(line, budget_for("satpass")) - else: - # DM format: compact with exact degrees - aos_str = _format_time_24h(aos_epoch) - los_str = _format_time_24h(los_epoch) - tz = _tz_abbr(aos_epoch) - return (f"{sat_name} {aos_str}\u2013{los_str} {tz} " - f"max {int(max_el)}\u00B0 " - f"{compass_seg}") - - -def _map_severity(max_el: float) -> str: - """Map max elevation to severity word.""" - if max_el >= 60: - return "immediate" - if max_el >= 45: - return "priority" - return "routine" - - -def _canonical_id(norad_id: int, aos_epoch: int) -> str: - """Generate consolidated canonical event ID (observer-independent).""" - bucket = aos_epoch // 3600 - return f"{norad_id}:{bucket}" - - -def _check_rate_cap(conn, now: int, max_per_hour: int) -> tuple[bool, int]: - """Check if broadcast rate cap has been reached. - - Returns (allowed, suppressed_count) where suppressed_count is the - number of broadcasts already made in the current hour window. - """ - hour_start = (now // 3600) * 3600 - row = conn.execute( - "SELECT COUNT(*) AS cnt FROM satpass_events " - "WHERE last_broadcast_at >= ? AND last_broadcast_at IS NOT NULL", - (hour_start,), - ).fetchone() - count = row["cnt"] if row else 0 - return (count < max_per_hour, count) - - -def _cleanup_pending(conn, consolidated_id: str) -> None: - """Remove all pending rows for a consolidated ID.""" - conn.execute("DELETE FROM satpass_pending WHERE consolidated_id=?", - (consolidated_id,)) - - -def handle_satpass(envelope: dict, subject: str, - data: Optional[dict] = None, - now: Optional[int] = None) -> Optional[str]: - """Process a satellite pass event from Central. - - Per-observer arrivals are accumulated into satpass_pending table. - Returns None (suppressing immediate broadcast). - Consolidation ID is added to _pending_consolidation_ids for consumer - to schedule a 5s timer. - """ - if not isinstance(envelope, dict): - return None - - inner = envelope.get("data") or {} - adapter = inner.get("adapter") or "" - - # Only handle pass prediction adapters (wire names from Central) - if adapter not in ("n2yo_visualpasses", "satpass_predict"): - return None - - # Enabled gate: silently drop when disabled, log once at INFO - cfg = adapter_config.satpass - if not getattr(cfg, "enabled", False): - if not getattr(handle_satpass, "_disabled_logged", False): - logger.info("satpass disabled; sat pass events dropped") - handle_satpass._disabled_logged = True - return None - - d = inner.get("data") or {} - now = now if now is not None else _now() - - # Extract pass data - norad_id = _coerce_int(d.get("norad_id") or d.get("satid")) - sat_name = d.get("satellite_name") or f"SAT-{norad_id}" - observer = d.get("observer_name") or d.get("observer_slug") or "unknown" - max_el = _coerce_float(d.get("max_elevation_deg")) - aos_iso = d.get("aos_time") - los_iso = d.get("los_time") - # Compass directions: prefer precomputed _compass strings (n2yo path), - # fall back to converting raw azimuth degrees (satpass_predict path). - aos_compass = d.get("azimuth_at_aos_compass") or ( - _azimuth_to_compass(d["azimuth_at_aos"]) if d.get("azimuth_at_aos") is not None else "") - los_compass = d.get("azimuth_at_los_compass") or ( - _azimuth_to_compass(d["azimuth_at_los"]) if d.get("azimuth_at_los") is not None else "") - direction = d.get("azimuth_at_peak_compass") or ( - _azimuth_to_compass(d["azimuth_at_peak"]) if d.get("azimuth_at_peak") is not None else "") - # Use peak direction as fallback for aos_compass only if aos is still empty - aos_compass = aos_compass or direction or "" - - if norad_id is None or max_el is None: - logger.debug("satpass_handler: missing norad_id or max_elevation_deg") - return None - - aos_epoch = _parse_iso_epoch(aos_iso) - los_epoch = _parse_iso_epoch(los_iso) - - if aos_epoch is None: - logger.debug("satpass_handler: could not parse aos time") - return None - - # Staleness guard: reject passes whose window already ended - if los_epoch is not None and los_epoch < now: - logger.debug("satpass_handler: pass already ended (los %d < now %d), skipping", - los_epoch, now) - return None - - # AOS horizon guard: reject passes too far in the future (likely stale prediction) - max_horizon_h = float(getattr(cfg, "max_aos_horizon_hours", 24)) - if max_horizon_h > 0 and aos_epoch > now + max_horizon_h * 3600: - logger.debug( - "satpass_handler: AOS %d is %.1fh away, beyond %gh horizon; skipping", - aos_epoch, (aos_epoch - now) / 3600, max_horizon_h) - return None - - # Observer filter (empty = all) - observers = getattr(cfg, "observers", []) or [] - if observers and observer not in observers: - logger.debug("satpass_handler: observer %r not in configured list", observer) - return None - - # OPT-IN NORAD ID filter: empty list = broadcast NOTHING - norad_ids_raw = getattr(cfg, "norad_ids", []) or [] - if not norad_ids_raw: - if not getattr(handle_satpass, "_no_norad_ids_logged", False): - logger.info("satpass: no norad_ids configured; pass broadcasts disabled") - handle_satpass._no_norad_ids_logged = True - return None - # Coerce to int set — GUI may save as strings (["25544"]), wire - # delivers int. Accept both shapes forever. - allow_set = {int(x) for x in norad_ids_raw if str(x).strip().isdigit()} - if norad_id not in allow_set: - logger.debug("satpass_handler: norad_id %d not in configured list", norad_id) - return None - - # Elevation floor - min_el = float(getattr(cfg, "min_elevation", 30)) - if max_el < min_el: - logger.debug("satpass_handler: max_el %.1f below floor %.1f", max_el, min_el) - return None - - # Generate consolidated canonical ID (observer-independent) - consolidated_id = _canonical_id(norad_id, aos_epoch) - severity_word = _map_severity(max_el) - category_raw = inner.get("category") or "sat.pass" - - try: - conn = get_db() - except Exception: - logger.exception("satpass_handler: persistence unavailable") - return None - - # Log the per-observer event arrival - _log_event_returning_id( - conn, now=now, source="satpass", category=category_raw, - severity_word=severity_word, event_id_external=consolidated_id, - subject=subject, handled=0, - table_name="satpass_pending", table_pk=f"{consolidated_id}:{observer}") - - # Accumulate into pending table. due_at is the durable fire-time backstop - # (received_at + baseline delay) so a restart can reconstruct a consolidation - # timer for rows the in-memory scheduler would otherwise orphan. - due_at = now + CONSOLIDATION_DELAY - 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, peak_compass, received_at, " - "due_at) " - "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", - (consolidated_id, observer, sat_name, norad_id, max_el, - aos_epoch, los_epoch, aos_compass, los_compass, direction, now, - due_at)) - - # Signal consumer to schedule consolidation timer - _pending_consolidation_ids.add(consolidated_id) - - # Suppress immediate broadcast - return None - - -def gate_consolidated_pass(consolidated: dict, *, - now: int) -> tuple[str, dict] | None: - """Source-agnostic broadcast gate for an already-consolidated pass. - - Both the Central consumer path (`consolidate_satpass_pending`, which - merges buffered per-observer rows from `satpass_pending`) and the native - env.satpass adapter (which consolidates in-memory across observers in one - tick) call THIS single function so the broadcast decision is byte-identical - regardless of source. It deliberately does NOT touch `satpass_pending` — - that buffer is a Central-consumer implementation detail owned by the caller. - - `consolidated` is a dict describing one merged pass with keys: - consolidated_id (str, = {norad}:{aos_epoch//3600}), - norad_id, sat_name, max_elevation, - aos_epoch, los_epoch (int epoch seconds), - aos_compass, los_compass, peak_compass, - entry_observer, exit_observer, - observer_list (comma-joined observer slugs for the audit column). - - Applies, in order: dedup-vs-`satpass_events`, rate cap, wire build, - dry-run gate, `satpass_events` upsert, and the deferred `_attach_commit` - that upserts last/first_broadcast_at on delivery. Returns (wire, data) to - broadcast, or None if suppressed. `now` is threaded explicitly for - determinism (rate-cap window + first_seen_at). - """ - try: - conn = get_db() - except Exception: - logger.exception("satpass gate: persistence unavailable") - return None - - cfg = adapter_config.satpass - - consolidated_id = consolidated["consolidated_id"] - norad_id = consolidated["norad_id"] - sat_name = consolidated["sat_name"] - max_el = consolidated["max_elevation"] - aos_epoch = consolidated["aos_epoch"] - los_epoch = consolidated["los_epoch"] - aos_compass = consolidated["aos_compass"] - los_compass = consolidated["los_compass"] - peak_compass = consolidated.get("peak_compass") - entry_obs = consolidated.get("entry_observer") - exit_obs = consolidated.get("exit_observer") - observer_list = consolidated.get("observer_list") or (entry_obs or "") - - # Dedup against satpass_events - existing = conn.execute( - "SELECT last_broadcast_at FROM satpass_events WHERE event_id=?", - (consolidated_id,)).fetchone() - if existing and existing["last_broadcast_at"] is not None: - return None - - # Rate cap - max_per_hour = int(getattr(cfg, "max_broadcasts_per_hour", 4)) - allowed, count = _check_rate_cap(conn, now, max_per_hour) - if not allowed: - logger.info("satpass: rate cap reached (%d/%d), suppressing consolidated pass %s", - count, max_per_hour, consolidated_id) - return None - - # Build consolidated wire — always pass observer names for region context - 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, - entry_observer=entry_obs, exit_observer=exit_obs) - - # Dry-run gate - dry_run = getattr(cfg, "dry_run", True) - if dry_run: - logger.info("DRY-RUN would air (consolidated): %s", wire) - return None - - # Upsert consolidated record into satpass_events - _upsert_satpass(conn, event_id=consolidated_id, norad_id=norad_id, - sat_name=sat_name, observer=observer_list, - max_elevation=max_el, aos_at=aos_epoch, - los_at=los_epoch, payload_json=None, - first_seen_at=now, set_last_broadcast=False) - - # Prepare data dict with callbacks - severity_word = _map_severity(max_el) - data = {"_meshai_precomposed": True, "_severity_override": severity_word} - _attach_commit(data, event_id=consolidated_id, event_log_row_id=None) - - return wire, data - - -def consolidate_satpass_pending(consolidated_id: str) -> tuple[str, dict] | None: - """Called by consumer when 5s consolidation timer fires. - - Reads the buffered per-observer rows for this canonical id, merges them - across observers (earliest AOS / latest LOS / max-elevation observer - supplies max_elevation + peak_compass / entry+exit observers), then - delegates the actual broadcast decision to the shared, source-agnostic - `gate_consolidated_pass`. Pending rows are cleaned up afterward regardless - of the gate's decision (dedup, rate-cap, dry-run, and success all consume - the buffer identically, as before). - - Returns (wire_string, data_dict) or None if suppressed. - """ - try: - conn = get_db() - except Exception: - logger.exception("satpass consolidation: persistence unavailable") - return None - - rows = conn.execute( - "SELECT * FROM satpass_pending WHERE consolidated_id=?", - (consolidated_id,)).fetchall() - if not rows: - return None - - # Consolidate observers - sorted_by_aos = sorted(rows, key=lambda r: r["aos_at"]) - sorted_by_los = sorted(rows, key=lambda r: r["los_at"]) - entry = sorted_by_aos[0] # earliest AOS - exit_ = sorted_by_los[-1] # latest LOS - best = max(rows, key=lambda r: r["max_elevation"]) - - consolidated = { - "consolidated_id": consolidated_id, - "norad_id": best["norad_id"], - "sat_name": best["sat_name"], - "max_elevation": best["max_elevation"], - "aos_epoch": entry["aos_at"], - "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_observer": entry["observer"], - "exit_observer": exit_["observer"], - "observer_list": ",".join(r["observer"] for r in sorted_by_aos), - } - - result = gate_consolidated_pass(consolidated, now=_now()) - - # Clean up pending rows regardless of the gate's decision. - _cleanup_pending(conn, consolidated_id) - - return result - - -def _upsert_satpass(conn, *, event_id, norad_id, sat_name, observer, - max_elevation, aos_at, los_at, payload_json, - first_seen_at, set_last_broadcast=False, - broadcast_at=None) -> None: - """Insert or update satpass_events row.""" - existing = conn.execute( - "SELECT 1 FROM satpass_events WHERE event_id=?", (event_id,)).fetchone() - if existing is None: - conn.execute( - "INSERT INTO satpass_events(event_id, norad_id, sat_name, observer, " - "max_elevation, aos_at, los_at, payload_json, first_seen_at, " - "last_broadcast_at) VALUES (?,?,?,?,?,?,?,?,?,?)", - (event_id, norad_id, sat_name, observer, max_elevation, aos_at, - los_at, payload_json, first_seen_at, - broadcast_at if set_last_broadcast else None)) - else: - conn.execute( - "UPDATE satpass_events SET sat_name=?, max_elevation=?, " - "payload_json=? WHERE event_id=?", - (sat_name, max_elevation, payload_json, event_id)) - - -def _attach_commit(data: Optional[dict], *, event_id: str, - event_log_row_id: Optional[int]) -> None: - """Attach post-broadcast commit callback.""" - if not isinstance(data, dict): - return - - def _on_commit(committed_at: float) -> None: - try: - conn = get_db() - except Exception: - logger.exception("satpass commit: persistence unavailable") - return - conn.execute( - "UPDATE satpass_events SET last_broadcast_at=?, " - "first_broadcast_at=COALESCE(first_broadcast_at, ?) WHERE event_id=?", - (int(committed_at), int(committed_at), event_id)) - if event_log_row_id is not None: - conn.execute("UPDATE event_log SET handled=1 WHERE id=?", - (int(event_log_row_id),)) - - data["_on_broadcast_committed"] = _on_commit - data["_broadcast_audit"] = {"table": "satpass_events", "pk": event_id} - - -def _log_event_returning_id(conn, *, now, source, category, severity_word, - event_id_external, subject, handled, - table_name, table_pk) -> int: - """Insert event_log row and return its ID.""" - cur = conn.execute( - "INSERT INTO event_log(received_at, source, category, severity_word, " - "event_id_external, nats_subject, handled, table_name, table_pk) " - "VALUES (?,?,?,?,?,?,?,?,?)", - (now, source, category, severity_word, event_id_external, subject, - int(bool(handled)), table_name, table_pk)) - return int(cur.lastrowid) - - -# Schema for satpass_events table (run once at startup via persistence) -SCHEMA_SATPASS_EVENTS = """ -CREATE TABLE IF NOT EXISTS satpass_events ( - event_id TEXT PRIMARY KEY, - norad_id INTEGER, - sat_name TEXT, - observer TEXT, - max_elevation REAL, - aos_at INTEGER, - los_at INTEGER, - payload_json TEXT, - first_seen_at INTEGER, - first_broadcast_at INTEGER, - last_broadcast_at INTEGER -); -CREATE INDEX IF NOT EXISTS idx_satpass_norad ON satpass_events(norad_id); -CREATE INDEX IF NOT EXISTS idx_satpass_observer ON satpass_events(observer); -CREATE INDEX IF NOT EXISTS idx_satpass_aos ON satpass_events(aos_at); -""" - -SCHEMA_SATPASS_PENDING = """ -CREATE TABLE IF NOT EXISTS satpass_pending ( - consolidated_id TEXT NOT NULL, - observer TEXT NOT NULL, - sat_name TEXT, - norad_id INTEGER, - max_elevation REAL, - aos_at INTEGER, - los_at INTEGER, - aos_compass TEXT, - los_compass TEXT, - peak_compass TEXT, - received_at INTEGER, - due_at INTEGER, - PRIMARY KEY (consolidated_id, observer) -); -""" - - -def load_pending_schedule() -> list[tuple[str, int]]: - """Return [(consolidated_id, due_at)] for every cid with pending rows. - - Used by the consumer's startup sweep to reconstruct consolidation timers - that were lost with the in-memory scheduler on restart. One entry per - distinct consolidated_id, keyed on the EARLIEST due_at across its observer - rows (MIN) so the reconstructed fire time matches the live timer, which is - armed off the first arrival and never re-armed for later observers. - - A row written before due_at existed (pre-v22, or a partial write) has - due_at IS NULL; COALESCE falls it back to received_at + baseline delay so - such a row is still recoverable rather than silently stranded. - """ - try: - conn = get_db() - except Exception: - logger.exception("satpass sweep: persistence unavailable") - return [] - rows = conn.execute( - "SELECT consolidated_id, " - "MIN(COALESCE(due_at, received_at + ?)) AS due_at " - "FROM satpass_pending GROUP BY consolidated_id", - (CONSOLIDATION_DELAY,), - ).fetchall() - out: list[tuple[str, int]] = [] - for r in rows: - try: - cid = r["consolidated_id"] - due = r["due_at"] - if cid is None or due is None: - continue - out.append((str(cid), int(due))) - except Exception: - logger.exception("satpass sweep: skipping malformed pending row") - return out diff --git a/work/meshai/commands/satpass_cmd.py b/work/meshai/commands/satpass_cmd.py index 554a3c4..1eee40e 100644 --- a/work/meshai/commands/satpass_cmd.py +++ b/work/meshai/commands/satpass_cmd.py @@ -105,7 +105,7 @@ class SatpassCommand(CommandHandler): tles = [] if norad_ids: - from meshai.central.tle_handler import get_tle_by_norad + from meshai.env.satellite.tle_store import get_tle_by_norad for nid in norad_ids: tle = get_tle_by_norad(nid, conn=conn) if tle: @@ -114,11 +114,11 @@ class SatpassCommand(CommandHandler): id_str = ", ".join(str(n) for n in norad_ids) return f"No fresh TLE for NORAD {id_str}. TLE cache may be empty." elif sat_name_query: - from meshai.central.tle_handler import search_tle_by_name + from meshai.env.satellite.tle_store import search_tle_by_name # Try exact NORAD ID first try: exact_id = int(sat_name_query) - from meshai.central.tle_handler import get_tle_by_norad + from meshai.env.satellite.tle_store import get_tle_by_norad tle = get_tle_by_norad(exact_id, conn=conn) if tle: tles = [tle] @@ -141,7 +141,7 @@ class SatpassCommand(CommandHandler): # Compute passes for each satellite try: - from meshai.central.pass_predictor import compute_passes, azimuth_to_compass + from meshai.env.satellite.pass_predictor import compute_passes, azimuth_to_compass except ImportError: return "Pass predictor not available (sgp4 missing?)." @@ -163,7 +163,7 @@ class SatpassCommand(CommandHandler): continue for p in passes: - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format 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) diff --git a/work/meshai/env/satellite/__init__.py b/work/meshai/env/satellite/__init__.py new file mode 100644 index 0000000..d8dc24c --- /dev/null +++ b/work/meshai/env/satellite/__init__.py @@ -0,0 +1,16 @@ +"""Native satellite pass prediction + TLE storage. + +Relocated from `meshai.central` (the retired Central NATS-consumer service) +during the Central ripout — this code was always the LIVE prediction/format/ +storage logic, just stranded next to a dead consumer. It now lives beside +its only caller, `meshai.env.satpass` (the native SGP4 pass adapter) and +`meshai.env.tle_fetch` (the native Celestrak TLE fetcher). + +Modules: + pass_predictor — SGP4 pass computation (compute_passes, PassInfo, ...) + pass_format — wire formatting + the shared broadcast gate + (format_pass, gate_consolidated_pass, ...) + tle_store — sat_tles upsert/read helpers (upsert_tle, get_fresh_tles, + get_tle_by_norad, search_tle_by_name) +""" +from __future__ import annotations diff --git a/work/meshai/env/satellite/pass_format.py b/work/meshai/env/satellite/pass_format.py new file mode 100644 index 0000000..12900f6 --- /dev/null +++ b/work/meshai/env/satellite/pass_format.py @@ -0,0 +1,425 @@ +"""Satellite pass wire formatting + the shared source-agnostic broadcast gate. + +Relocated from `meshai.central.satpass_handler` (the retired Central +NATS-consumer service) during the Central ripout. This module now has one +caller: the native SGP4 adapter (`meshai.env.satpass`), which computes ALL +observers for a satellite in a single `tick()`, consolidates across +observers in-memory, and calls `gate_consolidated_pass` directly — no +`satpass_pending` buffer, no consumer/timer. + +Severity mapping: + 4 = immediate (>= 60 deg max elevation) + 3 = priority (>= 45 deg max elevation) + <= 2 = routine + +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} +""" +from __future__ import annotations + +import logging +import re +from datetime import datetime, timezone +from typing import Optional +from zoneinfo import ZoneInfo + +from meshai.adapter_config import adapter_config +from meshai.notifications.formatters._budget import budget_for, fit_to_budget +from meshai.persistence import get_db + +logger = logging.getLogger(__name__) + +# Mountain time for broadcast display +_TZ = ZoneInfo("America/Boise") + + +# 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: + return "?" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) + # Use %-I for no-leading-zero hour on Linux, fall back to %I + try: + return dt.strftime("%-I:%M") + except ValueError: + return dt.strftime("%I:%M").lstrip("0") + except Exception: + return "?" + + +def _format_ampm(epoch: Optional[int]) -> str: + """Return AM or PM for an epoch in America/Boise.""" + if epoch is None: + return "" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) + return dt.strftime("%p") + except Exception: + return "" + + +def _format_time_24h(epoch: Optional[int]) -> str: + """Format epoch to HH:MM local time string (24h).""" + if epoch is None: + return "?" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) + return dt.strftime("%H:%M") + except Exception: + return "?" + + +def _tz_abbr(epoch: Optional[int]) -> str: + """Return timezone abbreviation for an epoch in America/Boise.""" + if epoch is None: + return "MDT" + try: + dt = datetime.fromtimestamp(epoch, tz=_TZ) + return dt.strftime("%Z") + except Exception: + return "MDT" + + +def _azimuth_to_compass(az_deg: float) -> str: + """Convert azimuth in degrees to 8-point compass direction.""" + az = az_deg % 360 + dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"] + idx = int((az + 22.5) / 45) % 8 + return dirs[idx] + + +def _date_label(epoch: Optional[int]) -> str: + """Return a date qualifier for passes not happening today. + + Returns '' for today, 'tomorrow' for tomorrow, or 'Mon Jun 17' + for anything further out. + """ + if epoch is None: + return "" + try: + now_local = datetime.now(tz=_TZ) + pass_local = datetime.fromtimestamp(epoch, tz=_TZ) + delta_days = (pass_local.date() - now_local.date()).days + if delta_days == 0: + return "" + if delta_days == 1: + return " tomorrow" + return pass_local.strftime(" %a %b %-d") + except Exception: + return "" + + +def format_pass(*, sat_name: str, max_el: float, + aos_epoch: Optional[int], los_epoch: Optional[int], + aos_compass: str, los_compass: str, + broadcast: bool = True, + entry_observer: Optional[str] = None, + exit_observer: Optional[str] = None, + peak_compass: Optional[str] = None, + norad_id: Optional[int] = None) -> str: + """Unified pass formatter with mode switch. + + 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. 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: 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: + dur_min = 0 + + if broadcast: + name = _short_sat_name(norad_id, sat_name) + rise_str = _format_time_12h(aos_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) + + # 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}°" + if compass_seg: + core += f" {compass_seg}" + line = ( + f"\U0001F6F0️ {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(line, budget_for("satpass")) + else: + # DM format: compact with exact degrees + aos_str = _format_time_24h(aos_epoch) + los_str = _format_time_24h(los_epoch) + tz = _tz_abbr(aos_epoch) + return (f"{sat_name} {aos_str}–{los_str} {tz} " + f"max {int(max_el)}° " + f"{compass_seg}") + + +def _map_severity(max_el: float) -> str: + """Map max elevation to severity word.""" + if max_el >= 60: + return "immediate" + if max_el >= 45: + return "priority" + return "routine" + + +def _canonical_id(norad_id: int, aos_epoch: int) -> str: + """Generate consolidated canonical event ID (observer-independent).""" + bucket = aos_epoch // 3600 + return f"{norad_id}:{bucket}" + + +def _check_rate_cap(conn, now: int, max_per_hour: int) -> tuple[bool, int]: + """Check if broadcast rate cap has been reached. + + Returns (allowed, suppressed_count) where suppressed_count is the + number of broadcasts already made in the current hour window. + """ + hour_start = (now // 3600) * 3600 + row = conn.execute( + "SELECT COUNT(*) AS cnt FROM satpass_events " + "WHERE last_broadcast_at >= ? AND last_broadcast_at IS NOT NULL", + (hour_start,), + ).fetchone() + count = row["cnt"] if row else 0 + return (count < max_per_hour, count) + + +def gate_consolidated_pass(consolidated: dict, *, + now: int) -> tuple[str, dict] | None: + """Source-agnostic broadcast gate for an already-consolidated pass. + + The native env.satpass adapter (which consolidates in-memory across + observers in one tick) calls this single function so the broadcast + decision is deterministic and self-contained. + + `consolidated` is a dict describing one merged pass with keys: + consolidated_id (str, = {norad}:{aos_epoch//3600}), + norad_id, sat_name, max_elevation, + aos_epoch, los_epoch (int epoch seconds), + aos_compass, los_compass, peak_compass, + entry_observer, exit_observer, + observer_list (comma-joined observer slugs for the audit column). + + Applies, in order: dedup-vs-`satpass_events`, rate cap, wire build, + dry-run gate, `satpass_events` upsert, and the deferred `_attach_commit` + that upserts last/first_broadcast_at on delivery. Returns (wire, data) to + broadcast, or None if suppressed. `now` is threaded explicitly for + determinism (rate-cap window + first_seen_at). + """ + try: + conn = get_db() + except Exception: + logger.exception("satpass gate: persistence unavailable") + return None + + cfg = adapter_config.satpass + + consolidated_id = consolidated["consolidated_id"] + norad_id = consolidated["norad_id"] + sat_name = consolidated["sat_name"] + max_el = consolidated["max_elevation"] + aos_epoch = consolidated["aos_epoch"] + los_epoch = consolidated["los_epoch"] + aos_compass = consolidated["aos_compass"] + los_compass = consolidated["los_compass"] + peak_compass = consolidated.get("peak_compass") + entry_obs = consolidated.get("entry_observer") + exit_obs = consolidated.get("exit_observer") + observer_list = consolidated.get("observer_list") or (entry_obs or "") + + # Dedup against satpass_events + existing = conn.execute( + "SELECT last_broadcast_at FROM satpass_events WHERE event_id=?", + (consolidated_id,)).fetchone() + if existing and existing["last_broadcast_at"] is not None: + return None + + # Rate cap + max_per_hour = int(getattr(cfg, "max_broadcasts_per_hour", 4)) + allowed, count = _check_rate_cap(conn, now, max_per_hour) + if not allowed: + logger.info("satpass: rate cap reached (%d/%d), suppressing consolidated pass %s", + count, max_per_hour, consolidated_id) + return None + + # Build consolidated wire — always pass observer names for region context + 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, + entry_observer=entry_obs, exit_observer=exit_obs) + + # Dry-run gate + dry_run = getattr(cfg, "dry_run", True) + if dry_run: + logger.info("DRY-RUN would air (consolidated): %s", wire) + return None + + # Upsert consolidated record into satpass_events + _upsert_satpass(conn, event_id=consolidated_id, norad_id=norad_id, + sat_name=sat_name, observer=observer_list, + max_elevation=max_el, aos_at=aos_epoch, + los_at=los_epoch, payload_json=None, + first_seen_at=now, set_last_broadcast=False) + + # Prepare data dict with callbacks + severity_word = _map_severity(max_el) + data = {"_meshai_precomposed": True, "_severity_override": severity_word} + _attach_commit(data, event_id=consolidated_id, event_log_row_id=None) + + return wire, data + + +def _upsert_satpass(conn, *, event_id, norad_id, sat_name, observer, + max_elevation, aos_at, los_at, payload_json, + first_seen_at, set_last_broadcast=False, + broadcast_at=None) -> None: + """Insert or update satpass_events row.""" + existing = conn.execute( + "SELECT 1 FROM satpass_events WHERE event_id=?", (event_id,)).fetchone() + if existing is None: + conn.execute( + "INSERT INTO satpass_events(event_id, norad_id, sat_name, observer, " + "max_elevation, aos_at, los_at, payload_json, first_seen_at, " + "last_broadcast_at) VALUES (?,?,?,?,?,?,?,?,?,?)", + (event_id, norad_id, sat_name, observer, max_elevation, aos_at, + los_at, payload_json, first_seen_at, + broadcast_at if set_last_broadcast else None)) + else: + conn.execute( + "UPDATE satpass_events SET sat_name=?, max_elevation=?, " + "payload_json=? WHERE event_id=?", + (sat_name, max_elevation, payload_json, event_id)) + + +def _attach_commit(data: Optional[dict], *, event_id: str, + event_log_row_id: Optional[int]) -> None: + """Attach post-broadcast commit callback.""" + if not isinstance(data, dict): + return + + def _on_commit(committed_at: float) -> None: + try: + conn = get_db() + except Exception: + logger.exception("satpass commit: persistence unavailable") + return + conn.execute( + "UPDATE satpass_events SET last_broadcast_at=?, " + "first_broadcast_at=COALESCE(first_broadcast_at, ?) WHERE event_id=?", + (int(committed_at), int(committed_at), event_id)) + if event_log_row_id is not None: + conn.execute("UPDATE event_log SET handled=1 WHERE id=?", + (int(event_log_row_id),)) + + data["_on_broadcast_committed"] = _on_commit + data["_broadcast_audit"] = {"table": "satpass_events", "pk": event_id} diff --git a/work/meshai/central/pass_predictor.py b/work/meshai/env/satellite/pass_predictor.py similarity index 100% rename from work/meshai/central/pass_predictor.py rename to work/meshai/env/satellite/pass_predictor.py diff --git a/work/meshai/central/tle_handler.py b/work/meshai/env/satellite/tle_store.py similarity index 56% rename from work/meshai/central/tle_handler.py rename to work/meshai/env/satellite/tle_store.py index 742c150..133526f 100644 --- a/work/meshai/central/tle_handler.py +++ b/work/meshai/env/satellite/tle_store.py @@ -1,93 +1,35 @@ -"""TLE cache handler — consumes central.sat.tle.> and upserts sat_tles. +"""TLE cache storage — sat_tles upsert/read helpers. -Central publishes ~190 TLEs every ~4h on CENTRAL_SAT stream, subject -central.sat.tle.{norad_id}. Envelope payload path: - data.data.{norad_id, satellite_name, tle_line1, tle_line2, epoch} +Relocated from `meshai.central.tle_handler` (the retired Central +NATS-consumer service) during the Central ripout. `upsert_tle` is shared by +BOTH ingest paths that remain: the native Celestrak fetcher (`env.tle_fetch`) +and reads feed the native pass predictor (`env.satpass`) and the on-demand +`!satpass` command (`commands.satpass_cmd`). Upsert rule: latest-wins on epoch — skip if cached epoch >= incoming. Read-time staleness: callers exclude epoch older than 14 days. """ from __future__ import annotations -import logging import time from typing import Optional from meshai.persistence import get_db -logger = logging.getLogger(__name__) - # Rows with epoch older than this are stale (no tombstone upstream). STALE_DAYS = 14 -def handle_tle(envelope: dict, subject: str, - data: Optional[dict] = None, - now: Optional[int] = None) -> Optional[str]: - """Process a TLE update from Central. - - Always returns None — TLE updates are storage-only, never broadcast. - """ - if not isinstance(envelope, dict): - return None - - inner = envelope.get("data") or {} - adapter = inner.get("adapter") or "" - - # Enabled gate: silently drop when disabled, log once at INFO - try: - from meshai.adapter_config import adapter_config - if not getattr(adapter_config.satpass, "enabled", False): - if not getattr(handle_tle, "_disabled_logged", False): - logger.info("satpass disabled; sat TLE events dropped") - handle_tle._disabled_logged = True - return None - except Exception: - pass # adapter_config may not be initialised in tests - - # Accept both sat_tles and sat_passes adapter (Central may tag either) - d = inner.get("data") or {} - - norad_id = d.get("norad_id") - if norad_id is None: - return None - try: - norad_id = int(norad_id) - except (TypeError, ValueError): - return None - - name = d.get("satellite_name") or d.get("name") or f"SAT-{norad_id}" - line1 = d.get("tle_line1") or d.get("line1") - line2 = d.get("tle_line2") or d.get("line2") - epoch = d.get("epoch") - - if not line1 or not line2 or not epoch: - logger.debug("tle_handler: missing line1/line2/epoch for NORAD %s", norad_id) - return None - - now = now if now is not None else int(time.time()) - - try: - conn = get_db() - except Exception: - logger.exception("tle_handler: persistence unavailable") - return None - - upsert_tle(conn, norad_id, name, line1, line2, epoch, now=now) - - return None # storage-only, never broadcast - - def upsert_tle(conn, norad_id: int, name: str, line1: str, line2: str, epoch, now: Optional[int] = None) -> bool: """Upsert one TLE into sat_tles with latest-epoch-wins semantics. - Shared by BOTH ingest paths — the Central envelope handler - (`handle_tle`) and the native Celestrak fetcher (`env.tle_fetch`) — so - the predictor reads TLEs identically regardless of source. `epoch` is a - lexicographically-sortable string (ISO 8601 for Central, ISO 8601 - derived from the TLE line-1 epoch field for the native fetch); a cached - row whose epoch is >= the incoming epoch is left untouched. + Shared by BOTH ingest paths — historically the Central envelope handler + and the native Celestrak fetcher (`env.tle_fetch`), now just the native + fetcher — so the predictor reads TLEs identically regardless of source. + `epoch` is a lexicographically-sortable string (ISO 8601 derived from the + TLE line-1 epoch field); a cached row whose epoch is >= the incoming + epoch is left untouched. Returns True if a row was written (insert or update), False if the cached epoch was same-or-newer and the write was skipped. diff --git a/work/meshai/env/satpass.py b/work/meshai/env/satpass.py index 1dd6ceb..9894922 100644 --- a/work/meshai/env/satpass.py +++ b/work/meshai/env/satpass.py @@ -1,12 +1,11 @@ """Native SGP4 satellite-pass adapter — predicts + broadcasts locally. -The final piece of native satpass: unlike the Central path (which receives -one per-observer envelope at a time, buffers them in `satpass_pending`, and -relies on the Central consumer's timer to fire `consolidate_satpass_pending`), -this adapter computes ALL observers for a satellite in a SINGLE `tick()`. That -lets it consolidate across observers in-memory and gate synchronously — so it -needs NEITHER the `satpass_pending` buffer NOR the Central consumer/timer, -both of which are OFF in standalone (feed_source="native") mode. +Unlike the retired Central path (which received one per-observer envelope at +a time, buffered them in a `satpass_pending` table, and relied on a +consumer timer to fire a consolidation pass), this adapter computes ALL +observers for a satellite in a SINGLE `tick()`. That lets it consolidate +across observers in-memory and gate synchronously — no buffer table, no +timer, no consumer. Flow per tick (slow poll, ~15 min; passes are computed over `window_hours`): 1. Resolve target satellites from the shared `sat_tles` table (populated @@ -19,14 +18,14 @@ Flow per tick (slow poll, ~15 min; passes are computed over `window_hours`): "predict for everything the fetcher stocked". Broadcast volume is still bounded by the gate's rate cap + dry-run default. 2. For each satellite x each enabled observer, run the SGP4 predictor - (`central.pass_predictor.compute_passes`). + (`env.satellite.pass_predictor.compute_passes`). 3. Group every (observer, PassInfo) by the observer-independent canonical - id `{norad}:{aos_epoch//3600}` and consolidate across observers exactly - like `consolidate_satpass_pending`: earliest AOS, latest LOS, the - max-elevation observer supplies max_elevation + peak_compass, and the - entry/exit observers are the earliest-AOS / latest-LOS stations. - 4. Hand each consolidated pass to the SHARED, source-agnostic - `satpass_handler.gate_consolidated_pass`, which dedups vs + id `{norad}:{aos_epoch//3600}` and consolidate across observers: + earliest AOS, latest LOS, the max-elevation observer supplies + max_elevation + peak_compass, and the entry/exit observers are the + earliest-AOS / latest-LOS stations (see `_consolidate` below). + 4. Hand each consolidated pass to + `env.satellite.pass_format.gate_consolidated_pass`, which dedups vs `satpass_events`, applies the rate cap + dry-run, upserts the event row, and attaches the deferred commit. Staged results feed `get_events()` / `to_event()`. @@ -124,7 +123,7 @@ class SatpassAdapter: sat_tles with GOES/METEOR/FENGYUN etc.; the strict post-filter guarantees those never leak into predictions once a list is set. """ - from meshai.central.tle_handler import get_fresh_tles, get_tle_by_norad + from meshai.env.satellite.tle_store import get_fresh_tles, get_tle_by_norad if self._norad_ids: allowed = set(self._norad_ids) @@ -138,11 +137,10 @@ class SatpassAdapter: @staticmethod def _consolidate(cid: str, recs: list[dict]) -> dict: - """Merge per-observer records for one canonical pass. - - Mirrors `consolidate_satpass_pending`: earliest AOS, latest LOS, the - max-elevation observer supplies max_elevation + peak_compass, and the - entry/exit observers are the earliest-AOS / latest-LOS stations. + """Merge per-observer records for one canonical pass: earliest AOS, + latest LOS, the max-elevation observer supplies max_elevation + + peak_compass, and the entry/exit observers are the earliest-AOS / + latest-LOS stations. """ by_aos = sorted(recs, key=lambda r: r["aos_epoch"]) by_los = sorted(recs, key=lambda r: r["los_epoch"]) @@ -168,8 +166,8 @@ class SatpassAdapter: def _compute_staged(self, now_epoch: int) -> list[dict]: """Predict → consolidate → gate. Returns staged event dicts.""" - from meshai.central import satpass_handler as sh - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite import pass_format as sh + from meshai.env.satellite.pass_predictor import compute_passes from meshai.persistence.observer_locations import get_observers observers = get_observers() diff --git a/work/meshai/env/tle_fetch.py b/work/meshai/env/tle_fetch.py index 1e26c8a..67ace3e 100644 --- a/work/meshai/env/tle_fetch.py +++ b/work/meshai/env/tle_fetch.py @@ -1,9 +1,9 @@ """Celestrak TLE fetcher — native, keyless population of sat_tles. -Storage-only native adapter (like central.tle_handler): it does NOT emit -mesh Events. Its sole job is to keep the shared `sat_tles` table populated -with fresh two-line element sets so the native SGP4 pass-predictor (built -next) has current orbital data when satpass runs with feed_source="native". +Storage-only native adapter: it does NOT emit mesh Events. Its sole job is +to keep the shared `sat_tles` table populated with fresh two-line element +sets so the native SGP4 pass-predictor has current orbital data when +satpass runs with feed_source="native". Source: Celestrak GP API (https://celestrak.org/NORAD/elements/gp.php), FORMAT=tle (classic 3-line: name / line1 / line2). Two selector styles: @@ -15,11 +15,10 @@ Config (SatpassConfig): `tle_groups` (list of group names), `norad_ids` update ~daily so the default is 6h). Gated on feed_source=="native" via the normal EnvironmentalStore registration. -Upserts flow through `central.tle_handler.upsert_tle` so native and -Central ingestion share identical latest-epoch-wins semantics and the same -`sat_tles` columns. The TLE line-1 epoch field (columns 19-32) is parsed -into an ISO-8601 string so it is directly comparable with the ISO epochs -Central stores. +Upserts flow through `env.satellite.tle_store.upsert_tle`, the shared +latest-epoch-wins helper used by every writer of the `sat_tles` table. The +TLE line-1 epoch field (columns 19-32) is parsed into an ISO-8601 string so +epochs are directly comparable regardless of source. """ from __future__ import annotations @@ -204,7 +203,7 @@ class TLEFetchAdapter: if not records: return 0 from meshai.persistence import get_db - from meshai.central.tle_handler import upsert_tle + from meshai.env.satellite.tle_store import upsert_tle conn = get_db() now = int(time.time()) diff --git a/work/tests/test_satpass_broadcast_safety.py b/work/tests/test_satpass_broadcast_safety.py index 45a0bd7..07eb828 100644 --- a/work/tests/test_satpass_broadcast_safety.py +++ b/work/tests/test_satpass_broadcast_safety.py @@ -1,174 +1,41 @@ """Tests for satpass broadcast safety controls. -Covers all five incident-response requirements: - 1. Opt-in bird filter: empty norad_ids broadcasts nothing - 2. Rate cap: max_broadcasts_per_hour suppresses excess - 3. Dry-run mode: logs wire text, dispatches nothing +Covers the incident-response requirements that live in +`meshai.env.satellite.pass_format` (formatting rules) and REGISTRY defaults: + 3. Dry-run default: REGISTRY dry_run default is True 4. Elevation default: REGISTRY min_elevation = 30 - 5. Broadcast wire format: two-line, buckets, byte budget + 5. Broadcast wire format: single-line, numeric degrees, byte budget + 6. Clean format: short names, degrees, compass collapse, friendly observers + +The Central envelope-ingest path (`handle_satpass`, +`consolidate_satpass_pending`, and the opt-in-bird-filter / rate-cap / +dry-run / staleness-guard / norad-id-coercion logic that lived INSIDE +`handle_satpass`) was retired with the Central NATS consumer and deleted +2026-07. That logic has no live equivalent in the native path -- the native +SatpassAdapter (env/satpass.py) does its own norad/observer filtering at the +adapter level (config-driven, not per-envelope) and its own imminence gate +instead of a staleness guard; both are covered by tests/test_satpass_native.py. +Dedup, rate-cap, and dry-run behavior of the shared +`gate_consolidated_pass` gate are also exercised live via the native adapter +in tests/test_satpass_native.py (`test_second_tick_does_not_rebroadcast_after_commit`, +imminence tests, etc). Tests that only exercised the dead ingest path's +filters (opt-in bird filter mechanics, rate-cap-via-handle_satpass, +dry-run-via-handle_satpass, norad-id string/int coercion, staleness guard) +were deleted rather than ported, since their behavior no longer exists +anywhere to test. """ from __future__ import annotations -import json -import logging -import time -from datetime import datetime, timezone -from unittest.mock import MagicMock, patch +from datetime import datetime import pytest -# ── Helpers ────────────────────────────────────────────────────────── - -def _envelope(norad_id=25544, sat_name="ISS", observer="Boise", - max_el=75.0, aos="2026-06-12T03:32:00Z", - los="2026-06-12T03:38:00Z", - aos_compass="SW", los_compass="NE"): - """Build a CloudEvents envelope for a satellite pass.""" - return { - "specversion": "1.0", - "type": "central.sat.pass", - "source": "central", - "id": f"pass-{norad_id}-{aos}", - "data": { - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "severity": 0, - "data": { - "norad_id": norad_id, - "satellite_name": sat_name, - "observer_name": observer, - "max_elevation_deg": max_el, - "aos_time": aos, - "los_time": los, - "azimuth_at_peak_compass": "S", - "azimuth_at_aos_compass": aos_compass, - "azimuth_at_los_compass": los_compass, - "duration_s": 360, - } - } - } - - -def _enable_satpass_db(norad_ids=None, dry_run=False, max_per_hour=100): - """Set satpass config in the test DB.""" - from meshai.persistence import get_db - from meshai.adapter_config import invalidate_cache - conn = get_db() - conn.execute( - "UPDATE adapter_config SET value_json='true' " - "WHERE adapter='satpass' AND key='enabled'" - ) - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='dry_run'", - (json.dumps(dry_run),) - ) - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='max_broadcasts_per_hour'", - (json.dumps(max_per_hour),) - ) - if norad_ids is not None: - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='norad_ids'", - (json.dumps(norad_ids),) - ) - conn.execute( - "UPDATE adapter_config SET value_json='5' " - "WHERE adapter='satpass' AND key='min_elevation'" - ) - invalidate_cache() - - -def _clear_handler_flags(): - from meshai.central.satpass_handler import handle_satpass - for attr in ("_disabled_logged", "_no_norad_ids_logged"): - if hasattr(handle_satpass, attr): - 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 +# 1. OPT-IN BIRD FILTER — REGISTRY-only surviving check # ══════════════════════════════════════════════════════════════════════ class TestOptInBirdFilter: - """empty norad_ids = broadcast nothing; norad_ids=[25544] = ISS only.""" - - def test_empty_norad_ids_broadcasts_nothing(self): - """norad_ids=[] must broadcast nothing.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[], dry_run=False) - _clear_handler_flags() - - env = _envelope(norad_id=25544, max_el=80.0) - result = handle_satpass(env, "central.sat.pass.iss", data={}, - now=1718163120) - assert result is None - - def test_empty_norad_ids_logs_once(self, caplog): - """Empty norad_ids logs info message once.""" - from meshai.central.satpass_handler import handle_satpass - _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=1781235000) - handle_satpass(_envelope(norad_id=25544, max_el=80.0, - aos="2026-06-12T04:32:00Z"), - "central.sat.pass.iss", data={}, now=1781235000) - - matching = [r for r in caplog.records - if "no norad_ids configured" in r.message] - assert len(matching) == 1, "Should log exactly once" - - def test_norad_ids_25544_airs_iss_only(self): - """norad_ids=[25544] must air ISS passes, reject others.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=False) - _clear_handler_flags() - - # ISS pass should air - iss_env = _envelope(norad_id=25544, max_el=65.0) - 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 (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 = _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): """!satpass DM replies about any bird in TLE cache, regardless of broadcast norad_ids being empty.""" @@ -183,66 +50,13 @@ class TestOptInBirdFilter: # ══════════════════════════════════════════════════════════════════════ -# 2. RATE CAP +# 2. RATE CAP — DM-path isolation check # ══════════════════════════════════════════════════════════════════════ class TestRateCap: - """max_broadcasts_per_hour suppresses excess.""" - - def test_cap_suppresses_pass_n_plus_1(self): - """After max_broadcasts_per_hour broadcasts, next one is suppressed.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - - _enable_satpass_db(norad_ids=[25544, 28654, 99999, 99998, 99997], - dry_run=False, max_per_hour=4) - _clear_handler_flags() - - now = 1718200000 - hour_start = (now // 3600) * 3600 - conn = get_db() - - # Pre-insert 4 broadcast records in current hour window - for i in range(4): - eid = f"prefill:{i}:dummy:{hour_start // 3600}" - conn.execute( - "INSERT INTO satpass_events(event_id, norad_id, sat_name, " - "observer, max_elevation, aos_at, los_at, first_seen_at, " - "last_broadcast_at) VALUES (?,?,?,?,?,?,?,?,?)", - (eid, 99990 + i, f"SAT-{i}", "Boise", 60.0, - now - 300, now + 300, now - 600, hour_start + i) - ) - - # 5th broadcast should be suppressed - env = _envelope(norad_id=25544, max_el=70.0, - aos="2026-06-12T10:00:00Z") - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=now) - assert result is None, "5th broadcast should be suppressed by rate cap" - - def test_cap_logged_on_suppress(self, caplog): - """Rate cap suppression must log at INFO.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - - _enable_satpass_db(norad_ids=[25544], dry_run=False, max_per_hour=0) - _clear_handler_flags() - - # 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", - 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" - def test_cap_does_not_apply_to_dm_path(self): """Rate cap is broadcast-only, never affects DM replies.""" - # The DM command (satpass_cmd.py) does not call handle_satpass, + # The DM command (satpass_cmd.py) does not call gate_consolidated_pass, # it uses pass_predictor directly. Verify they're separate paths. import inspect from meshai.commands import satpass_cmd @@ -252,59 +66,10 @@ class TestRateCap: # ══════════════════════════════════════════════════════════════════════ -# 3. DRY-RUN MODE +# 3. DRY-RUN MODE — REGISTRY default # ══════════════════════════════════════════════════════════════════════ class TestDryRun: - """dry_run=True logs wire text, dispatches nothing.""" - - def test_dry_run_dispatches_nothing(self): - """dry_run=True must return None (no dispatch).""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=True) - _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) - assert result is None, "dry_run should suppress dispatch" - assert "_on_broadcast_committed" not in data, \ - "dry_run should not attach commit callback" - - def test_dry_run_logs_wire_text(self, caplog): - """dry_run=True must log the exact wire text at INFO.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=True) - _clear_handler_flags() - - with caplog.at_level(logging.INFO, logger="meshai.central.satpass_handler"): - env = _envelope(norad_id=25544, sat_name="ISS", max_el=70.0) - 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): ". - matching = [r for r in caplog.records - 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 - - def test_dry_run_false_dispatches(self): - """dry_run=False must dispatch normally.""" - from meshai.central.satpass_handler import handle_satpass - _enable_satpass_db(norad_ids=[25544], dry_run=False) - _clear_handler_flags() - - env = _envelope(norad_id=25544, max_el=70.0) - 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): """REGISTRY default for dry_run must be True.""" from meshai.adapter_config.defaults import REGISTRY @@ -331,11 +96,11 @@ class TestElevationDefault: # ══════════════════════════════════════════════════════════════════════ class TestBroadcastWireFormat: - """Two-line format, buckets, byte budget.""" + """Single-line format, numeric degrees, byte budget.""" def test_exact_single_line_example(self): """Formatter produces the exact single-line target format.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass # ISS, SW->NE, 6-minute window, rises 8:38 PM MDT, max 55 deg. from zoneinfo import ZoneInfo @@ -353,46 +118,16 @@ class TestBroadcastWireFormat: ) # 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). + # date qualifier ("Fri Jun 12" — 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)" + "\U0001F6F0️ ISS 8:38 PM MDT Fri Jun 12, " + "max 55° SW→NE (6 min)" ) - def test_bucket_overhead_at_60(self): - """max_el=60 should be 'overhead'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(60.0) == "overhead" - - def test_bucket_overhead_at_90(self): - """max_el=90 should be 'overhead'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(90.0) == "overhead" - - def test_bucket_high_pass_at_59(self): - """max_el=59 should be 'high pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(59.0) == "high pass" - - def test_bucket_high_pass_at_30(self): - """max_el=30 should be 'high pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(30.0) == "high pass" - - def test_bucket_low_pass_at_29(self): - """max_el=29 should be 'low pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(29.0) == "low pass" - - def test_bucket_low_pass_at_10(self): - """max_el=10 should be 'low pass'.""" - from meshai.central.satpass_handler import _elevation_bucket - assert _elevation_bucket(10.0) == "low pass" - def test_broadcast_byte_length_under_budget(self): """Broadcast wire message must be <= 120 bytes UTF-8.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") @@ -423,7 +158,7 @@ class TestBroadcastWireFormat: def test_dm_format_has_exact_degrees(self): """DM format must include exact degree number, not bucket.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") @@ -438,13 +173,13 @@ class TestBroadcastWireFormat: broadcast=False, ) - assert "max 75\u00B0" in wire + assert "max 75°" in wire assert "overhead" not in wire assert "high pass" not in wire def test_dm_format_single_line(self): """DM format is a single line.""" - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") @@ -488,164 +223,6 @@ class TestRegistryKeys: "opt-in" in spec["description"].lower() -class TestNoradIdTypeCoercion: - """norad_ids may arrive as strings from the GUI or ints from code. - The handler must accept both shapes forever.""" - - def test_string_norad_ids_matches_int_wire(self): - """norad_ids=["25544"] must match wire norad_id 25544 (int).""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["25544"], dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - 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): - """norad_ids=[25544, "22825"] must match both NORAD IDs.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544, "22825"], dry_run=False) - - # int in list, int on wire - env_iss = _envelope(norad_id=25544, max_el=65.0) - 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 = _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): - """Non-numeric entries in norad_ids must be silently skipped.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["25544", "not_a_number", "", None, "abc123"], - dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - # Must not raise, and the valid entry should still match - 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): - """If every entry is garbage, allow_set is empty and nothing matches.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["abc", "", "xyz"], dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - result = handle_satpass(env, "test.subject", data={}, now=1781235000) - assert result is None, "all-garbage norad_ids should match nothing" - - def test_pure_int_norad_ids_still_works(self): - """norad_ids=[25544] (pure int) must continue to work.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - env = _envelope(norad_id=25544, max_el=80.0) - 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): - """norad_ids=["25544"] must NOT match wire norad_id 99999.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=["25544"], dry_run=False) - - env = _envelope(norad_id=99999, max_el=80.0) - result = handle_satpass(env, "test.subject", data={}, now=1781235000) - assert result is None, "non-matching norad_id should be rejected" - - -class TestStalenessGuard: - """Reject passes whose window already ended; allow ongoing/future/None.""" - - def test_past_pass_rejected(self): - """Pass with los 10 min in the past produces no wire, no broadcast mark.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos = "2026-06-12T02:00:00Z" # well in the past - los_epoch = now - 600 # 10 min ago - los = datetime.fromtimestamp(los_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - env = _envelope(norad_id=25544, max_el=80.0, aos=aos, los=los) - result = handle_satpass(env, "test.subject", data={}, now=now) - assert result is None, "past pass should produce no wire" - - # Verify no broadcast mark in DB - conn = get_db() - row = conn.execute( - "SELECT last_broadcast_at FROM satpass_events WHERE norad_id=25544" - ).fetchone() - assert row is None or row["last_broadcast_at"] is None, \ - "past pass should not create a broadcast-marked DB row" - - def test_ongoing_pass_broadcasts(self): - """Ongoing pass (aos -2 min, los +5 min) must produce wire.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos_epoch = now - 120 # started 2 min ago - los_epoch = now + 300 # ends in 5 min - aos = datetime.fromtimestamp(aos_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - 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 = _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): - """Future pass (aos and los both in future) must produce wire.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos_epoch = now + 600 # starts in 10 min - los_epoch = now + 1200 # ends in 20 min - aos = datetime.fromtimestamp(aos_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - 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 = _ingest_and_consolidate(env, "test.subject", now=now) - assert result is not None, "future pass should broadcast" - - def test_none_los_falls_through(self): - """los_epoch=None must not be rejected by staleness guard.""" - from meshai.central.satpass_handler import handle_satpass - _clear_handler_flags() - _enable_satpass_db(norad_ids=[25544], dry_run=False) - - now = 1718200000 - aos_epoch = now + 600 - aos = datetime.fromtimestamp(aos_epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - - # Build envelope with no los_time - env = _envelope(norad_id=25544, max_el=60.0, aos=aos, los="") - # Patch los to None by removing los_time from inner data - env["data"]["data"]["los_time"] = None - - 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 - # It may still produce wire or be filtered by something else, - # 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 # ══════════════════════════════════════════════════════════════════════ @@ -655,7 +232,7 @@ class TestCleanBroadcastFormat: @staticmethod def _wire(**kw): - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass from zoneinfo import ZoneInfo tz = ZoneInfo("America/Boise") base = dict( diff --git a/work/tests/test_satpass_command.py b/work/tests/test_satpass_command.py index 4bac7ce..c17f46f 100644 --- a/work/tests/test_satpass_command.py +++ b/work/tests/test_satpass_command.py @@ -67,11 +67,17 @@ def _seed_tle(conn, *, norad_id, name, line1, line2, epoch, updated_at=None): class TestTLEUpsert: - """T1: TLE upsert latest-wins on epoch.""" + """T1: TLE upsert latest-wins on epoch. + + Exercises `upsert_tle` directly — the shared latest-wins primitive used + by every writer of `sat_tles` (native env.tle_fetch is the only one + left; the Central envelope ingest path that used to call it via + `tle_handler.handle_tle` was retired with Central). + """ def test_newer_epoch_updates(self): _enable_satpass() - from meshai.central.tle_handler import handle_tle + from meshai.env.satellite.tle_store import upsert_tle conn = get_db() now = int(time.time()) @@ -79,20 +85,9 @@ class TestTLEUpsert: _seed_tle(conn, norad_id=25544, name="ISS", line1="OLD1", line2="OLD2", epoch="2024-06-10T00:00:00Z") - # Send newer TLE - env = { - "data": { - "adapter": "celestrak_tle", - "data": { - "norad_id": 25544, - "satellite_name": "ISS (ZARYA)", - "tle_line1": "NEW1", - "tle_line2": "NEW2", - "epoch": "2024-06-15T00:00:00Z", - }, - } - } - handle_tle(env, "central.sat.tle.25544", now=now) + # Upsert a newer TLE + upsert_tle(conn, 25544, "ISS (ZARYA)", "NEW1", "NEW2", + "2024-06-15T00:00:00Z", now=now) row = conn.execute("SELECT line1, line2 FROM sat_tles WHERE norad_id=25544").fetchone() assert row["line1"] == "NEW1" @@ -100,7 +95,7 @@ class TestTLEUpsert: def test_older_epoch_skipped(self): _enable_satpass() - from meshai.central.tle_handler import handle_tle + from meshai.env.satellite.tle_store import upsert_tle conn = get_db() now = int(time.time()) @@ -108,49 +103,20 @@ class TestTLEUpsert: _seed_tle(conn, norad_id=25544, name="ISS", line1="CURRENT1", line2="CURRENT2", epoch="2024-06-15T00:00:00Z") - # Send older TLE — should be skipped - env = { - "data": { - "adapter": "celestrak_tle", - "data": { - "norad_id": 25544, - "satellite_name": "ISS (ZARYA)", - "tle_line1": "OLD1", - "tle_line2": "OLD2", - "epoch": "2024-06-10T00:00:00Z", - }, - } - } - handle_tle(env, "central.sat.tle.25544", now=now) + # Upsert an older TLE — should be skipped + written = upsert_tle(conn, 25544, "ISS (ZARYA)", "OLD1", "OLD2", + "2024-06-10T00:00:00Z", now=now) + assert written is False row = conn.execute("SELECT line1 FROM sat_tles WHERE norad_id=25544").fetchone() assert row["line1"] == "CURRENT1", "older epoch should not overwrite" - def test_returns_none_always(self): - """TLE handler is storage-only, never returns wire.""" - _enable_satpass() - from meshai.central.tle_handler import handle_tle - env = { - "data": { - "adapter": "celestrak_tle", - "data": { - "norad_id": 99999, - "satellite_name": "TEST", - "tle_line1": "L1", - "tle_line2": "L2", - "epoch": "2024-06-15T00:00:00Z", - }, - } - } - result = handle_tle(env, "central.sat.tle.99999") - assert result is None - class TestTLEStaleness: """T2: 14-day staleness exclusion at read time.""" def test_fresh_tle_returned(self): - from meshai.central.tle_handler import get_tle_by_norad + from meshai.env.satellite.tle_store import get_tle_by_norad conn = get_db() # Seed with recent epoch recent = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat() @@ -161,7 +127,7 @@ class TestTLEStaleness: assert tle["norad_id"] == 25544 def test_stale_tle_excluded(self): - from meshai.central.tle_handler import get_tle_by_norad + from meshai.env.satellite.tle_store import get_tle_by_norad conn = get_db() # Seed with 15-day old epoch stale = (datetime.now(timezone.utc) - timedelta(days=15)).isoformat() @@ -171,7 +137,7 @@ class TestTLEStaleness: assert tle is None, "stale TLE (>14 days) should be excluded" def test_search_excludes_stale(self): - from meshai.central.tle_handler import search_tle_by_name + from meshai.env.satellite.tle_store import search_tle_by_name conn = get_db() stale = (datetime.now(timezone.utc) - timedelta(days=15)).isoformat() _seed_tle(conn, norad_id=25544, name="ISS (ZARYA)", line1=ISS_LINE1, @@ -185,7 +151,7 @@ class TestPassPredictor: def test_iss_produces_passes(self): """ISS TLE for Boise should produce at least one pass in 24h.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes # Use a fixed time near the TLE epoch for best accuracy start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, @@ -194,7 +160,7 @@ class TestPassPredictor: def test_pass_max_elevation_reasonable(self): """Max elevation should be between min_el and 90°.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -204,7 +170,7 @@ class TestPassPredictor: def test_pass_aos_before_los(self): """AOS should be before LOS for every pass.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -214,7 +180,7 @@ class TestPassPredictor: def test_pass_duration_reasonable(self): """Pass durations should be positive; 30s step may merge adjacent passes.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -232,7 +198,7 @@ class TestPassPredictor: over Boise (43.6°N, 51.6° inclination orbit). We assert that at least one pass in 24h exceeds 30° — a conservative threshold. """ - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -243,7 +209,7 @@ class TestPassPredictor: def test_azimuth_range(self): """Azimuths should be in [0, 360) range.""" - from meshai.central.pass_predictor import compute_passes + from meshai.env.satellite.pass_predictor import compute_passes start = datetime(2024, 6, 16, 0, 0, 0, tzinfo=timezone.utc) passes = compute_passes(ISS_LINE1, ISS_LINE2, BOISE_LAT, BOISE_LON, window_h=24, min_el=10.0, now=start) @@ -252,7 +218,7 @@ class TestPassPredictor: assert 0 <= p.azimuth_at_los < 360, f"LOS azimuth {p.azimuth_at_los} out of range" def test_compass_conversion(self): - from meshai.central.pass_predictor import azimuth_to_compass + from meshai.env.satellite.pass_predictor import azimuth_to_compass assert azimuth_to_compass(0) == "N" assert azimuth_to_compass(45) == "NE" assert azimuth_to_compass(90) == "E" @@ -442,7 +408,7 @@ class TestReplyFormat: def test_line_format_matches_spec(self): """Lines should match 'NAME HH:MM–HH:MM TZ max XX° DIR→DIR'.""" - from meshai.central.pass_predictor import compute_passes, azimuth_to_compass, PassInfo + from meshai.env.satellite.pass_predictor import compute_passes, azimuth_to_compass, PassInfo from meshai.commands.satpass_cmd import SatpassCommand from zoneinfo import ZoneInfo diff --git a/work/tests/test_satpass_compass_fallback.py b/work/tests/test_satpass_compass_fallback.py deleted file mode 100644 index 29864ca..0000000 --- a/work/tests/test_satpass_compass_fallback.py +++ /dev/null @@ -1,263 +0,0 @@ -"""Tests for compass direction fallback on satpass_predict envelopes. - -Proves the fix for empty compass on satpass_predict broadcasts: - (a) satpass_predict-shape envelope (raw azimuth degrees, NO _compass fields) - produces non-empty compass directions in the wire message. - (b) n2yo-shape envelope with precomputed _compass strings is unchanged. - (c) Envelope with neither raw azimuths nor _compass strings produces - empty compass, no crash. - -Uses verbatim field shapes from the live AO-27 predict envelope. -""" -from __future__ import annotations - -import copy -import json - -import pytest - - -# ── Live AO-27 satpass_predict envelope (no _compass fields) ───────── - -AO27_PREDICT_ENVELOPE = { - "id": "filer:36122:2026-06-13T06:12:00+00:00", - "source": "central.echo6.co", - "type": "central.pass.satpass_predict.v1", - "time": "2026-06-13T06:12:00+00:00", - "datacontenttype": "application/json", - "centralschemaversion": "1.0", - "centralcategory": "pass.satpass_predict", - "centralseverity": 1, - "specversion": "1.0", - "data": { - "id": "filer:36122:2026-06-13T06:12:00+00:00", - "adapter": "satpass_predict", - "category": "pass.satpass_predict", - "time": "2026-06-13T06:12:00Z", - "expires": None, - "severity": 1, - "geo": { - "centroid": [-114.6, 42.57], - "bbox": None, - "regions": ["US-ID"], - "primary_region": "US-ID", - "geometry": None, - }, - "data": { - "observer_name": "Filer", - "observer_slug": "filer", - "observer_state": "ID", - "norad_id": 36122, - "satellite_name": "EYESAT A (AO-27)", - "aos_time": "2026-06-13T06:12:00+00:00", - "peak_time": "2026-06-13T06:18:00+00:00", - "los_time": "2026-06-13T06:24:00+00:00", - "max_elevation_deg": 62.3, - "azimuth_at_aos": 163.2, - "azimuth_at_peak": 245.0, - "azimuth_at_los": 348.7, - "duration_s": 720, - }, - }, -} - - -# ── n2yo envelope with precomputed _compass strings ────────────────── - -N2YO_ENVELOPE = { - "id": "filer:28654:2026-06-10T04:34:40+00:00", - "source": "central.echo6.co", - "type": "central.pass.n2yo_visualpasses.v1", - "time": "2026-06-10T04:41:35+00:00", - "datacontenttype": "application/json", - "centralschemaversion": "1.0", - "centralcategory": "pass.n2yo_visualpasses", - "centralseverity": 1, - "specversion": "1.0", - "data": { - "id": "filer:28654:2026-06-10T04:34:40+00:00", - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "time": "2026-06-10T04:41:35Z", - "expires": None, - "severity": 1, - "geo": { - "centroid": [-114.6, 42.57], - "bbox": None, - "regions": ["US-ID"], - "primary_region": "US-ID", - "geometry": None, - }, - "data": { - "observer_name": "Filer", - "observer_slug": "filer", - "observer_state": "ID", - "norad_id": 28654, - "satellite_name": "NOAA 18", - "aos_time": "2026-06-10T04:34:40+00:00", - "peak_time": "2026-06-10T04:41:35+00:00", - "los_time": "2026-06-10T04:48:30+00:00", - "max_elevation_deg": 22.69, - "magnitude": 6.7, - "azimuth_at_aos": 125.6, - "azimuth_at_aos_compass": "SE", - "azimuth_at_peak": 63.0, - "azimuth_at_peak_compass": "ENE", - "azimuth_at_los": 359.5, - "azimuth_at_los_compass": "N", - "duration_s": 630, - }, - }, -} - - -def _enable_satpass(norad_ids=None): - """Set satpass.enabled=true with permissive filters.""" - from meshai.persistence import get_db - from meshai.adapter_config import invalidate_cache - conn = get_db() - conn.execute( - "UPDATE adapter_config SET value_json='true' " - "WHERE adapter='satpass' AND key='enabled'" - ) - conn.execute( - "UPDATE adapter_config SET value_json='5' " - "WHERE adapter='satpass' AND key='min_elevation'" - ) - conn.execute( - "UPDATE adapter_config SET value_json='false' " - "WHERE adapter='satpass' AND key='dry_run'" - ) - if norad_ids is None: - norad_ids = [36122, 28654] - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='norad_ids'", - (json.dumps(norad_ids),) - ) - invalidate_cache() - - -def _clear_handler_flags(): - from meshai.central.satpass_handler import handle_satpass - for attr in ("_disabled_logged", "_no_norad_ids_logged"): - if hasattr(handle_satpass, attr): - 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(): - """satpass_predict envelope with only raw azimuth degrees produces - non-empty compass directions like SSE→N in wire output.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - _clear_handler_flags() - - now = 1781330520 # well before the AO-27 pass window - result = _ingest_and_consolidate( - AO27_PREDICT_ENVELOPE, - "central.sat.pass.us.id.filer", - now=now, - ) - assert result is not None, "handler returned None for satpass_predict envelope" - wire, _ = result - - # 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) - 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 ─────── - -def test_n2yo_precomputed_compass_unchanged(): - """n2yo envelope with _compass string fields uses those strings, - not raw azimuth conversion.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - _clear_handler_flags() - - now = 1781065800 # before NOAA-18 pass window - result = _ingest_and_consolidate( - N2YO_ENVELOPE, - "central.sat.pass.us.id.filer", - now=now, - ) - assert result is not None, "handler returned None for n2yo envelope" - wire, _ = result - - # 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}" - 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 ────────────── - -def test_no_compass_no_azimuth_no_crash(): - """Envelope with no _compass fields AND no raw azimuth fields - produces empty compass directions without crashing.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - _clear_handler_flags() - - env = copy.deepcopy(AO27_PREDICT_ENVELOPE) - d = env["data"]["data"] - # Remove all azimuth fields - for key in ("azimuth_at_aos", "azimuth_at_los", "azimuth_at_peak", - "azimuth_at_aos_compass", "azimuth_at_los_compass", - "azimuth_at_peak_compass"): - d.pop(key, None) - - now = 1781330520 - result = _ingest_and_consolidate( - env, - "central.sat.pass.us.id.filer", - now=now, - ) - assert result is not None, "handler crashed or returned None — should produce wire with empty compass" - wire, _ = result - - # 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_event_path.py b/work/tests/test_satpass_event_path.py deleted file mode 100644 index 78f8fdc..0000000 --- a/work/tests/test_satpass_event_path.py +++ /dev/null @@ -1,192 +0,0 @@ -"""Tests for satpass event path — wire adapter names route correctly. - -Verifies: - - celestrak_tle envelope routes to tle_handler and inserts sat_tles row - - n2yo_visualpasses envelope routes to satpass_handler - - satpass_predict envelope routes to satpass_handler - - adapter_config reads min_elevation (not min_elevation_deg) - - CENTRAL_ADAPTER_TO_SOURCE maps wire names to 'satpass' - - stale adapter names removed from dispatch -""" - -import json -import time - -import pytest - - -# ── Realistic wire envelopes ──────────────────────────────────────── - -CELESTRAK_TLE_ENVELOPE = { - "specversion": "1.0", - "id": "tle-25544-1718200000", - "source": "central", - "type": "central.sat.tle", - "data": { - "id": "tle-25544-1718200000", - "adapter": "celestrak_tle", - "category": "sat.tle", - "data": { - "norad_id": 25544, - "satellite_name": "ISS (ZARYA)", - "tle_line1": "1 25544U 98067A 26163.51782528 .00020000 00000-0 35000-3 0 9999", - "tle_line2": "2 25544 51.6416 247.4627 0006703 130.5360 325.0288 15.49815002 17", - "epoch": "2026-06-12T12:25:40Z", - }, - }, -} - -N2YO_PASS_ENVELOPE = { - "specversion": "1.0", - "id": "pass-25544-twinfalls-1718300000", - "source": "central", - "type": "central.sat.pass", - "data": { - "id": "pass-25544-twinfalls-1718300000", - "adapter": "n2yo_visualpasses", - "category": "sat.pass", - "data": { - "norad_id": 25544, - "sat_name": "ISS (ZARYA)", - "observer": "Twin Falls", - "max_elevation": 72.5, - "aos": "2026-06-13T04:15:00Z", - "los": "2026-06-13T04:21:30Z", - "direction": "visible", - }, - }, -} - -SATPASS_PREDICT_ENVELOPE = { - "specversion": "1.0", - "id": "pass-25544-boise-1718300500", - "source": "central", - "type": "central.sat.pass", - "data": { - "id": "pass-25544-boise-1718300500", - "adapter": "satpass_predict", - "category": "sat.pass", - "data": { - "norad_id": 25544, - "sat_name": "ISS (ZARYA)", - "observer": "Boise", - "max_elevation": 45.0, - "aos": "2026-06-13T04:20:00Z", - "los": "2026-06-13T04:26:00Z", - "direction": "visible", - }, - }, -} - - -def _enable_satpass(): - """Set satpass.enabled=true in the test DB.""" - from meshai.persistence import get_db - from meshai.adapter_config import invalidate_cache - conn = get_db() - conn.execute( - "UPDATE adapter_config SET value_json='true' " - "WHERE adapter='satpass' AND key='enabled'" - ) - invalidate_cache() - - -# ── TLE handler route ────────────────────────────────────────────── - -def test_tle_handler_inserts_sat_tles_row(): - """A celestrak_tle envelope must land a row in sat_tles.""" - from meshai.central.tle_handler import handle_tle - from meshai.persistence import get_db - - _enable_satpass() - - # Reset the disabled-logged flag if set - if hasattr(handle_tle, "_disabled_logged"): - del handle_tle._disabled_logged - - result = handle_tle( - CELESTRAK_TLE_ENVELOPE, - "central.sat.tle.25544", - now=int(time.time()), - ) - - # TLE handler always returns None (storage-only) - assert result is None - - conn = get_db() - row = conn.execute( - "SELECT norad_id, name, line1, line2, epoch FROM sat_tles WHERE norad_id=25544" - ).fetchone() - assert row is not None, "TLE row not inserted into sat_tles" - assert row["name"] == "ISS (ZARYA)" - assert row["line1"].startswith("1 25544U") - assert row["line2"].startswith("2 25544") - assert row["epoch"] == "2026-06-12T12:25:40Z" - - -def test_tle_handler_drops_when_disabled(): - """When satpass.enabled=false, TLE handler drops and returns None.""" - from meshai.central.tle_handler import handle_tle - from meshai.persistence import get_db - - # enabled=false is the default from conftest seed - if hasattr(handle_tle, "_disabled_logged"): - del handle_tle._disabled_logged - - result = handle_tle( - CELESTRAK_TLE_ENVELOPE, - "central.sat.tle.25544", - now=int(time.time()), - ) - assert result is None - - conn = get_db() - row = conn.execute( - "SELECT norad_id FROM sat_tles WHERE norad_id=25544" - ).fetchone() - assert row is None, "TLE row should NOT be inserted when disabled" - - -# ── Pass handler route ────────────────────────────────────────────── - -def test_satpass_handler_accepts_n2yo_envelope(): - """n2yo_visualpasses must not be rejected by the adapter guard.""" - inner = N2YO_PASS_ENVELOPE["data"] - adapter = inner.get("adapter") - assert adapter == "n2yo_visualpasses" - assert adapter in ("n2yo_visualpasses", "satpass_predict") - - -def test_satpass_handler_accepts_satpass_predict_envelope(): - """satpass_predict must not be rejected by the adapter guard.""" - inner = SATPASS_PREDICT_ENVELOPE["data"] - adapter = inner.get("adapter") - assert adapter == "satpass_predict" - assert adapter in ("n2yo_visualpasses", "satpass_predict") - - -def test_satpass_handler_rejects_wrong_adapter(): - """An envelope with adapter='celestrak_tle' must be rejected.""" - from meshai.central.satpass_handler import handle_satpass - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - result = handle_satpass(CELESTRAK_TLE_ENVELOPE, "central.sat.tle.25544") - assert result is None - - -# ── Config key consistency ────────────────────────────────────────── - -def test_registry_has_min_elevation_not_deg(): - """The REGISTRY key must be 'min_elevation', not 'min_elevation_deg'.""" - from meshai.adapter_config.defaults import REGISTRY - assert ("satpass", "min_elevation") in REGISTRY - assert ("satpass", "min_elevation_deg") not in REGISTRY - - -def test_handler_reads_min_elevation(): - """satpass_handler must read cfg.min_elevation (matching REGISTRY key).""" - import inspect - from meshai.central import satpass_handler - src = inspect.getsource(satpass_handler) - assert "min_elevation" in src - assert "min_elevation_deg" not in src diff --git a/work/tests/test_satpass_handler.py b/work/tests/test_satpass_handler.py index 104c8e9..7d30233 100644 --- a/work/tests/test_satpass_handler.py +++ b/work/tests/test_satpass_handler.py @@ -1,204 +1,29 @@ -"""v0.7 satpass_handler tests.""" +"""v0.7 satpass_handler tests. -import pytest -from unittest.mock import MagicMock, patch +The Central envelope-ingest path (`handle_satpass`, +`consolidate_satpass_pending`, and the observer/norad/elevation/staleness +filtering that lived inside them) was retired with the Central NATS +consumer and deleted 2026-07 when the still-live wire-formatting + gate +code (`format_pass`, `gate_consolidated_pass`) moved to +`meshai.env.satellite.pass_format`. That live code's coverage now lives in +tests/test_satpass_native.py (dedup, wire format, commit callback, via the +native SatpassAdapter path) and tests/test_satpass_broadcast_safety.py +(format_pass formatting rules called directly). What remains here is the +one test that calls `format_pass` directly and doesn't fit either of those +files' focus. +""" - -def _envelope(norad_id=25544, sat_name="ISS", observer="Boise", - max_el=75.0, aos="2026-06-12T03:32:00Z", - los="2026-06-12T03:38:00Z", direction="NW-SE", - aos_compass="SW", los_compass="NE"): - """Build a CloudEvents envelope for a satellite pass.""" - return { - "specversion": "1.0", - "type": "central.sat.pass", - "source": "central", - "id": f"pass-{norad_id}-{aos}", - "data": { - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "severity": 0, - "data": { - "norad_id": norad_id, - "satellite_name": sat_name, - "observer_name": observer, - "max_elevation_deg": max_el, - "aos_time": aos, - "los_time": los, - "azimuth_at_peak_compass": direction, - "azimuth_at_aos_compass": aos_compass, - "azimuth_at_los_compass": los_compass, - } - } - } - - -@pytest.fixture -def mock_db(): - """Mock database connection.""" - conn = MagicMock() - conn.execute.return_value.fetchone.return_value = None - conn.execute.return_value.lastrowid = 1 - with patch("meshai.central.satpass_handler.get_db", return_value=conn): - yield conn - - -@pytest.fixture -def mock_adapter_config(): - """Mock adapter_config.satpass.""" - cfg = MagicMock() - cfg.enabled = True - cfg.observers = [] # empty = all observers - cfg.min_elevation = 30 - 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 - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - 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_adapter_config): - """A pass with high elevation should broadcast (via consolidation).""" - env = _envelope(max_el=75.0) - result = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result is not None - 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.""" - from meshai.central.satpass_handler import handle_satpass - - mock_adapter_config.min_elevation = 30 - env = _envelope(max_el=25.0) - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - - assert result is None - - def test_observer_filter_blocks_mismatch(self, mock_db, mock_adapter_config): - """A pass for non-configured observer should be filtered.""" - from meshai.central.satpass_handler import handle_satpass - - mock_adapter_config.observers = ["Magic Valley"] - env = _envelope(observer="Boise") - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - - assert result is None - - def test_observer_filter_allows_match(self, mock_adapter_config): - """A pass for configured observer should broadcast.""" - mock_adapter_config.observers = ["Boise", "Magic Valley"] - env = _envelope(observer="Boise", max_el=45.0) - 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): - """NORAD ID filter should block non-matching satellites.""" - from meshai.central.satpass_handler import handle_satpass - - mock_adapter_config.norad_ids = [25544] # ISS only - env = _envelope(norad_id=12345, max_el=60.0) - result = handle_satpass(env, "central.sat.pass.other", data={}, now=1718163120) - - assert result is None - - def test_dedup_blocks_second_broadcast(self, mock_adapter_config): - """Second pass in same hour bucket should be deduplicated.""" - env = _envelope(max_el=60.0) - - # First pass consolidates into a real broadcast. - result1 = _ingest_and_consolidate(env, "central.sat.pass.iss", - now=1781235000) - assert result1 is not None - - # Simulate the broadcast committing (marks satpass_events broadcast). - _wire, data1 = result1 - data1["_on_broadcast_committed"](1781235010) - - # 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_adapter_config): - """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", - now=1781235000) - assert result is not None - wire, _ = result - - 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).""" - env = _envelope(max_el=60.0) - 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" - - def test_wrong_adapter_ignored(self, mock_db, mock_adapter_config): - """Envelope with wrong adapter should be ignored.""" - from meshai.central.satpass_handler import handle_satpass - - env = _envelope() - env["data"]["adapter"] = "some_other_adapter" - result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120) - - assert result is None +from __future__ import annotations # ============================================================================ # Budget-fit SAFETY CAP: a pathologically long satellite name must not push # the broadcast string past 140 chars. Calls format_pass directly (bypasses -# the broken consolidation path). +# the consolidation path entirely). # ============================================================================ def test_format_pass_worst_case_fits_140(): - from meshai.central.satpass_handler import format_pass + from meshai.env.satellite.pass_format import format_pass wire = format_pass( sat_name=("NOAA-19 EXPERIMENTAL SUPER LONG SATELLITE DESIGNATION " "PAYLOAD REVISION X PROTOTYPE FLIGHT MODEL SERIAL 00042"), diff --git a/work/tests/test_satpass_native.py b/work/tests/test_satpass_native.py index 72f6e2d..8a35ef1 100644 --- a/work/tests/test_satpass_native.py +++ b/work/tests/test_satpass_native.py @@ -1,13 +1,12 @@ """Tests for the native SGP4 satpass adapter (env.satpass). The native adapter computes every observer for a satellite in ONE tick, so it -consolidates in-memory and gates synchronously via the SHARED -`satpass_handler.gate_consolidated_pass` — with NO `satpass_pending` buffer and -NO Central consumer/timer. These tests monkeypatch `compute_passes` and +consolidates in-memory and gates synchronously via +`env.satellite.pass_format.gate_consolidated_pass` — with no buffer table and +no consumer/timer. These tests monkeypatch `compute_passes` and `get_observers` (no SGP4 / no network) and seed a fresh `sat_tles` row, then exercise: multi-observer consolidation, cross-tick dedup via `satpass_events`, -resilient empty cases, the precomposed aos→peak→los wire, and a guard that the -Central path still feeds the shared gate the correctly-merged consolidation. +resilient empty cases, and the precomposed aos→peak→los wire. """ from __future__ import annotations @@ -18,8 +17,8 @@ import pytest from meshai.env.satpass import SatpassAdapter from meshai.config import SatpassConfig -from meshai.central.pass_predictor import PassInfo -from meshai.central.tle_handler import upsert_tle +from meshai.env.satellite.pass_predictor import PassInfo +from meshai.env.satellite.tle_store import upsert_tle from meshai.persistence import get_db @@ -85,7 +84,7 @@ def _adapter(**overrides) -> SatpassAdapter: def _patch_predictor(monkeypatch, fake): - monkeypatch.setattr("meshai.central.pass_predictor.compute_passes", fake) + monkeypatch.setattr("meshai.env.satellite.pass_predictor.compute_passes", fake) def _patch_observers(monkeypatch, observers): @@ -253,63 +252,6 @@ def test_emitted_event_renders_aos_peak_los(monkeypatch): assert composed.startswith("\U0001F6F0") # satellite emoji -# ══════════════════════════════════════════════════════════════════════ -# 5. SHARED-GATE EXTRACTION GUARD (Central path still feeds the gate right) -# ══════════════════════════════════════════════════════════════════════ - -def test_central_consolidate_feeds_shared_gate_merged(monkeypatch): - """consolidate_satpass_pending must merge observers and hand the SAME - shared gate a correctly-consolidated dict (earliest AOS / latest LOS / - max-el observer / entry+exit / observer_list). Spying the gate proves the - Central path routes through the extracted, source-agnostic function.""" - import meshai.central.satpass_handler as sh - - conn = get_db() - cid = f"25544:{T0 // 3600}" - # Two pending rows for the same canonical id (boise entry, twin exit+peak). - rows = [ - (cid, "boise", "ISS", 25544, 40.0, T0, T0 + 300, "SW", "NW", "E", T0, T0 + 5), - (cid, "twin", "ISS", 25544, 70.0, T0 + 60, T0 + 400, "W", "NE", "S", T0, T0 + 5), - ] - for r in rows: - 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, peak_compass, " - "received_at, due_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", r) - - captured = {} - - def _spy(consolidated, *, now): - captured["c"] = consolidated - captured["now"] = now - return None # short-circuit: no broadcast side effects - - monkeypatch.setattr(sh, "gate_consolidated_pass", _spy) - - result = sh.consolidate_satpass_pending(cid) - assert result is None - - c = captured["c"] - assert c["consolidated_id"] == cid - assert c["norad_id"] == 25544 - assert c["max_elevation"] == 70.0 # twin (higher) - assert c["aos_epoch"] == T0 # boise earliest AOS - assert c["los_epoch"] == T0 + 400 # twin latest LOS - assert c["aos_compass"] == "SW" # boise - assert c["los_compass"] == "NE" # twin - assert c["peak_compass"] == "S" # twin (max-el observer) - assert c["entry_observer"] == "boise" - assert c["exit_observer"] == "twin" - assert c["observer_list"] == "boise,twin" - - # Pending buffer is drained regardless of the gate's decision. - left = conn.execute( - "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?", - (cid,)).fetchone() - assert left["n"] == 0 - - # ══════════════════════════════════════════════════════════════════════ # 6. IMMINENCE BROADCAST TRIGGER (satpass's "just received" analog) # ══════════════════════════════════════════════════════════════════════ diff --git a/work/tests/test_satpass_persisted_timer.py b/work/tests/test_satpass_persisted_timer.py index d38d2bf..8e5fee8 100644 --- a/work/tests/test_satpass_persisted_timer.py +++ b/work/tests/test_satpass_persisted_timer.py @@ -1,4 +1,4 @@ -"""Tests for the satpass persisted-timer `due_at` column. +"""Tests for the satpass persisted-timer `due_at` column (schema only). Pending satellite-pass consolidations used to be scheduled only as in-memory asyncio TimerHandles, so a restart orphaned any satpass_pending rows: the row @@ -11,62 +11,24 @@ The Central NATS consumer (and `_sweep_pending_satpass` with it) was retired 2026-07 -- the sweep's tests are gone with it. The native satpass path (env/satpass.py) never used the satpass_pending buffer or this sweep in the first place (it consolidates in-memory within a single tick), so nothing -live is affected. What remains here: - - `due_at` is persisted on the normal ingest path (satpass_handler.py, - still live -- shared by both paths historically, now native-only) +live is affected. + +The ingest path that wrote `due_at` (`central.satpass_handler.handle_satpass`) +was itself dead -- its only caller was the already-retired Central consumer +-- and was deleted in the same pass that relocated the still-live satellite +code to `env.satellite` (2026-07). Its due_at-persistence test is gone with +it; the `satpass_pending` table (and its due_at/peak_compass columns) are +now unused by any live code path, but the migrations that created them are +left in place (schema changes are out of scope for that pass). What remains +here: - SCHEMA_VERSION == 26 and the v22 migration (which added the due_at column) still applies cleanly on a fresh DB """ from __future__ import annotations -import json -import time - import pytest from meshai.persistence import get_db, init_db, SCHEMA_VERSION -from meshai.adapter_config import invalidate_cache - - -# ── helpers ─────────────────────────────────────────────────────────── - -def _enable_satpass_db(norad_ids=(25544,), dry_run=True): - """Enable satpass and set opt-in norad_ids in the test DB.""" - conn = get_db() - conn.execute("UPDATE adapter_config SET value_json='true' " - "WHERE adapter='satpass' AND key='enabled'") - conn.execute("UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='dry_run'", - (json.dumps(bool(dry_run)),)) - conn.execute("UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='norad_ids'", - (json.dumps(list(norad_ids)),)) - invalidate_cache() - - -def _ingest_envelope(norad_id=25544, observer="Boise", max_el=72.5, - aos="2026-06-12T03:32:00Z", los="2026-06-12T03:38:00Z"): - return { - "specversion": "1.0", - "type": "central.sat.pass", - "source": "central", - "id": f"pass-{norad_id}-{aos}", - "data": { - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "data": { - "norad_id": norad_id, - "satellite_name": "ISS", - "observer_name": observer, - "max_elevation_deg": max_el, - "aos_time": aos, - "los_time": los, - "azimuth_at_peak_compass": "S", - "azimuth_at_aos_compass": "SW", - "azimuth_at_los_compass": "NE", - }, - }, - } # ── schema / migration ─────────────────────────────────────────────── @@ -94,29 +56,3 @@ def test_v22_migration_applies_and_adds_due_at_column(tmp_path, monkeypatch): assert "due_at" in cols close_thread_connection() persistence_db._initialised.discard(db) - - -# ── due_at persisted on normal ingest ──────────────────────────────── - -def test_due_at_persisted_on_normal_ingest(): - """handle_satpass writes due_at = received_at + CONSOLIDATION_DELAY.""" - from meshai.central.satpass_handler import ( - handle_satpass, CONSOLIDATION_DELAY, _parse_iso_epoch) - _enable_satpass_db(norad_ids=[25544], dry_run=True) - for attr in ("_disabled_logged", "_no_norad_ids_logged"): - if hasattr(handle_satpass, attr): - delattr(handle_satpass, attr) - - env = _ingest_envelope() - aos_epoch = _parse_iso_epoch("2026-06-12T03:32:00Z") - now = aos_epoch - 300 # inside horizon, before los - assert handle_satpass(env, "central.sat.pass.iss", now=now) is None - - conn = get_db() - row = conn.execute( - "SELECT received_at, due_at FROM satpass_pending " - "WHERE norad_id=25544").fetchone() - assert row is not None, "ingest did not write a pending row" - assert row["due_at"] is not None - assert row["due_at"] == row["received_at"] + CONSOLIDATION_DELAY - assert row["due_at"] == now + CONSOLIDATION_DELAY diff --git a/work/tests/test_satpass_wire_fields.py b/work/tests/test_satpass_wire_fields.py deleted file mode 100644 index 5bc5fd0..0000000 --- a/work/tests/test_satpass_wire_fields.py +++ /dev/null @@ -1,262 +0,0 @@ -"""Tests for satpass_handler wire field name reads. - -Uses the verbatim live NOAA-18 envelope captured from Central NATS -(central.sat.pass.us.id.filer, 2026-06-10). Proves: - 1. Handler extracts correct norad_id, satellite_name, observer_name, - max_elevation_deg, aos_time, los_time from the actual wire format. - 2. satpass_events row is inserted with correct values. - 3. Envelope missing norad_id is rejected (returns None). - 4. Wire message format includes the correct extracted values. - 5. Category mapping: pass.n2yo_visualpasses -> sat_pass. -""" -from __future__ import annotations - -import json -import time - -import pytest - - -# ── Verbatim live NOAA-18 envelope from Central NATS ──────────────── - -NOAA18_ENVELOPE = { - "id": "filer:28654:2026-06-10T04:34:40+00:00", - "source": "central.echo6.co", - "type": "central.pass.n2yo_visualpasses.v1", - "time": "2026-06-10T04:41:35+00:00", - "datacontenttype": "application/json", - "centralschemaversion": "1.0", - "centralcategory": "pass.n2yo_visualpasses", - "centralseverity": 1, - "specversion": "1.0", - "data": { - "id": "filer:28654:2026-06-10T04:34:40+00:00", - "adapter": "n2yo_visualpasses", - "category": "pass.n2yo_visualpasses", - "time": "2026-06-10T04:41:35Z", - "expires": None, - "severity": 1, - "geo": { - "centroid": [-114.6, 42.57], - "bbox": None, - "regions": ["US-ID"], - "primary_region": "US-ID", - "geometry": None, - }, - "data": { - "observer_name": "Filer", - "observer_slug": "filer", - "observer_state": "ID", - "norad_id": 28654, - "satellite_name": "NOAA 18", - "aos_time": "2026-06-10T04:34:40+00:00", - "peak_time": "2026-06-10T04:41:35+00:00", - "los_time": "2026-06-10T04:48:30+00:00", - "max_elevation_deg": 22.69, - "magnitude": 6.7, - "azimuth_at_aos": 125.6, - "azimuth_at_aos_compass": "SE", - "azimuth_at_peak": 63.0, - "azimuth_at_peak_compass": "ENE", - "azimuth_at_los": 359.5, - "azimuth_at_los_compass": "N", - "duration_s": 630, - }, - }, -} - - -def _enable_satpass(norad_ids=None): - """Set satpass.enabled=true in the test DB.""" - from meshai.persistence import get_db - from meshai.adapter_config import invalidate_cache - conn = get_db() - conn.execute( - "UPDATE adapter_config SET value_json='true' " - "WHERE adapter='satpass' AND key='enabled'" - ) - # Set min_elevation low enough to accept this 22.69 deg pass - conn.execute( - "UPDATE adapter_config SET value_json='5' " - "WHERE adapter='satpass' AND key='min_elevation'" - ) - # Disable dry_run for tests that expect wire output - conn.execute( - "UPDATE adapter_config SET value_json='false' " - "WHERE adapter='satpass' AND key='dry_run'" - ) - # Set norad_ids (must be non-empty for opt-in) - if norad_ids is None: - norad_ids = [28654] - conn.execute( - "UPDATE adapter_config SET value_json=? " - "WHERE adapter='satpass' AND key='norad_ids'", - (json.dumps(norad_ids),) - ) - 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(): - """Verbatim NOAA-18 envelope inserts row with correct field values.""" - from meshai.central.satpass_handler import handle_satpass - from meshai.persistence import get_db - - _enable_satpass() - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - - now = 1781065800 # before NOAA18 envelope los_time - result = _ingest_and_consolidate( - NOAA18_ENVELOPE, - "central.sat.pass.us.id.filer", - now=now, - ) - assert result is not None, "handler returned None -- field extraction failed" - wire, _ = result - - conn = get_db() - rows = conn.execute( - "SELECT norad_id, sat_name, observer, max_elevation, aos_at, los_at " - "FROM satpass_events WHERE norad_id=28654" - ).fetchall() - assert len(rows) >= 1, "no satpass_events row for norad_id=28654" - row = rows[0] - - assert row["norad_id"] == 28654 - assert row["sat_name"] == "NOAA 18" - assert row["observer"] == "Filer" - assert abs(row["max_elevation"] - 22.69) < 0.01 - # aos_time = 2026-06-10T04:34:40+00:00 -> epoch - assert row["aos_at"] is not None - assert row["los_at"] is not None - # los must be after aos - assert row["los_at"] > row["aos_at"] - - -def test_noaa18_wire_message_format(): - """Wire message includes satellite name, direction in new 2-line format.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - - result = _ingest_and_consolidate( - NOAA18_ENVELOPE, - "central.sat.pass.us.id.filer", - now=1781065800, # before NOAA18 envelope los_time - ) - assert result is not None - wire, _ = result - - # 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(): - """Envelope with norad_id removed returns None.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - - # Deep copy and remove norad_id - import copy - env = copy.deepcopy(NOAA18_ENVELOPE) - del env["data"]["data"]["norad_id"] - - wire = handle_satpass( - env, - "central.sat.pass.us.id.filer", - data={}, - now=1781065800, # before NOAA18 envelope los_time - ) - assert wire is None, "handler should reject envelope without norad_id" - - -def test_missing_max_elevation_deg_rejected(): - """Envelope with max_elevation_deg removed returns None.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - - import copy - env = copy.deepcopy(NOAA18_ENVELOPE) - del env["data"]["data"]["max_elevation_deg"] - - wire = handle_satpass( - env, - "central.sat.pass.us.id.filer", - data={}, - now=1781065800, # before NOAA18 envelope los_time - ) - assert wire is None, "handler should reject envelope without max_elevation_deg" - - -def test_observer_fallback_to_slug(): - """When observer_name is absent, falls back to observer_slug.""" - from meshai.central.satpass_handler import handle_satpass - - _enable_satpass() - if hasattr(handle_satpass, "_disabled_logged"): - del handle_satpass._disabled_logged - if hasattr(handle_satpass, "_no_norad_ids_logged"): - del handle_satpass._no_norad_ids_logged - - import copy - env = copy.deepcopy(NOAA18_ENVELOPE) - del env["data"]["data"]["observer_name"] - # observer_slug = "filer" still present - - result = _ingest_and_consolidate( - env, - "central.sat.pass.us.id.filer", - now=1781065800, # before NOAA18 envelope los_time - ) - assert result is not None - # Observer name stored in DB, not in broadcast wire format - from meshai.persistence import get_db - conn = get_db() - row = conn.execute( - "SELECT observer FROM satpass_events WHERE norad_id=28654" - ).fetchone() - assert row is not None - assert row["observer"] == "filer" diff --git a/work/tests/test_tle_fetch.py b/work/tests/test_tle_fetch.py index 86b599e..134a844 100644 --- a/work/tests/test_tle_fetch.py +++ b/work/tests/test_tle_fetch.py @@ -15,7 +15,7 @@ from meshai.env.tle_fetch import ( parse_tle_block, parse_tle_epoch, ) -from meshai.central.tle_handler import get_tle_by_norad +from meshai.env.satellite.tle_store import get_tle_by_norad from meshai.config import SatpassConfig from meshai.persistence import get_db