158 lines
5 KiB
Python
158 lines
5 KiB
Python
"""Get details of upcoming space launches."""
|
|
|
|
import json
|
|
import os
|
|
import typing
|
|
from datetime import datetime
|
|
|
|
import httpx
|
|
|
|
Launch = dict[str, typing.Any]
|
|
Summary = dict[str, typing.Any]
|
|
|
|
|
|
async 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}
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(url, params=params)
|
|
open(filename, "w").write(r.text)
|
|
data = r.json()
|
|
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"]),
|
|
}
|
|
|
|
|
|
async def get_launches(rocket_dir: str, limit: int = 200) -> 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 not existing or (now - existing[0][0]).seconds > 3600: # one hour
|
|
try:
|
|
return await next_launch_api(rocket_dir, limit=limit)
|
|
except httpx.ReadTimeout:
|
|
pass
|
|
|
|
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"]]
|