diff --git a/agenda/thespacedevs.py b/agenda/thespacedevs.py index 638ea54..e4f1b7a 100644 --- a/agenda/thespacedevs.py +++ b/agenda/thespacedevs.py @@ -16,6 +16,7 @@ Summary = dict[str, typing.Any] ttl = 60 * 60 * 2 # two hours LIMIT = 500 +ACTIVE_CREWED_FLIGHTS_CACHE_FILE = "active_crewed_flights.json" def next_launch_api_data(rocket_dir: str, limit: int = LIMIT) -> StrDict | None: @@ -42,6 +43,173 @@ def next_launch_api(rocket_dir: str, limit: int = LIMIT) -> list[Summary] | None return [summarize_launch(launch) for launch in data["results"]] +def parse_api_datetime(value: typing.Any) -> datetime | None: + """Parse API datetime strings into datetime objects.""" + if not isinstance(value, str): + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +def is_crewed_spaceflight(flight: Launch) -> bool: + """Return True when a spaceflight is crewed/human-rated.""" + spacecraft_config = get_nested(flight, ["spacecraft", "spacecraft_config"]) + if isinstance(spacecraft_config, dict) and spacecraft_config.get("human_rated"): + return True + + mission_type = get_nested(flight, ["launch", "mission", "type"]) + if isinstance(mission_type, str) and "human" in mission_type.lower(): + return True + + return False + + +def is_active_crewed_spaceflight(flight: Launch, now: datetime) -> bool: + """Return True when a crewed spaceflight is active (mission not yet ended).""" + launch = flight.get("launch") + if not isinstance(launch, dict): + return False + + if not is_crewed_spaceflight(flight): + return False + + launch_net = parse_api_datetime(launch.get("net")) + if launch_net and launch_net > now: + return False + + mission_end = parse_api_datetime(flight.get("mission_end")) + if mission_end and mission_end <= now: + return False + + spacecraft_in_space = get_nested(flight, ["spacecraft", "in_space"]) + if spacecraft_in_space is True: + return True + + if mission_end is None: + landing_success = get_nested(flight, ["landing", "success"]) + if landing_success is True: + return False + return True + + return True + + +def active_crewed_flights_api(limit: int = LIMIT) -> list[Summary] | None: + """ + Get active crewed spaceflights from the SpaceDevs API. + + The API does not reliably expose a direct filter for active flights, so this + paginates through results and applies local filtering. + """ + now = datetime.now().astimezone() + url = "https://ll.thespacedevs.com/2.2.0/spacecraft/flight/" + params: dict[str, str | int] = {"limit": limit} + launches: list[Summary] = [] + seen_slugs: set[str] = set() + page = 0 + max_pages = 20 + + while url and page < max_pages: + r = requests.get(url, params=params if page == 0 else None, timeout=30) + if not r.ok: + return None + try: + data: StrDict = r.json() + except requests.exceptions.JSONDecodeError: + return None + + results = data.get("results") + if not isinstance(results, list): + break + + for flight in results: + if not isinstance(flight, dict): + continue + if not is_active_crewed_spaceflight(flight, now): + continue + + launch = flight.get("launch") + if not isinstance(launch, dict): + continue + + launch_summary = summarize_launch(typing.cast(Launch, launch)) + slug = launch_summary.get("slug") + if not isinstance(slug, str): + continue + if slug in seen_slugs: + continue + seen_slugs.add(slug) + launches.append(launch_summary) + + next_url = data.get("next") + url = next_url if isinstance(next_url, str) else "" + params = {} + page += 1 + + return launches + + +def get_active_crewed_flights_cache_filename(rocket_dir: str) -> str: + """Path of the active crewed flights cache file.""" + return os.path.join(rocket_dir, ACTIVE_CREWED_FLIGHTS_CACHE_FILE) + + +def load_active_crewed_flights_cache( + rocket_dir: str, +) -> tuple[datetime, list[Summary]] | None: + """Load active crewed flights cache file.""" + filename = get_active_crewed_flights_cache_filename(rocket_dir) + if not os.path.exists(filename): + return None + + try: + cache_data = json.load(open(filename)) + except (json.JSONDecodeError, OSError): + return None + + updated_str = cache_data.get("updated") + updated = parse_api_datetime(updated_str) + results = cache_data.get("results") + if not updated or not isinstance(results, list): + return None + + summary_results: list[Summary] = [r for r in results if isinstance(r, dict)] + return (updated.replace(tzinfo=None), summary_results) + + +def write_active_crewed_flights_cache(rocket_dir: str, launches: list[Summary]) -> None: + """Write active crewed flights cache file.""" + filename = get_active_crewed_flights_cache_filename(rocket_dir) + payload: StrDict = { + "updated": datetime.now().isoformat(), + "results": launches, + } + with open(filename, "w") as f: + json.dump(payload, f) + + +def get_active_crewed_flights( + rocket_dir: str, refresh: bool = False +) -> list[Summary] | None: + """Get active crewed flights with cache and API fallback.""" + now = datetime.now() + cached = load_active_crewed_flights_cache(rocket_dir) + if cached and not refresh and (now - cached[0]).seconds <= ttl: + return cached[1] + + try: + active_flights = active_crewed_flights_api() + if active_flights is not None: + write_active_crewed_flights_cache(rocket_dir, active_flights) + return active_flights + except Exception: + pass + + return cached[1] if cached else [] + + def format_time(time_str: str, net_precision: str) -> tuple[str, str | None]: """Format time based on precision.""" dt = datetime.strptime(time_str, "%Y-%m-%dT%H:%M:%SZ") @@ -176,7 +344,20 @@ def get_launches( if refresh or not existing or (now - existing[0][0]).seconds > ttl: try: - return next_launch_api(rocket_dir, limit=limit) + upcoming = next_launch_api(rocket_dir, limit=limit) + if upcoming is None: + raise RuntimeError("unable to fetch upcoming launches") + active_crewed = get_active_crewed_flights(rocket_dir, refresh=refresh) or [] + by_slug = { + typing.cast(str, launch["slug"]): launch + for launch in upcoming + if isinstance(launch.get("slug"), str) + } + for launch in active_crewed: + slug = launch.get("slug") + if isinstance(slug, str) and slug not in by_slug: + by_slug[slug] = launch + return sorted(by_slug.values(), key=lambda launch: str(launch.get("net"))) except Exception: pass # fallback to cached version @@ -184,7 +365,18 @@ def get_launches( filename = os.path.join(rocket_dir, f) data = json.load(open(filename)) - return [summarize_launch(launch) for launch in data["results"]] + upcoming = [summarize_launch(launch) for launch in data["results"]] + active_crewed = get_active_crewed_flights(rocket_dir, refresh=refresh) or [] + by_slug = { + typing.cast(str, launch["slug"]): launch + for launch in upcoming + if isinstance(launch.get("slug"), str) + } + for launch in active_crewed: + slug = launch.get("slug") + if isinstance(slug, str) and slug not in by_slug: + by_slug[slug] = launch + return sorted(by_slug.values(), key=lambda launch: str(launch.get("net"))) def format_date(dt: datetime) -> str: diff --git a/tests/test_thespacedevs.py b/tests/test_thespacedevs.py index 369c556..ee97d87 100644 --- a/tests/test_thespacedevs.py +++ b/tests/test_thespacedevs.py @@ -2,7 +2,13 @@ import deepdiff import pytest -from agenda.thespacedevs import format_launch_changes +from datetime import datetime, timezone + +from agenda.thespacedevs import ( + format_launch_changes, + is_active_crewed_spaceflight, + is_crewed_spaceflight, +) # --- Helper Functions for Tests --- @@ -458,7 +464,9 @@ def test_mission_name_change(): old_launch = {"name": "Starship | Flight 9"} new_launch = {"name": "Starship | Flight 10"} diff = deepdiff.DeepDiff(old_launch, new_launch) - expected = "• Mission name changed from 'Starship | Flight 9' to 'Starship | Flight 10'" + expected = ( + "• Mission name changed from 'Starship | Flight 9' to 'Starship | Flight 10'" + ) assert format_launch_changes(diff) == expected @@ -610,3 +618,60 @@ def test_multiple_changes(): assert len(result_lines) == len(expected_changes) for expected_line in expected_changes: assert expected_line in result_lines + + +def test_is_crewed_spaceflight_uses_human_rated() -> None: + """Crewed flight detection should use human-rated spacecraft config.""" + flight = { + "spacecraft": {"spacecraft_config": {"human_rated": True}}, + "launch": {"mission": {"type": "Communications"}}, + } + assert is_crewed_spaceflight(flight) + + +def test_is_active_crewed_spaceflight_future_mission_end() -> None: + """Crewed flight with mission end in the future should be active.""" + now = datetime(2026, 2, 21, 12, 0, tzinfo=timezone.utc) + flight = { + "mission_end": "2026-02-22T12:00:00Z", + "spacecraft": { + "in_space": True, + "spacecraft_config": {"human_rated": True}, + }, + "launch": { + "net": "2026-02-20T10:00:00Z", + "mission": {"type": "Human Exploration"}, + }, + "landing": {"success": None}, + } + assert is_active_crewed_spaceflight(flight, now) + + +def test_is_active_crewed_spaceflight_ended_mission() -> None: + """Crewed flight should be inactive after mission end time.""" + now = datetime(2026, 2, 21, 12, 0, tzinfo=timezone.utc) + flight = { + "mission_end": "2026-02-21T09:00:00Z", + "spacecraft": { + "in_space": False, + "spacecraft_config": {"human_rated": True}, + }, + "launch": {"net": "2026-02-20T10:00:00Z", "mission": {"type": "Tourism"}}, + "landing": {"success": True}, + } + assert not is_active_crewed_spaceflight(flight, now) + + +def test_is_active_crewed_spaceflight_excludes_future_launches() -> None: + """Crewed flight record should not be active before launch.""" + now = datetime(2026, 2, 21, 12, 0, tzinfo=timezone.utc) + flight = { + "mission_end": None, + "spacecraft": { + "in_space": True, + "spacecraft_config": {"human_rated": True}, + }, + "launch": {"net": "2026-02-22T10:00:00Z", "mission": {"type": "Tourism"}}, + "landing": {"success": None}, + } + assert not is_active_crewed_spaceflight(flight, now)