341 lines
12 KiB
Python
341 lines
12 KiB
Python
"""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
|