Compare commits
No commits in common. "cb370b93781c7d0e2ef7f25a7da3de0243ac2d36" and "df7fd2fb9e3845eb7638ba95d985ac88c51b8a4d" have entirely different histories.
cb370b9378
...
df7fd2fb9e
9 changed files with 717 additions and 795 deletions
|
|
@ -1,12 +1,8 @@
|
|||
"""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]:
|
||||
|
|
@ -26,52 +22,3 @@ 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,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,207 +0,0 @@
|
|||
"""Prepare conference lists, country filters, series summaries, and timelines."""
|
||||
|
||||
import decimal
|
||||
import os.path
|
||||
import typing
|
||||
from collections import defaultdict
|
||||
from datetime import date, timedelta
|
||||
|
||||
import yaml
|
||||
|
||||
import agenda.conference
|
||||
from agenda.types import StrDict, Trip
|
||||
|
||||
|
||||
def build_conference_list(data_dir: str, trips: list[Trip]) -> list[StrDict]:
|
||||
"""Build conference list."""
|
||||
filepath = os.path.join(data_dir, "conferences.yaml")
|
||||
items: list[StrDict] = yaml.safe_load(open(filepath))
|
||||
series_lookup = agenda.conference.load_series(data_dir)
|
||||
conference_trip_lookup = {}
|
||||
|
||||
for trip in trips:
|
||||
for trip_conf in trip.conferences:
|
||||
key = (trip_conf["start"], trip_conf["name"])
|
||||
conference_trip_lookup[key] = trip
|
||||
|
||||
for conf in items:
|
||||
conf.update(agenda.conference.validate_conference_date_fields(conf))
|
||||
|
||||
price = conf.get("price")
|
||||
if price:
|
||||
conf["price"] = decimal.Decimal(price)
|
||||
|
||||
series_id = conf.get("series")
|
||||
if isinstance(series_id, str):
|
||||
conf["series_detail"] = series_lookup.get(series_id)
|
||||
|
||||
if "start" in conf:
|
||||
key = (conf["start"], conf["name"])
|
||||
if this_trip := conference_trip_lookup.get(key):
|
||||
conf["linked_trip"] = this_trip
|
||||
|
||||
items.sort(key=lambda item: item["sort_date"])
|
||||
return items
|
||||
|
||||
|
||||
def conference_country_code(conf: StrDict) -> str | None:
|
||||
"""Return normalized alpha-2 country code for a conference."""
|
||||
country = conf.get("country")
|
||||
if not isinstance(country, str):
|
||||
return None
|
||||
|
||||
code = country.strip().lower()
|
||||
return code if len(code) == 2 else None
|
||||
|
||||
|
||||
def normalize_country_filter(value: str | None) -> str | None:
|
||||
"""Normalize and validate a country query parameter."""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
code = value.strip().lower()
|
||||
if len(code) != 2:
|
||||
return None
|
||||
|
||||
return code if agenda.get_country(code) else None
|
||||
|
||||
|
||||
def filter_conferences_by_country(
|
||||
items: list[StrDict], country_code: str | None
|
||||
) -> list[StrDict]:
|
||||
"""Filter conferences by country when a country code is provided."""
|
||||
if not country_code:
|
||||
return items
|
||||
|
||||
return [item for item in items if conference_country_code(item) == country_code]
|
||||
|
||||
|
||||
def conference_country_options(items: list[StrDict]) -> list[StrDict]:
|
||||
"""Return country options for the conference country filter."""
|
||||
counts: defaultdict[str, int] = defaultdict(int)
|
||||
for item in items:
|
||||
code = conference_country_code(item)
|
||||
if code and agenda.get_country(code):
|
||||
counts[code] += 1
|
||||
|
||||
options: list[StrDict] = []
|
||||
for code, count in counts.items():
|
||||
country = agenda.get_country(code)
|
||||
if not country:
|
||||
continue
|
||||
options.append(
|
||||
{
|
||||
"code": code,
|
||||
"name": country.name,
|
||||
"flag": country.flag,
|
||||
"count": count,
|
||||
}
|
||||
)
|
||||
|
||||
options.sort(key=lambda item: str(item["name"]))
|
||||
return options
|
||||
|
||||
|
||||
def build_conference_series_list(
|
||||
series_lookup: dict[str, agenda.conference.ConferenceSeries],
|
||||
conferences: list[StrDict],
|
||||
today: date,
|
||||
) -> list[StrDict]:
|
||||
"""Build conference series list with conference counts."""
|
||||
|
||||
series_items: list[StrDict] = []
|
||||
for series_id, series in series_lookup.items():
|
||||
linked = [conf for conf in conferences if conf.get("series") == series_id]
|
||||
latest = max((conf["sort_date"] for conf in linked), default=None)
|
||||
next_conf = next(
|
||||
(conf for conf in linked if conf["latest_date"] >= today), None
|
||||
)
|
||||
attended = any(conf.get("going") or conf.get("linked_trip") for conf in linked)
|
||||
item: StrDict = {
|
||||
"id": series_id,
|
||||
**series,
|
||||
"count": len(linked),
|
||||
"latest": latest,
|
||||
"next_conf": next_conf,
|
||||
"attended": attended,
|
||||
}
|
||||
series_items.append(item)
|
||||
|
||||
series_items.sort(key=lambda item: str(item["name"]).lower())
|
||||
return series_items
|
||||
|
||||
|
||||
def build_conference_timeline(
|
||||
current: list[StrDict], future: list[StrDict], today: date, days: int = 90
|
||||
) -> dict[str, typing.Any] | None:
|
||||
"""Build data for a Gantt-style timeline of upcoming conferences."""
|
||||
timeline_start = today
|
||||
timeline_end = today + timedelta(days=days)
|
||||
|
||||
visible = [
|
||||
c
|
||||
for c in (current + future)
|
||||
if c["has_exact_dates"]
|
||||
and c["start_date"] <= timeline_end
|
||||
and c["end_date"] >= today
|
||||
]
|
||||
if not visible:
|
||||
return None
|
||||
|
||||
visible.sort(key=lambda c: c["start_date"])
|
||||
|
||||
# Greedy interval-coloring: assign each conference a lane (row)
|
||||
lane_ends: list[date] = []
|
||||
conf_data = []
|
||||
for conf in visible:
|
||||
lane = next(
|
||||
(i for i, end in enumerate(lane_ends) if end < conf["start_date"]),
|
||||
len(lane_ends),
|
||||
)
|
||||
if lane == len(lane_ends):
|
||||
lane_ends.append(conf["end_date"])
|
||||
else:
|
||||
lane_ends[lane] = conf["end_date"]
|
||||
|
||||
start_off = max((conf["start_date"] - timeline_start).days, 0)
|
||||
end_off = min((conf["end_date"] - timeline_start).days + 1, days)
|
||||
left_pct = round(start_off / days * 100, 2)
|
||||
width_pct = max(round((end_off - start_off) / days * 100, 2), 0.5)
|
||||
|
||||
conf_data.append(
|
||||
{
|
||||
"name": conf["name"],
|
||||
"url": conf.get("url"),
|
||||
"lane": lane,
|
||||
"left_pct": left_pct,
|
||||
"width_pct": width_pct,
|
||||
"key": f"{conf['start_date'].isoformat()}|{conf['name']}",
|
||||
"label": (
|
||||
f"{conf['name']} ({conf['start_date'].strftime('%-d %b')})"
|
||||
if conf["start_date"] == conf["end_date"]
|
||||
else (
|
||||
f"{conf['name']}"
|
||||
f" ({conf['start_date'].strftime('%-d %b')}–"
|
||||
f"{conf['end_date'].strftime('%-d %b')})"
|
||||
)
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
# Month markers for x-axis labels
|
||||
months = []
|
||||
d = today.replace(day=1)
|
||||
while d <= timeline_end:
|
||||
off = max((d - timeline_start).days, 0)
|
||||
months.append(
|
||||
{"label": d.strftime("%b %Y"), "left_pct": round(off / days * 100, 2)}
|
||||
)
|
||||
# advance to next month
|
||||
d = (d.replace(day=28) + timedelta(days=4)).replace(day=1)
|
||||
|
||||
return {
|
||||
"confs": conf_data,
|
||||
"lane_count": len(lane_ends),
|
||||
"months": months,
|
||||
"days": days,
|
||||
}
|
||||
|
|
@ -182,30 +182,3 @@ 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
|
||||
|
|
|
|||
|
|
@ -151,38 +151,3 @@ 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())
|
||||
|
|
|
|||
|
|
@ -1,61 +0,0 @@
|
|||
"""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
|
||||
|
|
@ -1,341 +0,0 @@
|
|||
"""Resolve trip destination timezones and local journey display times."""
|
||||
|
||||
import functools
|
||||
import importlib
|
||||
import json
|
||||
import os.path
|
||||
import typing
|
||||
from collections import defaultdict
|
||||
from datetime import date, datetime, timedelta, timezone
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import pytz
|
||||
from pycountry.db import Country
|
||||
|
||||
import agenda.trip
|
||||
from agenda.types import StrDict, Trip
|
||||
|
||||
|
||||
def _timezone_name_from_datetime(value: typing.Any) -> str | None:
|
||||
"""Get IANA timezone name from a datetime value if available."""
|
||||
if not isinstance(value, datetime) or value.tzinfo is None:
|
||||
return None
|
||||
|
||||
key = getattr(value.tzinfo, "key", None)
|
||||
if isinstance(key, str):
|
||||
return key
|
||||
|
||||
zone = getattr(value.tzinfo, "zone", None)
|
||||
if isinstance(zone, str):
|
||||
return zone
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _format_offset_from_uk(offset_minutes: int) -> str:
|
||||
"""Format offset from UK in +/-HH:MM."""
|
||||
if offset_minutes == 0:
|
||||
return "No difference"
|
||||
sign = "+" if offset_minutes > 0 else "-"
|
||||
hours, mins = divmod(abs(offset_minutes), 60)
|
||||
return f"{sign}{hours:02d}:{mins:02d} vs UK"
|
||||
|
||||
|
||||
def _trip_offset_minutes(
|
||||
trip_start: date, trip_end: date, destination_timezone: str
|
||||
) -> list[int]:
|
||||
"""Unique UTC offset differences vs UK across the trip date range."""
|
||||
destination_tz = ZoneInfo(destination_timezone)
|
||||
uk_timezone = ZoneInfo("Europe/London")
|
||||
current = trip_start
|
||||
offsets: set[int] = set()
|
||||
|
||||
while current <= trip_end:
|
||||
instant = datetime(
|
||||
current.year, current.month, current.day, 12, tzinfo=timezone.utc
|
||||
)
|
||||
destination_offset = instant.astimezone(destination_tz).utcoffset()
|
||||
uk_offset = instant.astimezone(uk_timezone).utcoffset()
|
||||
if destination_offset is not None and uk_offset is not None:
|
||||
offsets.add(int((destination_offset - uk_offset).total_seconds() // 60))
|
||||
current += timedelta(days=1)
|
||||
|
||||
return sorted(offsets)
|
||||
|
||||
|
||||
def _format_trip_offset_display(offsets: list[int]) -> str:
|
||||
"""Format trip-range offsets; include variation if DST changes during trip."""
|
||||
if not offsets:
|
||||
return "Timezone unknown"
|
||||
if len(offsets) == 1:
|
||||
return _format_offset_from_uk(offsets[0])
|
||||
return (
|
||||
"Varies during trip: "
|
||||
f"{_format_offset_from_uk(offsets[0])} to {_format_offset_from_uk(offsets[-1])}"
|
||||
)
|
||||
|
||||
|
||||
def _timezone_from_coordinates(latitude: float, longitude: float) -> str | None:
|
||||
"""Resolve IANA timezone name from coordinates."""
|
||||
timezone_finder = _get_timezone_finder()
|
||||
if timezone_finder is None:
|
||||
return None
|
||||
|
||||
for method_name in ("timezone_at", "certain_timezone_at", "closest_timezone_at"):
|
||||
finder_method = getattr(timezone_finder, method_name, None)
|
||||
if not callable(finder_method):
|
||||
continue
|
||||
tz_name = finder_method(lng=longitude, lat=latitude)
|
||||
if isinstance(tz_name, str):
|
||||
return tz_name
|
||||
return None
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=1)
|
||||
def _get_timezone_finder() -> typing.Any:
|
||||
"""Get timezone finder instance if dependency is available."""
|
||||
try:
|
||||
timezonefinder_module = importlib.import_module("timezonefinder")
|
||||
except ModuleNotFoundError:
|
||||
return None
|
||||
|
||||
timezone_finder_cls = getattr(timezonefinder_module, "TimezoneFinder", None)
|
||||
if timezone_finder_cls is None:
|
||||
return None
|
||||
|
||||
return timezone_finder_cls()
|
||||
|
||||
|
||||
def _coordinates_from_location(location: typing.Any) -> tuple[float, float] | None:
|
||||
"""Extract latitude/longitude from a location mapping."""
|
||||
if not isinstance(location, dict):
|
||||
return None
|
||||
|
||||
latitude = location.get("latitude")
|
||||
longitude = location.get("longitude")
|
||||
if not isinstance(latitude, (int, float)) or not isinstance(
|
||||
longitude, (int, float)
|
||||
):
|
||||
return None
|
||||
|
||||
return (float(latitude), float(longitude))
|
||||
|
||||
|
||||
def _route_endpoints_for_car_item(
|
||||
item: StrDict,
|
||||
data_dir: str,
|
||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None],
|
||||
) -> tuple[tuple[float, float], tuple[float, float]] | None:
|
||||
"""Return route endpoints for a car journey when available."""
|
||||
route_filename = item.get("geojson_filename")
|
||||
if not isinstance(route_filename, str):
|
||||
return None
|
||||
|
||||
if route_filename not in route_cache:
|
||||
geojson_text = agenda.trip.read_geojson(
|
||||
data_dir, os.path.join("car_routes", route_filename)
|
||||
)
|
||||
endpoints = agenda.trip.geojson_route_endpoints(json.loads(geojson_text))
|
||||
route_cache[route_filename] = endpoints
|
||||
|
||||
return route_cache[route_filename]
|
||||
|
||||
|
||||
def _timezone_name_for_car_item(
|
||||
item: StrDict,
|
||||
data_dir: str,
|
||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None],
|
||||
timezone_cache: dict[tuple[float, float], str | None],
|
||||
) -> str | None:
|
||||
"""Resolve the local timezone for a car journey from endpoints."""
|
||||
candidate_coords: list[tuple[float, float]] = []
|
||||
|
||||
for field in ("from_location", "to_location"):
|
||||
coord = _coordinates_from_location(item.get(field))
|
||||
if coord is not None:
|
||||
candidate_coords.append(coord)
|
||||
|
||||
if not candidate_coords:
|
||||
endpoints = _route_endpoints_for_car_item(item, data_dir, route_cache)
|
||||
if endpoints is not None:
|
||||
candidate_coords.extend(endpoints)
|
||||
|
||||
for coord in candidate_coords:
|
||||
if coord not in timezone_cache:
|
||||
timezone_cache[coord] = _timezone_from_coordinates(coord[0], coord[1])
|
||||
timezone_name = timezone_cache[coord]
|
||||
if timezone_name:
|
||||
return timezone_name
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def localize_trip_car_journey_display_times(trip: Trip, data_dir: str) -> None:
|
||||
"""Add route-local display timestamps for car journeys on the trip page."""
|
||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None] = {}
|
||||
timezone_cache: dict[tuple[float, float], str | None] = {}
|
||||
|
||||
for item in trip.travel:
|
||||
if item.get("type") != "car":
|
||||
continue
|
||||
|
||||
depart = item.get("depart")
|
||||
arrive = item.get("arrive")
|
||||
if not isinstance(depart, datetime) or not isinstance(arrive, datetime):
|
||||
continue
|
||||
|
||||
timezone_name = _timezone_name_for_car_item(
|
||||
item, data_dir, route_cache, timezone_cache
|
||||
)
|
||||
if not timezone_name:
|
||||
continue
|
||||
|
||||
item["display_depart"] = depart.astimezone(ZoneInfo(timezone_name))
|
||||
item["display_arrive"] = arrive.astimezone(ZoneInfo(timezone_name))
|
||||
|
||||
|
||||
def get_destination_timezones(trip: Trip) -> list[StrDict]:
|
||||
"""Build destination timezone metadata for the trip page."""
|
||||
per_location: dict[tuple[str, str], list[str]] = defaultdict(list)
|
||||
location_coords: dict[tuple[str, str], tuple[float, float]] = {}
|
||||
for item in trip.accommodation + trip.conferences + trip.events:
|
||||
location = item.get("location")
|
||||
country = item.get("country")
|
||||
if not isinstance(location, str) or not isinstance(country, str):
|
||||
continue
|
||||
|
||||
key = (location, country.lower())
|
||||
timezone_name = item.get("timezone")
|
||||
if isinstance(timezone_name, str):
|
||||
per_location[key].append(timezone_name)
|
||||
|
||||
latitude = item.get("latitude")
|
||||
longitude = item.get("longitude")
|
||||
if isinstance(latitude, (int, float)) and isinstance(longitude, (int, float)):
|
||||
location_coords[key] = (float(latitude), float(longitude))
|
||||
|
||||
for field in (
|
||||
"from",
|
||||
"to",
|
||||
"date",
|
||||
"start",
|
||||
"end",
|
||||
"attend_start",
|
||||
"attend_end",
|
||||
):
|
||||
candidate = _timezone_name_from_datetime(item.get(field))
|
||||
if candidate:
|
||||
per_location[key].append(candidate)
|
||||
|
||||
# Also collect airport locations from flights, for transit countries
|
||||
flight_locations: list[tuple[str, Country]] = []
|
||||
seen_flight_keys: set[tuple[str, str]] = set()
|
||||
for item in trip.travel:
|
||||
if item.get("type") != "flight":
|
||||
continue
|
||||
for airport_key in ("from_airport", "to_airport"):
|
||||
airport = item.get(airport_key)
|
||||
if not isinstance(airport, dict):
|
||||
continue
|
||||
city = airport.get("city")
|
||||
country_code = airport.get("country")
|
||||
if not isinstance(city, str) or not isinstance(country_code, str):
|
||||
continue
|
||||
if country_code == "gb":
|
||||
continue
|
||||
key = (city, country_code.lower())
|
||||
lat = airport.get("latitude")
|
||||
lon = airport.get("longitude")
|
||||
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
||||
location_coords.setdefault(key, (float(lat), float(lon)))
|
||||
if key not in seen_flight_keys:
|
||||
seen_flight_keys.add(key)
|
||||
flight_country = agenda.get_country(country_code)
|
||||
if flight_country:
|
||||
flight_locations.append((city, flight_country))
|
||||
|
||||
existing_location_keys = {(loc, c.alpha_2.lower()) for loc, c in trip.locations()}
|
||||
all_locations = list(trip.locations()) + [
|
||||
(city, country)
|
||||
for city, country in flight_locations
|
||||
if (city, country.alpha_2.lower()) not in existing_location_keys
|
||||
]
|
||||
|
||||
destination_times: list[StrDict] = []
|
||||
trip_end = trip.end or trip.start
|
||||
|
||||
for location, country in all_locations:
|
||||
country_code = country.alpha_2.lower()
|
||||
key = (location, country_code)
|
||||
timezone_name = None
|
||||
|
||||
for candidate in per_location.get(key, []):
|
||||
try:
|
||||
ZoneInfo(candidate)
|
||||
timezone_name = candidate
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not timezone_name and key in location_coords:
|
||||
latitude, longitude = location_coords[key]
|
||||
coordinate_timezone = _timezone_from_coordinates(latitude, longitude)
|
||||
if coordinate_timezone:
|
||||
timezone_name = coordinate_timezone
|
||||
|
||||
if not timezone_name:
|
||||
country_timezones = pytz.country_timezones.get(country_code, [])
|
||||
if len(country_timezones) == 1:
|
||||
timezone_name = country_timezones[0]
|
||||
|
||||
offset_display = "Timezone unknown"
|
||||
if timezone_name:
|
||||
offset_display = _format_trip_offset_display(
|
||||
_trip_offset_minutes(trip.start, trip_end, timezone_name)
|
||||
)
|
||||
|
||||
destination_times.append(
|
||||
{
|
||||
"location": location,
|
||||
"country_name": country.name,
|
||||
"country_flag": country.flag,
|
||||
"timezone": timezone_name,
|
||||
"offset_display": offset_display,
|
||||
}
|
||||
)
|
||||
|
||||
grouped: list[StrDict] = []
|
||||
grouped_index: dict[tuple[str, str, str | None], int] = {}
|
||||
for item in destination_times:
|
||||
group_key = (item["country_name"], item["country_flag"], item["timezone"])
|
||||
if group_key in grouped_index:
|
||||
existing = grouped[grouped_index[group_key]]
|
||||
existing_locations = typing.cast(list[str], existing["locations"])
|
||||
existing_locations.append(typing.cast(str, item["location"]))
|
||||
existing["location_count"] = (
|
||||
typing.cast(int, existing["location_count"]) + 1
|
||||
)
|
||||
continue
|
||||
|
||||
grouped_index[group_key] = len(grouped)
|
||||
grouped.append(
|
||||
{
|
||||
**item,
|
||||
"locations": [item["location"]],
|
||||
"location_count": 1,
|
||||
}
|
||||
)
|
||||
|
||||
for item in grouped:
|
||||
location_count = typing.cast(int, item["location_count"])
|
||||
country_name = typing.cast(str, item["country_name"])
|
||||
country_flag = typing.cast(str, item["country_flag"])
|
||||
if location_count > 1:
|
||||
label = f"{country_name} ({location_count} locations)"
|
||||
else:
|
||||
label = f"{item['location']} ({country_name})"
|
||||
if trip.show_flags:
|
||||
label = f"{label} {country_flag}"
|
||||
item["destination_label"] = label
|
||||
|
||||
return grouped
|
||||
|
|
@ -6,13 +6,14 @@ from types import SimpleNamespace
|
|||
|
||||
import yaml
|
||||
|
||||
import agenda.conference_list
|
||||
import agenda.fx
|
||||
import agenda.trip
|
||||
import web_view
|
||||
|
||||
|
||||
def test_build_conference_list_supports_inexact_dates(tmp_path: typing.Any) -> None:
|
||||
def test_build_conference_list_supports_inexact_dates(
|
||||
tmp_path: typing.Any, monkeypatch: typing.Any
|
||||
) -> None:
|
||||
"""Conference list should include tentative and approximate dates."""
|
||||
conferences = [
|
||||
{
|
||||
|
|
@ -56,7 +57,10 @@ def test_build_conference_list_supports_inexact_dates(tmp_path: typing.Any) -> N
|
|||
encoding="utf-8",
|
||||
)
|
||||
|
||||
items = agenda.conference_list.build_conference_list(str(tmp_path), [])
|
||||
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
||||
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
||||
|
||||
items = web_view.build_conference_list()
|
||||
|
||||
assert [item["name"] for item in items] == ["FOSDEM 2027", "PyCascades 2027"]
|
||||
assert items[0]["date_status"] == "tentative"
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ import flask
|
|||
|
||||
import agenda.trip
|
||||
import agenda.trip_schengen
|
||||
import agenda.trip_timezones
|
||||
import agenda.weather
|
||||
import web_view
|
||||
from agenda.types import Trip
|
||||
|
|
@ -347,7 +346,7 @@ def test_trip_page_renders_car_times_in_local_route_timezone() -> None:
|
|||
mock.patch.object(web_view, "get_trip_list", return_value=[trip]),
|
||||
mock.patch.object(agenda.weather, "get_trip_weather", return_value=[]),
|
||||
mock.patch.object(
|
||||
agenda.trip_timezones,
|
||||
web_view,
|
||||
"_timezone_from_coordinates",
|
||||
return_value="America/New_York",
|
||||
),
|
||||
|
|
|
|||
775
web_view.py
775
web_view.py
|
|
@ -2,25 +2,32 @@
|
|||
|
||||
"""Web page to show upcoming events."""
|
||||
|
||||
import decimal
|
||||
import functools
|
||||
import importlib
|
||||
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
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import flask
|
||||
import pytz
|
||||
from pycountry.db import Country
|
||||
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
|
||||
import agenda.data
|
||||
import agenda.error_mail
|
||||
import agenda.fx
|
||||
|
|
@ -29,13 +36,12 @@ 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__)
|
||||
|
|
@ -364,20 +370,211 @@ def travel_list() -> str:
|
|||
)
|
||||
|
||||
|
||||
def build_conference_list() -> list[StrDict]:
|
||||
"""Build conference list."""
|
||||
data_dir = app.config["PERSONAL_DATA"]
|
||||
filepath = os.path.join(data_dir, "conferences.yaml")
|
||||
items: list[StrDict] = yaml.safe_load(open(filepath))
|
||||
series_lookup = agenda.conference.load_series(data_dir)
|
||||
conference_trip_lookup = {}
|
||||
|
||||
for trip in agenda.trip.build_trip_list():
|
||||
for trip_conf in trip.conferences:
|
||||
key = (trip_conf["start"], trip_conf["name"])
|
||||
conference_trip_lookup[key] = trip
|
||||
|
||||
for conf in items:
|
||||
conf.update(agenda.conference.validate_conference_date_fields(conf))
|
||||
|
||||
price = conf.get("price")
|
||||
if price:
|
||||
conf["price"] = decimal.Decimal(price)
|
||||
|
||||
series_id = conf.get("series")
|
||||
if isinstance(series_id, str):
|
||||
conf["series_detail"] = series_lookup.get(series_id)
|
||||
|
||||
if "start" in conf:
|
||||
key = (conf["start"], conf["name"])
|
||||
if this_trip := conference_trip_lookup.get(key):
|
||||
conf["linked_trip"] = this_trip
|
||||
|
||||
items.sort(key=lambda item: item["sort_date"])
|
||||
return items
|
||||
|
||||
|
||||
def conference_country_code(conf: StrDict) -> str | None:
|
||||
"""Return normalized alpha-2 country code for a conference."""
|
||||
country = conf.get("country")
|
||||
if not isinstance(country, str):
|
||||
return None
|
||||
|
||||
code = country.strip().lower()
|
||||
return code if len(code) == 2 else None
|
||||
|
||||
|
||||
def normalize_country_filter(value: str | None) -> str | None:
|
||||
"""Normalize and validate a country query parameter."""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
code = value.strip().lower()
|
||||
if len(code) != 2:
|
||||
return None
|
||||
|
||||
return code if agenda.get_country(code) else None
|
||||
|
||||
|
||||
def filter_conferences_by_country(
|
||||
items: list[StrDict], country_code: str | None
|
||||
) -> list[StrDict]:
|
||||
"""Filter conferences by country when a country code is provided."""
|
||||
if not country_code:
|
||||
return items
|
||||
|
||||
return [item for item in items if conference_country_code(item) == country_code]
|
||||
|
||||
|
||||
def conference_country_options(items: list[StrDict]) -> list[StrDict]:
|
||||
"""Return country options for the conference country filter."""
|
||||
counts: defaultdict[str, int] = defaultdict(int)
|
||||
for item in items:
|
||||
code = conference_country_code(item)
|
||||
if code and agenda.get_country(code):
|
||||
counts[code] += 1
|
||||
|
||||
options: list[StrDict] = []
|
||||
for code, count in counts.items():
|
||||
country = agenda.get_country(code)
|
||||
if not country:
|
||||
continue
|
||||
options.append(
|
||||
{
|
||||
"code": code,
|
||||
"name": country.name,
|
||||
"flag": country.flag,
|
||||
"count": count,
|
||||
}
|
||||
)
|
||||
|
||||
options.sort(key=lambda item: str(item["name"]))
|
||||
return options
|
||||
|
||||
|
||||
def build_conference_series_list() -> list[StrDict]:
|
||||
"""Build conference series list with conference counts."""
|
||||
data_dir = app.config["PERSONAL_DATA"]
|
||||
series_lookup = agenda.conference.load_series(data_dir)
|
||||
conferences = build_conference_list()
|
||||
|
||||
series_items: list[StrDict] = []
|
||||
for series_id, series in series_lookup.items():
|
||||
linked = [conf for conf in conferences if conf.get("series") == series_id]
|
||||
latest = max((conf["sort_date"] for conf in linked), default=None)
|
||||
next_conf = next(
|
||||
(conf for conf in linked if conf["latest_date"] >= date.today()), None
|
||||
)
|
||||
attended = any(conf.get("going") or conf.get("linked_trip") for conf in linked)
|
||||
item: StrDict = {
|
||||
"id": series_id,
|
||||
**series,
|
||||
"count": len(linked),
|
||||
"latest": latest,
|
||||
"next_conf": next_conf,
|
||||
"attended": attended,
|
||||
}
|
||||
series_items.append(item)
|
||||
|
||||
series_items.sort(key=lambda item: str(item["name"]).lower())
|
||||
return series_items
|
||||
|
||||
|
||||
def build_conference_timeline(
|
||||
current: list[StrDict], future: list[StrDict], today: date, days: int = 90
|
||||
) -> dict[str, typing.Any] | None:
|
||||
"""Build data for a Gantt-style timeline of upcoming conferences."""
|
||||
timeline_start = today
|
||||
timeline_end = today + timedelta(days=days)
|
||||
|
||||
visible = [
|
||||
c
|
||||
for c in (current + future)
|
||||
if c["has_exact_dates"]
|
||||
and c["start_date"] <= timeline_end
|
||||
and c["end_date"] >= today
|
||||
]
|
||||
if not visible:
|
||||
return None
|
||||
|
||||
visible.sort(key=lambda c: c["start_date"])
|
||||
|
||||
# Greedy interval-coloring: assign each conference a lane (row)
|
||||
lane_ends: list[date] = []
|
||||
conf_data = []
|
||||
for conf in visible:
|
||||
lane = next(
|
||||
(i for i, end in enumerate(lane_ends) if end < conf["start_date"]),
|
||||
len(lane_ends),
|
||||
)
|
||||
if lane == len(lane_ends):
|
||||
lane_ends.append(conf["end_date"])
|
||||
else:
|
||||
lane_ends[lane] = conf["end_date"]
|
||||
|
||||
start_off = max((conf["start_date"] - timeline_start).days, 0)
|
||||
end_off = min((conf["end_date"] - timeline_start).days + 1, days)
|
||||
left_pct = round(start_off / days * 100, 2)
|
||||
width_pct = max(round((end_off - start_off) / days * 100, 2), 0.5)
|
||||
|
||||
conf_data.append(
|
||||
{
|
||||
"name": conf["name"],
|
||||
"url": conf.get("url"),
|
||||
"lane": lane,
|
||||
"left_pct": left_pct,
|
||||
"width_pct": width_pct,
|
||||
"key": f"{conf['start_date'].isoformat()}|{conf['name']}",
|
||||
"label": (
|
||||
f"{conf['name']} ({conf['start_date'].strftime('%-d %b')})"
|
||||
if conf["start_date"] == conf["end_date"]
|
||||
else (
|
||||
f"{conf['name']}"
|
||||
f" ({conf['start_date'].strftime('%-d %b')}–"
|
||||
f"{conf['end_date'].strftime('%-d %b')})"
|
||||
)
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
# Month markers for x-axis labels
|
||||
months = []
|
||||
d = today.replace(day=1)
|
||||
while d <= timeline_end:
|
||||
off = max((d - timeline_start).days, 0)
|
||||
months.append(
|
||||
{"label": d.strftime("%b %Y"), "left_pct": round(off / days * 100, 2)}
|
||||
)
|
||||
# advance to next month
|
||||
d = (d.replace(day=28) + timedelta(days=4)).replace(day=1)
|
||||
|
||||
return {
|
||||
"confs": conf_data,
|
||||
"lane_count": len(lane_ends),
|
||||
"months": months,
|
||||
"days": days,
|
||||
}
|
||||
|
||||
|
||||
@app.route("/conference")
|
||||
def conference_list() -> str:
|
||||
"""Page showing a list of conferences."""
|
||||
today = date.today()
|
||||
items = agenda.conference_list.build_conference_list(
|
||||
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
||||
)
|
||||
country_filter = agenda.conference_list.normalize_country_filter(
|
||||
flask.request.args.get("country")
|
||||
)
|
||||
country_options = agenda.conference_list.conference_country_options(
|
||||
items = build_conference_list()
|
||||
country_filter = normalize_country_filter(flask.request.args.get("country"))
|
||||
country_options = conference_country_options(
|
||||
[conf for conf in items if conf["latest_date"] >= today]
|
||||
)
|
||||
items = agenda.conference_list.filter_conferences_by_country(items, country_filter)
|
||||
items = filter_conferences_by_country(items, country_filter)
|
||||
|
||||
current = [
|
||||
conf
|
||||
|
|
@ -390,7 +587,7 @@ def conference_list() -> str:
|
|||
conf for conf in items if conf not in current and conf["latest_date"] >= today
|
||||
]
|
||||
|
||||
timeline = agenda.conference_list.build_conference_timeline(current, future, today)
|
||||
timeline = build_conference_timeline(current, future, today)
|
||||
|
||||
return flask.render_template(
|
||||
"conference_list.html",
|
||||
|
|
@ -409,16 +606,12 @@ def conference_list() -> str:
|
|||
def past_conference_list() -> str:
|
||||
"""Page showing a list of conferences."""
|
||||
today = date.today()
|
||||
items = agenda.conference_list.build_conference_list(
|
||||
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
||||
)
|
||||
country_filter = agenda.conference_list.normalize_country_filter(
|
||||
flask.request.args.get("country")
|
||||
)
|
||||
country_options = agenda.conference_list.conference_country_options(
|
||||
items = build_conference_list()
|
||||
country_filter = normalize_country_filter(flask.request.args.get("country"))
|
||||
country_options = conference_country_options(
|
||||
[conf for conf in items if conf["latest_date"] < today]
|
||||
)
|
||||
items = agenda.conference_list.filter_conferences_by_country(items, country_filter)
|
||||
items = filter_conferences_by_country(items, country_filter)
|
||||
return flask.render_template(
|
||||
"conference_list.html",
|
||||
past=[conf for conf in items if conf["latest_date"] < today],
|
||||
|
|
@ -435,13 +628,7 @@ def conference_series_list() -> str:
|
|||
"""Page showing conference series."""
|
||||
return flask.render_template(
|
||||
"conference_series_list.html",
|
||||
series_list=agenda.conference_list.build_conference_series_list(
|
||||
agenda.conference.load_series(app.config["PERSONAL_DATA"]),
|
||||
agenda.conference_list.build_conference_list(
|
||||
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
||||
),
|
||||
date.today(),
|
||||
),
|
||||
series_list=build_conference_series_list(),
|
||||
get_country=agenda.get_country,
|
||||
)
|
||||
|
||||
|
|
@ -455,11 +642,7 @@ def conference_series_page(series_id: str) -> str:
|
|||
flask.abort(404)
|
||||
|
||||
conferences = [
|
||||
conf
|
||||
for conf in agenda.conference_list.build_conference_list(
|
||||
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
||||
)
|
||||
if conf.get("series") == series_id
|
||||
conf for conf in build_conference_list() if conf.get("series") == series_id
|
||||
]
|
||||
return flask.render_template(
|
||||
"conference_series.html",
|
||||
|
|
@ -475,9 +658,7 @@ def conference_series_page(series_id: str) -> str:
|
|||
@app.route("/conference/ical")
|
||||
def conference_ical() -> werkzeug.Response:
|
||||
"""Return all conferences as an iCalendar feed."""
|
||||
items = agenda.conference_list.build_conference_list(
|
||||
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
||||
)
|
||||
items = build_conference_list()
|
||||
ical_data = agenda.conference_ical.build_conference_ical(items)
|
||||
response = flask.Response(ical_data, mimetype="text/calendar")
|
||||
response.headers["Content-Disposition"] = "inline; filename=conferences.ics"
|
||||
|
|
@ -490,13 +671,51 @@ def accommodation_list() -> str:
|
|||
data_dir = app.config["PERSONAL_DATA"]
|
||||
items = travel.parse_yaml("accommodation", data_dir)
|
||||
|
||||
context = agenda.accommodation.prepare_accommodation_list(
|
||||
items, agenda.trip.build_trip_list(), uk_tz.localize(datetime.now())
|
||||
# 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 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",
|
||||
**context,
|
||||
past=past,
|
||||
current=current,
|
||||
future=future,
|
||||
year_stats=sorted_year_stats,
|
||||
get_country=agenda.get_country,
|
||||
fx_rate=agenda.fx.get_rates(app.config),
|
||||
)
|
||||
|
|
@ -518,6 +737,41 @@ 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."""
|
||||
|
||||
|
|
@ -539,6 +793,33 @@ 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."""
|
||||
|
|
@ -550,19 +831,17 @@ def trip_past_list() -> str:
|
|||
"trip/list.html",
|
||||
heading="Past trips",
|
||||
trips=reversed(past),
|
||||
trip_school_holiday_map=agenda.holidays.trip_school_holiday_map(
|
||||
past, app.config["DATA_DIR"]
|
||||
),
|
||||
trip_school_holiday_map=trip_school_holiday_map(past),
|
||||
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=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),
|
||||
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),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -614,9 +893,7 @@ def trip_future_list() -> str:
|
|||
"trip/list.html",
|
||||
heading="Future trips",
|
||||
trips=shown,
|
||||
trip_school_holiday_map=agenda.holidays.trip_school_holiday_map(
|
||||
shown, app.config["DATA_DIR"]
|
||||
),
|
||||
trip_school_holiday_map=trip_school_holiday_map(shown),
|
||||
trip_weather_map=trip_weather_map,
|
||||
coordinates=coordinates,
|
||||
routes=routes,
|
||||
|
|
@ -624,12 +901,10 @@ 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=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),
|
||||
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),
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -666,6 +941,331 @@ def get_prev_current_and_next_trip(
|
|||
return (prev_trip, current_trip, next_trip)
|
||||
|
||||
|
||||
def _timezone_name_from_datetime(value: typing.Any) -> str | None:
|
||||
"""Get IANA timezone name from a datetime value if available."""
|
||||
if not isinstance(value, datetime) or value.tzinfo is None:
|
||||
return None
|
||||
|
||||
key = getattr(value.tzinfo, "key", None)
|
||||
if isinstance(key, str):
|
||||
return key
|
||||
|
||||
zone = getattr(value.tzinfo, "zone", None)
|
||||
if isinstance(zone, str):
|
||||
return zone
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _format_offset_from_uk(offset_minutes: int) -> str:
|
||||
"""Format offset from UK in +/-HH:MM."""
|
||||
if offset_minutes == 0:
|
||||
return "No difference"
|
||||
sign = "+" if offset_minutes > 0 else "-"
|
||||
hours, mins = divmod(abs(offset_minutes), 60)
|
||||
return f"{sign}{hours:02d}:{mins:02d} vs UK"
|
||||
|
||||
|
||||
def _trip_offset_minutes(
|
||||
trip_start: date, trip_end: date, destination_timezone: str
|
||||
) -> list[int]:
|
||||
"""Unique UTC offset differences vs UK across the trip date range."""
|
||||
destination_tz = ZoneInfo(destination_timezone)
|
||||
uk_timezone = ZoneInfo("Europe/London")
|
||||
current = trip_start
|
||||
offsets: set[int] = set()
|
||||
|
||||
while current <= trip_end:
|
||||
instant = datetime(
|
||||
current.year, current.month, current.day, 12, tzinfo=timezone.utc
|
||||
)
|
||||
destination_offset = instant.astimezone(destination_tz).utcoffset()
|
||||
uk_offset = instant.astimezone(uk_timezone).utcoffset()
|
||||
if destination_offset is not None and uk_offset is not None:
|
||||
offsets.add(int((destination_offset - uk_offset).total_seconds() // 60))
|
||||
current += timedelta(days=1)
|
||||
|
||||
return sorted(offsets)
|
||||
|
||||
|
||||
def _format_trip_offset_display(offsets: list[int]) -> str:
|
||||
"""Format trip-range offsets; include variation if DST changes during trip."""
|
||||
if not offsets:
|
||||
return "Timezone unknown"
|
||||
if len(offsets) == 1:
|
||||
return _format_offset_from_uk(offsets[0])
|
||||
return (
|
||||
"Varies during trip: "
|
||||
f"{_format_offset_from_uk(offsets[0])} to {_format_offset_from_uk(offsets[-1])}"
|
||||
)
|
||||
|
||||
|
||||
def _timezone_from_coordinates(latitude: float, longitude: float) -> str | None:
|
||||
"""Resolve IANA timezone name from coordinates."""
|
||||
timezone_finder = _get_timezone_finder()
|
||||
if timezone_finder is None:
|
||||
return None
|
||||
|
||||
for method_name in ("timezone_at", "certain_timezone_at", "closest_timezone_at"):
|
||||
finder_method = getattr(timezone_finder, method_name, None)
|
||||
if not callable(finder_method):
|
||||
continue
|
||||
tz_name = finder_method(lng=longitude, lat=latitude)
|
||||
if isinstance(tz_name, str):
|
||||
return tz_name
|
||||
return None
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=1)
|
||||
def _get_timezone_finder() -> typing.Any:
|
||||
"""Get timezone finder instance if dependency is available."""
|
||||
try:
|
||||
timezonefinder_module = importlib.import_module("timezonefinder")
|
||||
except ModuleNotFoundError:
|
||||
return None
|
||||
|
||||
timezone_finder_cls = getattr(timezonefinder_module, "TimezoneFinder", None)
|
||||
if timezone_finder_cls is None:
|
||||
return None
|
||||
|
||||
return timezone_finder_cls()
|
||||
|
||||
|
||||
def _coordinates_from_location(location: typing.Any) -> tuple[float, float] | None:
|
||||
"""Extract latitude/longitude from a location mapping."""
|
||||
if not isinstance(location, dict):
|
||||
return None
|
||||
|
||||
latitude = location.get("latitude")
|
||||
longitude = location.get("longitude")
|
||||
if not isinstance(latitude, (int, float)) or not isinstance(
|
||||
longitude, (int, float)
|
||||
):
|
||||
return None
|
||||
|
||||
return (float(latitude), float(longitude))
|
||||
|
||||
|
||||
def _route_endpoints_for_car_item(
|
||||
item: StrDict,
|
||||
data_dir: str,
|
||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None],
|
||||
) -> tuple[tuple[float, float], tuple[float, float]] | None:
|
||||
"""Return route endpoints for a car journey when available."""
|
||||
route_filename = item.get("geojson_filename")
|
||||
if not isinstance(route_filename, str):
|
||||
return None
|
||||
|
||||
if route_filename not in route_cache:
|
||||
geojson_text = agenda.trip.read_geojson(
|
||||
data_dir, os.path.join("car_routes", route_filename)
|
||||
)
|
||||
endpoints = agenda.trip.geojson_route_endpoints(json.loads(geojson_text))
|
||||
route_cache[route_filename] = endpoints
|
||||
|
||||
return route_cache[route_filename]
|
||||
|
||||
|
||||
def _timezone_name_for_car_item(
|
||||
item: StrDict,
|
||||
data_dir: str,
|
||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None],
|
||||
timezone_cache: dict[tuple[float, float], str | None],
|
||||
) -> str | None:
|
||||
"""Resolve the local timezone for a car journey from endpoints."""
|
||||
candidate_coords: list[tuple[float, float]] = []
|
||||
|
||||
for field in ("from_location", "to_location"):
|
||||
coord = _coordinates_from_location(item.get(field))
|
||||
if coord is not None:
|
||||
candidate_coords.append(coord)
|
||||
|
||||
if not candidate_coords:
|
||||
endpoints = _route_endpoints_for_car_item(item, data_dir, route_cache)
|
||||
if endpoints is not None:
|
||||
candidate_coords.extend(endpoints)
|
||||
|
||||
for coord in candidate_coords:
|
||||
if coord not in timezone_cache:
|
||||
timezone_cache[coord] = _timezone_from_coordinates(coord[0], coord[1])
|
||||
timezone_name = timezone_cache[coord]
|
||||
if timezone_name:
|
||||
return timezone_name
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def localize_trip_car_journey_display_times(trip: Trip, data_dir: str) -> None:
|
||||
"""Add route-local display timestamps for car journeys on the trip page."""
|
||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None] = {}
|
||||
timezone_cache: dict[tuple[float, float], str | None] = {}
|
||||
|
||||
for item in trip.travel:
|
||||
if item.get("type") != "car":
|
||||
continue
|
||||
|
||||
depart = item.get("depart")
|
||||
arrive = item.get("arrive")
|
||||
if not isinstance(depart, datetime) or not isinstance(arrive, datetime):
|
||||
continue
|
||||
|
||||
timezone_name = _timezone_name_for_car_item(
|
||||
item, data_dir, route_cache, timezone_cache
|
||||
)
|
||||
if not timezone_name:
|
||||
continue
|
||||
|
||||
item["display_depart"] = depart.astimezone(ZoneInfo(timezone_name))
|
||||
item["display_arrive"] = arrive.astimezone(ZoneInfo(timezone_name))
|
||||
|
||||
|
||||
def get_destination_timezones(trip: Trip) -> list[StrDict]:
|
||||
"""Build destination timezone metadata for the trip page."""
|
||||
per_location: dict[tuple[str, str], list[str]] = defaultdict(list)
|
||||
location_coords: dict[tuple[str, str], tuple[float, float]] = {}
|
||||
for item in trip.accommodation + trip.conferences + trip.events:
|
||||
location = item.get("location")
|
||||
country = item.get("country")
|
||||
if not isinstance(location, str) or not isinstance(country, str):
|
||||
continue
|
||||
|
||||
key = (location, country.lower())
|
||||
timezone_name = item.get("timezone")
|
||||
if isinstance(timezone_name, str):
|
||||
per_location[key].append(timezone_name)
|
||||
|
||||
latitude = item.get("latitude")
|
||||
longitude = item.get("longitude")
|
||||
if isinstance(latitude, (int, float)) and isinstance(longitude, (int, float)):
|
||||
location_coords[key] = (float(latitude), float(longitude))
|
||||
|
||||
for field in (
|
||||
"from",
|
||||
"to",
|
||||
"date",
|
||||
"start",
|
||||
"end",
|
||||
"attend_start",
|
||||
"attend_end",
|
||||
):
|
||||
candidate = _timezone_name_from_datetime(item.get(field))
|
||||
if candidate:
|
||||
per_location[key].append(candidate)
|
||||
|
||||
# Also collect airport locations from flights, for transit countries
|
||||
flight_locations: list[tuple[str, Country]] = []
|
||||
seen_flight_keys: set[tuple[str, str]] = set()
|
||||
for item in trip.travel:
|
||||
if item.get("type") != "flight":
|
||||
continue
|
||||
for airport_key in ("from_airport", "to_airport"):
|
||||
airport = item.get(airport_key)
|
||||
if not isinstance(airport, dict):
|
||||
continue
|
||||
city = airport.get("city")
|
||||
country_code = airport.get("country")
|
||||
if not isinstance(city, str) or not isinstance(country_code, str):
|
||||
continue
|
||||
if country_code == "gb":
|
||||
continue
|
||||
key = (city, country_code.lower())
|
||||
lat = airport.get("latitude")
|
||||
lon = airport.get("longitude")
|
||||
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
||||
location_coords.setdefault(key, (float(lat), float(lon)))
|
||||
if key not in seen_flight_keys:
|
||||
seen_flight_keys.add(key)
|
||||
flight_country = agenda.get_country(country_code)
|
||||
if flight_country:
|
||||
flight_locations.append((city, flight_country))
|
||||
|
||||
existing_location_keys = {(loc, c.alpha_2.lower()) for loc, c in trip.locations()}
|
||||
all_locations = list(trip.locations()) + [
|
||||
(city, country)
|
||||
for city, country in flight_locations
|
||||
if (city, country.alpha_2.lower()) not in existing_location_keys
|
||||
]
|
||||
|
||||
destination_times: list[StrDict] = []
|
||||
trip_end = trip.end or trip.start
|
||||
|
||||
for location, country in all_locations:
|
||||
country_code = country.alpha_2.lower()
|
||||
key = (location, country_code)
|
||||
timezone_name = None
|
||||
|
||||
for candidate in per_location.get(key, []):
|
||||
try:
|
||||
ZoneInfo(candidate)
|
||||
timezone_name = candidate
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not timezone_name and key in location_coords:
|
||||
latitude, longitude = location_coords[key]
|
||||
coordinate_timezone = _timezone_from_coordinates(latitude, longitude)
|
||||
if coordinate_timezone:
|
||||
timezone_name = coordinate_timezone
|
||||
|
||||
if not timezone_name:
|
||||
country_timezones = pytz.country_timezones.get(country_code, [])
|
||||
if len(country_timezones) == 1:
|
||||
timezone_name = country_timezones[0]
|
||||
|
||||
offset_display = "Timezone unknown"
|
||||
if timezone_name:
|
||||
offset_display = _format_trip_offset_display(
|
||||
_trip_offset_minutes(trip.start, trip_end, timezone_name)
|
||||
)
|
||||
|
||||
destination_times.append(
|
||||
{
|
||||
"location": location,
|
||||
"country_name": country.name,
|
||||
"country_flag": country.flag,
|
||||
"timezone": timezone_name,
|
||||
"offset_display": offset_display,
|
||||
}
|
||||
)
|
||||
|
||||
grouped: list[StrDict] = []
|
||||
grouped_index: dict[tuple[str, str, str | None], int] = {}
|
||||
for item in destination_times:
|
||||
group_key = (item["country_name"], item["country_flag"], item["timezone"])
|
||||
if group_key in grouped_index:
|
||||
existing = grouped[grouped_index[group_key]]
|
||||
existing_locations = typing.cast(list[str], existing["locations"])
|
||||
existing_locations.append(typing.cast(str, item["location"]))
|
||||
existing["location_count"] = (
|
||||
typing.cast(int, existing["location_count"]) + 1
|
||||
)
|
||||
continue
|
||||
|
||||
grouped_index[group_key] = len(grouped)
|
||||
grouped.append(
|
||||
{
|
||||
**item,
|
||||
"locations": [item["location"]],
|
||||
"location_count": 1,
|
||||
}
|
||||
)
|
||||
|
||||
for item in grouped:
|
||||
location_count = typing.cast(int, item["location_count"])
|
||||
country_name = typing.cast(str, item["country_name"])
|
||||
country_flag = typing.cast(str, item["country_flag"])
|
||||
if location_count > 1:
|
||||
label = f"{country_name} ({location_count} locations)"
|
||||
else:
|
||||
label = f"{item['location']} ({country_name})"
|
||||
if trip.show_flags:
|
||||
label = f"{label} {country_flag}"
|
||||
item["destination_label"] = label
|
||||
|
||||
return grouped
|
||||
|
||||
|
||||
@app.route("/trip/<start>")
|
||||
def trip_page(start: str) -> str:
|
||||
"""Individual trip page."""
|
||||
|
|
@ -700,9 +1300,7 @@ def trip_page(start: str) -> str:
|
|||
trip,
|
||||
cache_only=True,
|
||||
)
|
||||
agenda.trip_timezones.localize_trip_car_journey_display_times(
|
||||
trip, app.config["PERSONAL_DATA"]
|
||||
)
|
||||
localize_trip_car_journey_display_times(trip, app.config["PERSONAL_DATA"])
|
||||
|
||||
return flask.render_template(
|
||||
"trip_page.html",
|
||||
|
|
@ -716,7 +1314,7 @@ def trip_page(start: str) -> str:
|
|||
format_list_with_ampersand=format_list_with_ampersand,
|
||||
holidays=agenda.holidays.get_trip_holidays(trip),
|
||||
school_holidays=agenda.holidays.get_trip_school_holidays(trip),
|
||||
destination_times=agenda.trip_timezones.get_destination_timezones(trip),
|
||||
destination_times=get_destination_timezones(trip),
|
||||
human_readable_delta=agenda.utils.human_readable_delta,
|
||||
trip_weather=trip_weather,
|
||||
conference_free_days=agenda.trip.conference_free_days(trip),
|
||||
|
|
@ -739,7 +1337,54 @@ def trip_debug_page(start: str) -> str:
|
|||
# Add Schengen compliance information
|
||||
trip = agenda.trip_schengen.add_schengen_compliance_to_trip(trip)
|
||||
|
||||
trip_json, trip_yaml = agenda.trip_debug.serialize_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)
|
||||
|
||||
return flask.render_template(
|
||||
"trip_debug.html",
|
||||
|
|
@ -803,12 +1448,10 @@ def trip_stats() -> str:
|
|||
return flask.render_template(
|
||||
"trip/stats.html",
|
||||
count=len(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),
|
||||
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),
|
||||
yearly_stats=yearly_stats,
|
||||
overall_stats=overall_stats,
|
||||
conferences=conferences,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue