diff --git a/services/meteor_tracking/__init__.py b/services/meteor_tracking/__init__.py index 0564393..c6f9cc8 100644 --- a/services/meteor_tracking/__init__.py +++ b/services/meteor_tracking/__init__.py @@ -10,9 +10,20 @@ """ # Core modules (no external dependencies beyond stdlib) +from .hopi_circles import ( + SearchCircle, + SearchPattern, + generate_hopi_circles, +) +from .lexicon_prayers import ( + LexiconFormatter, + generate_prayer_of_finding, + generate_prayer_of_watching, + generate_status_prayer, +) from .shower_calendar import ( - ShowerCalendar, MeteorShower, + ShowerCalendar, get_current_shower, get_next_major_shower, ) @@ -21,38 +32,53 @@ calculate_trajectory, is_visible_from, ) -from .hopi_circles import ( - SearchPattern, - SearchCircle, - generate_hopi_circles, -) from .watch_manager import ( - WatchWindow, - WatchManager, WatchIntensity, + WatchManager, WatchRequestParser, -) -from .lexicon_prayers import ( - generate_prayer_of_finding, - generate_prayer_of_watching, - generate_status_prayer, - LexiconFormatter, + WatchWindow, ) # Lazy imports for aiohttp-dependent modules def get_fireball_clients(): """Get fireball API clients (requires aiohttp).""" - from .fireball_client import CNEOSClient, Fireball, AMSClient, AMSFireball + from .fireball_client import AMSClient, AMSFireball, CNEOSClient, Fireball return CNEOSClient, Fireball, AMSClient, AMSFireball def get_meteor_service(): """Get meteor tracking service (requires aiohttp).""" - from .meteor_service import MeteorTrackingService, MeteorConfig, MeteorAlert + from .meteor_service import MeteorAlert, MeteorConfig, MeteorTrackingService return MeteorTrackingService, MeteorConfig, MeteorAlert +def get_close_approach_client(): + """Get NASA/JPL CNEOS Close Approach Data client (requires aiohttp).""" + from .close_approach_client import ( + CADClient, + CloseApproach, + ThreatLevel, + estimate_diameter_from_h, + fetch_upcoming_approaches, + generate_approach_prayer, + ) + return ( + CADClient, + CloseApproach, + ThreatLevel, + estimate_diameter_from_h, + fetch_upcoming_approaches, + generate_approach_prayer, + ) + + +def get_neo_feed_client(): + """Get NASA NEO Feed API client (requires aiohttp).""" + from .neo_feed_client import NEOFeedClient + return NEOFeedClient + + __all__ = [ # Shower calendar 'ShowerCalendar', diff --git a/services/meteor_tracking/close_approach_client.py b/services/meteor_tracking/close_approach_client.py new file mode 100644 index 0000000..fb3a3de --- /dev/null +++ b/services/meteor_tracking/close_approach_client.py @@ -0,0 +1,373 @@ +""" +NIGHTWATCH Close Approach Data Client +NASA CNEOS Close Approach API for near-Earth object monitoring. + +Tracks asteroids and comets approaching Earth, providing early warning +for potential atmospheric entry events. This fills the gap between +fireball detection (post-entry) and close approach awareness (pre-entry). + +API Documentation: https://ssd-api.jpl.nasa.gov/doc/cad.html +""" + +import logging +import math +from dataclasses import dataclass +from datetime import datetime, timedelta +from enum import Enum +from typing import List, Optional + +import aiohttp + +logger = logging.getLogger("NIGHTWATCH.MeteorTracking") + + +class ThreatLevel(Enum): + """Threat classification for close approaches.""" + ROUTINE = "routine" # > 10 LD, small object + NOTABLE = "notable" # < 10 LD or large object + SIGNIFICANT = "significant" # < 5 LD and detectable size + WATCH = "watch" # < 2 LD or potentially hazardous + ALERT = "alert" # < 1 LD or imminent entry predicted + + +@dataclass +class CloseApproach: + """A near-Earth object close approach event.""" + designation: str # Object designation (e.g., "2024 BX1") + close_approach_date: datetime # Date/time of closest approach + distance_au: float # Nominal miss distance in AU + distance_ld: float # Miss distance in lunar distances + distance_km: float # Miss distance in km + relative_velocity_km_s: float # Relative velocity at close approach + absolute_magnitude_h: float | None # Absolute magnitude (H) + estimated_diameter_m_min: float | None # Minimum estimated diameter + estimated_diameter_m_max: float | None # Maximum estimated diameter + is_potentially_hazardous: bool # PHA flag + orbit_id: str | None # Orbit solution ID + fullname: str | None # Full object name + + @property + def threat_level(self) -> ThreatLevel: + """Classify threat level based on distance and size.""" + if self.distance_ld < 1.0 or self.is_potentially_hazardous: + return ThreatLevel.ALERT + if self.distance_ld < 2.0: + return ThreatLevel.WATCH + if self.distance_ld < 5.0 and self.estimated_diameter_m_max and self.estimated_diameter_m_max > 10: + return ThreatLevel.SIGNIFICANT + if self.distance_ld < 10.0 or (self.estimated_diameter_m_max and self.estimated_diameter_m_max > 50): + return ThreatLevel.NOTABLE + return ThreatLevel.ROUTINE + + @property + def estimated_diameter_str(self) -> str: + """Format diameter range as string.""" + if self.estimated_diameter_m_min is not None and self.estimated_diameter_m_max is not None: + if self.estimated_diameter_m_max < 1: + return f"{self.estimated_diameter_m_min*100:.0f}-{self.estimated_diameter_m_max*100:.0f} cm" + if self.estimated_diameter_m_max < 1000: + return f"{self.estimated_diameter_m_min:.0f}-{self.estimated_diameter_m_max:.0f} m" + return f"{self.estimated_diameter_m_min/1000:.1f}-{self.estimated_diameter_m_max/1000:.1f} km" + return "unknown" + + @property + def approach_id(self) -> str: + """Generate unique ID for this approach.""" + date_str = self.close_approach_date.strftime('%Y%m%d') + return f"cad_{self.designation.replace(' ', '_')}_{date_str}" + + @property + def lexicon_str(self) -> str: + """Format for Lexicon prayer output.""" + return ( + f"{self.designation} approach {self.distance_ld:.1f}LD " + f"at {self.relative_velocity_km_s:.1f}km/s " + f"({self.threat_level.value})" + ) + + def hours_until_approach(self) -> float: + """Hours until closest approach (negative if past).""" + delta = self.close_approach_date - datetime.utcnow() + return delta.total_seconds() / 3600 + + +def estimate_diameter_from_h(h_mag: float) -> tuple[float, float]: + """ + Estimate asteroid diameter range from absolute magnitude (H). + + Uses the standard relation: D = 1329 / sqrt(albedo) * 10^(-H/5) + Assumes albedo range of 0.05 (dark) to 0.25 (bright). + + Args: + h_mag: Absolute magnitude (H) + + Returns: + (min_diameter_m, max_diameter_m) tuple + """ + # D(km) = 1329 / sqrt(albedo) * 10^(-H/5) + d_bright_km = 1329.0 / math.sqrt(0.25) * (10 ** (-h_mag / 5)) + d_dark_km = 1329.0 / math.sqrt(0.05) * (10 ** (-h_mag / 5)) + return d_bright_km * 1000, d_dark_km * 1000 # Convert to meters + + +class CADClient: + """ + Async client for NASA CNEOS Close Approach Data API. + + Monitors near-Earth objects making close approaches to Earth. + This provides the pre-entry awareness that complements the + CNEOS Fireball API (post-entry detection). + + API: https://ssd-api.jpl.nasa.gov/cad.api + No API key required. Rate limiting applies. + """ + + BASE_URL = "https://ssd-api.jpl.nasa.gov/cad.api" + + # 1 Lunar Distance in AU + LD_TO_AU = 0.00257 + AU_TO_KM = 149_597_870.7 + + def __init__(self, session: aiohttp.ClientSession | None = None): + self._session = session + self._owns_session = session is None + + async def _get_session(self) -> aiohttp.ClientSession: + if self._session is None: + self._session = aiohttp.ClientSession( + headers={'User-Agent': 'NIGHTWATCH/1.0 (observatory-neo-tracking)'} + ) + return self._session + + async def close(self): + if self._owns_session and self._session: + await self._session.close() + self._session = None + + async def fetch_close_approaches( + self, + date_min: datetime | None = None, + date_max: datetime | None = None, + dist_max_au: float = 0.05, + min_h_mag: float | None = None, + sort: str = "dist", + limit: int = 50 + ) -> list[CloseApproach]: + """ + Fetch close approach data from CNEOS CAD API. + + Args: + date_min: Start of date range (default: now) + date_max: End of date range (default: +7 days) + dist_max_au: Maximum distance in AU (0.05 AU ~ 19.5 LD) + min_h_mag: Minimum absolute magnitude (filter small objects) + sort: Sort field ('dist', 'date', 'h') + limit: Maximum results + + Returns: + List of CloseApproach objects sorted by distance + """ + if date_min is None: + date_min = datetime.utcnow() + if date_max is None: + date_max = datetime.utcnow() + timedelta(days=7) + + params = { + 'date-min': date_min.strftime('%Y-%m-%d'), + 'date-max': date_max.strftime('%Y-%m-%d'), + 'dist-max': f'{dist_max_au}', + 'sort': sort, + 'limit': limit, + } + + if min_h_mag is not None: + params['h-max'] = str(min_h_mag) + + try: + session = await self._get_session() + async with session.get( + self.BASE_URL, + params=params, + timeout=aiohttp.ClientTimeout(total=30) + ) as response: + response.raise_for_status() + data = await response.json() + return self._parse_approaches(data) + + except aiohttp.ClientError as e: + logger.error(f"CNEOS CAD API error: {e}") + return [] + + async def fetch_today(self) -> list[CloseApproach]: + """Fetch close approaches for today and tomorrow.""" + now = datetime.utcnow() + return await self.fetch_close_approaches( + date_min=now, + date_max=now + timedelta(days=2), + dist_max_au=0.05 + ) + + async def fetch_watch_level(self, days: int = 7) -> list[CloseApproach]: + """Fetch approaches at WATCH level or higher within N days.""" + approaches = await self.fetch_close_approaches( + date_max=datetime.utcnow() + timedelta(days=days), + dist_max_au=0.013, # ~5 LD + ) + return [a for a in approaches if a.threat_level.value in ('watch', 'alert', 'significant')] + + def _parse_approaches(self, data: dict) -> list[CloseApproach]: + """Parse CAD API response into CloseApproach objects.""" + approaches = [] + + if 'data' not in data or 'fields' not in data: + logger.warning(f"CAD API returned unexpected format: {list(data.keys())}") + return approaches + + fields = data['fields'] + field_map = {field: idx for idx, field in enumerate(fields)} + + for row in data['data']: + try: + approach = self._parse_row(row, field_map) + if approach: + approaches.append(approach) + except (IndexError, KeyError, ValueError) as e: + logger.debug(f"Parse error for CAD row: {e}") + continue + + return approaches + + def _parse_row(self, row: list, field_map: dict) -> CloseApproach | None: + """Parse a single row from the CAD API response.""" + designation = self._get_field(row, field_map, 'des', '') + if not designation: + return None + + # Parse distance + dist_au = self._parse_float(self._get_field(row, field_map, 'dist')) + if dist_au is None: + return None + + dist_ld = dist_au / self.LD_TO_AU + dist_km = dist_au * self.AU_TO_KM + + # Parse velocity + v_rel = self._parse_float(self._get_field(row, field_map, 'v_rel')) + + # Parse absolute magnitude and estimate diameter + h_mag = self._parse_float(self._get_field(row, field_map, 'h')) + diameter_min, diameter_max = None, None + if h_mag is not None: + diameter_min, diameter_max = estimate_diameter_from_h(h_mag) + + # Parse date + cd_str = self._get_field(row, field_map, 'cd', '') + close_date = self._parse_cad_datetime(cd_str) + + # Check fullname + fullname = self._get_field(row, field_map, 'fullname') + + # Determine PHA status from orbit_id or size + is_pha = False + if h_mag is not None and h_mag <= 22.0 and dist_au < 0.05: + is_pha = True # Conservative: large + close = potentially hazardous + + return CloseApproach( + designation=designation, + close_approach_date=close_date, + distance_au=dist_au, + distance_ld=dist_ld, + distance_km=dist_km, + relative_velocity_km_s=v_rel or 0.0, + absolute_magnitude_h=h_mag, + estimated_diameter_m_min=diameter_min, + estimated_diameter_m_max=diameter_max, + is_potentially_hazardous=is_pha, + orbit_id=self._get_field(row, field_map, 'orbit_id'), + fullname=fullname.strip() if fullname else None + ) + + def _get_field(self, row: list, field_map: dict, field_name: str, default=None): + """Safely get a field value from a row.""" + idx = field_map.get(field_name) + if idx is not None and idx < len(row): + return row[idx] + return default + + def _parse_float(self, value) -> float | None: + if value is None or value == '': + return None + try: + return float(value) + except (ValueError, TypeError): + return None + + def _parse_cad_datetime(self, date_str: str) -> datetime: + """Parse CAD API datetime format (YYYY-Mon-DD HH:MM or variants).""" + formats = [ + '%Y-%b-%d %H:%M', # 2026-Mar-21 14:30 + '%Y-%m-%d %H:%M', # 2026-03-21 14:30 + '%Y-%b-%d', # 2026-Mar-21 + '%Y-%m-%d', # 2026-03-21 + ] + for fmt in formats: + try: + return datetime.strptime(date_str.strip(), fmt) + except ValueError: + continue + logger.debug(f"Could not parse CAD date: {date_str}") + return datetime.utcnow() + + +def generate_approach_prayer(approaches: list[CloseApproach]) -> str: + """ + Generate a Lexicon-style report for close approaches. + + nightwatch-approach-scan. + """ + lines = ["nightwatch-approach-scan."] + lines.append(f"varek: {datetime.utcnow().strftime('%Y-%m-%d %H:%M')} UTC") + lines.append("") + + if not approaches: + lines.append("neo-clear: no significant approaches") + lines.append("") + lines.append("velmu-sky-quiet.") + return "\n".join(lines) + + # Group by threat level + for level in (ThreatLevel.ALERT, ThreatLevel.WATCH, ThreatLevel.SIGNIFICANT, ThreatLevel.NOTABLE): + level_approaches = [a for a in approaches if a.threat_level == level] + if not level_approaches: + continue + + lines.append(f"--- {level.value.upper()} ---") + for a in level_approaches: + hours = a.hours_until_approach() + time_str = f"in {hours:.0f}h" if hours > 0 else f"{abs(hours):.0f}h ago" + lines.append(f" {a.designation}: {a.distance_ld:.1f} LD, " + f"{a.relative_velocity_km_s:.1f} km/s, " + f"~{a.estimated_diameter_str}, " + f"{time_str}") + lines.append("") + + routine_count = sum(1 for a in approaches if a.threat_level == ThreatLevel.ROUTINE) + if routine_count > 0: + lines.append(f"routine-passes: {routine_count}") + lines.append("") + + lines.append("presa-sky-aware.") + lines.append("do-good-us.") + return "\n".join(lines) + + +async def fetch_upcoming_approaches(days: int = 7, dist_max_au: float = 0.05) -> list[CloseApproach]: + """Convenience function to fetch upcoming close approaches.""" + client = CADClient() + try: + return await client.fetch_close_approaches( + date_max=datetime.utcnow() + timedelta(days=days), + dist_max_au=dist_max_au + ) + finally: + await client.close() diff --git a/services/meteor_tracking/neo_feed_client.py b/services/meteor_tracking/neo_feed_client.py new file mode 100644 index 0000000..6fd470c --- /dev/null +++ b/services/meteor_tracking/neo_feed_client.py @@ -0,0 +1,187 @@ +""" +NIGHTWATCH NASA NEO Feed API Client +Secondary data source via api.nasa.gov/neo/rest/v1/feed. + +Complements the CAD API in close_approach_client.py: +- CAD API: CNEOS close approach predictions (no auth needed) +- NEO Feed: NASA NeoWs with PHA flags and diameter estimates (DEMO_KEY or API key) + +The NEO Feed provides the is_potentially_hazardous_asteroid flag directly, +which the CAD API does not include. Combining both gives more complete data. + +API Documentation: https://api.nasa.gov/ (NeoWs section) +""" + +import logging +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from typing import List, Optional + +import aiohttp + +from .close_approach_client import CloseApproach, ThreatLevel + +logger = logging.getLogger("NIGHTWATCH.MeteorTracking") + + +# Conversion constants +AU_TO_KM = 149_597_870.7 +LD_TO_AU = 0.00257 + + +class NEOFeedClient: + """ + Async client for NASA NEO Feed API (NeoWs). + + Provides near-Earth object data with official PHA classification. + Requires API key (DEMO_KEY available for low-rate testing). + + API: https://api.nasa.gov/neo/rest/v1/feed + Rate limit: 30 req/hour (DEMO_KEY), 1000 req/hour (registered key) + """ + + BASE_URL = "https://api.nasa.gov/neo/rest/v1/feed" + + def __init__( + self, + api_key: str = "DEMO_KEY", + session: aiohttp.ClientSession | None = None + ): + self.api_key = api_key + self._session = session + self._owns_session = session is None + + async def _get_session(self) -> aiohttp.ClientSession: + if self._session is None: + self._session = aiohttp.ClientSession( + headers={'User-Agent': 'NIGHTWATCH/1.0 (observatory-neo-tracking)'} + ) + return self._session + + async def close(self): + if self._owns_session and self._session: + await self._session.close() + self._session = None + + async def fetch_neo_feed( + self, + start_date: date | None = None, + end_date: date | None = None + ) -> list[CloseApproach]: + """ + Fetch NEO feed data from NASA API. + + Note: Feed API limited to 7-day windows. + + Args: + start_date: Start date (default: today) + end_date: End date (default: 7 days from start, max 7 day span) + + Returns: + List of CloseApproach objects (same type as CAD client) + """ + if start_date is None: + start_date = date.today() + if end_date is None: + end_date = start_date + timedelta(days=7) + + # API limits to 7-day windows + if (end_date - start_date).days > 7: + end_date = start_date + timedelta(days=7) + + params = { + 'start_date': start_date.isoformat(), + 'end_date': end_date.isoformat(), + 'api_key': self.api_key, + } + + try: + session = await self._get_session() + async with session.get( + self.BASE_URL, + params=params, + timeout=aiohttp.ClientTimeout(total=30) + ) as response: + if response.status == 429: + logger.warning("NEO Feed API rate limit exceeded") + return [] + response.raise_for_status() + data = await response.json() + return self._parse_neo_feed(data) + + except aiohttp.ClientError as e: + logger.error(f"NEO Feed API error: {e}") + return [] + + def _parse_neo_feed(self, data: dict) -> list[CloseApproach]: + """Parse NEO Feed API response into CloseApproach objects.""" + approaches = [] + neo_objects = data.get('near_earth_objects', {}) + + for date_str, objects in neo_objects.items(): + for obj in objects: + try: + approach = self._parse_neo_object(obj) + if approach: + approaches.append(approach) + except (KeyError, ValueError, TypeError) as e: + logger.debug(f"Parse error for NEO object: {e}") + continue + + approaches.sort(key=lambda a: a.close_approach_date) + return approaches + + def _parse_neo_object(self, obj: dict) -> CloseApproach | None: + """Parse a single NEO object from the feed.""" + close_approach_data = obj.get('close_approach_data', []) + if not close_approach_data: + return None + + ca = close_approach_data[0] + + # Diameter estimates + diameter = obj.get('estimated_diameter', {}) + meters = diameter.get('meters', {}) + d_min = meters.get('estimated_diameter_min') + d_max = meters.get('estimated_diameter_max') + + # Distance + miss_distance = ca.get('miss_distance', {}) + dist_au = float(miss_distance.get('astronomical', 0)) + dist_km = float(miss_distance.get('kilometers', 0)) + dist_ld = dist_au / LD_TO_AU if dist_au else 0 + + # Velocity + rel_velocity = ca.get('relative_velocity', {}) + v_km_s = float(rel_velocity.get('kilometers_per_second', 0)) + + return CloseApproach( + designation=obj.get('neo_reference_id', 'unknown'), + close_approach_date=self._parse_datetime( + ca.get('close_approach_date_full', ca.get('close_approach_date', '')) + ), + distance_au=dist_au, + distance_ld=dist_ld, + distance_km=dist_km, + relative_velocity_km_s=v_km_s, + absolute_magnitude_h=obj.get('absolute_magnitude_h'), + estimated_diameter_m_min=d_min, + estimated_diameter_m_max=d_max, + is_potentially_hazardous=obj.get('is_potentially_hazardous_asteroid', False), + orbit_id=None, + fullname=obj.get('name', ''), + ) + + def _parse_datetime(self, date_str: str) -> datetime: + """Parse NEO Feed date formats.""" + formats = [ + '%Y-%b-%d %H:%M', + '%Y-%m-%d %H:%M', + '%Y-%m-%d', + ] + for fmt in formats: + try: + return datetime.strptime(date_str.strip(), fmt) + except ValueError: + continue + return datetime.utcnow() diff --git a/tests/unit/test_neo_client.py b/tests/unit/test_neo_client.py new file mode 100644 index 0000000..12a6a29 --- /dev/null +++ b/tests/unit/test_neo_client.py @@ -0,0 +1,557 @@ +""" +NIGHTWATCH Close Approach Data Client Tests +presa-nightwatch. velmu-sky. watching-what-approaches. + +Tests for the NASA JPL CAD API client that tracks asteroid/comet +close approaches to Earth. +""" + +import pytest +import math +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +from services.meteor_tracking.close_approach_client import ( + CloseApproach, + CADClient, + ThreatLevel, + estimate_diameter_from_h, + generate_approach_prayer, + fetch_upcoming_approaches, +) + + +# ========================================================================= +# Test data fixtures +# ========================================================================= + +def make_approach(**overrides) -> CloseApproach: + """Factory for CloseApproach test objects.""" + defaults = dict( + designation="2026 EG1", + close_approach_date=datetime(2026, 3, 12, 9, 15), + distance_au=0.00133, + distance_ld=0.518, + distance_km=198_925.0, + relative_velocity_km_s=9.62, + absolute_magnitude_h=27.5, + estimated_diameter_m_min=5.3, + estimated_diameter_m_max=11.8, + is_potentially_hazardous=False, + orbit_id="23", + fullname="(2026 EG1)", + ) + defaults.update(overrides) + return CloseApproach(**defaults) + + +# Sample API response matching JPL CAD format +SAMPLE_API_RESPONSE = { + "signature": {"version": "1.5", "source": "NASA/JPL SBDB Close Approach Data API"}, + "count": "3", + "fields": ["des", "orbit_id", "jd", "cd", "dist", "dist_min", "dist_max", + "v_rel", "v_inf", "t_sigma_f", "h", "diameter", "diameter_sigma", + "fullname"], + "data": [ + ["2026 EG1", "23", "2460747.885", "2026-Mar-12 09:15", "0.00133", + "0.00125", "0.00140", "9.62", "9.50", "00:03", "27.5", None, None, + " (2026 EG1)"], + ["2026 FU", "15", "2460758.321", "2026-Mar-22 19:42", "0.00987", + "0.00950", "0.01020", "12.44", "12.30", "00:08", "24.1", None, None, + " (2026 FU)"], + ["2026 FK", "8", "2460758.654", "2026-Mar-22 03:41", "0.01234", + "0.01200", "0.01280", "8.75", "8.60", "00:12", "25.8", None, None, + " (2026 FK)"], + ], +} + +EMPTY_API_RESPONSE = { + "signature": {"version": "1.5", "source": "NASA/JPL"}, + "count": "0", + "fields": ["des", "orbit_id", "jd", "cd", "dist", "dist_min", "dist_max", + "v_rel", "v_inf", "t_sigma_f", "h", "diameter", "diameter_sigma", + "fullname"], + "data": [], +} + + +# ========================================================================= +# CloseApproach dataclass tests +# ========================================================================= + +class TestCloseApproach: + """Test CloseApproach dataclass properties.""" + + def test_threat_level_alert_close_distance(self): + """< 1 LD = ALERT.""" + ca = make_approach(distance_ld=0.5, is_potentially_hazardous=False) + assert ca.threat_level == ThreatLevel.ALERT + + def test_threat_level_alert_pha(self): + """PHA flag = ALERT regardless of distance.""" + ca = make_approach(distance_ld=5.0, is_potentially_hazardous=True) + assert ca.threat_level == ThreatLevel.ALERT + + def test_threat_level_watch(self): + """< 2 LD = WATCH.""" + ca = make_approach(distance_ld=1.5, is_potentially_hazardous=False) + assert ca.threat_level == ThreatLevel.WATCH + + def test_threat_level_significant(self): + """< 5 LD + detectable size = SIGNIFICANT.""" + ca = make_approach( + distance_ld=3.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=20.0, + ) + assert ca.threat_level == ThreatLevel.SIGNIFICANT + + def test_threat_level_notable_distance(self): + """< 10 LD = NOTABLE.""" + ca = make_approach( + distance_ld=7.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=5.0, + ) + assert ca.threat_level == ThreatLevel.NOTABLE + + def test_threat_level_notable_large(self): + """Large object (> 50m) = NOTABLE even at distance.""" + ca = make_approach( + distance_ld=15.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=60.0, + ) + assert ca.threat_level == ThreatLevel.NOTABLE + + def test_threat_level_routine(self): + """> 10 LD + small = ROUTINE.""" + ca = make_approach( + distance_ld=25.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=5.0, + ) + assert ca.threat_level == ThreatLevel.ROUTINE + + def test_estimated_diameter_str_meters(self): + ca = make_approach(estimated_diameter_m_min=5.3, estimated_diameter_m_max=11.8) + assert "5-12 m" in ca.estimated_diameter_str or "5" in ca.estimated_diameter_str + + def test_estimated_diameter_str_cm(self): + ca = make_approach(estimated_diameter_m_min=0.01, estimated_diameter_m_max=0.05) + assert "cm" in ca.estimated_diameter_str + + def test_estimated_diameter_str_km(self): + ca = make_approach(estimated_diameter_m_min=1200, estimated_diameter_m_max=2500) + assert "km" in ca.estimated_diameter_str + + def test_estimated_diameter_str_unknown(self): + ca = make_approach(estimated_diameter_m_min=None, estimated_diameter_m_max=None) + assert ca.estimated_diameter_str == "unknown" + + def test_approach_id_format(self): + ca = make_approach( + designation="2026 EG1", + close_approach_date=datetime(2026, 3, 12), + ) + assert ca.approach_id == "cad_2026_EG1_20260312" + + def test_approach_id_uniqueness(self): + ca1 = make_approach(designation="2026 EG1") + ca2 = make_approach(designation="2026 FU") + assert ca1.approach_id != ca2.approach_id + + def test_lexicon_str(self): + ca = make_approach() + s = ca.lexicon_str + assert "2026 EG1" in s + assert "LD" in s + assert "km/s" in s + + def test_hours_until_approach_future(self): + future = datetime.utcnow() + timedelta(hours=5) + ca = make_approach(close_approach_date=future) + hours = ca.hours_until_approach() + assert 4.9 < hours < 5.1 + + def test_hours_until_approach_past(self): + past = datetime.utcnow() - timedelta(hours=3) + ca = make_approach(close_approach_date=past) + hours = ca.hours_until_approach() + assert -3.1 < hours < -2.9 + + +# ========================================================================= +# Diameter estimation tests +# ========================================================================= + +class TestDiameterEstimation: + """Test the H magnitude to diameter conversion.""" + + def test_estimate_large_asteroid(self): + """H=17 should give ~1-3 km diameter.""" + d_min, d_max = estimate_diameter_from_h(17.0) + assert d_min > 500 # > 500m + assert d_max < 5000 # < 5km + + def test_estimate_medium_asteroid(self): + """H=22 should give ~100-200m range.""" + d_min, d_max = estimate_diameter_from_h(22.0) + assert d_min > 50 + assert d_max < 500 + + def test_estimate_small_asteroid(self): + """H=27 should give ~5-15m range.""" + d_min, d_max = estimate_diameter_from_h(27.0) + assert d_min > 1 + assert d_max < 50 + + def test_estimate_tiny_asteroid(self): + """H=30 should give very small (< 5m).""" + d_min, d_max = estimate_diameter_from_h(30.0) + assert d_max < 10 + + def test_dark_gives_larger_diameter(self): + """Lower albedo (dark) should estimate larger diameter.""" + d_min, d_max = estimate_diameter_from_h(25.0) + assert d_max > d_min # d_max uses dark albedo + + def test_consistent_with_apophis(self): + """ + Cross-check: Apophis (H=19.7) is ~370m. + Our estimate range should bracket that. + """ + d_min, d_max = estimate_diameter_from_h(19.7) + assert d_min < 370 + assert d_max > 370 + + +# ========================================================================= +# CADClient parsing tests +# ========================================================================= + +class TestCADClientParsing: + """Test API response parsing (no network calls).""" + + def test_parse_sample_response(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + assert len(approaches) == 3 + assert approaches[0].designation == "2026 EG1" + assert approaches[1].designation == "2026 FU" + assert approaches[2].designation == "2026 FK" + + def test_parse_distances(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + eg1 = approaches[0] + assert eg1.distance_au == 0.00133 + # distance_ld should be derived from AU + assert eg1.distance_ld > 0 + # distance_km should be derived from AU + expected_km = 0.00133 * CADClient.AU_TO_KM + assert abs(eg1.distance_km - expected_km) < 1.0 + + def test_parse_velocity(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + assert approaches[0].relative_velocity_km_s == 9.62 + + def test_parse_h_magnitude(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + assert approaches[0].absolute_magnitude_h == 27.5 + assert approaches[1].absolute_magnitude_h == 24.1 + + def test_parse_fullname_stripped(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + assert approaches[0].fullname == "(2026 EG1)" + + def test_parse_diameter_estimation(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + eg1 = approaches[0] + assert eg1.estimated_diameter_m_min is not None + assert eg1.estimated_diameter_m_max is not None + assert eg1.estimated_diameter_m_min < eg1.estimated_diameter_m_max + + def test_parse_date_format(self): + client = CADClient() + approaches = client._parse_approaches(SAMPLE_API_RESPONSE) + eg1 = approaches[0] + assert eg1.close_approach_date.year == 2026 + assert eg1.close_approach_date.month == 3 + assert eg1.close_approach_date.day == 12 + + def test_parse_empty_response(self): + client = CADClient() + assert client._parse_approaches(EMPTY_API_RESPONSE) == [] + + def test_parse_malformed_response(self): + client = CADClient() + assert client._parse_approaches({}) == [] + assert client._parse_approaches({"fields": []}) == [] + assert client._parse_approaches({"data": []}) == [] + + def test_parse_row_missing_designation(self): + """Row missing designation is skipped.""" + client = CADClient() + bad_response = { + "fields": ["des", "orbit_id", "cd", "dist", "v_rel", "h", "fullname"], + "data": [ + ["", "1", "2026-Mar-12 09:15", "0.001", "10.0", "25", "Test"], + ["2026 YY", "1", "2026-Mar-12 09:15", "0.001", "10.0", "25", "Good"], + ], + } + approaches = client._parse_approaches(bad_response) + assert len(approaches) == 1 + assert approaches[0].designation == "2026 YY" + + def test_parse_row_missing_distance(self): + """Row missing distance is skipped.""" + client = CADClient() + bad_response = { + "fields": ["des", "orbit_id", "cd", "dist", "v_rel", "h", "fullname"], + "data": [ + ["2026 XX", "1", "2026-Mar-12 09:15", None, "10.0", "25", "Test"], + ["2026 YY", "1", "2026-Mar-12 09:15", "0.001", "10.0", "25", "Good"], + ], + } + approaches = client._parse_approaches(bad_response) + assert len(approaches) == 1 + assert approaches[0].designation == "2026 YY" + + +# ========================================================================= +# Date parsing tests +# ========================================================================= + +class TestCADDateParsing: + """Test JPL CAD date format variations.""" + + def test_parse_standard_jpl_format(self): + client = CADClient() + dt = client._parse_cad_datetime("2026-Mar-23 09:15") + assert dt.year == 2026 + assert dt.month == 3 + assert dt.day == 23 + assert dt.hour == 9 + assert dt.minute == 15 + + def test_parse_numeric_date_format(self): + client = CADClient() + dt = client._parse_cad_datetime("2026-03-23 09:15") + assert dt.month == 3 + assert dt.day == 23 + + def test_parse_date_only_jpl(self): + client = CADClient() + dt = client._parse_cad_datetime("2026-Mar-23") + assert dt.month == 3 + assert dt.day == 23 + + def test_parse_date_only_numeric(self): + client = CADClient() + dt = client._parse_cad_datetime("2026-03-23") + assert dt.month == 3 + + def test_parse_unparseable_date(self): + """Unparseable date should return current time, not crash.""" + client = CADClient() + dt = client._parse_cad_datetime("not-a-date") + assert isinstance(dt, datetime) + + +# ========================================================================= +# Approach prayer formatting tests +# ========================================================================= + +class TestApproachPrayer: + """Test Lexicon-style approach prayer formatting.""" + + def test_prayer_structure_with_approaches(self): + approaches = [make_approach(distance_ld=0.5)] # ALERT level + prayer = generate_approach_prayer(approaches) + assert "nightwatch-approach-scan." in prayer + assert "varek:" in prayer + assert "ALERT" in prayer + assert "2026 EG1" in prayer + assert "presa-sky-aware." in prayer + assert "do-good-us." in prayer + + def test_prayer_empty_approaches(self): + prayer = generate_approach_prayer([]) + assert "nightwatch-approach-scan." in prayer + assert "neo-clear" in prayer + assert "velmu-sky-quiet." in prayer + + def test_prayer_groups_by_threat(self): + approaches = [ + make_approach(designation="Close1", distance_ld=0.5), + make_approach(designation="Watch1", distance_ld=1.5, is_potentially_hazardous=False), + make_approach( + designation="Sig1", distance_ld=3.0, + is_potentially_hazardous=False, estimated_diameter_m_max=20.0, + ), + ] + prayer = generate_approach_prayer(approaches) + assert "ALERT" in prayer + assert "WATCH" in prayer + assert "SIGNIFICANT" in prayer + + def test_prayer_routine_count(self): + approaches = [ + make_approach( + designation=f"Routine{i}", + distance_ld=25.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=5.0, + ) + for i in range(3) + ] + prayer = generate_approach_prayer(approaches) + assert "routine-passes: 3" in prayer + + +# ========================================================================= +# Async client method tests (mocked network) +# ========================================================================= + +class TestCADClientAsync: + """Test async client methods with mocked HTTP.""" + + @pytest.mark.asyncio + async def test_fetch_close_approaches(self): + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.raise_for_status = MagicMock() + mock_response.json = AsyncMock(return_value=SAMPLE_API_RESPONSE) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=False) + + mock_session = AsyncMock() + mock_session.get = MagicMock(return_value=mock_response) + + client = CADClient(session=mock_session) + approaches = await client.fetch_close_approaches() + + assert len(approaches) == 3 + assert approaches[0].designation == "2026 EG1" + mock_session.get.assert_called_once() + + @pytest.mark.asyncio + async def test_fetch_today(self): + mock_response = AsyncMock() + mock_response.status = 200 + mock_response.raise_for_status = MagicMock() + mock_response.json = AsyncMock(return_value=EMPTY_API_RESPONSE) + mock_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_response.__aexit__ = AsyncMock(return_value=False) + + mock_session = AsyncMock() + mock_session.get = MagicMock(return_value=mock_response) + + client = CADClient(session=mock_session) + approaches = await client.fetch_today() + assert approaches == [] + + @pytest.mark.asyncio + async def test_fetch_handles_network_error(self): + import aiohttp + + mock_session = AsyncMock() + mock_session.get = MagicMock( + side_effect=aiohttp.ClientError("Connection failed") + ) + + client = CADClient(session=mock_session) + approaches = await client.fetch_close_approaches() + assert approaches == [] + + @pytest.mark.asyncio + async def test_close_owned_session(self): + client = CADClient() + mock_session = AsyncMock() + client._session = mock_session + client._owns_session = True + + await client.close() + mock_session.close.assert_called_once() + assert client._session is None + + @pytest.mark.asyncio + async def test_close_borrowed_session(self): + mock_session = AsyncMock() + client = CADClient(session=mock_session) + + await client.close() + mock_session.close.assert_not_called() + + +# ========================================================================= +# Integration-style tests (pure logic, no network) +# ========================================================================= + +class TestCADIntegration: + """Test CAD client in context of NIGHTWATCH workflow.""" + + def test_approach_to_prayer_flow(self): + """Full flow: CloseApproach -> generate_approach_prayer.""" + approach = make_approach( + distance_ld=0.518, + is_potentially_hazardous=False, + ) + + assert approach.threat_level == ThreatLevel.ALERT # < 1 LD + prayer = generate_approach_prayer([approach]) + assert "ALERT" in prayer + assert "2026 EG1" in prayer + + def test_multiple_approaches_sorted_by_threat(self): + approaches = [ + make_approach( + designation="Far", + distance_ld=25.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=5.0, + ), + make_approach(designation="Close", distance_ld=0.3), + make_approach( + designation="Mid", + distance_ld=8.0, + is_potentially_hazardous=False, + estimated_diameter_m_max=5.0, + ), + ] + + threat_order = { + ThreatLevel.ALERT: 0, + ThreatLevel.WATCH: 1, + ThreatLevel.SIGNIFICANT: 2, + ThreatLevel.NOTABLE: 3, + ThreatLevel.ROUTINE: 4, + } + sorted_approaches = sorted( + approaches, + key=lambda a: threat_order.get(a.threat_level, 99), + ) + + assert sorted_approaches[0].designation == "Close" + assert sorted_approaches[-1].designation == "Far" + + def test_approach_ids_unique_across_batch(self): + approaches = [ + make_approach( + designation=f"2026 {chr(65+i)}{chr(65+i)}", + close_approach_date=datetime(2026, 3, 12 + i), + ) + for i in range(10) + ] + + ids = [a.approach_id for a in approaches] + assert len(ids) == len(set(ids)) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/unit/test_neo_feed_client.py b/tests/unit/test_neo_feed_client.py new file mode 100644 index 0000000..4632182 --- /dev/null +++ b/tests/unit/test_neo_feed_client.py @@ -0,0 +1,245 @@ +""" +NIGHTWATCH NEO Feed Client Tests +Tests for NASA NEO Feed API client (secondary data source). + +presa-nightwatch. velmu-test. +""" + +import pytest +from datetime import datetime, timedelta +from unittest.mock import AsyncMock, MagicMock + +from services.meteor_tracking.close_approach_client import CloseApproach, ThreatLevel +from services.meteor_tracking.neo_feed_client import NEOFeedClient + + +# ========================================================================= +# Test Data Fixtures +# ========================================================================= + +SAMPLE_NEO_FEED_RESPONSE = { + 'element_count': 2, + 'near_earth_objects': { + '2026-03-23': [ + { + 'neo_reference_id': '2026FU', + 'name': '(2026 FU)', + 'absolute_magnitude_h': 27.0, + 'estimated_diameter': { + 'meters': { + 'estimated_diameter_min': 10.0, + 'estimated_diameter_max': 22.0, + } + }, + 'is_potentially_hazardous_asteroid': False, + 'close_approach_data': [ + { + 'close_approach_date': '2026-03-23', + 'close_approach_date_full': '2026-Mar-23 12:00', + 'miss_distance': { + 'astronomical': '0.00246', + 'kilometers': '367918', + 'lunar': '0.957', + }, + 'relative_velocity': { + 'kilometers_per_second': '8.5', + 'kilometers_per_hour': '30600', + } + } + ] + }, + { + 'neo_reference_id': '3200', + 'name': '3200 Phaethon', + 'absolute_magnitude_h': 14.6, + 'estimated_diameter': { + 'meters': { + 'estimated_diameter_min': 4600.0, + 'estimated_diameter_max': 5200.0, + } + }, + 'is_potentially_hazardous_asteroid': True, + 'close_approach_data': [ + { + 'close_approach_date': '2026-03-23', + 'close_approach_date_full': '2026-Mar-23 18:00', + 'miss_distance': { + 'astronomical': '0.5', + 'kilometers': '74798935', + 'lunar': '194.6', + }, + 'relative_velocity': { + 'kilometers_per_second': '25.0', + 'kilometers_per_hour': '90000', + } + } + ] + } + ] + } +} + + +# ========================================================================= +# NEO Feed Client Tests +# ========================================================================= + +class TestNEOFeedClient: + """Test NASA NEO Feed API client.""" + + def test_init_default(self): + """Test default initialization with DEMO_KEY.""" + client = NEOFeedClient() + assert client.api_key == "DEMO_KEY" + + def test_init_custom_key(self): + """Test initialization with custom API key.""" + client = NEOFeedClient(api_key="my_key_123") + assert client.api_key == "my_key_123" + + @pytest.mark.asyncio + async def test_parse_neo_feed(self): + """Test parsing NEO Feed API response.""" + client = NEOFeedClient() + approaches = client._parse_neo_feed(SAMPLE_NEO_FEED_RESPONSE) + + assert len(approaches) == 2 + # Should be sorted by approach date + assert approaches[0].close_approach_date <= approaches[1].close_approach_date + + @pytest.mark.asyncio + async def test_parse_neo_feed_properties(self): + """Test that parsed NEO objects have correct properties.""" + client = NEOFeedClient() + approaches = client._parse_neo_feed(SAMPLE_NEO_FEED_RESPONSE) + + fu = next(a for a in approaches if a.designation == '2026FU') + assert fu.fullname == '(2026 FU)' + assert fu.distance_au == pytest.approx(0.00246) + assert fu.distance_km == pytest.approx(367918.0) + assert fu.relative_velocity_km_s == pytest.approx(8.5) + assert fu.estimated_diameter_m_min == 10.0 + assert fu.estimated_diameter_m_max == 22.0 + assert fu.is_potentially_hazardous is False + + @pytest.mark.asyncio + async def test_parse_neo_feed_pha(self): + """Test PHA flag is correctly parsed from NEO Feed.""" + client = NEOFeedClient() + approaches = client._parse_neo_feed(SAMPLE_NEO_FEED_RESPONSE) + + phaethon = next(a for a in approaches if a.designation == '3200') + assert phaethon.is_potentially_hazardous is True + # PHA at any distance should be ALERT + assert phaethon.threat_level == ThreatLevel.ALERT + + @pytest.mark.asyncio + async def test_parse_neo_feed_empty(self): + """Test parsing empty feed response.""" + client = NEOFeedClient() + approaches = client._parse_neo_feed({'near_earth_objects': {}}) + assert len(approaches) == 0 + + @pytest.mark.asyncio + async def test_parse_neo_feed_no_approach_data(self): + """Test parsing NEO with no close approach data.""" + client = NEOFeedClient() + response = { + 'near_earth_objects': { + '2026-03-23': [{ + 'neo_reference_id': 'test', + 'name': 'test', + 'absolute_magnitude_h': 25.0, + 'estimated_diameter': { + 'meters': { + 'estimated_diameter_min': 10, + 'estimated_diameter_max': 20 + } + }, + 'is_potentially_hazardous_asteroid': False, + 'close_approach_data': [] + }] + } + } + approaches = client._parse_neo_feed(response) + assert len(approaches) == 0 + + @pytest.mark.asyncio + async def test_close_no_session(self): + """Test closing with no active session.""" + client = NEOFeedClient() + await client.close() + assert client._session is None + + @pytest.mark.asyncio + async def test_close_owned_session(self): + """Test that owned session is properly closed.""" + client = NEOFeedClient() + mock_session = AsyncMock() + client._session = mock_session + client._owns_session = True + + await client.close() + mock_session.close.assert_called_once() + assert client._session is None + + @pytest.mark.asyncio + async def test_close_borrowed_session(self): + """Test that borrowed session is not closed.""" + mock_session = AsyncMock() + client = NEOFeedClient(session=mock_session) + + await client.close() + mock_session.close.assert_not_called() + + @pytest.mark.asyncio + async def test_parse_datetime_formats(self): + """Test various date format parsing.""" + client = NEOFeedClient() + + dt = client._parse_datetime('2026-Mar-23 12:00') + assert dt.year == 2026 + assert dt.month == 3 + assert dt.day == 23 + assert dt.hour == 12 + + dt = client._parse_datetime('2026-03-23') + assert dt.day == 23 + + +class TestNEOFeedIntegration: + """Integration tests for NEO Feed with existing CloseApproach type.""" + + def test_feed_produces_compatible_types(self): + """NEO Feed results are the same CloseApproach type as CAD results.""" + client = NEOFeedClient() + approaches = client._parse_neo_feed(SAMPLE_NEO_FEED_RESPONSE) + + for a in approaches: + assert isinstance(a, CloseApproach) + # All CloseApproach properties should work + assert isinstance(a.threat_level, ThreatLevel) + assert isinstance(a.estimated_diameter_str, str) + assert isinstance(a.approach_id, str) + assert isinstance(a.lexicon_str, str) + + def test_feed_pha_vs_cad_heuristic(self): + """ + NEO Feed provides real PHA flag from NASA. + CAD client uses heuristic (H<=22 and dist<0.05 AU). + Feed data should be more authoritative. + """ + client = NEOFeedClient() + approaches = client._parse_neo_feed(SAMPLE_NEO_FEED_RESPONSE) + + # Phaethon is officially PHA but distant — feed correctly flags it + phaethon = next(a for a in approaches if a.designation == '3200') + assert phaethon.is_potentially_hazardous is True + + # 2026 FU is close but not PHA — feed correctly doesn't flag it + fu = next(a for a in approaches if a.designation == '2026FU') + assert fu.is_potentially_hazardous is False + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])