Compare commits

...

8 commits

Author SHA1 Message Date
Edward Betts b5188771be Ignore types in geopy.distance 2024-08-04 09:23:34 +02:00
Edward Betts b9adb3d15e Move timedelta_display() function to agenda.utils 2024-08-04 09:23:13 +02:00
Edward Betts 15c5053e44 Show page render time on events list 2024-08-04 12:01:54 +08:00
Edward Betts 11bc0419b3 Reorder accommodation list 2024-08-04 11:45:13 +08:00
Edward Betts 7169d1ba27 Improve support for adding a new currency 2024-08-04 11:44:44 +08:00
Edward Betts fb65b4d6fb Split code out into rocket_launch_events() 2024-08-03 14:58:32 +08:00
Edward Betts 7cdb6903fc Make mypy happier 2024-08-03 14:51:30 +08:00
Edward Betts f423fcdcbe Split up train and flight loading
Reduce complexity of train and flight loading functions by splitting
code out into separate functions.
2024-08-03 14:49:21 +08:00
10 changed files with 104 additions and 70 deletions

View file

@ -133,6 +133,39 @@ class AgendaData(typing.TypedDict, total=False):
errors: list[tuple[str, Exception]]
def rocket_launch_events(rockets: list[thespacedevs.Summary]) -> list[Event]:
"""Rocket launch events."""
events: list[Event] = []
for launch in rockets:
dt = None
net_precision = launch["net_precision"]
skip = {"Year", "Month", "Quarter", "Fiscal Year"}
if net_precision == "Day":
dt = datetime.strptime(launch["net"], "%Y-%m-%dT%H:%M:%SZ").date()
elif (
net_precision
and net_precision not in skip
and "Year" not in net_precision
and launch["t0_time"]
):
dt = pytz.utc.localize(
datetime.strptime(launch["net"], "%Y-%m-%dT%H:%M:%SZ")
)
if not dt:
continue
rocket_name = (
f'{launch["rocket"]["full_name"]}: '
+ f'{launch["mission_name"] or "[no mission]"}'
)
e = Event(name="rocket", date=dt, title=rocket_name)
events.append(e)
return events
async def get_data(now: datetime, config: flask.config.Config) -> AgendaData:
"""Get data to display on agenda dashboard."""
data_dir = config["DATA_DIR"]
@ -230,34 +263,7 @@ async def get_data(now: datetime, config: flask.config.Config) -> AgendaData:
events += meetup.get_events(my_data)
events += hn.whoishiring(last_year, next_year)
events += carnival.rio_carnival_events(last_year, next_year)
for launch in rockets:
dt = None
net_precision = launch["net_precision"]
skip = {"Year", "Month", "Quarter", "Fiscal Year"}
if net_precision == "Day":
dt = datetime.strptime(launch["net"], "%Y-%m-%dT%H:%M:%SZ").date()
elif (
net_precision
and net_precision not in skip
and "Year" not in net_precision
and launch["t0_time"]
):
dt = pytz.utc.localize(
datetime.strptime(launch["net"], "%Y-%m-%dT%H:%M:%SZ")
)
if not dt:
continue
rocket_name = (
f'{launch["rocket"]["full_name"]}: '
+ f'{launch["mission_name"] or "[no mission]"}'
)
e = Event(name="rocket", date=dt, title=rocket_name)
events.append(e)
events += rocket_launch_events(rockets)
events += [Event(name="today", date=today)]
busy_events = [

View file

@ -44,10 +44,16 @@ async def get_gbpusd(config: flask.config.Config) -> Decimal:
return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"])
def read_cached_rates(filename: str, currencies: list[str]) -> dict[str, Decimal]:
def read_cached_rates(
filename: str | None, currencies: list[str]
) -> dict[str, Decimal]:
"""Read FX rates from cache."""
if filename is None:
return {}
with open(filename) as file:
data = json.load(file, parse_float=Decimal)
return {
cur: Decimal(data["quotes"][f"GBP{cur}"])
for cur in currencies
@ -69,7 +75,9 @@ def get_rates(config: flask.config.Config) -> dict[str, Decimal]:
currency_string = ",".join(sorted(currencies))
file_suffix = f"{currency_string}_to_GBP.json"
existing_data = os.listdir(fx_dir)
existing_files = [f for f in existing_data if f.endswith(file_suffix)]
existing_files = [f for f in existing_data if f.endswith(".json")]
full_path: str | None = None
if existing_files:
recent_filename = max(existing_files)
@ -77,7 +85,7 @@ def get_rates(config: flask.config.Config) -> dict[str, Decimal]:
delta = now - recent
full_path = os.path.join(fx_dir, recent_filename)
if delta < timedelta(hours=12):
if not recent_filename.endswith(file_suffix) or delta < timedelta(hours=12):
return read_cached_rates(full_path, currencies)
url = "http://api.exchangerate.host/live"

View file

@ -3,26 +3,14 @@
from datetime import timedelta, timezone
import dateutil.tz
import exchange_calendars
import pandas
import exchange_calendars # type: ignore
import pandas # type: ignore
from . import utils
here = dateutil.tz.tzlocal()
def timedelta_display(delta: timedelta) -> str:
"""Format timedelta as a human readable string."""
total_seconds = int(delta.total_seconds())
days, remainder = divmod(total_seconds, 24 * 60 * 60)
hours, remainder = divmod(remainder, 60 * 60)
mins, secs = divmod(remainder, 60)
return " ".join(
f"{v:>3} {label}"
for v, label in ((days, "days"), (hours, "hrs"), (mins, "mins"))
if v
)
def open_and_close() -> list[str]:
"""Stock markets open and close times."""
# The trading calendars code is slow, maybe there is a faster way to do this
@ -40,11 +28,11 @@ def open_and_close() -> list[str]:
if cal.is_open_on_minute(now_local):
next_close = cal.next_close(now).tz_convert(here)
next_close = next_close.replace(minute=round(next_close.minute, -1))
delta_close = timedelta_display(next_close - now_local)
delta_close = utils.timedelta_display(next_close - now_local)
prev_open = cal.previous_open(now).tz_convert(here)
prev_open = prev_open.replace(minute=round(prev_open.minute, -1))
delta_open = timedelta_display(now_local - prev_open)
delta_open = utils.timedelta_display(now_local - prev_open)
msg = (
f"{label:>6} market opened {delta_open} ago, "
@ -54,7 +42,7 @@ def open_and_close() -> list[str]:
ts = cal.next_open(now)
ts = ts.replace(minute=round(ts.minute, -1))
ts = ts.tz_convert(here)
delta = timedelta_display(ts - now_local)
delta = utils.timedelta_display(ts - now_local)
msg = f"{label:>6} market opens in {delta}" + (
f" ({ts:%H:%M})" if (ts - now_local) < timedelta(days=1) else ""
)

View file

@ -7,7 +7,7 @@ import typing
import flask
import yaml
from geopy.distance import geodesic
from geopy.distance import geodesic # type: ignore
from .types import Event, StrDict

View file

@ -28,6 +28,19 @@ def load_travel(travel_type: str, plural: str, data_dir: str) -> list[StrDict]:
return items
def process_train_leg(
leg: StrDict,
by_name: StrDict,
route_distances: travel.RouteDistances | None,
) -> None:
"""Process train leg."""
assert leg["from"] in by_name and leg["to"] in by_name
leg["from_station"], leg["to_station"] = by_name[leg["from"]], by_name[leg["to"]]
if route_distances:
travel.add_leg_route_distance(leg, route_distances)
def load_trains(
data_dir: str, route_distances: travel.RouteDistances | None = None
) -> list[StrDict]:
@ -45,13 +58,7 @@ def load_trains(
train["to_station"] = by_name[train["to"]]
for leg in train["legs"]:
assert leg["from"] in by_name
assert leg["to"] in by_name
leg["from_station"] = by_name[leg["from"]]
leg["to_station"] = by_name[leg["to"]]
if route_distances:
travel.add_leg_route_distance(leg, route_distances)
process_train_leg(leg, by_name=by_name, route_distances=route_distances)
if all("distance" in leg for leg in train["legs"]):
train["distance"] = sum(leg["distance"] for leg in train["legs"])
@ -90,6 +97,20 @@ def depart_datetime(item: StrDict) -> datetime:
return datetime.combine(depart, time.min).replace(tzinfo=ZoneInfo("UTC"))
def process_flight(
flight: StrDict, iata: dict[str, str], airports: list[StrDict]
) -> None:
"""Add airport detail, airline name and distance to flight."""
if flight["from"] in airports:
flight["from_airport"] = airports[flight["from"]]
if flight["to"] in airports:
flight["to_airport"] = airports[flight["to"]]
if "airline" in flight:
flight["airline_name"] = iata.get(flight["airline"], "[unknown]")
flight["distance"] = travel.flight_distance(flight)
def load_flight_bookings(data_dir: str) -> list[StrDict]:
"""Load flight bookings."""
bookings = load_travel("flight", "flights", data_dir)
@ -98,14 +119,7 @@ def load_flight_bookings(data_dir: str) -> list[StrDict]:
airports = travel.parse_yaml("airports", data_dir)
for booking in bookings:
for flight in booking["flights"]:
if flight["from"] in airports:
flight["from_airport"] = airports[flight["from"]]
if flight["to"] in airports:
flight["to_airport"] = airports[flight["to"]]
if "airline" in flight:
flight["airline_name"] = iata.get(flight["airline"], "[unknown]")
flight["distance"] = travel.flight_distance(flight)
process_flight(flight, iata, airports)
return bookings
@ -272,7 +286,7 @@ def read_geojson(data_dir: str, filename: str) -> str:
def get_trip_routes(trip: Trip) -> list[StrDict]:
"""Get routes for given trip to show on map."""
routes = []
routes: list[StrDict] = []
seen_geojson = set()
for t in trip.travel:
if t["type"] == "ferry":

View file

@ -2,7 +2,7 @@
import os
import typing
from datetime import date, datetime, timezone
from datetime import date, datetime, timedelta, timezone
from time import time
@ -28,6 +28,20 @@ def as_datetime(d: datetime | date) -> datetime:
raise TypeError(f"Unsupported type: {type(d)}")
def timedelta_display(delta: timedelta) -> str:
"""Format timedelta as a human readable string."""
total_seconds = int(delta.total_seconds())
days, remainder = divmod(total_seconds, 24 * 60 * 60)
hours, remainder = divmod(remainder, 60 * 60)
mins, secs = divmod(remainder, 60)
return " ".join(
f"{v} {label}"
for v, label in ((days, "days"), (hours, "hrs"), (mins, "mins"))
if v
)
def human_readable_delta(future_date: date) -> str | None:
"""
Calculate the human-readable time delta for a given future date.

View file

@ -43,9 +43,9 @@
</ul>
<div class="grid-container">
{{ section("Past", past) }}
{{ section("Current", current) }}
{{ section("Future", future) }}
{{ section("Past", past) }}
</div>
</div>

View file

@ -150,6 +150,7 @@
{% for name, seconds in timings %}
<li>{{ name }} took {{ "%.1f" | format(seconds) }} seconds</li>
{% endfor %}
<li>Render time: {{ "%.1f" | format(render_time) }} seconds</li>
</ul>
</div>

View file

@ -38,7 +38,7 @@
<div class="grid-item text-end">
{% if item.price and item.currency %}
<span class="badge bg-info text-nowrap">{{ "{:,d}".format(item.price | int) }} {{ item.currency }}</span>
{% if item.currency != "GBP" %}
{% if item.currency != "GBP" and item.currency in fx_rate %}
<span class="badge bg-info text-nowrap">{{ "{:,.2f}".format(item.price / fx_rate[item.currency]) }} GBP</span>
{% endif %}
{% elif item.free %}

View file

@ -7,6 +7,7 @@ import inspect
import operator
import os.path
import sys
import time
import traceback
from collections import defaultdict
from datetime import date, datetime, timedelta
@ -70,6 +71,7 @@ def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str,
@app.route("/")
async def index() -> str:
"""Index page."""
t0 = time.time()
now = datetime.now()
data = await agenda.data.get_data(now, app.config)
@ -88,6 +90,7 @@ async def index() -> str:
fullcalendar_events=calendar.build_events(events),
start_event_list=date.today() - timedelta(days=1),
end_event_list=date.today() + timedelta(days=365 * 2),
render_time=(time.time() - t0),
**data,
)