diff --git a/agenda/schengen.py b/agenda/schengen.py
index 7f1d355..016cc01 100644
--- a/agenda/schengen.py
+++ b/agenda/schengen.py
@@ -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 = {
diff --git a/agenda/trip.py b/agenda/trip.py
index 9aad250..978434d 100644
--- a/agenda/trip.py
+++ b/agenda/trip.py
@@ -3,14 +3,14 @@
import decimal
import os
import typing
-from datetime import date
+from datetime import date, datetime, time
+from zoneinfo import ZoneInfo
import flask
import yaml
-from agenda import travel, trip_schengen
+from agenda import travel
from agenda.types import StrDict, Trip
-from agenda.utils import depart_datetime
class Airline(typing.TypedDict, total=False):
@@ -122,6 +122,18 @@ 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:
@@ -397,9 +409,7 @@ 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(
{
@@ -414,9 +424,7 @@ 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(
{
@@ -436,9 +444,7 @@ 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),
}
)
@@ -490,33 +496,3 @@ 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
diff --git a/agenda/trip_schengen.py b/agenda/trip_schengen.py
index 6e34044..8d3cd8b 100644
--- a/agenda/trip_schengen.py
+++ b/agenda/trip_schengen.py
@@ -7,37 +7,26 @@ from datetime import date, timedelta
import flask
from . import get_country, trip
-from .schengen import (
- SCHENGEN_COUNTRIES,
- calculate_schengen_time,
- extract_schengen_stays_from_travel,
-)
+from .schengen import calculate_schengen_time, extract_schengen_stays_from_travel
from .types import SchengenCalculation, SchengenStay, StrDict, 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:
+def add_schengen_compliance_to_trip(trip_obj: 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.travel)
+ calculation = calculate_schengen_time(trip_obj.travel)
# Add the calculation to the trip object
- trip.schengen_compliance = calculation
+ trip_obj.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.start}: {e}"
+ f"Failed to calculate Schengen compliance for trip {trip_obj.start}: {e}"
)
- trip.schengen_compliance = None
+ trip_obj.schengen_compliance = None
- return trip
+ return trip_obj
def get_schengen_compliance_for_all_trips(
@@ -138,9 +127,7 @@ 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 for trip in trip.build_trip_list(data_dir) if trip_includes_schengen(trip)
- ]
+ trip_list = trip.build_trip_list(data_dir)
# Calculate current compliance with trip information
all_travel_items = []
diff --git a/agenda/types.py b/agenda/types.py
index d6041ed..d1ecdda 100644
--- a/agenda/types.py
+++ b/agenda/types.py
@@ -55,43 +55,6 @@ 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."""
@@ -104,7 +67,7 @@ class Trip:
flight_bookings: list[StrDict] = field(default_factory=list)
name: str | None = None
private: bool = False
- schengen_compliance: SchengenCalculation | None = None
+ schengen_compliance: typing.Optional["SchengenCalculation"] = None
@property
def title(self) -> str:
@@ -446,3 +409,39 @@ 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)
diff --git a/agenda/utils.py b/agenda/utils.py
index 15c9100..5f8573c 100644
--- a/agenda/utils.py
+++ b/agenda/utils.py
@@ -2,10 +2,8 @@
import os
import typing
-from datetime import date, datetime, time, timedelta, timezone
-from zoneinfo import ZoneInfo
-
-from .types import StrDict
+from datetime import date, datetime, timedelta, timezone
+from time import time
def as_date(d: datetime | date) -> date:
@@ -120,15 +118,3 @@ 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"))
diff --git a/templates/schengen_report.html b/templates/schengen_report.html
index b34c02d..9222785 100644
--- a/templates/schengen_report.html
+++ b/templates/schengen_report.html
@@ -162,8 +162,8 @@
| {{ trip_date.strftime('%Y-%m-%d') }} |
- {% for trip in trip_list if trip.start == trip_date %}
- {{ trip.title }} {{ trip.country_flags }}
+ {% for trip_obj in trip_list if trip_obj.start == trip_date %}
+ {{ trip_obj.title }}
{% endfor %}
|
diff --git a/templates/trip/stats.html b/templates/trip/stats.html
index 883cb0d..b31cb58 100644
--- a/templates/trip/stats.html
+++ b/templates/trip/stats.html
@@ -20,7 +20,7 @@
{% endfor %}
- {% for year, year_stats in yearly_stats | dictsort(reverse=True) %}
+ {% for year, year_stats in yearly_stats | dictsort %}
{% set countries = year_stats.countries | sort(attribute="name") %}
{{ year }}
Trips in {{ year }}: {{ year_stats.count }}
diff --git a/web_view.py b/web_view.py
index 8e41aef..bc04d08 100755
--- a/web_view.py
+++ b/web_view.py
@@ -57,7 +57,6 @@ 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(
@@ -73,6 +72,19 @@ 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."""
@@ -94,7 +106,7 @@ async def index() -> str:
today=today,
events=events,
get_country=agenda.get_country,
- current_trip=agenda.trip.get_current_trip(today),
+ current_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),
@@ -417,6 +429,23 @@ 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."""
@@ -448,17 +477,15 @@ 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 get_trip_list() if (item.end or item.start) < today]
+
+ past = [item for item in trip_list if (item.end or item.start) < today]
+
coordinates, routes = agenda.trip.get_coordinates_and_routes(past)
return flask.render_template(
@@ -480,7 +507,8 @@ def trip_past_list() -> str:
@app.route("/trip/future")
def trip_future_list() -> str:
"""Page showing a list of future trips."""
- trip_list = get_trip_list()
+ route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
+ trip_list = get_trip_list(route_distances)
today = date.today()
current = [
@@ -512,7 +540,7 @@ def trip_future_list() -> str:
@app.route("/trip/text")
def trip_list_text() -> str:
"""Page showing a list of trips."""
- trip_list = agenda.trip.get_trip_list()
+ trip_list = get_trip_list()
today = date.today()
future = [item for item in trip_list if item.start > today]
@@ -545,7 +573,8 @@ def get_prev_current_and_next_trip(
@app.route("/trip/")
def trip_page(start: str) -> str:
"""Individual trip page."""
- trip_list = get_trip_list()
+ route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
+ trip_list = get_trip_list(route_distances)
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
if not trip:
@@ -587,7 +616,8 @@ def trip_debug_page(start: str) -> str:
if not flask.g.user.is_authenticated:
flask.abort(401)
- trip_list = get_trip_list()
+ route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
+ trip_list = get_trip_list(route_distances)
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
if not trip:
@@ -683,7 +713,8 @@ def birthday_list() -> str:
@app.route("/trip/stats")
def trip_stats() -> str:
"""Travel stats: distance and price by year and travel type."""
- trip_list = get_trip_list()
+ route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
+ trip_list = get_trip_list(route_distances)
conferences = sum(len(item.conferences) for item in trip_list)