Compare commits
No commits in common. "3747e83adeba3c99f69c6e14ef748d80a4cbccd6" and "442ae9846377d02eb710baa8ce504bc2cb2a306c" have entirely different histories.
3747e83ade
...
442ae98463
16 changed files with 30 additions and 1216 deletions
|
|
@ -1,474 +0,0 @@
|
||||||
"""Import recorded car journeys from a Google Timeline export."""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import argparse
|
|
||||||
import json
|
|
||||||
import typing
|
|
||||||
from collections import defaultdict
|
|
||||||
from dataclasses import dataclass
|
|
||||||
from datetime import date, datetime, timedelta
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from geopy.distance import geodesic # type: ignore
|
|
||||||
|
|
||||||
from . import car_journey_yaml, trip as trip_module, utils
|
|
||||||
|
|
||||||
PERSONAL_DATA_DIR = Path("~/src/personal-data").expanduser()
|
|
||||||
VEHICLE_ACTIVITY_TYPES = frozenset({"IN_PASSENGER_VEHICLE", "IN_VEHICLE"})
|
|
||||||
ACTIVITY_SPLIT_GAP = timedelta(hours=2)
|
|
||||||
|
|
||||||
LatLon = tuple[float, float]
|
|
||||||
LonLat = tuple[float, float]
|
|
||||||
StrDict = dict[str, typing.Any]
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
|
||||||
class TimelinePoint:
|
|
||||||
"""A single Google Timeline breadcrumb point."""
|
|
||||||
|
|
||||||
time: datetime
|
|
||||||
location: LatLon
|
|
||||||
|
|
||||||
@property
|
|
||||||
def lonlat(self) -> LonLat:
|
|
||||||
"""Return point as GeoJSON lon/lat."""
|
|
||||||
return (self.location[1], self.location[0])
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
|
||||||
class VehicleActivity:
|
|
||||||
"""A vehicle activity segment plus its breadcrumb points."""
|
|
||||||
|
|
||||||
index: int
|
|
||||||
mode: str
|
|
||||||
start_time: datetime
|
|
||||||
end_time: datetime
|
|
||||||
start: LatLon
|
|
||||||
end: LatLon
|
|
||||||
points: tuple[TimelinePoint, ...]
|
|
||||||
|
|
||||||
@property
|
|
||||||
def path(self) -> tuple[LatLon, ...]:
|
|
||||||
"""Return a deduplicated path."""
|
|
||||||
coords = [point.location for point in self.points]
|
|
||||||
if not coords:
|
|
||||||
coords = [self.start, self.end]
|
|
||||||
elif coords[0] != self.start:
|
|
||||||
coords.insert(0, self.start)
|
|
||||||
if coords[-1] != self.end:
|
|
||||||
coords.append(self.end)
|
|
||||||
return tuple(dedupe_consecutive(coords))
|
|
||||||
|
|
||||||
@property
|
|
||||||
def depart_time(self) -> datetime:
|
|
||||||
"""Prefer the first breadcrumb time when available."""
|
|
||||||
return self.points[0].time if self.points else self.start_time
|
|
||||||
|
|
||||||
@property
|
|
||||||
def arrive_time(self) -> datetime:
|
|
||||||
"""Prefer the last breadcrumb time when available."""
|
|
||||||
return self.points[-1].time if self.points else self.end_time
|
|
||||||
|
|
||||||
@property
|
|
||||||
def leading_path(self) -> tuple[LatLon, ...]:
|
|
||||||
"""Return the first few coordinates for start matching."""
|
|
||||||
return self.path[: min(3, len(self.path))]
|
|
||||||
|
|
||||||
@property
|
|
||||||
def trailing_path(self) -> tuple[LatLon, ...]:
|
|
||||||
"""Return the last few coordinates for end matching."""
|
|
||||||
return self.path[-min(3, len(self.path)) :]
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
|
||||||
class TimelineCarJourney:
|
|
||||||
"""A YAML-ready car journey row with a GeoJSON path."""
|
|
||||||
|
|
||||||
trip: date
|
|
||||||
depart: datetime
|
|
||||||
arrive: datetime
|
|
||||||
route: str
|
|
||||||
from_label: str
|
|
||||||
to_label: str
|
|
||||||
path: tuple[LatLon, ...]
|
|
||||||
mode: str
|
|
||||||
|
|
||||||
def as_yaml_item(self) -> StrDict:
|
|
||||||
"""Return the YAML mapping."""
|
|
||||||
return {
|
|
||||||
"trip": self.trip,
|
|
||||||
"depart": self.depart,
|
|
||||||
"arrive": self.arrive,
|
|
||||||
"route": self.route,
|
|
||||||
"from": self.from_label,
|
|
||||||
"to": self.to_label,
|
|
||||||
}
|
|
||||||
|
|
||||||
def as_geojson(self) -> StrDict:
|
|
||||||
"""Return a GeoJSON feature for this journey."""
|
|
||||||
return {
|
|
||||||
"type": "Feature",
|
|
||||||
"properties": {
|
|
||||||
"source": "google_timeline",
|
|
||||||
"mode": self.mode,
|
|
||||||
"start_time": self.depart.isoformat(),
|
|
||||||
"end_time": self.arrive.isoformat(),
|
|
||||||
},
|
|
||||||
"geometry": {
|
|
||||||
"type": "LineString",
|
|
||||||
"coordinates": [[lon, lat] for lat, lon in self.path],
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def parse_latlng(value: str) -> LatLon:
|
|
||||||
"""Parse Google's ``lat°, lon°`` format."""
|
|
||||||
latitude, longitude = value.replace("°", "").split(",", 1)
|
|
||||||
return (float(latitude.strip()), float(longitude.strip()))
|
|
||||||
|
|
||||||
|
|
||||||
def days_spanned(start: datetime, end: datetime) -> list[date]:
|
|
||||||
"""Return all dates touched by a datetime range."""
|
|
||||||
day = start.date()
|
|
||||||
days = [day]
|
|
||||||
while day < end.date():
|
|
||||||
day += timedelta(days=1)
|
|
||||||
days.append(day)
|
|
||||||
return days
|
|
||||||
|
|
||||||
|
|
||||||
def dedupe_consecutive(coords: list[LatLon]) -> list[LatLon]:
|
|
||||||
"""Drop repeated adjacent coordinates."""
|
|
||||||
deduped: list[LatLon] = []
|
|
||||||
for coord in coords:
|
|
||||||
if deduped and deduped[-1] == coord:
|
|
||||||
continue
|
|
||||||
deduped.append(coord)
|
|
||||||
return deduped
|
|
||||||
|
|
||||||
|
|
||||||
def load_timeline_path_points(data: StrDict) -> dict[date, list[TimelinePoint]]:
|
|
||||||
"""Group all timelinePath points by local date."""
|
|
||||||
points_by_day: dict[date, list[TimelinePoint]] = defaultdict(list)
|
|
||||||
for segment in data.get("semanticSegments", []):
|
|
||||||
timeline_path = segment.get("timelinePath")
|
|
||||||
if not isinstance(timeline_path, list):
|
|
||||||
continue
|
|
||||||
for item in timeline_path:
|
|
||||||
if not isinstance(item, dict):
|
|
||||||
continue
|
|
||||||
point = item.get("point")
|
|
||||||
time_value = item.get("time")
|
|
||||||
if not isinstance(point, str) or not isinstance(time_value, str):
|
|
||||||
continue
|
|
||||||
breadcrumb = TimelinePoint(
|
|
||||||
time=datetime.fromisoformat(time_value),
|
|
||||||
location=parse_latlng(point),
|
|
||||||
)
|
|
||||||
points_by_day[breadcrumb.time.date()].append(breadcrumb)
|
|
||||||
for breadcrumbs in points_by_day.values():
|
|
||||||
breadcrumbs.sort(key=lambda item: item.time)
|
|
||||||
return points_by_day
|
|
||||||
|
|
||||||
|
|
||||||
def activity_points(
|
|
||||||
start: datetime,
|
|
||||||
end: datetime,
|
|
||||||
points_by_day: dict[date, list[TimelinePoint]],
|
|
||||||
) -> tuple[TimelinePoint, ...]:
|
|
||||||
"""Return breadcrumb points within an activity window."""
|
|
||||||
selected: list[TimelinePoint] = []
|
|
||||||
for day in days_spanned(start, end):
|
|
||||||
for point in points_by_day.get(day, []):
|
|
||||||
if start <= point.time <= end:
|
|
||||||
selected.append(point)
|
|
||||||
selected.sort(key=lambda item: item.time)
|
|
||||||
return tuple(selected)
|
|
||||||
|
|
||||||
|
|
||||||
def load_vehicle_activities(timeline_path: Path) -> list[VehicleActivity]:
|
|
||||||
"""Load vehicle activities and attach breadcrumb points."""
|
|
||||||
data = typing.cast(StrDict, json.loads(timeline_path.read_text()))
|
|
||||||
points_by_day = load_timeline_path_points(data)
|
|
||||||
activities: list[VehicleActivity] = []
|
|
||||||
next_index = 0
|
|
||||||
for index, segment in enumerate(data.get("semanticSegments", [])):
|
|
||||||
activity = segment.get("activity")
|
|
||||||
if not isinstance(activity, dict):
|
|
||||||
continue
|
|
||||||
mode = activity.get("topCandidate", {}).get("type")
|
|
||||||
start_location = activity.get("start", {}).get("latLng")
|
|
||||||
end_location = activity.get("end", {}).get("latLng")
|
|
||||||
start_time = segment.get("startTime")
|
|
||||||
end_time = segment.get("endTime")
|
|
||||||
if (
|
|
||||||
mode not in VEHICLE_ACTIVITY_TYPES
|
|
||||||
or not isinstance(start_location, str)
|
|
||||||
or not isinstance(end_location, str)
|
|
||||||
or not isinstance(start_time, str)
|
|
||||||
or not isinstance(end_time, str)
|
|
||||||
):
|
|
||||||
continue
|
|
||||||
|
|
||||||
depart = datetime.fromisoformat(start_time)
|
|
||||||
arrive = datetime.fromisoformat(end_time)
|
|
||||||
start_latlon = parse_latlng(start_location)
|
|
||||||
end_latlon = parse_latlng(end_location)
|
|
||||||
activity = VehicleActivity(
|
|
||||||
index=index,
|
|
||||||
mode=typing.cast(str, mode),
|
|
||||||
start_time=depart,
|
|
||||||
end_time=arrive,
|
|
||||||
start=start_latlon,
|
|
||||||
end=end_latlon,
|
|
||||||
points=activity_points(
|
|
||||||
depart,
|
|
||||||
arrive,
|
|
||||||
points_by_day,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
for split_activity in split_activity_on_gaps(activity):
|
|
||||||
activities.append(
|
|
||||||
VehicleActivity(
|
|
||||||
index=next_index,
|
|
||||||
mode=split_activity.mode,
|
|
||||||
start_time=split_activity.start_time,
|
|
||||||
end_time=split_activity.end_time,
|
|
||||||
start=split_activity.start,
|
|
||||||
end=split_activity.end,
|
|
||||||
points=split_activity.points,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
next_index += 1
|
|
||||||
return activities
|
|
||||||
|
|
||||||
|
|
||||||
def activity_matches_date(activity: VehicleActivity, target: date) -> bool:
|
|
||||||
"""Return whether an activity overlaps a date."""
|
|
||||||
return target in days_spanned(activity.start_time, activity.end_time)
|
|
||||||
|
|
||||||
|
|
||||||
def split_activity_on_gaps(activity: VehicleActivity) -> list[VehicleActivity]:
|
|
||||||
"""Split an activity when breadcrumbs contain long pauses."""
|
|
||||||
if len(activity.points) < 2:
|
|
||||||
return [activity]
|
|
||||||
|
|
||||||
chunks: list[list[TimelinePoint]] = [[]]
|
|
||||||
for point in activity.points:
|
|
||||||
current = chunks[-1]
|
|
||||||
if current and point.time - current[-1].time > ACTIVITY_SPLIT_GAP:
|
|
||||||
chunks.append([])
|
|
||||||
current = chunks[-1]
|
|
||||||
current.append(point)
|
|
||||||
|
|
||||||
if len(chunks) == 1:
|
|
||||||
return [activity]
|
|
||||||
|
|
||||||
split: list[VehicleActivity] = []
|
|
||||||
for chunk in chunks:
|
|
||||||
split.append(
|
|
||||||
VehicleActivity(
|
|
||||||
index=activity.index,
|
|
||||||
mode=activity.mode,
|
|
||||||
start_time=chunk[0].time,
|
|
||||||
end_time=chunk[-1].time,
|
|
||||||
start=chunk[0].location,
|
|
||||||
end=chunk[-1].location,
|
|
||||||
points=tuple(chunk),
|
|
||||||
)
|
|
||||||
)
|
|
||||||
return split
|
|
||||||
|
|
||||||
|
|
||||||
def distance_to_coords_km(coords: tuple[LatLon, ...], target: LatLon) -> float:
|
|
||||||
"""Return the closest coordinate distance to a target coordinate."""
|
|
||||||
return min(float(geodesic(coord, target).km) for coord in coords)
|
|
||||||
|
|
||||||
|
|
||||||
def select_activity_slice(
|
|
||||||
activities: list[VehicleActivity],
|
|
||||||
start_target: LatLon,
|
|
||||||
end_target: LatLon,
|
|
||||||
) -> list[VehicleActivity]:
|
|
||||||
"""Select the contiguous activity slice that best matches a planned leg."""
|
|
||||||
if not activities:
|
|
||||||
raise ValueError("no vehicle activities available for route matching")
|
|
||||||
|
|
||||||
best_slice: tuple[float, int, int] | None = None
|
|
||||||
for start_index, activity in enumerate(activities):
|
|
||||||
start_distance = distance_to_coords_km(activity.leading_path, start_target)
|
|
||||||
for end_index in range(start_index, len(activities)):
|
|
||||||
end_distance = distance_to_coords_km(
|
|
||||||
activities[end_index].trailing_path,
|
|
||||||
end_target,
|
|
||||||
)
|
|
||||||
score = start_distance + end_distance
|
|
||||||
if (
|
|
||||||
best_slice is None
|
|
||||||
or score < best_slice[0]
|
|
||||||
or (
|
|
||||||
score == best_slice[0]
|
|
||||||
and end_index - start_index < best_slice[2] - best_slice[1]
|
|
||||||
)
|
|
||||||
):
|
|
||||||
best_slice = (score, start_index, end_index)
|
|
||||||
|
|
||||||
assert best_slice is not None
|
|
||||||
return activities[best_slice[1] : best_slice[2] + 1]
|
|
||||||
|
|
||||||
|
|
||||||
def stop_labels(
|
|
||||||
from_label: str,
|
|
||||||
to_label: str,
|
|
||||||
count: int,
|
|
||||||
) -> list[tuple[str, str]]:
|
|
||||||
"""Return start/end labels for a split route."""
|
|
||||||
if count == 1:
|
|
||||||
return [(from_label, to_label)]
|
|
||||||
labels: list[tuple[str, str]] = []
|
|
||||||
for index in range(count):
|
|
||||||
start = from_label if index == 0 else f"Drive stop {index}"
|
|
||||||
end = to_label if index == count - 1 else f"Drive stop {index + 1}"
|
|
||||||
labels.append((start, end))
|
|
||||||
return labels
|
|
||||||
|
|
||||||
|
|
||||||
def route_name(activity: VehicleActivity) -> str:
|
|
||||||
"""Return a stable route filename stem for an imported activity."""
|
|
||||||
return "timeline_" + activity.depart_time.strftime("%Y%m%d_%H%M%S")
|
|
||||||
|
|
||||||
|
|
||||||
def build_timeline_journeys(
|
|
||||||
timeline_path: Path,
|
|
||||||
data_dir: Path,
|
|
||||||
today: date | None = None,
|
|
||||||
) -> list[TimelineCarJourney]:
|
|
||||||
"""Generate split past car journeys from timeline data."""
|
|
||||||
if today is None:
|
|
||||||
today = date.today()
|
|
||||||
|
|
||||||
activities = load_vehicle_activities(timeline_path)
|
|
||||||
existing = car_journey_yaml.load_yaml_list(data_dir / "car_journeys.yaml")
|
|
||||||
generated: list[TimelineCarJourney] = []
|
|
||||||
|
|
||||||
for item in existing:
|
|
||||||
trip_date = utils.as_date(item["trip"])
|
|
||||||
if trip_date >= today:
|
|
||||||
continue
|
|
||||||
|
|
||||||
route = item.get("route")
|
|
||||||
if not isinstance(route, str):
|
|
||||||
raise ValueError(f"car journey route must be a string: {item!r}")
|
|
||||||
|
|
||||||
route_path = car_journey_yaml.car_route_path(data_dir, route)
|
|
||||||
existing_geojson = car_journey_yaml.read_geojson(route_path)
|
|
||||||
endpoints = trip_module.geojson_route_endpoints(existing_geojson)
|
|
||||||
if endpoints is None:
|
|
||||||
raise ValueError(f"could not read route endpoints from {route_path}")
|
|
||||||
|
|
||||||
from_label, to_label = trip_module.car_route_labels(item)
|
|
||||||
if not isinstance(from_label, str) or not isinstance(to_label, str):
|
|
||||||
raise ValueError(f"could not determine car route labels for {item!r}")
|
|
||||||
|
|
||||||
journey_date = utils.as_date(item["depart"])
|
|
||||||
matches = [
|
|
||||||
activity
|
|
||||||
for activity in activities
|
|
||||||
if activity_matches_date(activity, journey_date)
|
|
||||||
]
|
|
||||||
selected = select_activity_slice(matches, endpoints[0], endpoints[1])
|
|
||||||
labels = stop_labels(from_label, to_label, len(selected))
|
|
||||||
for activity, (start_label, end_label) in zip(selected, labels, strict=True):
|
|
||||||
generated.append(
|
|
||||||
TimelineCarJourney(
|
|
||||||
trip=trip_date,
|
|
||||||
depart=activity.depart_time,
|
|
||||||
arrive=activity.arrive_time,
|
|
||||||
route=route_name(activity),
|
|
||||||
from_label=start_label,
|
|
||||||
to_label=end_label,
|
|
||||||
path=activity.path,
|
|
||||||
mode=activity.mode,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
return generated
|
|
||||||
|
|
||||||
|
|
||||||
def write_timeline_routes(
|
|
||||||
data_dir: Path,
|
|
||||||
journeys: list[TimelineCarJourney],
|
|
||||||
) -> int:
|
|
||||||
"""Write GeoJSON files for generated journeys."""
|
|
||||||
route_dir = data_dir / "car_routes"
|
|
||||||
route_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
written = 0
|
|
||||||
for journey in journeys:
|
|
||||||
path = route_dir / f"{journey.route}.geojson"
|
|
||||||
path.write_text(json.dumps(journey.as_geojson(), separators=(",", ":")) + "\n")
|
|
||||||
written += 1
|
|
||||||
return written
|
|
||||||
|
|
||||||
|
|
||||||
def rewrite_car_journeys_yaml(
|
|
||||||
data_dir: Path,
|
|
||||||
journeys: list[TimelineCarJourney],
|
|
||||||
today: date | None = None,
|
|
||||||
) -> tuple[int, int]:
|
|
||||||
"""Replace past YAML rows with timeline-derived journeys."""
|
|
||||||
if today is None:
|
|
||||||
today = date.today()
|
|
||||||
|
|
||||||
path = data_dir / "car_journeys.yaml"
|
|
||||||
existing = car_journey_yaml.load_yaml_list(path)
|
|
||||||
future_rows = [item for item in existing if utils.as_date(item["trip"]) >= today]
|
|
||||||
generated_rows = [journey.as_yaml_item() for journey in journeys]
|
|
||||||
combined = generated_rows + future_rows
|
|
||||||
combined.sort(key=car_journey_yaml.journey_key)
|
|
||||||
path.write_text(car_journey_yaml.dump_yaml_list_with_blank_lines(combined))
|
|
||||||
replaced = len(existing) - len(future_rows)
|
|
||||||
return (replaced, len(generated_rows))
|
|
||||||
|
|
||||||
|
|
||||||
def import_from_timeline(
|
|
||||||
timeline_path: Path,
|
|
||||||
data_dir: Path = PERSONAL_DATA_DIR,
|
|
||||||
today: date | None = None,
|
|
||||||
) -> tuple[int, int, int]:
|
|
||||||
"""Write route files and replace past YAML rows."""
|
|
||||||
journeys = build_timeline_journeys(timeline_path, data_dir, today=today)
|
|
||||||
routes_written = write_timeline_routes(data_dir, journeys)
|
|
||||||
rows_replaced, rows_written = rewrite_car_journeys_yaml(
|
|
||||||
data_dir, journeys, today=today
|
|
||||||
)
|
|
||||||
return (routes_written, rows_replaced, rows_written)
|
|
||||||
|
|
||||||
|
|
||||||
def parse_date(value: str) -> date:
|
|
||||||
"""Parse an ISO date."""
|
|
||||||
return date.fromisoformat(value)
|
|
||||||
|
|
||||||
|
|
||||||
def main(argv: list[str] | None = None) -> int:
|
|
||||||
"""CLI entrypoint."""
|
|
||||||
parser = argparse.ArgumentParser(
|
|
||||||
description=(
|
|
||||||
"Replace past car_journeys.yaml rows with routes split from a "
|
|
||||||
"Google Timeline export."
|
|
||||||
)
|
|
||||||
)
|
|
||||||
parser.add_argument("timeline", type=Path)
|
|
||||||
parser.add_argument("--data-dir", type=Path, default=PERSONAL_DATA_DIR)
|
|
||||||
parser.add_argument("--today", type=parse_date, default=None)
|
|
||||||
args = parser.parse_args(argv)
|
|
||||||
|
|
||||||
routes_written, rows_replaced, rows_written = import_from_timeline(
|
|
||||||
args.timeline,
|
|
||||||
data_dir=args.data_dir,
|
|
||||||
today=args.today,
|
|
||||||
)
|
|
||||||
print(f"Wrote {routes_written} route GeoJSON file(s)")
|
|
||||||
print(f"Replaced {rows_replaced} past car journey row(s)")
|
|
||||||
print(f"Wrote {rows_written} split timeline journey row(s)")
|
|
||||||
return 0
|
|
||||||
|
|
@ -9,16 +9,12 @@ import re
|
||||||
import typing
|
import typing
|
||||||
import unicodedata
|
import unicodedata
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from datetime import date, datetime
|
from datetime import date
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
import openrouteservice # type: ignore[import-untyped]
|
||||||
import yaml
|
import yaml
|
||||||
|
|
||||||
try:
|
|
||||||
import openrouteservice # type: ignore[import-not-found]
|
|
||||||
except ImportError: # pragma: no cover - optional dependency in tests
|
|
||||||
openrouteservice = None
|
|
||||||
|
|
||||||
from . import trip as trip_module
|
from . import trip as trip_module
|
||||||
from . import utils
|
from . import utils
|
||||||
|
|
||||||
|
|
@ -604,8 +600,6 @@ def car_journeys_for_trip(
|
||||||
|
|
||||||
def openrouteservice_fetcher(api_key: str) -> RouteFetcher:
|
def openrouteservice_fetcher(api_key: str) -> RouteFetcher:
|
||||||
"""Return a route fetcher backed by openrouteservice."""
|
"""Return a route fetcher backed by openrouteservice."""
|
||||||
if openrouteservice is None:
|
|
||||||
raise ValueError("openrouteservice must be installed")
|
|
||||||
client = openrouteservice.Client(key=api_key)
|
client = openrouteservice.Client(key=api_key)
|
||||||
|
|
||||||
def fetch(start: LonLat, end: LonLat) -> GeoJSON:
|
def fetch(start: LonLat, end: LonLat) -> GeoJSON:
|
||||||
|
|
@ -645,15 +639,15 @@ def write_route_files(
|
||||||
return written
|
return written
|
||||||
|
|
||||||
|
|
||||||
def journey_key(item: StrDict) -> tuple[date, datetime, datetime, str]:
|
def journey_key(item: StrDict) -> tuple[date, date, date, str]:
|
||||||
"""Return a stable key for duplicate detection and sorting."""
|
"""Return a stable key for duplicate detection and sorting."""
|
||||||
route = item.get("route")
|
route = item.get("route")
|
||||||
if not isinstance(route, str):
|
if not isinstance(route, str):
|
||||||
raise ValueError(f"car journey route must be a string: {item!r}")
|
raise ValueError(f"car journey route must be a string: {item!r}")
|
||||||
return (
|
return (
|
||||||
utils.as_date(item["trip"]),
|
utils.as_date(item["trip"]),
|
||||||
utils.as_datetime(item["depart"]),
|
utils.as_date(item["depart"]),
|
||||||
utils.as_datetime(item["arrive"]),
|
utils.as_date(item["arrive"]),
|
||||||
trip_module.route_filename_without_extension(route),
|
trip_module.route_filename_without_extension(route),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,15 +1,10 @@
|
||||||
"""Trip statistic functions."""
|
"""Trip statistic functions."""
|
||||||
|
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
import typing
|
from typing import Counter, Mapping
|
||||||
from typing import TYPE_CHECKING, Counter, Mapping
|
|
||||||
|
|
||||||
import agenda
|
import agenda
|
||||||
|
from agenda.types import StrDict, Trip, airport_label
|
||||||
if TYPE_CHECKING:
|
|
||||||
from agenda.types import Trip
|
|
||||||
|
|
||||||
StrDict = dict[str, typing.Any]
|
|
||||||
|
|
||||||
|
|
||||||
def travel_legs(trip: Trip, stats: StrDict) -> None:
|
def travel_legs(trip: Trip, stats: StrDict) -> None:
|
||||||
|
|
@ -23,8 +18,6 @@ def travel_legs(trip: Trip, stats: StrDict) -> None:
|
||||||
stats["co2_by_transport_type"].setdefault(transport_type, 0)
|
stats["co2_by_transport_type"].setdefault(transport_type, 0)
|
||||||
stats["co2_by_transport_type"][transport_type] += leg["co2_kg"]
|
stats["co2_by_transport_type"][transport_type] += leg["co2_kg"]
|
||||||
if leg["type"] == "flight":
|
if leg["type"] == "flight":
|
||||||
from agenda.types import airport_label
|
|
||||||
|
|
||||||
stats.setdefault("flight_count", 0)
|
stats.setdefault("flight_count", 0)
|
||||||
stats.setdefault("airlines", Counter())
|
stats.setdefault("airlines", Counter())
|
||||||
stats.setdefault("airports", Counter())
|
stats.setdefault("airports", Counter())
|
||||||
|
|
@ -76,8 +69,6 @@ def calculate_overall_stats(yearly_stats: dict[int, StrDict]) -> StrDict:
|
||||||
"stations": Counter(),
|
"stations": Counter(),
|
||||||
"flight_count": 0,
|
"flight_count": 0,
|
||||||
"train_count": 0,
|
"train_count": 0,
|
||||||
"co2_kg": 0.0,
|
|
||||||
"co2_by_transport_type": {},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
for year_stats in yearly_stats.values():
|
for year_stats in yearly_stats.values():
|
||||||
|
|
@ -89,12 +80,6 @@ def calculate_overall_stats(yearly_stats: dict[int, StrDict]) -> StrDict:
|
||||||
overall["stations"] += year_stats["stations"]
|
overall["stations"] += year_stats["stations"]
|
||||||
overall["flight_count"] += year_stats.get("flight_count", 0)
|
overall["flight_count"] += year_stats.get("flight_count", 0)
|
||||||
overall["train_count"] += year_stats.get("train_count", 0)
|
overall["train_count"] += year_stats.get("train_count", 0)
|
||||||
overall["co2_kg"] += year_stats.get("co2_kg", 0)
|
|
||||||
for transport_type, co2_kg in year_stats.get(
|
|
||||||
"co2_by_transport_type", {}
|
|
||||||
).items():
|
|
||||||
overall["co2_by_transport_type"].setdefault(transport_type, 0.0)
|
|
||||||
overall["co2_by_transport_type"][transport_type] += co2_kg
|
|
||||||
|
|
||||||
return overall
|
return overall
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -17,12 +17,6 @@ from agenda import conference, ical, travel, trip_schengen
|
||||||
from agenda.types import StrDict, Trip, TripElement
|
from agenda.types import StrDict, Trip, TripElement
|
||||||
from agenda.utils import as_date, as_datetime, depart_datetime
|
from agenda.utils import as_date, as_datetime, depart_datetime
|
||||||
|
|
||||||
TRAIN_CO2_KG_PER_KM = 0.037
|
|
||||||
COACH_CO2_KG_PER_KM = 0.027
|
|
||||||
FERRY_CO2_KG_PER_KM = 0.02254
|
|
||||||
BUS_CO2_KG_PER_KM = 0.1
|
|
||||||
CAR_CO2_KG_PER_KM = 0.218
|
|
||||||
|
|
||||||
|
|
||||||
class Airline(typing.TypedDict, total=False):
|
class Airline(typing.TypedDict, total=False):
|
||||||
"""Airline."""
|
"""Airline."""
|
||||||
|
|
@ -197,11 +191,13 @@ def load_trains(
|
||||||
if route_distances:
|
if route_distances:
|
||||||
travel.add_leg_route_distance(leg, route_distances)
|
travel.add_leg_route_distance(leg, route_distances)
|
||||||
|
|
||||||
|
# Calculate CO2 emissions for train leg (0.037 kg CO2e per passenger per km)
|
||||||
if "distance" in leg:
|
if "distance" in leg:
|
||||||
leg["co2_kg"] = leg["distance"] * TRAIN_CO2_KG_PER_KM
|
leg["co2_kg"] = leg["distance"] * 0.037
|
||||||
|
|
||||||
if all("distance" in leg for leg in train["legs"]):
|
if all("distance" in leg for leg in train["legs"]):
|
||||||
train["distance"] = sum(leg["distance"] for leg in train["legs"])
|
train["distance"] = sum(leg["distance"] for leg in train["legs"])
|
||||||
|
# Calculate total CO2 for entire train journey
|
||||||
train["co2_kg"] = sum(leg["co2_kg"] for leg in train["legs"])
|
train["co2_kg"] = sum(leg["co2_kg"] for leg in train["legs"])
|
||||||
|
|
||||||
return trains
|
return trains
|
||||||
|
|
@ -224,8 +220,9 @@ def load_ferries(
|
||||||
if route_distances:
|
if route_distances:
|
||||||
travel.add_leg_route_distance(item, route_distances)
|
travel.add_leg_route_distance(item, route_distances)
|
||||||
|
|
||||||
|
# Calculate CO2 emissions for ferry (0.02254 kg CO2e per passenger per km)
|
||||||
if "distance" in item:
|
if "distance" in item:
|
||||||
item["co2_kg"] = item["distance"] * FERRY_CO2_KG_PER_KM
|
item["co2_kg"] = item["distance"] * 0.02254
|
||||||
|
|
||||||
geojson = from_terminal["routes"].get(item["to"])
|
geojson = from_terminal["routes"].get(item["to"])
|
||||||
if geojson:
|
if geojson:
|
||||||
|
|
@ -273,12 +270,7 @@ def load_coaches(
|
||||||
) -> list[StrDict]:
|
) -> list[StrDict]:
|
||||||
"""Load coaches."""
|
"""Load coaches."""
|
||||||
return load_road_transport(
|
return load_road_transport(
|
||||||
"coach",
|
"coach", "coaches", "coach_stations", data_dir, 0.027, route_distances
|
||||||
"coaches",
|
|
||||||
"coach_stations",
|
|
||||||
data_dir,
|
|
||||||
COACH_CO2_KG_PER_KM,
|
|
||||||
route_distances,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -287,7 +279,7 @@ def load_buses(
|
||||||
) -> list[StrDict]:
|
) -> list[StrDict]:
|
||||||
"""Load buses."""
|
"""Load buses."""
|
||||||
return load_road_transport(
|
return load_road_transport(
|
||||||
"bus", "buses", "bus_stops", data_dir, BUS_CO2_KG_PER_KM, route_distances
|
"bus", "buses", "bus_stops", data_dir, 0.1, route_distances
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -427,8 +419,6 @@ def load_cars(data_dir: str) -> list[StrDict]:
|
||||||
|
|
||||||
if "distance" not in item:
|
if "distance" not in item:
|
||||||
item["distance"] = geojson_route_distance_km(geojson_data)
|
item["distance"] = geojson_route_distance_km(geojson_data)
|
||||||
if "distance" in item and "co2_kg" not in item:
|
|
||||||
item["co2_kg"] = item["distance"] * CAR_CO2_KG_PER_KM
|
|
||||||
|
|
||||||
endpoints = geojson_route_endpoints(geojson_data)
|
endpoints = geojson_route_endpoints(geojson_data)
|
||||||
from_label, to_label = car_route_labels(item)
|
from_label, to_label = car_route_labels(item)
|
||||||
|
|
|
||||||
|
|
@ -58,12 +58,6 @@ def airport_label(airport: StrDict) -> str:
|
||||||
return f"{name} ({airport['iata']})"
|
return f"{name} ({airport['iata']})"
|
||||||
|
|
||||||
|
|
||||||
def is_generated_drive_stop(label: str) -> bool:
|
|
||||||
"""Return whether a car stop label is synthetic timeline filler."""
|
|
||||||
prefix = "Drive stop "
|
|
||||||
return label.startswith(prefix) and label[len(prefix) :].isdigit()
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class SchengenStay:
|
class SchengenStay:
|
||||||
"""Represents a stay in the Schengen area."""
|
"""Represents a stay in the Schengen area."""
|
||||||
|
|
@ -136,13 +130,8 @@ class Trip:
|
||||||
if not (depart := (travel["depart"] and utils.as_date(travel["depart"]))):
|
if not (depart := (travel["depart"] and utils.as_date(travel["depart"]))):
|
||||||
continue
|
continue
|
||||||
for when, from_or_to in ((self.start, "from"), (self.end, "to")):
|
for when, from_or_to in ((self.start, "from"), (self.end, "to")):
|
||||||
if depart == when:
|
if depart != when and travel[from_or_to] not in titles:
|
||||||
continue
|
titles.append(travel[from_or_to])
|
||||||
label = travel.get(from_or_to)
|
|
||||||
if not isinstance(label, str) or is_generated_drive_stop(label):
|
|
||||||
continue
|
|
||||||
if label not in titles:
|
|
||||||
titles.append(label)
|
|
||||||
return titles
|
return titles
|
||||||
|
|
||||||
@property
|
@property
|
||||||
|
|
|
||||||
|
|
@ -6,7 +6,7 @@ from datetime import date, datetime, time, timedelta, timezone
|
||||||
from time import time as unixtime
|
from time import time as unixtime
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
StrDict = dict[str, typing.Any]
|
from .types import StrDict
|
||||||
|
|
||||||
|
|
||||||
def as_date(d: datetime | date) -> date:
|
def as_date(d: datetime | date) -> date:
|
||||||
|
|
|
||||||
|
|
@ -1,15 +0,0 @@
|
||||||
#!/usr/bin/python3
|
|
||||||
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
|
|
||||||
SCRIPT_PATH = os.path.realpath(__file__)
|
|
||||||
SCRIPT_DIR = os.path.dirname(SCRIPT_PATH)
|
|
||||||
REPO_ROOT = os.path.dirname(SCRIPT_DIR)
|
|
||||||
if REPO_ROOT not in sys.path:
|
|
||||||
sys.path.insert(0, REPO_ROOT)
|
|
||||||
|
|
||||||
from agenda.car_journey_timeline import main
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
raise SystemExit(main())
|
|
||||||
|
|
@ -212,28 +212,6 @@ tr.conf-hl > td {
|
||||||
<div class="container-fluid mt-2">
|
<div class="container-fluid mt-2">
|
||||||
<h1>Conferences</h1>
|
<h1>Conferences</h1>
|
||||||
|
|
||||||
{% if country_options %}
|
|
||||||
<form method="get" class="row g-2 align-items-end mb-3">
|
|
||||||
<div class="col-auto">
|
|
||||||
<label for="country-filter" class="form-label mb-1">Country</label>
|
|
||||||
<select id="country-filter" name="country" class="form-select form-select-sm" onchange="this.form.submit()">
|
|
||||||
<option value="">All countries</option>
|
|
||||||
{% for option in country_options %}
|
|
||||||
<option value="{{ option.code }}"{% if option.code == selected_country %} selected{% endif %}>
|
|
||||||
{{ option.flag }} {{ option.name }}
|
|
||||||
</option>
|
|
||||||
{% endfor %}
|
|
||||||
</select>
|
|
||||||
</div>
|
|
||||||
<div class="col-auto">
|
|
||||||
<button type="submit" class="btn btn-sm btn-primary">Filter</button>
|
|
||||||
{% if selected_country %}
|
|
||||||
<a class="btn btn-sm btn-outline-secondary" href="{{ url_for(request.endpoint) }}">Clear</a>
|
|
||||||
{% endif %}
|
|
||||||
</div>
|
|
||||||
</form>
|
|
||||||
{% endif %}
|
|
||||||
|
|
||||||
{{ render_timeline(timeline) }}
|
{{ render_timeline(timeline) }}
|
||||||
|
|
||||||
<table class="table table-sm table-hover align-middle">
|
<table class="table table-sm table-hover align-middle">
|
||||||
|
|
|
||||||
|
|
@ -6,14 +6,6 @@
|
||||||
|
|
||||||
{% block title %}{{ heading }} - Edward Betts{% endblock %}
|
{% block title %}{{ heading }} - Edward Betts{% endblock %}
|
||||||
|
|
||||||
{% block style %}
|
|
||||||
<style>
|
|
||||||
.co2-chart {
|
|
||||||
height: 240px;
|
|
||||||
}
|
|
||||||
</style>
|
|
||||||
{% endblock %}
|
|
||||||
|
|
||||||
{% macro stat_list(id, label, counter, show_top=5) %}
|
{% macro stat_list(id, label, counter, show_top=5) %}
|
||||||
<div class="mb-2">
|
<div class="mb-2">
|
||||||
<strong>{{ label }}:</strong> {{ counter | count }}
|
<strong>{{ label }}:</strong> {{ counter | count }}
|
||||||
|
|
@ -32,14 +24,6 @@
|
||||||
</div>
|
</div>
|
||||||
{% endmacro %}
|
{% endmacro %}
|
||||||
|
|
||||||
{% macro format_co2(co2_kg) %}
|
|
||||||
{% if co2_kg >= 1000 %}
|
|
||||||
{{ "{:,.2f}".format(co2_kg / 1000.0) }} tonnes
|
|
||||||
{% else %}
|
|
||||||
{{ "{:,.0f}".format(co2_kg) }} kg
|
|
||||||
{% endif %}
|
|
||||||
{% endmacro %}
|
|
||||||
|
|
||||||
{% block content %}
|
{% block content %}
|
||||||
<div class="container-fluid">
|
<div class="container-fluid">
|
||||||
<h1>Trip statistics</h1>
|
<h1>Trip statistics</h1>
|
||||||
|
|
@ -55,21 +39,10 @@
|
||||||
{% for transport_type, distance in distances_by_transport_type %}
|
{% for transport_type, distance in distances_by_transport_type %}
|
||||||
<div class="ms-3">{{ transport_type | title }}: {{ format_distance(distance) }}</div>
|
<div class="ms-3">{{ transport_type | title }}: {{ format_distance(distance) }}</div>
|
||||||
{% endfor %}
|
{% endfor %}
|
||||||
{% if total_co2_kg %}
|
|
||||||
<div class="mt-2">CO₂: {{ format_co2(total_co2_kg) }}</div>
|
|
||||||
{% endif %}
|
|
||||||
{% for transport_type, co2_kg in co2_by_transport_type %}
|
|
||||||
<div class="ms-3">{{ transport_type | title }} CO₂: {{ format_co2(co2_kg) }}</div>
|
|
||||||
{% endfor %}
|
|
||||||
</div>
|
</div>
|
||||||
<div class="col-md-6">
|
<div class="col-md-6">
|
||||||
<div>Flight segments: {{ overall_stats.flight_count }}</div>
|
<div>Flight segments: {{ overall_stats.flight_count }}</div>
|
||||||
<div>Train segments: {{ overall_stats.train_count }}</div>
|
<div>Train segments: {{ overall_stats.train_count }}</div>
|
||||||
{% if co2_by_transport_type %}
|
|
||||||
<div class="co2-chart mt-3">
|
|
||||||
<canvas id="overall-co2-chart"></canvas>
|
|
||||||
</div>
|
|
||||||
{% endif %}
|
|
||||||
</div>
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
|
@ -106,12 +79,12 @@
|
||||||
<div>Flight segments: {{ year_stats.flight_count or 0 }}</div>
|
<div>Flight segments: {{ year_stats.flight_count or 0 }}</div>
|
||||||
<div>Train segments: {{ year_stats.train_count or 0 }}</div>
|
<div>Train segments: {{ year_stats.train_count or 0 }}</div>
|
||||||
{% if year_stats.co2_kg %}
|
{% if year_stats.co2_kg %}
|
||||||
<div>CO₂: {{ format_co2(year_stats.co2_kg) }}</div>
|
<div>CO₂:
|
||||||
{% for transport_type, co2_kg in year_stats.co2_by_transport_type.items() %}
|
{% if year_stats.co2_kg >= 1000 %}
|
||||||
<div class="ms-3">{{ transport_type | title }} CO₂: {{ format_co2(co2_kg) }}</div>
|
{{ "{:,.2f}".format(year_stats.co2_kg / 1000.0) }} tonnes
|
||||||
{% endfor %}
|
{% else %}
|
||||||
<div class="co2-chart mt-3">
|
{{ "{:,.0f}".format(year_stats.co2_kg) }} kg
|
||||||
<canvas id="co2-chart-{{ year }}"></canvas>
|
{% endif %}
|
||||||
</div>
|
</div>
|
||||||
{% endif %}
|
{% endif %}
|
||||||
</div>
|
</div>
|
||||||
|
|
@ -146,72 +119,3 @@
|
||||||
{% endfor %}
|
{% endfor %}
|
||||||
</div>
|
</div>
|
||||||
{% endblock %}
|
{% endblock %}
|
||||||
|
|
||||||
{% block scripts %}
|
|
||||||
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
|
|
||||||
<script>
|
|
||||||
const co2ChartColors = [
|
|
||||||
"#4e79a7",
|
|
||||||
"#f28e2b",
|
|
||||||
"#e15759",
|
|
||||||
"#76b7b2",
|
|
||||||
"#59a14f",
|
|
||||||
"#edc948",
|
|
||||||
"#b07aa1",
|
|
||||||
"#ff9da7",
|
|
||||||
"#9c755f",
|
|
||||||
"#bab0ac",
|
|
||||||
];
|
|
||||||
|
|
||||||
function titleCase(value) {
|
|
||||||
return value.replace(/\b\w/g, char => char.toUpperCase());
|
|
||||||
}
|
|
||||||
|
|
||||||
function co2TooltipLabel(context) {
|
|
||||||
const label = context.label || "";
|
|
||||||
const value = context.parsed || 0;
|
|
||||||
return `${label}: ${value.toLocaleString(undefined, { maximumFractionDigits: 1 })} kg`;
|
|
||||||
}
|
|
||||||
|
|
||||||
function renderCo2Chart(id, values) {
|
|
||||||
const canvas = document.getElementById(id);
|
|
||||||
if (!canvas || !values || values.length === 0) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
new Chart(canvas.getContext("2d"), {
|
|
||||||
type: "pie",
|
|
||||||
data: {
|
|
||||||
labels: values.map(item => titleCase(item[0])),
|
|
||||||
datasets: [{
|
|
||||||
data: values.map(item => item[1]),
|
|
||||||
backgroundColor: co2ChartColors,
|
|
||||||
}],
|
|
||||||
},
|
|
||||||
options: {
|
|
||||||
responsive: true,
|
|
||||||
maintainAspectRatio: false,
|
|
||||||
plugins: {
|
|
||||||
legend: {
|
|
||||||
position: "right",
|
|
||||||
},
|
|
||||||
tooltip: {
|
|
||||||
callbacks: {
|
|
||||||
label: co2TooltipLabel,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
renderCo2Chart("overall-co2-chart", {{ co2_by_transport_type | tojson }});
|
|
||||||
{% for year, year_stats in yearly_stats | dictsort(reverse=True) %}
|
|
||||||
{% set year_co2_by_transport_type = year_stats.co2_by_transport_type | default({}) %}
|
|
||||||
renderCo2Chart(
|
|
||||||
"co2-chart-{{ year }}",
|
|
||||||
{{ year_co2_by_transport_type.items() | list | tojson }}
|
|
||||||
);
|
|
||||||
{% endfor %}
|
|
||||||
</script>
|
|
||||||
{% endblock %}
|
|
||||||
|
|
|
||||||
|
|
@ -369,8 +369,6 @@
|
||||||
|
|
||||||
{% elif e.element_type in ("coach", "bus", "car") %}
|
{% elif e.element_type in ("coach", "bus", "car") %}
|
||||||
{% set item = e.detail %}
|
{% set item = e.detail %}
|
||||||
{% set display_depart = item.display_depart if item.display_depart is defined else item.depart %}
|
|
||||||
{% set display_arrive = item.display_arrive if item.display_arrive is defined else item.arrive %}
|
|
||||||
<div class="trip-transport-card my-1">
|
<div class="trip-transport-card my-1">
|
||||||
<div class="card-body">
|
<div class="card-body">
|
||||||
<h5 class="card-title">
|
<h5 class="card-title">
|
||||||
|
|
@ -379,14 +377,14 @@
|
||||||
{% if item.operator %}<small class="text-muted fw-normal">{{ item.operator }}</small>{% endif %}
|
{% if item.operator %}<small class="text-muted fw-normal">{{ item.operator }}</small>{% endif %}
|
||||||
</h5>
|
</h5>
|
||||||
<p class="card-text">
|
<p class="card-text">
|
||||||
{% if display_depart.hour is defined and display_arrive.hour is defined %}
|
{% if item.depart.hour is defined and item.arrive.hour is defined %}
|
||||||
{{ display_depart.strftime("%H:%M") }} → {{ display_arrive.strftime("%H:%M") }}
|
{{ item.depart.strftime("%H:%M") }} → {{ item.arrive.strftime("%H:%M") }}
|
||||||
{% endif %}
|
{% endif %}
|
||||||
{% if item.class %}
|
{% if item.class %}
|
||||||
<span class="badge bg-info text-nowrap">{{ item.class }}</span>
|
<span class="badge bg-info text-nowrap">{{ item.class }}</span>
|
||||||
{% endif %}
|
{% endif %}
|
||||||
{% if display_depart.hour is defined and display_arrive.hour is defined %}
|
{% if item.depart.hour is defined and item.arrive.hour is defined %}
|
||||||
<span class="text-muted">🕒{{ trip_duration(display_depart, display_arrive) }}</span>
|
<span class="text-muted">🕒{{ trip_duration(item.depart, item.arrive) }}</span>
|
||||||
{% endif %}
|
{% endif %}
|
||||||
{% if item.distance %}
|
{% if item.distance %}
|
||||||
<span class="text-muted">🛤️ {{ "{:,.0f} km".format(item.distance) }}</span>
|
<span class="text-muted">🛤️ {{ "{:,.0f} km".format(item.distance) }}</span>
|
||||||
|
|
|
||||||
|
|
@ -1,152 +0,0 @@
|
||||||
"""Tests for importing car journeys from a Google Timeline export."""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import json
|
|
||||||
from datetime import date, datetime, timezone
|
|
||||||
from pathlib import Path
|
|
||||||
import typing
|
|
||||||
|
|
||||||
from agenda import car_journey_timeline, car_journey_yaml
|
|
||||||
|
|
||||||
|
|
||||||
def test_journey_key_orders_split_same_day_rows() -> None:
|
|
||||||
"""Datetime rows should sort chronologically within a single day."""
|
|
||||||
earlier = {
|
|
||||||
"trip": date(2025, 7, 4),
|
|
||||||
"depart": datetime(2025, 7, 4, 8, 0, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2025, 7, 4, 8, 30, tzinfo=timezone.utc),
|
|
||||||
"route": "a",
|
|
||||||
}
|
|
||||||
later = {
|
|
||||||
"trip": date(2025, 7, 4),
|
|
||||||
"depart": datetime(2025, 7, 4, 9, 0, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2025, 7, 4, 9, 30, tzinfo=timezone.utc),
|
|
||||||
"route": "b",
|
|
||||||
}
|
|
||||||
|
|
||||||
assert car_journey_yaml.journey_key(earlier) < car_journey_yaml.journey_key(later)
|
|
||||||
|
|
||||||
|
|
||||||
def write_json(path: Path, data: dict[str, typing.Any]) -> None:
|
|
||||||
"""Write JSON test data."""
|
|
||||||
path.write_text(json.dumps(data))
|
|
||||||
|
|
||||||
|
|
||||||
def test_import_from_timeline_splits_stopovers(tmp_path: Path) -> None:
|
|
||||||
"""Past car journeys should be split into multiple timeline segments."""
|
|
||||||
data_dir = tmp_path
|
|
||||||
route_dir = data_dir / "car_routes"
|
|
||||||
route_dir.mkdir()
|
|
||||||
(data_dir / "car_journeys.yaml").write_text("""---
|
|
||||||
|
|
||||||
- trip: 2025-07-04
|
|
||||||
depart: 2025-07-04
|
|
||||||
arrive: 2025-07-04
|
|
||||||
route: PCH_to_Far
|
|
||||||
|
|
||||||
- trip: 2026-07-16
|
|
||||||
depart: 2026-07-16
|
|
||||||
arrive: 2026-07-16
|
|
||||||
route: Future_to_Elsewhere
|
|
||||||
""")
|
|
||||||
write_json(
|
|
||||||
route_dir / "PCH_to_Far.geojson",
|
|
||||||
{
|
|
||||||
"type": "Feature",
|
|
||||||
"properties": {},
|
|
||||||
"geometry": {
|
|
||||||
"type": "LineString",
|
|
||||||
"coordinates": [[-2.6, 51.4], [-4.1, 50.3]],
|
|
||||||
},
|
|
||||||
},
|
|
||||||
)
|
|
||||||
write_json(
|
|
||||||
route_dir / "Future_to_Elsewhere.geojson",
|
|
||||||
{
|
|
||||||
"type": "Feature",
|
|
||||||
"properties": {},
|
|
||||||
"geometry": {
|
|
||||||
"type": "LineString",
|
|
||||||
"coordinates": [[0.0, 0.0], [1.0, 1.0]],
|
|
||||||
},
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
timeline_path = tmp_path / "Timeline.json"
|
|
||||||
write_json(
|
|
||||||
timeline_path,
|
|
||||||
{
|
|
||||||
"semanticSegments": [
|
|
||||||
{
|
|
||||||
"startTime": "2025-07-04T08:00:00+00:00",
|
|
||||||
"endTime": "2025-07-04T10:00:00+00:00",
|
|
||||||
"timelinePath": [
|
|
||||||
{
|
|
||||||
"point": "51.4000°, -2.6000°",
|
|
||||||
"time": "2025-07-04T08:00:00+00:00",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"point": "51.0000°, -3.1000°",
|
|
||||||
"time": "2025-07-04T08:30:00+00:00",
|
|
||||||
},
|
|
||||||
],
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"startTime": "2025-07-04T08:00:00+00:00",
|
|
||||||
"endTime": "2025-07-04T08:30:00+00:00",
|
|
||||||
"activity": {
|
|
||||||
"start": {"latLng": "51.4000°, -2.6000°"},
|
|
||||||
"end": {"latLng": "51.0000°, -3.1000°"},
|
|
||||||
"topCandidate": {"type": "IN_PASSENGER_VEHICLE"},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"startTime": "2025-07-04T10:00:00+00:00",
|
|
||||||
"endTime": "2025-07-04T12:00:00+00:00",
|
|
||||||
"timelinePath": [
|
|
||||||
{
|
|
||||||
"point": "51.0000°, -3.1000°",
|
|
||||||
"time": "2025-07-04T10:00:00+00:00",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"point": "50.3000°, -4.1000°",
|
|
||||||
"time": "2025-07-04T11:00:00+00:00",
|
|
||||||
},
|
|
||||||
],
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"startTime": "2025-07-04T10:00:00+00:00",
|
|
||||||
"endTime": "2025-07-04T11:00:00+00:00",
|
|
||||||
"activity": {
|
|
||||||
"start": {"latLng": "51.0000°, -3.1000°"},
|
|
||||||
"end": {"latLng": "50.3000°, -4.1000°"},
|
|
||||||
"topCandidate": {"type": "IN_PASSENGER_VEHICLE"},
|
|
||||||
},
|
|
||||||
},
|
|
||||||
]
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
routes_written, rows_replaced, rows_written = (
|
|
||||||
car_journey_timeline.import_from_timeline(
|
|
||||||
timeline_path, data_dir=data_dir, today=date(2026, 7, 9)
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
assert routes_written == 2
|
|
||||||
assert rows_replaced == 1
|
|
||||||
assert rows_written == 2
|
|
||||||
|
|
||||||
updated = car_journey_yaml.load_yaml_list(data_dir / "car_journeys.yaml")
|
|
||||||
assert len(updated) == 3
|
|
||||||
assert updated[0]["from"] == "PCH"
|
|
||||||
assert updated[0]["to"] == "Drive stop 1"
|
|
||||||
assert updated[1]["from"] == "Drive stop 1"
|
|
||||||
assert updated[1]["to"] == "Far"
|
|
||||||
assert updated[2]["route"] == "Future_to_Elsewhere"
|
|
||||||
|
|
||||||
generated_paths = sorted(route_dir.glob("timeline_*.geojson"))
|
|
||||||
assert len(generated_paths) == 2
|
|
||||||
generated_geojson = json.loads(generated_paths[0].read_text())
|
|
||||||
assert generated_geojson["geometry"]["type"] == "LineString"
|
|
||||||
|
|
@ -6,7 +6,6 @@ from types import SimpleNamespace
|
||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
|
|
||||||
import agenda.fx
|
|
||||||
import agenda.trip
|
import agenda.trip
|
||||||
import web_view
|
import web_view
|
||||||
|
|
||||||
|
|
@ -122,84 +121,3 @@ def test_conference_series_pages(tmp_path: typing.Any, monkeypatch: typing.Any)
|
||||||
assert b"attended" in index_response.data
|
assert b"attended" in index_response.data
|
||||||
assert detail_response.status_code == 200
|
assert detail_response.status_code == 200
|
||||||
assert b"trip: Seattle Python trip" in detail_response.data
|
assert b"trip: Seattle Python trip" in detail_response.data
|
||||||
|
|
||||||
|
|
||||||
def test_conference_page_filters_by_country(
|
|
||||||
tmp_path: typing.Any, monkeypatch: typing.Any
|
|
||||||
) -> None:
|
|
||||||
"""Conference page should filter upcoming conferences by country code."""
|
|
||||||
conferences = [
|
|
||||||
{
|
|
||||||
"name": "UK Mapping Conf 2099",
|
|
||||||
"topic": "Maps",
|
|
||||||
"location": "London",
|
|
||||||
"country": "GB",
|
|
||||||
"start": date(2099, 5, 1),
|
|
||||||
"end": date(2099, 5, 2),
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"name": "US Python Conf 2099",
|
|
||||||
"topic": "Python",
|
|
||||||
"location": "Pittsburgh",
|
|
||||||
"country": "US",
|
|
||||||
"start": date(2099, 6, 1),
|
|
||||||
"end": date(2099, 6, 2),
|
|
||||||
},
|
|
||||||
]
|
|
||||||
(tmp_path / "conferences.yaml").write_text(
|
|
||||||
yaml.safe_dump(conferences), encoding="utf-8"
|
|
||||||
)
|
|
||||||
|
|
||||||
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
||||||
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
|
||||||
monkeypatch.setattr(agenda.fx, "get_rates", lambda config: {})
|
|
||||||
|
|
||||||
web_view.app.config["TESTING"] = True
|
|
||||||
with web_view.app.test_client() as client:
|
|
||||||
response = client.get("/conference?country=gb")
|
|
||||||
|
|
||||||
assert response.status_code == 200
|
|
||||||
assert b"UK Mapping Conf 2099" in response.data
|
|
||||||
assert b"US Python Conf 2099" not in response.data
|
|
||||||
assert b'<option value="gb" selected>' in response.data
|
|
||||||
assert b'href="/conference"' in response.data
|
|
||||||
|
|
||||||
|
|
||||||
def test_past_conference_page_filters_by_country(
|
|
||||||
tmp_path: typing.Any, monkeypatch: typing.Any
|
|
||||||
) -> None:
|
|
||||||
"""Past conference page should filter conferences by country code."""
|
|
||||||
conferences = [
|
|
||||||
{
|
|
||||||
"name": "Past UK Conf",
|
|
||||||
"topic": "Maps",
|
|
||||||
"location": "London",
|
|
||||||
"country": "GB",
|
|
||||||
"start": date(2001, 5, 1),
|
|
||||||
"end": date(2001, 5, 2),
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"name": "Past US Conf",
|
|
||||||
"topic": "Python",
|
|
||||||
"location": "Pittsburgh",
|
|
||||||
"country": "US",
|
|
||||||
"start": date(2001, 6, 1),
|
|
||||||
"end": date(2001, 6, 2),
|
|
||||||
},
|
|
||||||
]
|
|
||||||
(tmp_path / "conferences.yaml").write_text(
|
|
||||||
yaml.safe_dump(conferences), encoding="utf-8"
|
|
||||||
)
|
|
||||||
|
|
||||||
monkeypatch.setitem(web_view.app.config, "PERSONAL_DATA", str(tmp_path))
|
|
||||||
monkeypatch.setattr(agenda.trip, "build_trip_list", lambda: [])
|
|
||||||
monkeypatch.setattr(agenda.fx, "get_rates", lambda config: {})
|
|
||||||
|
|
||||||
web_view.app.config["TESTING"] = True
|
|
||||||
with web_view.app.test_client() as client:
|
|
||||||
response = client.get("/conference/past?country=us")
|
|
||||||
|
|
||||||
assert response.status_code == 200
|
|
||||||
assert b"Past US Conf" in response.data
|
|
||||||
assert b"Past UK Conf" not in response.data
|
|
||||||
assert b'href="/conference/past"' in response.data
|
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import pathlib
|
import pathlib
|
||||||
from datetime import date, datetime, timezone
|
from datetime import date
|
||||||
|
|
||||||
import agenda.trip
|
import agenda.trip
|
||||||
import pytest
|
import pytest
|
||||||
|
|
@ -203,9 +203,6 @@ def test_load_cars_infers_route_labels_and_home_marker(
|
||||||
assert cars[0]["to"] == "EMF"
|
assert cars[0]["to"] == "EMF"
|
||||||
assert cars[0]["geojson_filename"] == "PCH_to_EMF"
|
assert cars[0]["geojson_filename"] == "PCH_to_EMF"
|
||||||
assert cars[0]["distance"] > 0
|
assert cars[0]["distance"] > 0
|
||||||
assert cars[0]["co2_kg"] == pytest.approx(
|
|
||||||
cars[0]["distance"] * agenda.trip.CAR_CO2_KG_PER_KM
|
|
||||||
)
|
|
||||||
assert cars[0]["from_location"] == {
|
assert cars[0]["from_location"] == {
|
||||||
"name": "PCH",
|
"name": "PCH",
|
||||||
"type": "home",
|
"type": "home",
|
||||||
|
|
@ -277,49 +274,3 @@ def test_get_trip_routes_includes_car_geojson() -> None:
|
||||||
"geojson_filename": "car_routes/PCH_to_EMF",
|
"geojson_filename": "car_routes/PCH_to_EMF",
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
def test_trip_title_ignores_generated_drive_stop_labels() -> None:
|
|
||||||
"""Fallback trip titles should ignore synthetic split-car stop labels."""
|
|
||||||
trip = Trip(
|
|
||||||
start=date(2025, 12, 29),
|
|
||||||
travel=[
|
|
||||||
{
|
|
||||||
"type": "car",
|
|
||||||
"depart": datetime(2025, 12, 29, 11, 4, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2025, 12, 29, 11, 10, tzinfo=timezone.utc),
|
|
||||||
"from": "PCH",
|
|
||||||
"to": "Drive stop 1",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"type": "car",
|
|
||||||
"depart": datetime(2025, 12, 29, 11, 27, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2025, 12, 29, 11, 40, tzinfo=timezone.utc),
|
|
||||||
"from": "Drive stop 1",
|
|
||||||
"to": "Drive stop 2",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"type": "car",
|
|
||||||
"depart": datetime(2025, 12, 29, 14, 6, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2025, 12, 29, 15, 22, tzinfo=timezone.utc),
|
|
||||||
"from": "Drive stop 2",
|
|
||||||
"to": "St Ives",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"type": "car",
|
|
||||||
"depart": datetime(2026, 1, 5, 12, 46, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2026, 1, 5, 13, 0, tzinfo=timezone.utc),
|
|
||||||
"from": "St Ives",
|
|
||||||
"to": "Drive stop 1",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"type": "car",
|
|
||||||
"depart": datetime(2026, 1, 5, 16, 58, tzinfo=timezone.utc),
|
|
||||||
"arrive": datetime(2026, 1, 5, 17, 29, tzinfo=timezone.utc),
|
|
||||||
"from": "Drive stop 3",
|
|
||||||
"to": "PCH",
|
|
||||||
},
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
assert trip.title == "St Ives"
|
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,6 @@
|
||||||
"""Regression tests for trip page route wiring and rendering."""
|
"""Regression tests for trip page route wiring and rendering."""
|
||||||
|
|
||||||
from datetime import date, datetime, timedelta, timezone
|
from datetime import date, datetime
|
||||||
import typing
|
import typing
|
||||||
from unittest import mock
|
from unittest import mock
|
||||||
|
|
||||||
|
|
@ -307,57 +307,3 @@ def test_trip_page_uses_bed_icon_for_overnight_train_coach() -> None:
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
page = response.data.decode()
|
page = response.data.decode()
|
||||||
assert "🛏️ Coach S, Seat 4" in page
|
assert "🛏️ Coach S, Seat 4" in page
|
||||||
|
|
||||||
|
|
||||||
def test_trip_page_renders_car_times_in_local_route_timezone() -> None:
|
|
||||||
"""Car journeys should display in the route's local timezone."""
|
|
||||||
trip = Trip(
|
|
||||||
start=date(2023, 8, 23),
|
|
||||||
travel=[
|
|
||||||
{
|
|
||||||
"type": "car",
|
|
||||||
"depart": datetime(
|
|
||||||
2023, 8, 24, 5, 22, tzinfo=timezone(timedelta(hours=1))
|
|
||||||
),
|
|
||||||
"arrive": datetime(
|
|
||||||
2023, 8, 24, 6, 26, tzinfo=timezone(timedelta(hours=1))
|
|
||||||
),
|
|
||||||
"from": "JFK",
|
|
||||||
"to": "Rockaway",
|
|
||||||
"distance": 102.0,
|
|
||||||
"from_location": {
|
|
||||||
"name": "JFK",
|
|
||||||
"country": "us",
|
|
||||||
"latitude": 40.64,
|
|
||||||
"longitude": -73.78,
|
|
||||||
},
|
|
||||||
"to_location": {
|
|
||||||
"name": "Rockaway",
|
|
||||||
"country": "us",
|
|
||||||
"latitude": 40.907,
|
|
||||||
"longitude": -74.554,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
with web_view.app.app_context():
|
|
||||||
with (
|
|
||||||
mock.patch.object(web_view, "get_trip_list", return_value=[trip]),
|
|
||||||
mock.patch.object(agenda.weather, "get_trip_weather", return_value=[]),
|
|
||||||
mock.patch.object(
|
|
||||||
web_view,
|
|
||||||
"_timezone_from_coordinates",
|
|
||||||
return_value="America/New_York",
|
|
||||||
),
|
|
||||||
):
|
|
||||||
web_view.app.config["TESTING"] = True
|
|
||||||
|
|
||||||
with web_view.app.test_client() as client:
|
|
||||||
response = client.get("/trip/2023-08-23")
|
|
||||||
|
|
||||||
assert response.status_code == 200
|
|
||||||
page = response.data.decode()
|
|
||||||
assert "JFK → Rockaway" in page
|
|
||||||
assert "00:22 → 01:26" in page
|
|
||||||
assert "05:22 → 06:26" not in page
|
|
||||||
|
|
|
||||||
|
|
@ -3,7 +3,6 @@
|
||||||
from datetime import date
|
from datetime import date
|
||||||
|
|
||||||
import agenda
|
import agenda
|
||||||
import pytest
|
|
||||||
from agenda.stats import calculate_yearly_stats
|
from agenda.stats import calculate_yearly_stats
|
||||||
from agenda.types import Trip
|
from agenda.types import Trip
|
||||||
|
|
||||||
|
|
@ -36,22 +35,3 @@ def test_new_country_respects_previously_visited() -> None:
|
||||||
|
|
||||||
yearly_stats = calculate_yearly_stats(trips, {"CZ"})
|
yearly_stats = calculate_yearly_stats(trips, {"CZ"})
|
||||||
assert "new_countries" not in yearly_stats[2024]
|
assert "new_countries" not in yearly_stats[2024]
|
||||||
|
|
||||||
|
|
||||||
def test_yearly_stats_tracks_co2_by_transport_type() -> None:
|
|
||||||
"""CO₂ stats should include a per-transport-type breakdown."""
|
|
||||||
trip = Trip(
|
|
||||||
start=date(2024, 5, 1),
|
|
||||||
travel=[
|
|
||||||
{"type": "car", "distance": 100.0, "co2_kg": 21.8},
|
|
||||||
{"type": "bus", "distance": 50.0, "co2_kg": 5.0},
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
yearly_stats = calculate_yearly_stats([trip])
|
|
||||||
|
|
||||||
assert yearly_stats[2024]["co2_kg"] == pytest.approx(26.8)
|
|
||||||
assert yearly_stats[2024]["co2_by_transport_type"] == {
|
|
||||||
"car": 21.8,
|
|
||||||
"bus": 5.0,
|
|
||||||
}
|
|
||||||
|
|
|
||||||
180
web_view.py
180
web_view.py
|
|
@ -3,7 +3,6 @@
|
||||||
"""Web page to show upcoming events."""
|
"""Web page to show upcoming events."""
|
||||||
|
|
||||||
import decimal
|
import decimal
|
||||||
import functools
|
|
||||||
import hashlib
|
import hashlib
|
||||||
import importlib
|
import importlib
|
||||||
import inspect
|
import inspect
|
||||||
|
|
@ -401,64 +400,6 @@ def build_conference_list() -> list[StrDict]:
|
||||||
return items
|
return items
|
||||||
|
|
||||||
|
|
||||||
def conference_country_code(conf: StrDict) -> str | None:
|
|
||||||
"""Return normalized alpha-2 country code for a conference."""
|
|
||||||
country = conf.get("country")
|
|
||||||
if not isinstance(country, str):
|
|
||||||
return None
|
|
||||||
|
|
||||||
code = country.strip().lower()
|
|
||||||
return code if len(code) == 2 else None
|
|
||||||
|
|
||||||
|
|
||||||
def normalize_country_filter(value: str | None) -> str | None:
|
|
||||||
"""Normalize and validate a country query parameter."""
|
|
||||||
if not value:
|
|
||||||
return None
|
|
||||||
|
|
||||||
code = value.strip().lower()
|
|
||||||
if len(code) != 2:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return code if agenda.get_country(code) else None
|
|
||||||
|
|
||||||
|
|
||||||
def filter_conferences_by_country(
|
|
||||||
items: list[StrDict], country_code: str | None
|
|
||||||
) -> list[StrDict]:
|
|
||||||
"""Filter conferences by country when a country code is provided."""
|
|
||||||
if not country_code:
|
|
||||||
return items
|
|
||||||
|
|
||||||
return [item for item in items if conference_country_code(item) == country_code]
|
|
||||||
|
|
||||||
|
|
||||||
def conference_country_options(items: list[StrDict]) -> list[StrDict]:
|
|
||||||
"""Return country options for the conference country filter."""
|
|
||||||
counts: defaultdict[str, int] = defaultdict(int)
|
|
||||||
for item in items:
|
|
||||||
code = conference_country_code(item)
|
|
||||||
if code and agenda.get_country(code):
|
|
||||||
counts[code] += 1
|
|
||||||
|
|
||||||
options: list[StrDict] = []
|
|
||||||
for code, count in counts.items():
|
|
||||||
country = agenda.get_country(code)
|
|
||||||
if not country:
|
|
||||||
continue
|
|
||||||
options.append(
|
|
||||||
{
|
|
||||||
"code": code,
|
|
||||||
"name": country.name,
|
|
||||||
"flag": country.flag,
|
|
||||||
"count": count,
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
options.sort(key=lambda item: str(item["name"]))
|
|
||||||
return options
|
|
||||||
|
|
||||||
|
|
||||||
def build_conference_series_list() -> list[StrDict]:
|
def build_conference_series_list() -> list[StrDict]:
|
||||||
"""Build conference series list with conference counts."""
|
"""Build conference series list with conference counts."""
|
||||||
data_dir = app.config["PERSONAL_DATA"]
|
data_dir = app.config["PERSONAL_DATA"]
|
||||||
|
|
@ -650,11 +591,6 @@ def conference_list() -> str:
|
||||||
"""Page showing a list of conferences."""
|
"""Page showing a list of conferences."""
|
||||||
today = date.today()
|
today = date.today()
|
||||||
items = build_conference_list()
|
items = build_conference_list()
|
||||||
country_filter = normalize_country_filter(flask.request.args.get("country"))
|
|
||||||
country_options = conference_country_options(
|
|
||||||
[conf for conf in items if conf["latest_date"] >= today]
|
|
||||||
)
|
|
||||||
items = filter_conferences_by_country(items, country_filter)
|
|
||||||
|
|
||||||
current = [
|
current = [
|
||||||
conf
|
conf
|
||||||
|
|
@ -676,8 +612,6 @@ def conference_list() -> str:
|
||||||
timeline=timeline,
|
timeline=timeline,
|
||||||
today=today,
|
today=today,
|
||||||
get_country=agenda.get_country,
|
get_country=agenda.get_country,
|
||||||
selected_country=country_filter,
|
|
||||||
country_options=country_options,
|
|
||||||
fx_rate=agenda.fx.get_rates(app.config),
|
fx_rate=agenda.fx.get_rates(app.config),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -686,19 +620,11 @@ def conference_list() -> str:
|
||||||
def past_conference_list() -> str:
|
def past_conference_list() -> str:
|
||||||
"""Page showing a list of conferences."""
|
"""Page showing a list of conferences."""
|
||||||
today = date.today()
|
today = date.today()
|
||||||
items = build_conference_list()
|
|
||||||
country_filter = normalize_country_filter(flask.request.args.get("country"))
|
|
||||||
country_options = conference_country_options(
|
|
||||||
[conf for conf in items if conf["latest_date"] < today]
|
|
||||||
)
|
|
||||||
items = filter_conferences_by_country(items, country_filter)
|
|
||||||
return flask.render_template(
|
return flask.render_template(
|
||||||
"conference_list.html",
|
"conference_list.html",
|
||||||
past=[conf for conf in items if conf["latest_date"] < today],
|
past=[conf for conf in build_conference_list() if conf["latest_date"] < today],
|
||||||
today=today,
|
today=today,
|
||||||
get_country=agenda.get_country,
|
get_country=agenda.get_country,
|
||||||
selected_country=country_filter,
|
|
||||||
country_options=country_options,
|
|
||||||
fx_rate=agenda.fx.get_rates(app.config),
|
fx_rate=agenda.fx.get_rates(app.config),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -842,16 +768,6 @@ 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 sum_co2_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
|
|
||||||
"""Sum CO₂ by transport type."""
|
|
||||||
co2_by_transport_type: defaultdict[str, float] = defaultdict(float)
|
|
||||||
for trip in trips:
|
|
||||||
for transport_type, co2_kg in trip.co2_by_transport_type():
|
|
||||||
co2_by_transport_type[transport_type] += co2_kg
|
|
||||||
|
|
||||||
return list(co2_by_transport_type.items())
|
|
||||||
|
|
||||||
|
|
||||||
def get_home_weather() -> list[StrDict]:
|
def get_home_weather() -> list[StrDict]:
|
||||||
"""Get Bristol home weather forecast, with date objects added for templates."""
|
"""Get Bristol home weather forecast, with date objects added for templates."""
|
||||||
from datetime import date as date_type
|
from datetime import date as date_type
|
||||||
|
|
@ -922,7 +838,6 @@ def trip_past_list() -> str:
|
||||||
total_distance=calc_total_distance(past),
|
total_distance=calc_total_distance(past),
|
||||||
total_co2_kg=calc_total_co2_kg(past),
|
total_co2_kg=calc_total_co2_kg(past),
|
||||||
distances_by_transport_type=sum_distances_by_transport_type(past),
|
distances_by_transport_type=sum_distances_by_transport_type(past),
|
||||||
co2_by_transport_type=sum_co2_by_transport_type(past),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -985,7 +900,6 @@ def trip_future_list() -> str:
|
||||||
total_distance=calc_total_distance(current + future),
|
total_distance=calc_total_distance(current + future),
|
||||||
total_co2_kg=calc_total_co2_kg(current + future),
|
total_co2_kg=calc_total_co2_kg(current + future),
|
||||||
distances_by_transport_type=sum_distances_by_transport_type(current + future),
|
distances_by_transport_type=sum_distances_by_transport_type(current + future),
|
||||||
co2_by_transport_type=sum_co2_by_transport_type(current + future),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -1097,7 +1011,6 @@ def _timezone_from_coordinates(latitude: float, longitude: float) -> str | None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
@functools.lru_cache(maxsize=1)
|
|
||||||
def _get_timezone_finder() -> typing.Any:
|
def _get_timezone_finder() -> typing.Any:
|
||||||
"""Get timezone finder instance if dependency is available."""
|
"""Get timezone finder instance if dependency is available."""
|
||||||
try:
|
try:
|
||||||
|
|
@ -1112,94 +1025,6 @@ def _get_timezone_finder() -> typing.Any:
|
||||||
return timezone_finder_cls()
|
return timezone_finder_cls()
|
||||||
|
|
||||||
|
|
||||||
def _coordinates_from_location(location: typing.Any) -> tuple[float, float] | None:
|
|
||||||
"""Extract latitude/longitude from a location mapping."""
|
|
||||||
if not isinstance(location, dict):
|
|
||||||
return None
|
|
||||||
|
|
||||||
latitude = location.get("latitude")
|
|
||||||
longitude = location.get("longitude")
|
|
||||||
if not isinstance(latitude, (int, float)) or not isinstance(
|
|
||||||
longitude, (int, float)
|
|
||||||
):
|
|
||||||
return None
|
|
||||||
|
|
||||||
return (float(latitude), float(longitude))
|
|
||||||
|
|
||||||
|
|
||||||
def _route_endpoints_for_car_item(
|
|
||||||
item: StrDict,
|
|
||||||
data_dir: str,
|
|
||||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None],
|
|
||||||
) -> tuple[tuple[float, float], tuple[float, float]] | None:
|
|
||||||
"""Return route endpoints for a car journey when available."""
|
|
||||||
route_filename = item.get("geojson_filename")
|
|
||||||
if not isinstance(route_filename, str):
|
|
||||||
return None
|
|
||||||
|
|
||||||
if route_filename not in route_cache:
|
|
||||||
geojson_text = agenda.trip.read_geojson(
|
|
||||||
data_dir, os.path.join("car_routes", route_filename)
|
|
||||||
)
|
|
||||||
endpoints = agenda.trip.geojson_route_endpoints(json.loads(geojson_text))
|
|
||||||
route_cache[route_filename] = endpoints
|
|
||||||
|
|
||||||
return route_cache[route_filename]
|
|
||||||
|
|
||||||
|
|
||||||
def _timezone_name_for_car_item(
|
|
||||||
item: StrDict,
|
|
||||||
data_dir: str,
|
|
||||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None],
|
|
||||||
timezone_cache: dict[tuple[float, float], str | None],
|
|
||||||
) -> str | None:
|
|
||||||
"""Resolve the local timezone for a car journey from endpoints."""
|
|
||||||
candidate_coords: list[tuple[float, float]] = []
|
|
||||||
|
|
||||||
for field in ("from_location", "to_location"):
|
|
||||||
coord = _coordinates_from_location(item.get(field))
|
|
||||||
if coord is not None:
|
|
||||||
candidate_coords.append(coord)
|
|
||||||
|
|
||||||
if not candidate_coords:
|
|
||||||
endpoints = _route_endpoints_for_car_item(item, data_dir, route_cache)
|
|
||||||
if endpoints is not None:
|
|
||||||
candidate_coords.extend(endpoints)
|
|
||||||
|
|
||||||
for coord in candidate_coords:
|
|
||||||
if coord not in timezone_cache:
|
|
||||||
timezone_cache[coord] = _timezone_from_coordinates(coord[0], coord[1])
|
|
||||||
timezone_name = timezone_cache[coord]
|
|
||||||
if timezone_name:
|
|
||||||
return timezone_name
|
|
||||||
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def localize_trip_car_journey_display_times(trip: Trip, data_dir: str) -> None:
|
|
||||||
"""Add route-local display timestamps for car journeys on the trip page."""
|
|
||||||
route_cache: dict[str, tuple[tuple[float, float], tuple[float, float]] | None] = {}
|
|
||||||
timezone_cache: dict[tuple[float, float], str | None] = {}
|
|
||||||
|
|
||||||
for item in trip.travel:
|
|
||||||
if item.get("type") != "car":
|
|
||||||
continue
|
|
||||||
|
|
||||||
depart = item.get("depart")
|
|
||||||
arrive = item.get("arrive")
|
|
||||||
if not isinstance(depart, datetime) or not isinstance(arrive, datetime):
|
|
||||||
continue
|
|
||||||
|
|
||||||
timezone_name = _timezone_name_for_car_item(
|
|
||||||
item, data_dir, route_cache, timezone_cache
|
|
||||||
)
|
|
||||||
if not timezone_name:
|
|
||||||
continue
|
|
||||||
|
|
||||||
item["display_depart"] = depart.astimezone(ZoneInfo(timezone_name))
|
|
||||||
item["display_arrive"] = arrive.astimezone(ZoneInfo(timezone_name))
|
|
||||||
|
|
||||||
|
|
||||||
def get_destination_timezones(trip: Trip) -> list[StrDict]:
|
def get_destination_timezones(trip: Trip) -> list[StrDict]:
|
||||||
"""Build destination timezone metadata for the trip page."""
|
"""Build destination timezone metadata for the trip page."""
|
||||||
per_location: dict[tuple[str, str], list[str]] = defaultdict(list)
|
per_location: dict[tuple[str, str], list[str]] = defaultdict(list)
|
||||||
|
|
@ -1378,7 +1203,6 @@ def trip_page(start: str) -> str:
|
||||||
trip,
|
trip,
|
||||||
cache_only=True,
|
cache_only=True,
|
||||||
)
|
)
|
||||||
localize_trip_car_journey_display_times(trip, app.config["PERSONAL_DATA"])
|
|
||||||
|
|
||||||
return flask.render_template(
|
return flask.render_template(
|
||||||
"trip_page.html",
|
"trip_page.html",
|
||||||
|
|
@ -1527,9 +1351,7 @@ def trip_stats() -> str:
|
||||||
"trip/stats.html",
|
"trip/stats.html",
|
||||||
count=len(trip_list),
|
count=len(trip_list),
|
||||||
total_distance=calc_total_distance(trip_list),
|
total_distance=calc_total_distance(trip_list),
|
||||||
total_co2_kg=calc_total_co2_kg(trip_list),
|
|
||||||
distances_by_transport_type=sum_distances_by_transport_type(trip_list),
|
distances_by_transport_type=sum_distances_by_transport_type(trip_list),
|
||||||
co2_by_transport_type=sum_co2_by_transport_type(trip_list),
|
|
||||||
yearly_stats=yearly_stats,
|
yearly_stats=yearly_stats,
|
||||||
overall_stats=overall_stats,
|
overall_stats=overall_stats,
|
||||||
conferences=conferences,
|
conferences=conferences,
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue