Edward Betts
29d5145b87
Simplify the location tracking function by extracting travel data directly from trip objects instead of requiring separate YAML file parameters. Changes: - Remove airport, train, and ferry location helper functions that required separate YAML data lookups - Update get_location_for_date signature to only take target_date and trips - Extract flight/train/ferry details directly from trip.travel items - Use embedded airport/station/terminal objects from trip data - Remove YAML file parsing from weekends function - Update test calls to use new simplified signature This eliminates duplicate data loading and simplifies the API while maintaining all existing functionality. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
656 lines
26 KiB
Python
656 lines
26 KiB
Python
"""Identify busy events and gaps when nothing is scheduled."""
|
|
|
|
import itertools
|
|
import typing
|
|
from datetime import date, datetime, timedelta
|
|
|
|
import flask
|
|
import pycountry
|
|
|
|
from . import events_yaml, get_country, travel
|
|
from .event import Event
|
|
from .types import StrDict, Trip
|
|
|
|
|
|
def busy_event(e: Event) -> bool:
|
|
"""Busy."""
|
|
if e.name not in {
|
|
"event",
|
|
"accommodation",
|
|
"conference",
|
|
"transport",
|
|
"meetup",
|
|
"party",
|
|
"trip",
|
|
"hackathon",
|
|
}:
|
|
return False
|
|
|
|
if e.title in ("IA UK board meeting", "Mill Road Winter Fair"):
|
|
return False
|
|
|
|
if e.name == "conference" and not e.going:
|
|
return False
|
|
if not e.title:
|
|
return True
|
|
if e.title == "LHG Run Club" or "Third Thursday Social" in e.title:
|
|
return False
|
|
|
|
lc_title = e.title.lower()
|
|
return (
|
|
"rebels" not in lc_title
|
|
and "south west data social" not in lc_title
|
|
and "dorkbot" not in lc_title
|
|
)
|
|
|
|
|
|
def get_busy_events(
|
|
start: date, config: flask.config.Config, trips: list[Trip]
|
|
) -> list[Event]:
|
|
"""Find busy events from a year ago to two years in the future."""
|
|
last_year = start - timedelta(days=365)
|
|
next_year = start + timedelta(days=2 * 365)
|
|
|
|
my_data = config["PERSONAL_DATA"]
|
|
events = events_yaml.read(my_data, last_year, next_year, skip_trips=True)
|
|
|
|
for trip in trips:
|
|
event_type = "trip"
|
|
if trip.events and not trip.conferences:
|
|
event_type = trip.events[0]["name"]
|
|
elif len(trip.conferences) == 1 and trip.conferences[0].get("hackathon"):
|
|
event_type = "hackathon"
|
|
events.append(
|
|
Event(
|
|
name=event_type,
|
|
title=trip.title + " " + trip.country_flags,
|
|
date=trip.start,
|
|
end_date=trip.end,
|
|
url=flask.url_for("trip_page", start=trip.start.isoformat()),
|
|
)
|
|
)
|
|
|
|
busy_events = [
|
|
e
|
|
for e in sorted(events, key=lambda e: e.as_date)
|
|
if (e.as_date >= start or (e.end_date and e.end_as_date >= start))
|
|
and e.as_date < next_year
|
|
and busy_event(e)
|
|
]
|
|
|
|
return busy_events
|
|
|
|
|
|
def _parse_datetime_field(datetime_obj: datetime | date) -> tuple[datetime, date]:
|
|
"""Parse a datetime field that could be datetime object or string."""
|
|
if hasattr(datetime_obj, "date"):
|
|
return datetime_obj, datetime_obj.date()
|
|
elif isinstance(datetime_obj, str):
|
|
dt = datetime.fromisoformat(datetime_obj.replace("Z", "+00:00"))
|
|
return dt, dt.date()
|
|
else:
|
|
raise ValueError(f"Invalid datetime format: {datetime_obj}")
|
|
|
|
|
|
def _get_accommodation_location(
|
|
acc: StrDict, on_trip: bool = False
|
|
) -> tuple[str | None, pycountry.db.Country]:
|
|
"""Get location from accommodation data."""
|
|
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,
|
|
) -> 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"}
|
|
|
|
trip_most_recent_date = None
|
|
trip_most_recent_location = None
|
|
trip_most_recent_datetime = None
|
|
|
|
# Check flights within trip period
|
|
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 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_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 trip.accommodation:
|
|
if "from" in acc:
|
|
try:
|
|
_, acc_date = _parse_datetime_field(acc["from"])
|
|
except ValueError:
|
|
continue
|
|
|
|
# Only consider accommodations within this trip and before/on target date
|
|
if trip.start <= acc_date <= target_date:
|
|
# Accommodation takes precedence over flights on the same date
|
|
# or if it's genuinely more recent
|
|
if (
|
|
trip_most_recent_date is None
|
|
or acc_date > trip_most_recent_date
|
|
or acc_date == trip_most_recent_date
|
|
):
|
|
trip_most_recent_date = acc_date
|
|
trip_most_recent_location = _get_accommodation_location(
|
|
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
|
|
|
|
|
|
def _get_trip_location_by_progression(
|
|
trip: Trip, target_date: date
|
|
) -> tuple[str | None, pycountry.db.Country | None] | None:
|
|
"""Determine location based on trip progression and date."""
|
|
locations = trip.locations()
|
|
if not locations:
|
|
return None
|
|
|
|
# If only one location, use it (when on a trip, always show the location)
|
|
if len(locations) == 1:
|
|
city, country = locations[0]
|
|
return (city, country)
|
|
|
|
# Multiple locations: use progression through the trip
|
|
trip_duration = (trip.end - trip.start).days + 1
|
|
days_into_trip = (target_date - trip.start).days
|
|
|
|
# Simple progression: first half at first location, second half at last location
|
|
if days_into_trip <= trip_duration // 2:
|
|
city, country = locations[0]
|
|
else:
|
|
city, country = locations[-1]
|
|
|
|
return (city, country)
|
|
|
|
|
|
def _find_most_recent_travel_before_date(
|
|
target_date: date,
|
|
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"}
|
|
|
|
most_recent_location = None
|
|
most_recent_date = None
|
|
most_recent_datetime = None
|
|
|
|
# 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(
|
|
travel_item["arrive"]
|
|
)
|
|
except ValueError:
|
|
continue
|
|
|
|
if arrive_date <= target_date:
|
|
# Compare both date and time to handle same-day flights 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_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
|
|
)
|
|
|
|
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"),
|
|
)
|
|
|
|
# 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
|
|
|
|
|
|
def _check_return_home_heuristic(
|
|
target_date: date, trips: list[Trip]
|
|
) -> tuple[str | None, pycountry.db.Country | None] | None:
|
|
"""Check if should return home based on recent trips that have ended."""
|
|
for trip in trips:
|
|
if trip.end and trip.end < target_date:
|
|
locations = trip.locations()
|
|
if locations:
|
|
final_city, final_country = locations[-1]
|
|
final_alpha_2 = final_country.alpha_2
|
|
days_since_trip = (target_date - trip.end).days
|
|
|
|
# If trip ended in UK, you should be home now
|
|
if hasattr(final_country, "alpha_2") and final_country.alpha_2 == "GB":
|
|
return (None, get_country("gb"))
|
|
|
|
return None
|
|
|
|
|
|
def get_location_for_date(
|
|
target_date: date,
|
|
trips: list[Trip],
|
|
) -> 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 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
|
|
|
|
# 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)
|
|
return_home = _check_return_home_heuristic(target_date, trips)
|
|
if return_home:
|
|
return return_home
|
|
|
|
# Return most recent location or default to home
|
|
if recent_travel:
|
|
return recent_travel
|
|
|
|
return (None, get_country("gb"))
|
|
|
|
|
|
def weekends(
|
|
start: date, busy_events: list[Event], trips: list[Trip], data_dir: str
|
|
) -> typing.Sequence[StrDict]:
|
|
"""Next ten weekends."""
|
|
weekday = start.weekday()
|
|
|
|
# Calculate the difference to the next or previous Saturday
|
|
if weekday == 6: # Sunday
|
|
start_date = start - timedelta(days=1)
|
|
else:
|
|
start_date = start + timedelta(days=(5 - weekday))
|
|
|
|
weekends_info = []
|
|
for i in range(52):
|
|
saturday = start_date + timedelta(weeks=i)
|
|
sunday = saturday + timedelta(days=1)
|
|
|
|
saturday_events = [
|
|
event
|
|
for event in busy_events
|
|
if event.end_date and event.as_date <= saturday <= event.end_as_date
|
|
]
|
|
sunday_events = [
|
|
event
|
|
for event in busy_events
|
|
if event.end_date and event.as_date <= sunday <= event.end_as_date
|
|
]
|
|
|
|
saturday_location = get_location_for_date(
|
|
saturday,
|
|
trips,
|
|
)
|
|
sunday_location = get_location_for_date(
|
|
sunday,
|
|
trips,
|
|
)
|
|
|
|
weekends_info.append(
|
|
{
|
|
"date": saturday,
|
|
"saturday": saturday_events,
|
|
"sunday": sunday_events,
|
|
"saturday_location": saturday_location,
|
|
"sunday_location": sunday_location,
|
|
}
|
|
)
|
|
|
|
return weekends_info
|
|
|
|
|
|
def find_gaps(events: list[Event], min_gap_days: int = 3) -> list[StrDict]:
|
|
"""Gaps of at least `min_gap_days` between events in a list of events."""
|
|
# Sort events by start date
|
|
|
|
gaps: list[tuple[date, date]] = []
|
|
previous_event_end = None
|
|
|
|
by_start_date = {
|
|
d: list(on_day)
|
|
for d, on_day in itertools.groupby(events, key=lambda e: e.as_date)
|
|
}
|
|
|
|
by_end_date = {
|
|
d: list(on_day)
|
|
for d, on_day in itertools.groupby(events, key=lambda e: e.end_as_date)
|
|
}
|
|
|
|
for event in events:
|
|
# Use start date for current event
|
|
start_date = event.as_date
|
|
|
|
# If previous event exists, calculate the gap
|
|
if previous_event_end:
|
|
gap_days = (start_date - previous_event_end).days
|
|
if gap_days >= (min_gap_days + 2):
|
|
start_end = (
|
|
previous_event_end + timedelta(days=1),
|
|
start_date - timedelta(days=1),
|
|
)
|
|
gaps.append(start_end)
|
|
|
|
# Update previous event end date
|
|
end = event.end_as_date
|
|
if not previous_event_end or end > previous_event_end:
|
|
previous_event_end = end
|
|
|
|
return [
|
|
{
|
|
"start": gap_start,
|
|
"end": gap_end,
|
|
"after": by_start_date[gap_end + timedelta(days=1)],
|
|
"before": by_end_date[gap_start - timedelta(days=1)],
|
|
}
|
|
for gap_start, gap_end in gaps
|
|
]
|