Compare commits

...

5 commits

8 changed files with 135 additions and 114 deletions

View file

@ -2,8 +2,8 @@
from datetime import date, datetime, timedelta
from .trip import depart_datetime
from .types import SchengenCalculation, SchengenStay, StrDict
from .utils import depart_datetime
# Schengen Area countries as of 2025
SCHENGEN_COUNTRIES = {

View file

@ -3,14 +3,14 @@
import decimal
import os
import typing
from datetime import date, datetime, time
from zoneinfo import ZoneInfo
from datetime import date
import flask
import yaml
from agenda import travel
from agenda import travel, trip_schengen
from agenda.types import StrDict, Trip
from agenda.utils import depart_datetime
class Airline(typing.TypedDict, total=False):
@ -122,18 +122,6 @@ def load_coaches(
return coaches
def depart_datetime(item: StrDict) -> datetime:
"""Return a datetime for this travel item.
If the travel item already has a datetime return that, otherwise if the
departure time is just a date return midnight UTC for that date.
"""
depart = item["depart"]
if isinstance(depart, datetime):
return depart
return datetime.combine(depart, time.min).replace(tzinfo=ZoneInfo("UTC"))
def process_flight(
flight: StrDict, by_iata: dict[str, Airline], airports: list[StrDict]
) -> None:
@ -409,7 +397,9 @@ def get_trip_routes(trip: Trip, data_dir: str) -> list[StrDict]:
# Use GeoJSON route when available, otherwise draw straight line
if t.get("geojson_filename"):
filename = os.path.join("coach_routes", t["geojson_filename"])
routes.append({"type": "coach", "key": key, "geojson_filename": filename})
routes.append(
{"type": "coach", "key": key, "geojson_filename": filename}
)
else:
routes.append(
{
@ -424,7 +414,9 @@ def get_trip_routes(trip: Trip, data_dir: str) -> list[StrDict]:
for leg in t["legs"]:
train_from, train_to = leg["from_station"], leg["to_station"]
geojson_filename = train_from.get("routes", {}).get(train_to["name"])
key = "_".join(["train"] + sorted([train_from["name"], train_to["name"]]))
key = "_".join(
["train"] + sorted([train_from["name"], train_to["name"]])
)
if not geojson_filename:
routes.append(
{
@ -444,7 +436,9 @@ def get_trip_routes(trip: Trip, data_dir: str) -> list[StrDict]:
{
"type": "train",
"key": key,
"geojson_filename": os.path.join("train_routes", geojson_filename),
"geojson_filename": os.path.join(
"train_routes", geojson_filename
),
}
)
@ -496,3 +490,33 @@ def get_coordinates_and_routes(
route["geojson"] = read_geojson(data_dir, route.pop("geojson_filename"))
return (coordinates, routes)
def get_trip_list(
route_distances: travel.RouteDistances | None = None,
) -> list[Trip]:
"""Get list of trips respecting current authentication status."""
trips = [
trip
for trip in build_trip_list(route_distances=route_distances)
if flask.g.user.is_authenticated or not trip.private
]
# Add Schengen compliance information to each trip
for trip in trips:
trip_schengen.add_schengen_compliance_to_trip(trip)
return trips
def get_current_trip(today: date) -> Trip | None:
"""Get current trip."""
trip_list = get_trip_list(route_distances=None)
current = [
item
for item in trip_list
if item.start <= today and (item.end or item.start) >= today
]
assert len(current) < 2
return current[0] if current else None

View file

@ -7,26 +7,37 @@ from datetime import date, timedelta
import flask
from . import get_country, trip
from .schengen import calculate_schengen_time, extract_schengen_stays_from_travel
from .schengen import (
SCHENGEN_COUNTRIES,
calculate_schengen_time,
extract_schengen_stays_from_travel,
)
from .types import SchengenCalculation, SchengenStay, StrDict, Trip
def add_schengen_compliance_to_trip(trip_obj: Trip) -> Trip:
def trip_includes_schengen(trip: Trip) -> bool:
return bool({c.alpha_2.lower() for c in trip.countries} & SCHENGEN_COUNTRIES)
def add_schengen_compliance_to_trip(trip: Trip) -> Trip:
"""Add Schengen compliance information to a trip object."""
if not trip_includes_schengen(trip):
return trip
try:
# Calculate Schengen compliance for the trip
calculation = calculate_schengen_time(trip_obj.travel)
calculation = calculate_schengen_time(trip.travel)
# Add the calculation to the trip object
trip_obj.schengen_compliance = calculation
trip.schengen_compliance = calculation
except Exception as e:
# Log the error but don't fail the trip loading
logging.warning(
f"Failed to calculate Schengen compliance for trip {trip_obj.start}: {e}"
f"Failed to calculate Schengen compliance for trip {trip.start}: {e}"
)
trip_obj.schengen_compliance = None
trip.schengen_compliance = None
return trip_obj
return trip
def get_schengen_compliance_for_all_trips(
@ -127,7 +138,9 @@ def schengen_dashboard_data(data_dir: str | None = None) -> dict[str, typing.Any
data_dir = flask.current_app.config["PERSONAL_DATA"]
# Load all trips
trip_list = trip.build_trip_list(data_dir)
trip_list = [
trip for trip in trip.build_trip_list(data_dir) if trip_includes_schengen(trip)
]
# Calculate current compliance with trip information
all_travel_items = []

View file

@ -55,6 +55,43 @@ def airport_label(airport: StrDict) -> str:
return f"{name} ({airport['iata']})"
@dataclass
class SchengenStay:
"""Represents a stay in the Schengen area."""
entry_date: date
exit_date: date | None # None if currently in Schengen
country: str
days: int
trip_date: date | None = None # Trip start date for linking
trip_name: str | None = None # Trip name for display
def __post_init__(self) -> None:
"""Post init."""
if self.exit_date is None:
# Currently in Schengen, calculate days up to today
self.days = (date.today() - self.entry_date).days + 1
else:
self.days = (self.exit_date - self.entry_date).days + 1
@dataclass
class SchengenCalculation:
"""Result of Schengen time calculation."""
total_days_used: int
days_remaining: int
is_compliant: bool
current_180_day_period: tuple[date, date] # (start, end)
stays_in_period: SchengenStay
next_reset_date: typing.Optional[date] # When the 180-day window resets
@property
def days_over_limit(self) -> int:
"""Days over the 90-day limit."""
return max(0, self.total_days_used - 90)
@dataclass
class Trip:
"""Trip."""
@ -67,7 +104,7 @@ class Trip:
flight_bookings: list[StrDict] = field(default_factory=list)
name: str | None = None
private: bool = False
schengen_compliance: typing.Optional["SchengenCalculation"] = None
schengen_compliance: SchengenCalculation | None = None
@property
def title(self) -> str:
@ -409,39 +446,3 @@ class Holiday:
if self.local_name and self.local_name != self.name
else self.name
)
@dataclass
class SchengenStay:
"""Represents a stay in the Schengen area."""
entry_date: date
exit_date: typing.Optional[date] # None if currently in Schengen
country: str
days: int
trip_date: typing.Optional[date] = None # Trip start date for linking
trip_name: typing.Optional[str] = None # Trip name for display
def __post_init__(self) -> None:
if self.exit_date is None:
# Currently in Schengen, calculate days up to today
self.days = (date.today() - self.entry_date).days + 1
else:
self.days = (self.exit_date - self.entry_date).days + 1
@dataclass
class SchengenCalculation:
"""Result of Schengen time calculation."""
total_days_used: int
days_remaining: int
is_compliant: bool
current_180_day_period: tuple[date, date] # (start, end)
stays_in_period: list["SchengenStay"]
next_reset_date: typing.Optional[date] # When the 180-day window resets
@property
def days_over_limit(self) -> int:
"""Days over the 90-day limit."""
return max(0, self.total_days_used - 90)

View file

@ -2,8 +2,10 @@
import os
import typing
from datetime import date, datetime, timedelta, timezone
from time import time
from datetime import date, datetime, time, timedelta, timezone
from zoneinfo import ZoneInfo
from .types import StrDict
def as_date(d: datetime | date) -> date:
@ -118,3 +120,15 @@ async def time_function(
exception = e
end_time = time()
return name, result, end_time - start_time, exception
def depart_datetime(item: StrDict) -> datetime:
"""Return a datetime for this travel item.
If the travel item already has a datetime return that, otherwise if the
departure time is just a date return midnight UTC for that date.
"""
depart = item["depart"]
if isinstance(depart, datetime):
return depart
return datetime.combine(depart, time.min).replace(tzinfo=ZoneInfo("UTC"))

View file

@ -162,8 +162,8 @@
<td>{{ trip_date.strftime('%Y-%m-%d') }}</td>
<td>
<a href="{{ url_for('trip_page', start=trip_date.strftime('%Y-%m-%d')) }}" class="text-decoration-none">
{% for trip_obj in trip_list if trip_obj.start == trip_date %}
{{ trip_obj.title }}
{% for trip in trip_list if trip.start == trip_date %}
{{ trip.title }} {{ trip.country_flags }}
{% endfor %}
</a>
</td>

View file

@ -20,7 +20,7 @@
</div>
{% endfor %}
{% for year, year_stats in yearly_stats | dictsort %}
{% for year, year_stats in yearly_stats | dictsort(reverse=True) %}
{% set countries = year_stats.countries | sort(attribute="name") %}
<h4>{{ year }}</h4>
<div>Trips in {{ year }}: {{ year_stats.count }}</div>

View file

@ -57,6 +57,7 @@ def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str,
last_frame = list(traceback.walk_tb(current_traceback))[-1][0]
last_frame_args = inspect.getargs(last_frame.f_code)
assert tb._te.exc_type
return (
flask.render_template(
@ -72,19 +73,6 @@ def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str,
)
def get_current_trip(today: date) -> Trip | None:
"""Get current trip."""
trip_list = get_trip_list(route_distances=None)
current = [
item
for item in trip_list
if item.start <= today and (item.end or item.start) >= today
]
assert len(current) < 2
return current[0] if current else None
@app.route("/")
async def index() -> str:
"""Index page."""
@ -106,7 +94,7 @@ async def index() -> str:
today=today,
events=events,
get_country=agenda.get_country,
current_trip=get_current_trip(today),
current_trip=agenda.trip.get_current_trip(today),
start_event_list=date.today() - timedelta(days=1),
end_event_list=date.today() + timedelta(days=365 * 2),
render_time=(time.time() - t0),
@ -429,23 +417,6 @@ def accommodation_list() -> str:
)
def get_trip_list(
route_distances: agenda.travel.RouteDistances | None = None,
) -> list[Trip]:
"""Get list of trips respecting current authentication status."""
trips = [
trip
for trip in agenda.trip.build_trip_list(route_distances=route_distances)
if flask.g.user.is_authenticated or not trip.private
]
# Add Schengen compliance information to each trip
for trip in trips:
agenda.trip_schengen.add_schengen_compliance_to_trip(trip)
return trips
@app.route("/trip")
def trip_list() -> werkzeug.Response:
"""Trip list to redirect to future trip list."""
@ -477,15 +448,17 @@ def sum_distances_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]
return list(distances_by_transport_type.items())
def get_trip_list() -> list[Trip]:
"""Get trip list with route distances."""
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
return agenda.trip.get_trip_list(route_distances)
@app.route("/trip/past")
def trip_past_list() -> str:
"""Page showing a list of past trips."""
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
trip_list = get_trip_list(route_distances)
today = date.today()
past = [item for item in trip_list if (item.end or item.start) < today]
past = [item for item in get_trip_list() if (item.end or item.start) < today]
coordinates, routes = agenda.trip.get_coordinates_and_routes(past)
return flask.render_template(
@ -507,8 +480,7 @@ def trip_past_list() -> str:
@app.route("/trip/future")
def trip_future_list() -> str:
"""Page showing a list of future trips."""
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
trip_list = get_trip_list(route_distances)
trip_list = get_trip_list()
today = date.today()
current = [
@ -540,7 +512,7 @@ def trip_future_list() -> str:
@app.route("/trip/text")
def trip_list_text() -> str:
"""Page showing a list of trips."""
trip_list = get_trip_list()
trip_list = agenda.trip.get_trip_list()
today = date.today()
future = [item for item in trip_list if item.start > today]
@ -573,8 +545,7 @@ def get_prev_current_and_next_trip(
@app.route("/trip/<start>")
def trip_page(start: str) -> str:
"""Individual trip page."""
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
trip_list = get_trip_list(route_distances)
trip_list = get_trip_list()
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
if not trip:
@ -616,8 +587,7 @@ def trip_debug_page(start: str) -> str:
if not flask.g.user.is_authenticated:
flask.abort(401)
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
trip_list = get_trip_list(route_distances)
trip_list = get_trip_list()
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
if not trip:
@ -713,8 +683,7 @@ def birthday_list() -> str:
@app.route("/trip/stats")
def trip_stats() -> str:
"""Travel stats: distance and price by year and travel type."""
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
trip_list = get_trip_list(route_distances)
trip_list = get_trip_list()
conferences = sum(len(item.conferences) for item in trip_list)