diff --git a/agenda/busy.py b/agenda/busy.py index 4fcbeec..8452903 100644 --- a/agenda/busy.py +++ b/agenda/busy.py @@ -11,20 +11,6 @@ from . import events_yaml, get_country, travel from .event import Event from .types import StrDict, Trip -NEARBY_BALKAN_COUNTRIES = { - "GR", - "AL", - "XK", - "HR", - "SI", - "MK", - "BA", - "ME", - "RS", - "BG", - "RO", -} - def busy_event(e: Event) -> bool: """Busy.""" @@ -106,58 +92,20 @@ def _parse_datetime_field(datetime_obj: datetime | date) -> tuple[datetime, date raise ValueError(f"Invalid datetime format: {datetime_obj}") -def _get_airport_location( - airport_code: str, airports: StrDict, uk_airports: set[str], on_trip: bool = False -) -> tuple[str | None, pycountry.db.Country | None]: - """Get location from airport code.""" - if airport_code in uk_airports: - if on_trip: - # When on a trip, show the actual location even for UK airports - airport_info = airports.get(airport_code) - if airport_info: - location_name = airport_info.get( - "city", airport_info.get("name", "London") - ) - return (location_name, get_country("gb")) - else: - return ("London", get_country("gb")) - else: - # When not on a trip, UK airports mean home - return (None, get_country("gb")) - else: - # Non-UK airports - airport_info = airports.get(airport_code) - if airport_info: - location_name = airport_info.get( - "city", airport_info.get("name", airport_code) - ) - return (location_name, get_country(airport_info.get("country", "gb"))) - else: - return (airport_code, get_country("gb")) - - def _get_accommodation_location( acc: StrDict, on_trip: bool = False -) -> tuple[str | None, pycountry.db.Country | None]: +) -> tuple[str | None, pycountry.db.Country]: """Get location from accommodation data.""" - if acc.get("country") == "gb": - if on_trip: - # When on a trip, show the actual location even for UK accommodations - return (acc.get("location", "London"), get_country("gb")) - else: - # When not on a trip, UK accommodation means home - return (None, get_country("gb")) - else: - return (acc.get("location", "Unknown"), get_country(acc.get("country", "gb"))) + c = get_country(acc["country"]) + assert c + assert isinstance(acc["location"], str) + return (acc["location"] if on_trip else None, c) def _find_most_recent_travel_within_trip( trip: Trip, target_date: date, - bookings: list[StrDict], - accommodations: list[StrDict], - airports: StrDict, -) -> tuple[str | None, pycountry.db.Country | None] | None: +) -> tuple[str | None, pycountry.db.Country] | None: """Find the most recent travel location within a trip.""" uk_airports = {"LHR", "LGW", "STN", "LTN", "BRS", "BHX", "MAN", "EDI", "GLA"} @@ -166,39 +114,54 @@ def _find_most_recent_travel_within_trip( trip_most_recent_datetime = None # Check flights within trip period - for booking in bookings: - for flight in booking.get("flights", []): - if "arrive" in flight: - try: - arrive_datetime, arrive_date = _parse_datetime_field( - flight["arrive"] - ) - except ValueError: - continue + for travel_item in trip.travel: + if travel_item["type"] == "flight" and "arrive" in travel_item: + arrive_datetime, arrive_date = _parse_datetime_field(travel_item["arrive"]) - # Only consider flights within this trip and before target date - if trip.start <= arrive_date <= target_date: - # Compare both date and time to handle same-day flights correctly - if ( - trip_most_recent_date is None - or arrive_date > trip_most_recent_date - or ( - arrive_date == trip_most_recent_date - and ( - trip_most_recent_datetime is None - or arrive_datetime > trip_most_recent_datetime - ) + # Only consider flights within this trip and before target date + if not (trip.start <= arrive_date <= target_date): + continue + # Compare both date and time to handle same-day flights correctly + if ( + trip_most_recent_date is None + or arrive_date > trip_most_recent_date + or ( + arrive_date == trip_most_recent_date + and ( + trip_most_recent_datetime is None + or arrive_datetime > trip_most_recent_datetime + ) + ) + ): + trip_most_recent_date = arrive_date + trip_most_recent_datetime = arrive_datetime + destination_airport = travel_item["to"] + assert "to_airport" in travel_item + airport_info = travel_item["to_airport"] + airport_country = airport_info["country"] + if airport_country == "gb": + if destination_airport in uk_airports: + # UK airport while on trip - show actual location + location_name = airport_info.get( + "city", airport_info.get("name", "London") ) - ): - trip_most_recent_date = arrive_date - trip_most_recent_datetime = arrive_datetime - destination_airport = flight["to"] - trip_most_recent_location = _get_airport_location( - destination_airport, airports, uk_airports, on_trip=True + trip_most_recent_location = ( + location_name, + get_country("gb"), ) + else: + trip_most_recent_location = (None, get_country("gb")) + else: + location_name = airport_info.get( + "city", airport_info.get("name", destination_airport) + ) + trip_most_recent_location = ( + location_name, + get_country(airport_country), + ) # Check accommodations within trip period - for acc in accommodations: + for acc in trip.accommodation: if "from" in acc: try: _, acc_date = _parse_datetime_field(acc["from"]) @@ -219,6 +182,93 @@ def _find_most_recent_travel_within_trip( acc, on_trip=True ) + # Check trains within trip period + for travel_item in trip.travel: + if travel_item["type"] == "train": + for leg in travel_item.get("legs", []): + if "arrive" in leg: + try: + arrive_datetime, arrive_date = _parse_datetime_field( + leg["arrive"] + ) + except ValueError: + continue + + # Only consider trains within this trip and before target date + if trip.start <= arrive_date <= target_date: + # Compare both date and time to handle same-day arrivals correctly + if ( + trip_most_recent_date is None + or arrive_date > trip_most_recent_date + or ( + arrive_date == trip_most_recent_date + and ( + trip_most_recent_datetime is None + or arrive_datetime > trip_most_recent_datetime + ) + ) + ): + trip_most_recent_date = arrive_date + trip_most_recent_datetime = arrive_datetime + # For trains, we can get station info from to_station if available + destination = leg.get("to") + assert "to_station" in leg + station_info = leg["to_station"] + station_country = station_info["country"] + if station_country == "gb": + trip_most_recent_location = ( + destination, + get_country("gb"), + ) + else: + trip_most_recent_location = ( + destination, + get_country(station_country), + ) + + # Check ferries within trip period + for travel_item in trip.travel: + if travel_item["type"] == "ferry" and "arrive" in travel_item: + try: + arrive_datetime, arrive_date = _parse_datetime_field( + travel_item["arrive"] + ) + except ValueError: + continue + + # Only consider ferries within this trip and before target date + if trip.start <= arrive_date <= target_date: + # Compare both date and time to handle same-day arrivals correctly + if ( + trip_most_recent_date is None + or arrive_date > trip_most_recent_date + or ( + arrive_date == trip_most_recent_date + and ( + trip_most_recent_datetime is None + or arrive_datetime > trip_most_recent_datetime + ) + ) + ): + trip_most_recent_date = arrive_date + trip_most_recent_datetime = arrive_datetime + # For ferries, we can get terminal info from to_terminal if available + destination = travel_item.get("to") + assert "to_terminal" in travel_item + terminal_info = travel_item["to_terminal"] + terminal_country = terminal_info.get("country", "gb") + terminal_city = terminal_info.get("city", destination) + if terminal_country == "gb": + trip_most_recent_location = ( + terminal_city, + get_country("gb"), + ) + else: + trip_most_recent_location = ( + terminal_city, + get_country(terminal_country), + ) + return trip_most_recent_location @@ -250,9 +300,7 @@ def _get_trip_location_by_progression( def _find_most_recent_travel_before_date( target_date: date, - bookings: list[StrDict], - accommodations: list[StrDict], - airports: StrDict, + trips: list[Trip], ) -> tuple[str | None, pycountry.db.Country | None] | None: """Find the most recent travel location before a given date.""" uk_airports = {"LHR", "LGW", "STN", "LTN", "BRS", "BHX", "MAN", "EDI", "GLA"} @@ -261,13 +309,14 @@ def _find_most_recent_travel_before_date( most_recent_date = None most_recent_datetime = None - # Check flights - for booking in bookings: - for flight in booking.get("flights", []): - if "arrive" in flight: + # Check all travel across all trips + for trip in trips: + # Check flights + for travel_item in trip.travel: + if travel_item["type"] == "flight" and "arrive" in travel_item: try: arrive_datetime, arrive_date = _parse_datetime_field( - flight["arrive"] + travel_item["arrive"] ) except ValueError: continue @@ -287,26 +336,167 @@ def _find_most_recent_travel_before_date( ): most_recent_date = arrive_date most_recent_datetime = arrive_datetime - destination_airport = flight["to"] - most_recent_location = _get_airport_location( - destination_airport, airports, uk_airports, on_trip=False + destination_airport = travel_item["to"] + # For flights, determine if we're "on trip" based on whether this is within any trip period + on_trip = any( + t.start <= arrive_date <= (t.end or t.start) for t in trips ) - # Check accommodation - only override if accommodation is more recent - for acc in accommodations: - if "from" in acc: - try: - _, acc_date = _parse_datetime_field(acc["from"]) - except ValueError: - continue + if "to_airport" in travel_item: + airport_info = travel_item["to_airport"] + airport_country = airport_info.get("country", "gb") + if airport_country == "gb": + if not on_trip: + # When not on a trip, UK airports mean home + most_recent_location = (None, get_country("gb")) + else: + # When on a trip, show the actual location even for UK airports + location_name = airport_info.get( + "city", airport_info.get("name", "London") + ) + most_recent_location = ( + location_name, + get_country("gb"), + ) + else: + location_name = airport_info.get( + "city", + airport_info.get("name", destination_airport), + ) + most_recent_location = ( + location_name, + get_country(airport_country), + ) + else: + most_recent_location = ( + destination_airport, + get_country("gb"), + ) - if acc_date <= target_date: - # Only update if this accommodation is more recent than existing result - if most_recent_date is None or acc_date > most_recent_date: - most_recent_date = acc_date - most_recent_location = _get_accommodation_location( - acc, on_trip=False + # Check trains + elif travel_item["type"] == "train": + for leg in travel_item.get("legs", []): + if "arrive" in leg: + try: + arrive_datetime, arrive_date = _parse_datetime_field( + leg["arrive"] + ) + except ValueError: + continue + + if arrive_date <= target_date: + # Compare both date and time to handle same-day arrivals correctly + if ( + most_recent_date is None + or arrive_date > most_recent_date + or ( + arrive_date == most_recent_date + and ( + most_recent_datetime is None + or arrive_datetime > most_recent_datetime + ) + ) + ): + most_recent_date = arrive_date + most_recent_datetime = arrive_datetime + destination = leg.get("to") + on_trip = any( + t.start <= arrive_date <= (t.end or t.start) + for t in trips + ) + + if "to_station" in leg: + station_info = leg["to_station"] + station_country = station_info.get("country", "gb") + if station_country == "gb": + if not on_trip: + most_recent_location = ( + None, + get_country("gb"), + ) + else: + most_recent_location = ( + destination, + get_country("gb"), + ) + else: + most_recent_location = ( + destination, + get_country(station_country), + ) + else: + most_recent_location = ( + destination, + get_country("gb"), + ) + + # Check ferries + elif travel_item["type"] == "ferry" and "arrive" in travel_item: + try: + arrive_datetime, arrive_date = _parse_datetime_field( + travel_item["arrive"] ) + except ValueError: + continue + + if arrive_date <= target_date: + # Compare both date and time to handle same-day arrivals correctly + if ( + most_recent_date is None + or arrive_date > most_recent_date + or ( + arrive_date == most_recent_date + and ( + most_recent_datetime is None + or arrive_datetime > most_recent_datetime + ) + ) + ): + most_recent_date = arrive_date + most_recent_datetime = arrive_datetime + destination = travel_item.get("to") + on_trip = any( + t.start <= arrive_date <= (t.end or t.start) for t in trips + ) + + if "to_terminal" in travel_item: + terminal_info = travel_item["to_terminal"] + terminal_country = terminal_info.get("country", "gb") + terminal_city = terminal_info.get("city", destination) + if terminal_country == "gb": + if not on_trip: + most_recent_location = (None, get_country("gb")) + else: + most_recent_location = ( + terminal_city, + get_country("gb"), + ) + else: + most_recent_location = ( + terminal_city, + get_country(terminal_country), + ) + else: + most_recent_location = (destination, get_country("gb")) + + # Check accommodation - only override if accommodation is more recent + for acc in trip.accommodation: + if "from" in acc: + try: + _, acc_date = _parse_datetime_field(acc["from"]) + except ValueError: + continue + + if acc_date <= target_date: + # Only update if this accommodation is more recent than existing result + if most_recent_date is None or acc_date > most_recent_date: + most_recent_date = acc_date + on_trip = any( + t.start <= acc_date <= (t.end or t.start) for t in trips + ) + most_recent_location = _get_accommodation_location( + acc, on_trip=on_trip + ) return most_recent_location @@ -327,49 +517,33 @@ def _check_return_home_heuristic( if hasattr(final_country, "alpha_2") and final_country.alpha_2 == "GB": return (None, get_country("gb")) - # For short trips to nearby countries or international trips - # (ended >=1 day ago), assume returned home if no subsequent travel data - if days_since_trip >= 1 and ( - # European countries (close by rail/ferry) - final_alpha_2 in {"BE", "NL", "FR", "DE", "CH", "AT", "IT", "ES"} - # Nearby Balkan countries - or final_alpha_2 in NEARBY_BALKAN_COUNTRIES - # International trips (assume return home after trip ends) - or final_alpha_2 - in {"US", "CA", "IN", "JP", "CN", "AU", "NZ", "BR", "AR", "ZA"} - ): - return (None, get_country("gb")) - return None def get_location_for_date( target_date: date, trips: list[Trip], - bookings: list[StrDict], - accommodations: list[StrDict], - airports: StrDict, ) -> tuple[str | None, pycountry.db.Country | None]: """Get location (city, country) for a specific date using travel history.""" # First check if currently on a trip for trip in trips: - if trip.start <= target_date <= (trip.end or trip.start): - # For trips, find the most recent flight or accommodation within the trip period - trip_location = _find_most_recent_travel_within_trip( - trip, target_date, bookings, accommodations, airports - ) - if trip_location: - return trip_location + if not (trip.start <= target_date <= (trip.end or trip.start)): + continue + # For trips, find the most recent travel within the trip period + trip_location = _find_most_recent_travel_within_trip( + trip, + target_date, + ) + if trip_location: + return trip_location - # Fallback: determine location based on trip progression and date - progression_location = _get_trip_location_by_progression(trip, target_date) - if progression_location: - return progression_location + # Fallback: determine location based on trip progression and date + progression_location = _get_trip_location_by_progression(trip, target_date) + if progression_location: + return progression_location - # Find most recent flight or accommodation before this date - recent_travel = _find_most_recent_travel_before_date( - target_date, bookings, accommodations, airports - ) + # Find most recent travel before this date + recent_travel = _find_most_recent_travel_before_date(target_date, trips) # Check for recent trips that have ended - prioritize this over individual travel data # This handles cases where you're traveling home after a trip (e.g. stopovers, connections) @@ -396,11 +570,6 @@ def weekends( else: start_date = start + timedelta(days=(5 - weekday)) - # Parse YAML files once for all location lookups - bookings = travel.parse_yaml("flights", data_dir) - accommodations = travel.parse_yaml("accommodation", data_dir) - airports = travel.parse_yaml("airports", data_dir) - weekends_info = [] for i in range(52): saturday = start_date + timedelta(weeks=i) @@ -418,10 +587,12 @@ def weekends( ] saturday_location = get_location_for_date( - saturday, trips, bookings, accommodations, airports + saturday, + trips, ) sunday_location = get_location_for_date( - sunday, trips, bookings, accommodations, airports + sunday, + trips, ) weekends_info.append( diff --git a/templates/trip_debug.html b/templates/trip_debug.html new file mode 100644 index 0000000..7ccc06e --- /dev/null +++ b/templates/trip_debug.html @@ -0,0 +1,128 @@ +{% extends "base.html" %} + +{% block title %}Debug: {{ trip.title }} ({{ trip.start }}) - Edward Betts{% endblock %} + +{% block style %} + +{% endblock %} + +{% block content %} +