Move two functions to agenda/trip.py
This commit is contained in:
parent
f718535624
commit
b4126d04f8
4 changed files with 64 additions and 63 deletions
|
|
@ -2,8 +2,8 @@
|
||||||
|
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
|
|
||||||
from .trip import depart_datetime
|
|
||||||
from .types import SchengenCalculation, SchengenStay, StrDict
|
from .types import SchengenCalculation, SchengenStay, StrDict
|
||||||
|
from .utils import depart_datetime
|
||||||
|
|
||||||
# Schengen Area countries as of 2025
|
# Schengen Area countries as of 2025
|
||||||
SCHENGEN_COUNTRIES = {
|
SCHENGEN_COUNTRIES = {
|
||||||
|
|
|
||||||
|
|
@ -3,14 +3,14 @@
|
||||||
import decimal
|
import decimal
|
||||||
import os
|
import os
|
||||||
import typing
|
import typing
|
||||||
from datetime import date, datetime, time
|
from datetime import date
|
||||||
from zoneinfo import ZoneInfo
|
|
||||||
|
|
||||||
import flask
|
import flask
|
||||||
import yaml
|
import yaml
|
||||||
|
|
||||||
from agenda import travel
|
from agenda import travel, trip_schengen
|
||||||
from agenda.types import StrDict, Trip
|
from agenda.types import StrDict, Trip
|
||||||
|
from agenda.utils import depart_datetime
|
||||||
|
|
||||||
|
|
||||||
class Airline(typing.TypedDict, total=False):
|
class Airline(typing.TypedDict, total=False):
|
||||||
|
|
@ -122,18 +122,6 @@ def load_coaches(
|
||||||
return 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(
|
def process_flight(
|
||||||
flight: StrDict, by_iata: dict[str, Airline], airports: list[StrDict]
|
flight: StrDict, by_iata: dict[str, Airline], airports: list[StrDict]
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|
@ -502,3 +490,33 @@ def get_coordinates_and_routes(
|
||||||
route["geojson"] = read_geojson(data_dir, route.pop("geojson_filename"))
|
route["geojson"] = read_geojson(data_dir, route.pop("geojson_filename"))
|
||||||
|
|
||||||
return (coordinates, routes)
|
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
|
||||||
|
|
|
||||||
|
|
@ -2,8 +2,10 @@
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import typing
|
import typing
|
||||||
from datetime import date, datetime, timedelta, timezone
|
from datetime import date, datetime, time, timedelta, timezone
|
||||||
from time import time
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
|
from .types import StrDict
|
||||||
|
|
||||||
|
|
||||||
def as_date(d: datetime | date) -> date:
|
def as_date(d: datetime | date) -> date:
|
||||||
|
|
@ -118,3 +120,15 @@ async def time_function(
|
||||||
exception = e
|
exception = e
|
||||||
end_time = time()
|
end_time = time()
|
||||||
return name, result, end_time - start_time, exception
|
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"))
|
||||||
|
|
|
||||||
59
web_view.py
59
web_view.py
|
|
@ -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 = list(traceback.walk_tb(current_traceback))[-1][0]
|
||||||
last_frame_args = inspect.getargs(last_frame.f_code)
|
last_frame_args = inspect.getargs(last_frame.f_code)
|
||||||
|
assert tb._te.exc_type
|
||||||
|
|
||||||
return (
|
return (
|
||||||
flask.render_template(
|
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("/")
|
@app.route("/")
|
||||||
async def index() -> str:
|
async def index() -> str:
|
||||||
"""Index page."""
|
"""Index page."""
|
||||||
|
|
@ -106,7 +94,7 @@ async def index() -> str:
|
||||||
today=today,
|
today=today,
|
||||||
events=events,
|
events=events,
|
||||||
get_country=agenda.get_country,
|
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),
|
start_event_list=date.today() - timedelta(days=1),
|
||||||
end_event_list=date.today() + timedelta(days=365 * 2),
|
end_event_list=date.today() + timedelta(days=365 * 2),
|
||||||
render_time=(time.time() - t0),
|
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")
|
@app.route("/trip")
|
||||||
def trip_list() -> werkzeug.Response:
|
def trip_list() -> werkzeug.Response:
|
||||||
"""Trip list to redirect to future trip list."""
|
"""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())
|
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")
|
@app.route("/trip/past")
|
||||||
def trip_past_list() -> str:
|
def trip_past_list() -> str:
|
||||||
"""Page showing a list of past trips."""
|
"""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()
|
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)
|
coordinates, routes = agenda.trip.get_coordinates_and_routes(past)
|
||||||
|
|
||||||
return flask.render_template(
|
return flask.render_template(
|
||||||
|
|
@ -507,8 +480,7 @@ def trip_past_list() -> str:
|
||||||
@app.route("/trip/future")
|
@app.route("/trip/future")
|
||||||
def trip_future_list() -> str:
|
def trip_future_list() -> str:
|
||||||
"""Page showing a list of future trips."""
|
"""Page showing a list of future trips."""
|
||||||
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
|
trip_list = get_trip_list()
|
||||||
trip_list = get_trip_list(route_distances)
|
|
||||||
today = date.today()
|
today = date.today()
|
||||||
|
|
||||||
current = [
|
current = [
|
||||||
|
|
@ -540,7 +512,7 @@ def trip_future_list() -> str:
|
||||||
@app.route("/trip/text")
|
@app.route("/trip/text")
|
||||||
def trip_list_text() -> str:
|
def trip_list_text() -> str:
|
||||||
"""Page showing a list of trips."""
|
"""Page showing a list of trips."""
|
||||||
trip_list = get_trip_list()
|
trip_list = agenda.trip.get_trip_list()
|
||||||
today = date.today()
|
today = date.today()
|
||||||
future = [item for item in trip_list if item.start > 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>")
|
@app.route("/trip/<start>")
|
||||||
def trip_page(start: str) -> str:
|
def trip_page(start: str) -> str:
|
||||||
"""Individual trip page."""
|
"""Individual trip page."""
|
||||||
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
|
trip_list = get_trip_list()
|
||||||
trip_list = get_trip_list(route_distances)
|
|
||||||
|
|
||||||
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
|
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
|
||||||
if not trip:
|
if not trip:
|
||||||
|
|
@ -616,8 +587,7 @@ def trip_debug_page(start: str) -> str:
|
||||||
if not flask.g.user.is_authenticated:
|
if not flask.g.user.is_authenticated:
|
||||||
flask.abort(401)
|
flask.abort(401)
|
||||||
|
|
||||||
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
|
trip_list = get_trip_list()
|
||||||
trip_list = get_trip_list(route_distances)
|
|
||||||
|
|
||||||
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
|
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
|
||||||
if not trip:
|
if not trip:
|
||||||
|
|
@ -713,8 +683,7 @@ def birthday_list() -> str:
|
||||||
@app.route("/trip/stats")
|
@app.route("/trip/stats")
|
||||||
def trip_stats() -> str:
|
def trip_stats() -> str:
|
||||||
"""Travel stats: distance and price by year and travel type."""
|
"""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()
|
||||||
trip_list = get_trip_list(route_distances)
|
|
||||||
|
|
||||||
conferences = sum(len(item.conferences) for item in trip_list)
|
conferences = sum(len(item.conferences) for item in trip_list)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue