Keep active crewed missions on launches page

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:

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)