Compare commits
8 commits
a130a85a48
...
b5188771be
Author | SHA1 | Date | |
---|---|---|---|
Edward Betts | b5188771be | ||
Edward Betts | b9adb3d15e | ||
Edward Betts | 15c5053e44 | ||
Edward Betts | 11bc0419b3 | ||
Edward Betts | 7169d1ba27 | ||
Edward Betts | fb65b4d6fb | ||
Edward Betts | 7cdb6903fc | ||
Edward Betts | f423fcdcbe |
|
@ -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 = [
|
||||
|
|
14
agenda/fx.py
14
agenda/fx.py
|
@ -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"
|
||||
|
|
|
@ -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 ""
|
||||
)
|
||||
|
|
|
@ -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
|
||||
|
||||
|
|
|
@ -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":
|
||||
|
|
|
@ -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.
|
||||
|
|
|
@ -43,9 +43,9 @@
|
|||
</ul>
|
||||
|
||||
<div class="grid-container">
|
||||
{{ section("Past", past) }}
|
||||
{{ section("Current", current) }}
|
||||
{{ section("Future", future) }}
|
||||
{{ section("Past", past) }}
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
|
|
@ -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>
|
||||
|
|
|
@ -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 %}
|
||||
|
|
|
@ -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,
|
||||
)
|
||||
|
||||
|
|
Loading…
Reference in a new issue