agenda/agenda/thespacedevs.py

"""Get details of upcoming space launches."""

import json
import os
import typing
from datetime import datetime

import requests

Launch = dict[str, typing.Any]
Summary = dict[str, typing.Any]

ttl = 60 * 60 * 2  # two hours


def next_launch_api(rocket_dir: str, limit: int = 200) -> list[Launch]:
    """Get the next upcoming launches from the API."""
    now = datetime.now()
    filename = os.path.join(rocket_dir, now.strftime("%Y-%m-%d_%H:%M:%S.json"))
    url = "https://ll.thespacedevs.com/2.2.0/launch/upcoming/"

    params: dict[str, str | int] = {"limit": limit}
    r = requests.get(url, params=params)
    data = r.json()
    open(filename, "w").write(r.text)
    return [summarize_launch(launch) for launch in data["results"]]


def filename_timestamp(filename: str) -> tuple[datetime, str] | None:
    """Get datetime from filename."""
    try:
        ts = datetime.strptime(filename, "%Y-%m-%d_%H:%M:%S.json")
    except ValueError:
        return None
    return (ts, filename)


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")

    include_time = False
    # Format the date based on precision
    time_format: str | None
    field2 = None

    match net_precision:
        case "Year":
            time_format = "%Y"
        case "Month":
            time_format = "%b %Y"
        case "Week":
            time_format = "%d %b %Y"
            field2 = "(Week of)"
        case "Day" | "Hour" | "Minute":
            time_format = "%d %b %Y"
            include_time = net_precision in {"Hour", "Minute"}
        case "Second":
            time_format = "%d %b %Y"
            include_time = True
        case _ if net_precision and net_precision.startswith("Quarter "):
            time_format = f"Q{net_precision[-1]} %Y"
        case _:
            time_format = None

    if not time_format:
        return (repr(time_str), repr(net_precision))

    assert time_format

    formatted = dt.strftime(time_format)

    if include_time:
        return (formatted, dt.strftime("%H:%M"))
    else:
        return (formatted, field2)


launch_providers = {
    "Indian Space Research Organization": "ISRO",
    "United Launch Alliance": "ULA",
    "Payload Aerospace S.L.": "Payload Aerospace",
    "Russian Federal Space Agency (ROSCOSMOS)": "ROSCOSMOS",
    "China Aerospace Science and Technology Corporation": "CASC",
}


def get_nested(data: dict[str, typing.Any], keys: list[str]) -> typing.Any | None:
    """
    Safely get a nested value from a dictionary.

    Args:
        data (Dict[str, Any]): The dictionary to search.
        keys (List[str]): A list of keys for the nested lookup.

    Returns:
        Optional[Any]: The retrieved value, or None if any key is missing.
    """
    for key in keys:
        if data is None or key not in data:
            return None
        data = data[key]
    return data


def summarize_launch(launch: Launch) -> Summary:
    """Summarize rocket launch."""
    try:
        launch_provider = launch["launch_service_provider"]["name"]
        launch_provider_abbrev = launch_providers.get(launch_provider)
    except (TypeError, IndexError):
        launch_provider = None
        launch_provider_abbrev = None

    net_precision = typing.cast(str, get_nested(launch, ["net_precision", "name"]))
    t0_date, t0_time = format_time(launch["net"], net_precision)

    return {
        "name": launch.get("name"),
        "status": launch.get("status"),
        "net": launch.get("net"),
        "net_precision": net_precision,
        "t0_date": t0_date,
        "t0_time": t0_time,
        "window_start": launch.get("window_start"),
        "window_end": launch.get("window_end"),
        "launch_provider": launch_provider,
        "launch_provider_abbrev": launch_provider_abbrev,
        "launch_provider_type": get_nested(launch, ["launch_service_provider", "type"]),
        "rocket": launch["rocket"]["configuration"]["full_name"],
        "mission": launch.get("mission"),
        "mission_name": get_nested(launch, ["mission", "name"]),
        "pad_name": launch["pad"]["name"],
        "pad_wikipedia_url": launch["pad"]["wiki_url"],
        "location": launch["pad"]["location"]["name"],
        "country_code": launch["pad"]["country_code"],
        "orbit": get_nested(launch, ["mission", "orbit"]),
    }


def read_cached_launches(rocket_dir: str) -> list[Summary]:
    """Read the most recent cache of launches."""
    existing = [x for x in (filename_timestamp(f) for f in os.listdir(rocket_dir)) if x]

    existing.sort(reverse=True)
    f = existing[0][1]

    filename = os.path.join(rocket_dir, f)
    data = json.load(open(filename))
    return [summarize_launch(launch) for launch in data["results"]]


def get_launches(
    rocket_dir: str, limit: int = 200, refresh: bool = False
) -> list[Summary]:
    """Get rocket launches with caching."""
    now = datetime.now()
    existing = [x for x in (filename_timestamp(f) for f in os.listdir(rocket_dir)) if x]

    existing.sort(reverse=True)

    if refresh or not existing or (now - existing[0][0]).seconds > ttl:
        return next_launch_api(rocket_dir, limit=limit)

    f = existing[0][1]

    filename = os.path.join(rocket_dir, f)
    data = json.load(open(filename))
    return [summarize_launch(launch) for launch in data["results"]]