diff --git a/agenda/accommodation.py b/agenda/accommodation.py index 59e9d64..5061a83 100644 --- a/agenda/accommodation.py +++ b/agenda/accommodation.py @@ -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, + } diff --git a/agenda/holidays.py b/agenda/holidays.py index b0af818..5c91e95 100644 --- a/agenda/holidays.py +++ b/agenda/holidays.py @@ -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 diff --git a/agenda/stats.py b/agenda/stats.py index 91493c9..10f8981 100644 --- a/agenda/stats.py +++ b/agenda/stats.py @@ -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()) diff --git a/agenda/trip_debug.py b/agenda/trip_debug.py new file mode 100644 index 0000000..663a9b5 --- /dev/null +++ b/agenda/trip_debug.py @@ -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 diff --git a/web_view.py b/web_view.py index 3107763..b6fb42a 100755 --- a/web_view.py +++ b/web_view.py @@ -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,