Compare commits
No commits in common. "b5188771bee39ddf73c99f9f37730a893e7514d8" and "a130a85a488ee070ee252b58b04048d151b2029c" have entirely different histories.
b5188771be
...
a130a85a48
|
@ -133,39 +133,6 @@ class AgendaData(typing.TypedDict, total=False):
|
||||||
errors: list[tuple[str, Exception]]
|
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:
|
async def get_data(now: datetime, config: flask.config.Config) -> AgendaData:
|
||||||
"""Get data to display on agenda dashboard."""
|
"""Get data to display on agenda dashboard."""
|
||||||
data_dir = config["DATA_DIR"]
|
data_dir = config["DATA_DIR"]
|
||||||
|
@ -263,7 +230,34 @@ async def get_data(now: datetime, config: flask.config.Config) -> AgendaData:
|
||||||
events += meetup.get_events(my_data)
|
events += meetup.get_events(my_data)
|
||||||
events += hn.whoishiring(last_year, next_year)
|
events += hn.whoishiring(last_year, next_year)
|
||||||
events += carnival.rio_carnival_events(last_year, next_year)
|
events += carnival.rio_carnival_events(last_year, next_year)
|
||||||
events += rocket_launch_events(rockets)
|
|
||||||
|
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 += [Event(name="today", date=today)]
|
events += [Event(name="today", date=today)]
|
||||||
|
|
||||||
busy_events = [
|
busy_events = [
|
||||||
|
|
14
agenda/fx.py
14
agenda/fx.py
|
@ -44,16 +44,10 @@ async def get_gbpusd(config: flask.config.Config) -> Decimal:
|
||||||
return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"])
|
return typing.cast(Decimal, 1 / data["quotes"]["USDGBP"])
|
||||||
|
|
||||||
|
|
||||||
def read_cached_rates(
|
def read_cached_rates(filename: str, currencies: list[str]) -> dict[str, Decimal]:
|
||||||
filename: str | None, currencies: list[str]
|
|
||||||
) -> dict[str, Decimal]:
|
|
||||||
"""Read FX rates from cache."""
|
"""Read FX rates from cache."""
|
||||||
if filename is None:
|
|
||||||
return {}
|
|
||||||
|
|
||||||
with open(filename) as file:
|
with open(filename) as file:
|
||||||
data = json.load(file, parse_float=Decimal)
|
data = json.load(file, parse_float=Decimal)
|
||||||
|
|
||||||
return {
|
return {
|
||||||
cur: Decimal(data["quotes"][f"GBP{cur}"])
|
cur: Decimal(data["quotes"][f"GBP{cur}"])
|
||||||
for cur in currencies
|
for cur in currencies
|
||||||
|
@ -75,9 +69,7 @@ def get_rates(config: flask.config.Config) -> dict[str, Decimal]:
|
||||||
currency_string = ",".join(sorted(currencies))
|
currency_string = ",".join(sorted(currencies))
|
||||||
file_suffix = f"{currency_string}_to_GBP.json"
|
file_suffix = f"{currency_string}_to_GBP.json"
|
||||||
existing_data = os.listdir(fx_dir)
|
existing_data = os.listdir(fx_dir)
|
||||||
existing_files = [f for f in existing_data if f.endswith(".json")]
|
existing_files = [f for f in existing_data if f.endswith(file_suffix)]
|
||||||
|
|
||||||
full_path: str | None = None
|
|
||||||
|
|
||||||
if existing_files:
|
if existing_files:
|
||||||
recent_filename = max(existing_files)
|
recent_filename = max(existing_files)
|
||||||
|
@ -85,7 +77,7 @@ def get_rates(config: flask.config.Config) -> dict[str, Decimal]:
|
||||||
delta = now - recent
|
delta = now - recent
|
||||||
|
|
||||||
full_path = os.path.join(fx_dir, recent_filename)
|
full_path = os.path.join(fx_dir, recent_filename)
|
||||||
if not recent_filename.endswith(file_suffix) or delta < timedelta(hours=12):
|
if delta < timedelta(hours=12):
|
||||||
return read_cached_rates(full_path, currencies)
|
return read_cached_rates(full_path, currencies)
|
||||||
|
|
||||||
url = "http://api.exchangerate.host/live"
|
url = "http://api.exchangerate.host/live"
|
||||||
|
|
|
@ -3,14 +3,26 @@
|
||||||
from datetime import timedelta, timezone
|
from datetime import timedelta, timezone
|
||||||
|
|
||||||
import dateutil.tz
|
import dateutil.tz
|
||||||
import exchange_calendars # type: ignore
|
import exchange_calendars
|
||||||
import pandas # type: ignore
|
import pandas
|
||||||
|
|
||||||
from . import utils
|
|
||||||
|
|
||||||
here = dateutil.tz.tzlocal()
|
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]:
|
def open_and_close() -> list[str]:
|
||||||
"""Stock markets open and close times."""
|
"""Stock markets open and close times."""
|
||||||
# The trading calendars code is slow, maybe there is a faster way to do this
|
# The trading calendars code is slow, maybe there is a faster way to do this
|
||||||
|
@ -28,11 +40,11 @@ def open_and_close() -> list[str]:
|
||||||
if cal.is_open_on_minute(now_local):
|
if cal.is_open_on_minute(now_local):
|
||||||
next_close = cal.next_close(now).tz_convert(here)
|
next_close = cal.next_close(now).tz_convert(here)
|
||||||
next_close = next_close.replace(minute=round(next_close.minute, -1))
|
next_close = next_close.replace(minute=round(next_close.minute, -1))
|
||||||
delta_close = utils.timedelta_display(next_close - now_local)
|
delta_close = timedelta_display(next_close - now_local)
|
||||||
|
|
||||||
prev_open = cal.previous_open(now).tz_convert(here)
|
prev_open = cal.previous_open(now).tz_convert(here)
|
||||||
prev_open = prev_open.replace(minute=round(prev_open.minute, -1))
|
prev_open = prev_open.replace(minute=round(prev_open.minute, -1))
|
||||||
delta_open = utils.timedelta_display(now_local - prev_open)
|
delta_open = timedelta_display(now_local - prev_open)
|
||||||
|
|
||||||
msg = (
|
msg = (
|
||||||
f"{label:>6} market opened {delta_open} ago, "
|
f"{label:>6} market opened {delta_open} ago, "
|
||||||
|
@ -42,7 +54,7 @@ def open_and_close() -> list[str]:
|
||||||
ts = cal.next_open(now)
|
ts = cal.next_open(now)
|
||||||
ts = ts.replace(minute=round(ts.minute, -1))
|
ts = ts.replace(minute=round(ts.minute, -1))
|
||||||
ts = ts.tz_convert(here)
|
ts = ts.tz_convert(here)
|
||||||
delta = utils.timedelta_display(ts - now_local)
|
delta = timedelta_display(ts - now_local)
|
||||||
msg = f"{label:>6} market opens in {delta}" + (
|
msg = f"{label:>6} market opens in {delta}" + (
|
||||||
f" ({ts:%H:%M})" if (ts - now_local) < timedelta(days=1) else ""
|
f" ({ts:%H:%M})" if (ts - now_local) < timedelta(days=1) else ""
|
||||||
)
|
)
|
||||||
|
|
|
@ -7,7 +7,7 @@ import typing
|
||||||
|
|
||||||
import flask
|
import flask
|
||||||
import yaml
|
import yaml
|
||||||
from geopy.distance import geodesic # type: ignore
|
from geopy.distance import geodesic
|
||||||
|
|
||||||
from .types import Event, StrDict
|
from .types import Event, StrDict
|
||||||
|
|
||||||
|
|
|
@ -28,19 +28,6 @@ def load_travel(travel_type: str, plural: str, data_dir: str) -> list[StrDict]:
|
||||||
return items
|
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(
|
def load_trains(
|
||||||
data_dir: str, route_distances: travel.RouteDistances | None = None
|
data_dir: str, route_distances: travel.RouteDistances | None = None
|
||||||
) -> list[StrDict]:
|
) -> list[StrDict]:
|
||||||
|
@ -58,7 +45,13 @@ def load_trains(
|
||||||
train["to_station"] = by_name[train["to"]]
|
train["to_station"] = by_name[train["to"]]
|
||||||
|
|
||||||
for leg in train["legs"]:
|
for leg in train["legs"]:
|
||||||
process_train_leg(leg, by_name=by_name, route_distances=route_distances)
|
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)
|
||||||
|
|
||||||
if all("distance" in leg for leg in train["legs"]):
|
if all("distance" in leg for leg in train["legs"]):
|
||||||
train["distance"] = sum(leg["distance"] for leg in train["legs"])
|
train["distance"] = sum(leg["distance"] for leg in train["legs"])
|
||||||
|
@ -97,20 +90,6 @@ def depart_datetime(item: StrDict) -> datetime:
|
||||||
return datetime.combine(depart, time.min).replace(tzinfo=ZoneInfo("UTC"))
|
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]:
|
def load_flight_bookings(data_dir: str) -> list[StrDict]:
|
||||||
"""Load flight bookings."""
|
"""Load flight bookings."""
|
||||||
bookings = load_travel("flight", "flights", data_dir)
|
bookings = load_travel("flight", "flights", data_dir)
|
||||||
|
@ -119,7 +98,14 @@ def load_flight_bookings(data_dir: str) -> list[StrDict]:
|
||||||
airports = travel.parse_yaml("airports", data_dir)
|
airports = travel.parse_yaml("airports", data_dir)
|
||||||
for booking in bookings:
|
for booking in bookings:
|
||||||
for flight in booking["flights"]:
|
for flight in booking["flights"]:
|
||||||
process_flight(flight, iata, airports)
|
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)
|
||||||
return bookings
|
return bookings
|
||||||
|
|
||||||
|
|
||||||
|
@ -286,7 +272,7 @@ def read_geojson(data_dir: str, filename: str) -> str:
|
||||||
|
|
||||||
def get_trip_routes(trip: Trip) -> list[StrDict]:
|
def get_trip_routes(trip: Trip) -> list[StrDict]:
|
||||||
"""Get routes for given trip to show on map."""
|
"""Get routes for given trip to show on map."""
|
||||||
routes: list[StrDict] = []
|
routes = []
|
||||||
seen_geojson = set()
|
seen_geojson = set()
|
||||||
for t in trip.travel:
|
for t in trip.travel:
|
||||||
if t["type"] == "ferry":
|
if t["type"] == "ferry":
|
||||||
|
|
|
@ -2,7 +2,7 @@
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import typing
|
import typing
|
||||||
from datetime import date, datetime, timedelta, timezone
|
from datetime import date, datetime, timezone
|
||||||
from time import time
|
from time import time
|
||||||
|
|
||||||
|
|
||||||
|
@ -28,20 +28,6 @@ def as_datetime(d: datetime | date) -> datetime:
|
||||||
raise TypeError(f"Unsupported type: {type(d)}")
|
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:
|
def human_readable_delta(future_date: date) -> str | None:
|
||||||
"""
|
"""
|
||||||
Calculate the human-readable time delta for a given future date.
|
Calculate the human-readable time delta for a given future date.
|
||||||
|
|
|
@ -43,9 +43,9 @@
|
||||||
</ul>
|
</ul>
|
||||||
|
|
||||||
<div class="grid-container">
|
<div class="grid-container">
|
||||||
|
{{ section("Past", past) }}
|
||||||
{{ section("Current", current) }}
|
{{ section("Current", current) }}
|
||||||
{{ section("Future", future) }}
|
{{ section("Future", future) }}
|
||||||
{{ section("Past", past) }}
|
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
|
|
@ -150,7 +150,6 @@
|
||||||
{% for name, seconds in timings %}
|
{% for name, seconds in timings %}
|
||||||
<li>{{ name }} took {{ "%.1f" | format(seconds) }} seconds</li>
|
<li>{{ name }} took {{ "%.1f" | format(seconds) }} seconds</li>
|
||||||
{% endfor %}
|
{% endfor %}
|
||||||
<li>Render time: {{ "%.1f" | format(render_time) }} seconds</li>
|
|
||||||
|
|
||||||
</ul>
|
</ul>
|
||||||
</div>
|
</div>
|
||||||
|
|
|
@ -38,7 +38,7 @@
|
||||||
<div class="grid-item text-end">
|
<div class="grid-item text-end">
|
||||||
{% if item.price and item.currency %}
|
{% if item.price and item.currency %}
|
||||||
<span class="badge bg-info text-nowrap">{{ "{:,d}".format(item.price | int) }} {{ item.currency }}</span>
|
<span class="badge bg-info text-nowrap">{{ "{:,d}".format(item.price | int) }} {{ item.currency }}</span>
|
||||||
{% if item.currency != "GBP" and item.currency in fx_rate %}
|
{% if item.currency != "GBP" %}
|
||||||
<span class="badge bg-info text-nowrap">{{ "{:,.2f}".format(item.price / fx_rate[item.currency]) }} GBP</span>
|
<span class="badge bg-info text-nowrap">{{ "{:,.2f}".format(item.price / fx_rate[item.currency]) }} GBP</span>
|
||||||
{% endif %}
|
{% endif %}
|
||||||
{% elif item.free %}
|
{% elif item.free %}
|
||||||
|
|
|
@ -7,7 +7,6 @@ import inspect
|
||||||
import operator
|
import operator
|
||||||
import os.path
|
import os.path
|
||||||
import sys
|
import sys
|
||||||
import time
|
|
||||||
import traceback
|
import traceback
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
|
@ -71,7 +70,6 @@ def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str,
|
||||||
@app.route("/")
|
@app.route("/")
|
||||||
async def index() -> str:
|
async def index() -> str:
|
||||||
"""Index page."""
|
"""Index page."""
|
||||||
t0 = time.time()
|
|
||||||
now = datetime.now()
|
now = datetime.now()
|
||||||
data = await agenda.data.get_data(now, app.config)
|
data = await agenda.data.get_data(now, app.config)
|
||||||
|
|
||||||
|
@ -90,7 +88,6 @@ async def index() -> str:
|
||||||
fullcalendar_events=calendar.build_events(events),
|
fullcalendar_events=calendar.build_events(events),
|
||||||
start_event_list=date.today() - timedelta(days=1),
|
start_event_list=date.today() - timedelta(days=1),
|
||||||
end_event_list=date.today() + timedelta(days=365 * 2),
|
end_event_list=date.today() + timedelta(days=365 * 2),
|
||||||
render_time=(time.time() - t0),
|
|
||||||
**data,
|
**data,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue