diff --git a/agenda/conference_list.py b/agenda/conference_list.py new file mode 100644 index 0000000..fb4164f --- /dev/null +++ b/agenda/conference_list.py @@ -0,0 +1,207 @@ +"""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, + } diff --git a/agenda/trip_timezones.py b/agenda/trip_timezones.py new file mode 100644 index 0000000..e8b86b0 --- /dev/null +++ b/agenda/trip_timezones.py @@ -0,0 +1,341 @@ +"""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 diff --git a/tests/test_conference_list.py b/tests/test_conference_list.py index d1ccc87..6de3ee6 100644 --- a/tests/test_conference_list.py +++ b/tests/test_conference_list.py @@ -6,14 +6,13 @@ 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, monkeypatch: typing.Any -) -> None: +def test_build_conference_list_supports_inexact_dates(tmp_path: typing.Any) -> None: """Conference list should include tentative and approximate dates.""" conferences = [ { @@ -57,10 +56,7 @@ def test_build_conference_list_supports_inexact_dates( encoding="utf-8", ) - 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() + items = agenda.conference_list.build_conference_list(str(tmp_path), []) assert [item["name"] for item in items] == ["FOSDEM 2027", "PyCascades 2027"] assert items[0]["date_status"] == "tentative" diff --git a/tests/test_trip_page_route.py b/tests/test_trip_page_route.py index 6651bd9..9795969 100644 --- a/tests/test_trip_page_route.py +++ b/tests/test_trip_page_route.py @@ -8,6 +8,7 @@ import flask import agenda.trip import agenda.trip_schengen +import agenda.trip_timezones import agenda.weather import web_view from agenda.types import Trip @@ -346,7 +347,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( - web_view, + agenda.trip_timezones, "_timezone_from_coordinates", return_value="America/New_York", ), diff --git a/web_view.py b/web_view.py index a0f3968..3107763 100755 --- a/web_view.py +++ b/web_view.py @@ -2,9 +2,6 @@ """Web page to show upcoming events.""" -import decimal -import functools -import importlib import inspect import json import os.path @@ -15,11 +12,8 @@ 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 @@ -28,6 +22,7 @@ from werkzeug.middleware.proxy_fix import ProxyFix import agenda.conference import agenda.conference_ical +import agenda.conference_list import agenda.data import agenda.error_mail import agenda.fx @@ -37,6 +32,7 @@ import agenda.stats import agenda.thespacedevs import agenda.trip import agenda.trip_schengen +import agenda.trip_timezones import agenda.uk_school_holiday import agenda.utils import agenda.weather @@ -370,211 +366,20 @@ 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 = build_conference_list() - country_filter = normalize_country_filter(flask.request.args.get("country")) - country_options = conference_country_options( + 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( [conf for conf in items if conf["latest_date"] >= today] ) - items = filter_conferences_by_country(items, country_filter) + items = agenda.conference_list.filter_conferences_by_country(items, country_filter) current = [ conf @@ -587,7 +392,7 @@ def conference_list() -> str: conf for conf in items if conf not in current and conf["latest_date"] >= today ] - timeline = build_conference_timeline(current, future, today) + timeline = agenda.conference_list.build_conference_timeline(current, future, today) return flask.render_template( "conference_list.html", @@ -606,12 +411,16 @@ def conference_list() -> str: def past_conference_list() -> str: """Page showing a list of conferences.""" today = date.today() - items = build_conference_list() - country_filter = normalize_country_filter(flask.request.args.get("country")) - country_options = conference_country_options( + 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( [conf for conf in items if conf["latest_date"] < today] ) - items = filter_conferences_by_country(items, country_filter) + items = agenda.conference_list.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], @@ -628,7 +437,13 @@ def conference_series_list() -> str: """Page showing conference series.""" return flask.render_template( "conference_series_list.html", - series_list=build_conference_series_list(), + 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(), + ), get_country=agenda.get_country, ) @@ -642,7 +457,11 @@ def conference_series_page(series_id: str) -> str: flask.abort(404) conferences = [ - conf for conf in build_conference_list() if conf.get("series") == series_id + 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 ] return flask.render_template( "conference_series.html", @@ -658,7 +477,9 @@ 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 = build_conference_list() + items = agenda.conference_list.build_conference_list( + app.config["PERSONAL_DATA"], agenda.trip.build_trip_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" @@ -941,331 +762,6 @@ 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/") def trip_page(start: str) -> str: """Individual trip page.""" @@ -1300,7 +796,9 @@ def trip_page(start: str) -> str: trip, cache_only=True, ) - localize_trip_car_journey_display_times(trip, app.config["PERSONAL_DATA"]) + agenda.trip_timezones.localize_trip_car_journey_display_times( + trip, app.config["PERSONAL_DATA"] + ) return flask.render_template( "trip_page.html", @@ -1314,7 +812,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=get_destination_timezones(trip), + destination_times=agenda.trip_timezones.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),