From 6ed9e520ed02d7bfad607dd9a3d0964a56101097 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Tue, 3 Oct 2023 13:00:23 +0100 Subject: [PATCH] List rocket launches using new API Closes: #12 --- agenda/__init__.py | 5 +- agenda/thespacedevs.py | 116 +++++++++++++++++++++++++++++++++++++++++ templates/index.html | 44 ++++++---------- 3 files changed, 135 insertions(+), 30 deletions(-) create mode 100644 agenda/thespacedevs.py diff --git a/agenda/__init__.py b/agenda/__init__.py index 83c93b0..d90359c 100644 --- a/agenda/__init__.py +++ b/agenda/__init__.py @@ -16,7 +16,7 @@ import pytz import requests from dateutil.easter import easter -# from agenda import spacexdata +from agenda import thespacedevs warnings.simplefilter(action="ignore", category=FutureWarning) @@ -268,6 +268,8 @@ def get_us_holiday() -> dict[str, date | str]: def get_data() -> dict[str, str | object]: """Get data to display on agenda dashboard.""" + rocket_dir = os.path.join(data_dir, "thespacedevs") + reply = { "now": now, "gbpusd": get_gbpusd(), @@ -285,6 +287,7 @@ def get_data() -> dict[str, str | object]: "uk_financial_year_end": uk_financial_year_end(today), "xmas_last_posting_dates": xmas_last_posting_dates, "xmas_day": xmas_day(today), + "rockets": thespacedevs.get_launches(rocket_dir, limit=40), } return reply diff --git a/agenda/thespacedevs.py b/agenda/thespacedevs.py new file mode 100644 index 0000000..de2857d --- /dev/null +++ b/agenda/thespacedevs.py @@ -0,0 +1,116 @@ +import json +import os +import typing +from datetime import datetime + +import requests + +Launch = dict[str, typing.Any] +Summary = dict[str, typing.Any] + + +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) + 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 + if net_precision == "Month": + time_format = "%b %Y" + if net_precision == "Day": + time_format = "%d %b %Y" + if net_precision == "Hour": + time_format = "%d %b %Y" + include_time = True + if net_precision == "Minute": + time_format = "%d %b %Y" + include_time = True + elif net_precision == "Second": + time_format = "%d %b %Y" + include_time = True + elif net_precision.startswith("Quarter "): + time_format = f"Q{net_precision[-1]} %Y" + + assert time_format + + formatted = dt.strftime(time_format) + + if include_time: + return (formatted, dt.strftime("%H:%M")) + else: + return (formatted, None) + + +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 summarize_launch(launch: Launch) -> Summary: + """Summarize rocket launch.""" + launch_provider = launch["launch_service_provider"]["name"] + launch_provider_abbrev = launch_providers.get(launch_provider) + + t0_date, t0_time = format_time(launch["net"], launch["net_precision"]["name"]) + + return { + "name": launch["name"], + "status": launch["status"]["abbrev"], + "t0_date": t0_date, + "t0_time": t0_time, + "window_start": launch["window_start"], + "window_end": launch["window_end"], + "launch_provider": launch_provider, + "launch_provider_abbrev": launch_provider_abbrev, + "launch_provider_type": launch["launch_service_provider"]["type"], + "rocket": launch["rocket"]["configuration"]["name"], + "mission": launch["mission"]["name"] if launch["mission"] else "N/A", + "location": launch["pad"]["location"]["name"], + "country_code": launch["pad"]["country_code"], + } + + +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 next_launch_api(rocket_dir, limit=limit) + except ValueError: + print("*** SpaceX next launch error ***") + + 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"]] diff --git a/templates/index.html b/templates/index.html index 58963a7..91ced63 100644 --- a/templates/index.html +++ b/templates/index.html @@ -78,40 +78,26 @@

{{ market }}

{% endfor %} - {# -

SpaceX launches

+

Rocket launches

- {% for launch in spacex %} + {% for launch in rockets %} - {% if launch["date_precision"] == "day" %} - - - - {% else %} - - - - {% endif %} - - - - - + + + + + {% endfor %}
- {{ days(launch["when"].date()) }} - - {{ launch["when"].strftime("%a, %d %b") }} - - - {{ days_hours(launch["when"]) }} - - {{ launch["when"].strftime("%a, %d %b") }} - - {{ launch["when"].strftime("%H:%M") }} - {{ launch["name"] }}{{ launch["rocket"] }}{{ launch["payloads"] | join(" + ") }}{{ launch["date_precision"] }}
{{ launch.t0_date }} + + {% if launch.t0_time %}
{{ launch.t0_time }}{% endif %}		{{ launch.status }}{{ launch.rocket }}
{{launch.mission }}		 + {% if launch.launch_provider_abbrev %} + {{ launch.launch_provider_abbrev }} + {% else %} + {{ launch.launch_provider }} + {% endif %} + ({{ launch.launch_provider_type }}){{ launch.location }}
- #}