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]]
|
||||
|
||||
|
||||
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"]
|
||||
|
@ -263,7 +230,34 @@ 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)
|
||||
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)]
|
||||
|
||||
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"])
|
||||
|
||||
|
||||
def read_cached_rates(
|
||||
filename: str | None, currencies: list[str]
|
||||
) -> dict[str, Decimal]:
|
||||
def read_cached_rates(filename: str, 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
|
||||
|
@ -75,9 +69,7 @@ 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(".json")]
|
||||
|
||||
full_path: str | None = None
|
||||
existing_files = [f for f in existing_data if f.endswith(file_suffix)]
|
||||
|
||||
if existing_files:
|
||||
recent_filename = max(existing_files)
|
||||
|
@ -85,7 +77,7 @@ def get_rates(config: flask.config.Config) -> dict[str, Decimal]:
|
|||
delta = now - recent
|
||||
|
||||
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)
|
||||
|
||||
url = "http://api.exchangerate.host/live"
|
||||
|
|
|
@ -3,14 +3,26 @@
|
|||
from datetime import timedelta, timezone
|
||||
|
||||
import dateutil.tz
|
||||
import exchange_calendars # type: ignore
|
||||
import pandas # type: ignore
|
||||
|
||||
from . import utils
|
||||
import exchange_calendars
|
||||
import pandas
|
||||
|
||||
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
|
||||
|
@ -28,11 +40,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 = 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 = 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 = (
|
||||
f"{label:>6} market opened {delta_open} ago, "
|
||||
|
@ -42,7 +54,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 = utils.timedelta_display(ts - now_local)
|
||||
delta = 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 # type: ignore
|
||||
from geopy.distance import geodesic
|
||||
|
||||
from .types import Event, StrDict
|
||||
|
||||
|
|
|
@ -28,19 +28,6 @@ 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]:
|
||||
|
@ -58,7 +45,13 @@ def load_trains(
|
|||
train["to_station"] = by_name[train["to"]]
|
||||
|
||||
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"]):
|
||||
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"))
|
||||
|
||||
|
||||
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)
|
||||
|
@ -119,7 +98,14 @@ 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"]:
|
||||
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
|
||||
|
||||
|
||||
|
@ -286,7 +272,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: list[StrDict] = []
|
||||
routes = []
|
||||
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, timedelta, timezone
|
||||
from datetime import date, datetime, timezone
|
||||
from time import time
|
||||
|
||||
|
||||
|
@ -28,20 +28,6 @@ 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,7 +150,6 @@
|
|||
{% 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" 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>
|
||||
{% endif %}
|
||||
{% elif item.free %}
|
||||
|
|
|
@ -7,7 +7,6 @@ import inspect
|
|||
import operator
|
||||
import os.path
|
||||
import sys
|
||||
import time
|
||||
import traceback
|
||||
from collections import defaultdict
|
||||
from datetime import date, datetime, timedelta
|
||||
|
@ -71,7 +70,6 @@ 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)
|
||||
|
||||
|
@ -90,7 +88,6 @@ 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