edward/agenda
1
0
Fork 0
Code Issues 10 Pull requests Projects Releases Packages Wiki Activity

Extract accommodation, trip statistics, holiday mapping and debug helpers

Browse source
This commit is contained in:
Edward Betts 2026-09-09 12:08:13 +01:00
parent e3eb3e5af8
commit cb370b9378
5 changed files with 204 additions and 169 deletions
Download patch file Download diff file Expand all files Collapse all files

53
agenda/accommodation.py
View file

@ -1,8 +1,12 @@
"""Accommodation."""
from collections import defaultdict
from datetime import datetime, timedelta
import yaml
from .event import Event
from .types import StrDict, Trip
def get_events(filepath: str) -> list[Event]:
@ -22,3 +26,52 @@ def get_events(filepath: str) -> list[Event]:
)
for item in yaml.safe_load(f)
]
def prepare_accommodation_list(
items: list[StrDict], trips: list[Trip], now: datetime
) -> StrDict:
"""Link stays to trips and prepare date groups and annual night counts."""
# Create a dictionary to hold stats for each year
year_stats: defaultdict[int, dict[str, int]] = defaultdict(
lambda: {"total_nights": 0, "nights_abroad": 0}
)
# Calculate stats for each year
for stay in items:
current_date = stay["from"].date()
end_date = stay["to"].date()
while current_date < end_date:
year = current_date.year
year_stats[year]["total_nights"] += 1
if stay.get("country") != "gb":
year_stats[year]["nights_abroad"] += 1
current_date += timedelta(days=1)
# Sort the stats by year in descending order
sorted_year_stats = sorted(
year_stats.items(), key=lambda item: item[0], reverse=True
)
trip_lookup = {}
for trip in trips:
for trip_stay in trip.accommodation:
key = (trip_stay["from"], trip_stay["name"])
trip_lookup[key] = trip
for item in items:
key = (item["from"], item["name"])
if this_trip := trip_lookup.get(key):
item["linked_trip"] = this_trip
past = [conf for conf in items if conf["to"] < now]
current = [conf for conf in items if conf["from"] <= now and conf["to"] >= now]
future = [conf for conf in items if conf["from"] > now]
return {
"past": past,
"current": current,
"future": future,
"year_stats": sorted_year_stats,
}

27
agenda/holidays.py
View file

@ -182,3 +182,30 @@ def get_all(last_year: date, next_year: date, data_dir: str) -> list[Holiday]:
holiday_list += get_holidays(country, last_year, next_year)
return holiday_list
def trip_school_holiday_map(trips: list[Trip], data_dir: str) -> dict[str, list[Event]]:
"""Map trip-start ISO date to overlapping UK school holidays."""
if not trips:
return {}
starts = [trip.start for trip in trips]
ends = [trip.end or trip.start for trip in trips]
school_holidays = get_school_holidays(
min(starts),
max(ends),
data_dir,
)
result: dict[str, list[Event]] = {}
for trip in trips:
trip_end = trip.end or trip.start
overlaps = [
school_holiday
for school_holiday in school_holidays
if school_holiday.as_date <= trip_end
and school_holiday.end_as_date >= trip.start
]
result[trip.start.isoformat()] = overlaps
return result

35
agenda/stats.py
View file

@ -151,3 +151,38 @@ def calculate_yearly_stats(
travel_legs(trip, yearly_stats[year])
return dict(yearly_stats)
def calc_total_distance(trips: list[Trip]) -> float:
"""Total distance for trips."""
total = 0.0
for item in trips:
if dist := item.total_distance():
total += dist
return total
def calc_total_co2_kg(trips: list[Trip]) -> float:
"""Total CO₂ for trips."""
return sum(item.total_co2_kg() or 0.0 for item in trips)
def sum_distances_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
"""Sum distances by transport type."""
distances_by_transport_type: defaultdict[str, float] = defaultdict(float)
for trip in trips:
for transport_type, dist in trip.distances_by_transport_type():
distances_by_transport_type[transport_type] += dist
return list(distances_by_transport_type.items())
def sum_co2_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
"""Sum CO₂ by transport type."""
co2_by_transport_type: defaultdict[str, float] = defaultdict(float)
for trip in trips:
for transport_type, co2_kg in trip.co2_by_transport_type():
co2_by_transport_type[transport_type] += co2_kg
return list(co2_by_transport_type.items())

61
agenda/trip_debug.py Normal file
View file

@ -0,0 +1,61 @@
"""Serialize trip details for the debug page."""
import json
import yaml
from agenda.types import Trip
def serialize_trip(trip: Trip) -> tuple[str, str]:
"""Return JSON and YAML including computed trip properties."""
# Convert trip object to dictionary for display
trip_dict = {
"start": trip.start.isoformat(),
"name": trip.name,
"private": trip.private,
"travel": trip.travel,
"accommodation": trip.accommodation,
"conferences": trip.conferences,
"events": trip.events,
"flight_bookings": trip.flight_bookings,
"computed_properties": {
"title": trip.title,
"end": trip.end.isoformat() if trip.end else None,
"countries": [
{"name": c.name, "alpha_2": c.alpha_2, "flag": c.flag}
for c in trip.countries
],
"locations": [
{
"location": loc,
"country": {"name": country.name, "alpha_2": country.alpha_2},
}
for loc, country in trip.locations()
],
"total_distance": trip.total_distance(),
"total_co2_kg": trip.total_co2_kg(),
"distances_by_transport_type": trip.distances_by_transport_type(),
"co2_by_transport_type": trip.co2_by_transport_type(),
},
"schengen_compliance": (
{
"total_days_used": trip.schengen_compliance.total_days_used,
"days_remaining": trip.schengen_compliance.days_remaining,
"is_compliant": trip.schengen_compliance.is_compliant,
"current_180_day_period": [
trip.schengen_compliance.current_180_day_period[0].isoformat(),
trip.schengen_compliance.current_180_day_period[1].isoformat(),
],
"days_over_limit": trip.schengen_compliance.days_over_limit,
}
if trip.schengen_compliance
else None
),
}
# Convert to JSON for pretty printing
trip_json = json.dumps(trip_dict, indent=2, default=str)
trip_yaml = yaml.safe_dump(json.loads(trip_json), sort_keys=False)
return trip_json, trip_yaml

197
web_view.py
View file

@ -3,23 +3,21 @@
"""Web page to show upcoming events."""
import inspect
import json
import os.path
import sys
import time
import traceback
import typing
from collections import defaultdict
from datetime import date
from datetime import datetime, timedelta, timezone
import flask
import werkzeug
import werkzeug.debug.tbtools
import yaml
from authlib.integrations.flask_client import OAuth
from werkzeug.middleware.proxy_fix import ProxyFix
import agenda.accommodation
import agenda.conference
import agenda.conference_ical
import agenda.conference_list
@ -31,13 +29,13 @@ import agenda.meteors
import agenda.stats
import agenda.thespacedevs
import agenda.trip
import agenda.trip_debug
import agenda.trip_schengen
import agenda.trip_timezones
import agenda.uk_school_holiday
import agenda.utils
import agenda.weather
from agenda import calendar, format_list_with_ampersand, travel, uk_tz
from agenda.event import Event
from agenda.types import StrDict, Trip
app = flask.Flask(__name__)
@ -492,51 +490,13 @@ def accommodation_list() -> str:
data_dir = app.config["PERSONAL_DATA"]
items = travel.parse_yaml("accommodation", data_dir)
# Create a dictionary to hold stats for each year
year_stats: defaultdict[int, dict[str, int]] = defaultdict(
lambda: {"total_nights": 0, "nights_abroad": 0}
context = agenda.accommodation.prepare_accommodation_list(
items, agenda.trip.build_trip_list(), uk_tz.localize(datetime.now())
)
# Calculate stats for each year
for stay in items:
current_date = stay["from"].date()
end_date = stay["to"].date()
while current_date < end_date:
year = current_date.year
year_stats[year]["total_nights"] += 1
if stay.get("country") != "gb":
year_stats[year]["nights_abroad"] += 1
current_date += timedelta(days=1)
# Sort the stats by year in descending order
sorted_year_stats = sorted(
year_stats.items(), key=lambda item: item[0], reverse=True
)
trip_lookup = {}
for trip in agenda.trip.build_trip_list():
for trip_stay in trip.accommodation:
key = (trip_stay["from"], trip_stay["name"])
trip_lookup[key] = trip
for item in items:
key = (item["from"], item["name"])
if this_trip := trip_lookup.get(key):
item["linked_trip"] = this_trip
now = uk_tz.localize(datetime.now())
past = [conf for conf in items if conf["to"] < now]
current = [conf for conf in items if conf["from"] <= now and conf["to"] >= now]
future = [conf for conf in items if conf["from"] > now]
return flask.render_template(
"accommodation.html",
past=past,
current=current,
future=future,
year_stats=sorted_year_stats,
**context,
get_country=agenda.get_country,
fx_rate=agenda.fx.get_rates(app.config),
)
@ -558,41 +518,6 @@ def trip_ical() -> werkzeug.Response:
return response
def calc_total_distance(trips: list[Trip]) -> float:
"""Total distance for trips."""
total = 0.0
for item in trips:
if dist := item.total_distance():
total += dist
return total
def calc_total_co2_kg(trips: list[Trip]) -> float:
"""Total CO₂ for trips."""
return sum(item.total_co2_kg() or 0.0 for item in trips)
def sum_distances_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
"""Sum distances by transport type."""
distances_by_transport_type: defaultdict[str, float] = defaultdict(float)
for trip in trips:
for transport_type, dist in trip.distances_by_transport_type():
distances_by_transport_type[transport_type] += dist
return list(distances_by_transport_type.items())
def sum_co2_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
"""Sum CO₂ by transport type."""
co2_by_transport_type: defaultdict[str, float] = defaultdict(float)
for trip in trips:
for transport_type, co2_kg in trip.co2_by_transport_type():
co2_by_transport_type[transport_type] += co2_kg
return list(co2_by_transport_type.items())
def get_home_weather() -> list[StrDict]:
"""Get Bristol home weather forecast, with date objects added for templates."""
@ -614,33 +539,6 @@ def get_trip_list() -> list[Trip]:
return agenda.trip.get_trip_list(route_distances)
def trip_school_holiday_map(trips: list[Trip]) -> dict[str, list[Event]]:
"""Map trip-start ISO date to overlapping UK school holidays."""
if not trips:
return {}
starts = [trip.start for trip in trips]
ends = [trip.end or trip.start for trip in trips]
school_holidays = agenda.holidays.get_school_holidays(
min(starts),
max(ends),
app.config["DATA_DIR"],
)
result: dict[str, list[Event]] = {}
for trip in trips:
trip_end = trip.end or trip.start
overlaps = [
school_holiday
for school_holiday in school_holidays
if school_holiday.as_date <= trip_end
and school_holiday.end_as_date >= trip.start
]
result[trip.start.isoformat()] = overlaps
return result
@app.route("/trip/past")
def trip_past_list() -> str:
"""Page showing a list of past trips."""
@ -652,17 +550,19 @@ def trip_past_list() -> str:
"trip/list.html",
heading="Past trips",
trips=reversed(past),
trip_school_holiday_map=trip_school_holiday_map(past),
trip_school_holiday_map=agenda.holidays.trip_school_holiday_map(
past, app.config["DATA_DIR"]
),
coordinates=coordinates,
routes=routes,
today=today,
get_country=agenda.get_country,
format_list_with_ampersand=format_list_with_ampersand,
fx_rate=agenda.fx.get_rates(app.config),
total_distance=calc_total_distance(past),
total_co2_kg=calc_total_co2_kg(past),
distances_by_transport_type=sum_distances_by_transport_type(past),
co2_by_transport_type=sum_co2_by_transport_type(past),
total_distance=agenda.stats.calc_total_distance(past),
total_co2_kg=agenda.stats.calc_total_co2_kg(past),
distances_by_transport_type=agenda.stats.sum_distances_by_transport_type(past),
co2_by_transport_type=agenda.stats.sum_co2_by_transport_type(past),
)
@ -714,7 +614,9 @@ def trip_future_list() -> str:
"trip/list.html",
heading="Future trips",
trips=shown,
trip_school_holiday_map=trip_school_holiday_map(shown),
trip_school_holiday_map=agenda.holidays.trip_school_holiday_map(
shown, app.config["DATA_DIR"]
),
trip_weather_map=trip_weather_map,
coordinates=coordinates,
routes=routes,
@ -722,10 +624,12 @@ def trip_future_list() -> str:
get_country=agenda.get_country,
format_list_with_ampersand=format_list_with_ampersand,
fx_rate=agenda.fx.get_rates(app.config),
total_distance=calc_total_distance(current + future),
total_co2_kg=calc_total_co2_kg(current + future),
distances_by_transport_type=sum_distances_by_transport_type(current + future),
co2_by_transport_type=sum_co2_by_transport_type(current + future),
total_distance=agenda.stats.calc_total_distance(current + future),
total_co2_kg=agenda.stats.calc_total_co2_kg(current + future),
distances_by_transport_type=agenda.stats.sum_distances_by_transport_type(
current + future
),
co2_by_transport_type=agenda.stats.sum_co2_by_transport_type(current + future),
)
@ -835,54 +739,7 @@ def trip_debug_page(start: str) -> str:
# Add Schengen compliance information
trip = agenda.trip_schengen.add_schengen_compliance_to_trip(trip)
# Convert trip object to dictionary for display
trip_dict = {
"start": trip.start.isoformat(),
"name": trip.name,
"private": trip.private,
"travel": trip.travel,
"accommodation": trip.accommodation,
"conferences": trip.conferences,
"events": trip.events,
"flight_bookings": trip.flight_bookings,
"computed_properties": {
"title": trip.title,
"end": trip.end.isoformat() if trip.end else None,
"countries": [
{"name": c.name, "alpha_2": c.alpha_2, "flag": c.flag}
for c in trip.countries
],
"locations": [
{
"location": loc,
"country": {"name": country.name, "alpha_2": country.alpha_2},
}
for loc, country in trip.locations()
],
"total_distance": trip.total_distance(),
"total_co2_kg": trip.total_co2_kg(),
"distances_by_transport_type": trip.distances_by_transport_type(),
"co2_by_transport_type": trip.co2_by_transport_type(),
},
"schengen_compliance": (
{
"total_days_used": trip.schengen_compliance.total_days_used,
"days_remaining": trip.schengen_compliance.days_remaining,
"is_compliant": trip.schengen_compliance.is_compliant,
"current_180_day_period": [
trip.schengen_compliance.current_180_day_period[0].isoformat(),
trip.schengen_compliance.current_180_day_period[1].isoformat(),
],
"days_over_limit": trip.schengen_compliance.days_over_limit,
}
if trip.schengen_compliance
else None
),
}
# Convert to JSON for pretty printing
trip_json = json.dumps(trip_dict, indent=2, default=str)
trip_yaml = yaml.safe_dump(json.loads(trip_json), sort_keys=False)
trip_json, trip_yaml = agenda.trip_debug.serialize_trip(trip)
return flask.render_template(
"trip_debug.html",
@ -946,10 +803,12 @@ def trip_stats() -> str:
return flask.render_template(
"trip/stats.html",
count=len(trip_list),
total_distance=calc_total_distance(trip_list),
total_co2_kg=calc_total_co2_kg(trip_list),
distances_by_transport_type=sum_distances_by_transport_type(trip_list),
co2_by_transport_type=sum_co2_by_transport_type(trip_list),
total_distance=agenda.stats.calc_total_distance(trip_list),
total_co2_kg=agenda.stats.calc_total_co2_kg(trip_list),
distances_by_transport_type=agenda.stats.sum_distances_by_transport_type(
trip_list
),
co2_by_transport_type=agenda.stats.sum_co2_by_transport_type(trip_list),
yearly_stats=yearly_stats,
overall_stats=overall_stats,
conferences=conferences,