diff --git a/agenda/car_journey_timeline.py b/agenda/car_journey_timeline.py new file mode 100644 index 0000000..d3518ee --- /dev/null +++ b/agenda/car_journey_timeline.py @@ -0,0 +1,474 @@ +"""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 diff --git a/agenda/car_journey_yaml.py b/agenda/car_journey_yaml.py index cbf360a..e8cb1f9 100644 --- a/agenda/car_journey_yaml.py +++ b/agenda/car_journey_yaml.py @@ -9,12 +9,16 @@ import re import typing import unicodedata from dataclasses import dataclass -from datetime import date +from datetime import date, datetime from pathlib import Path -import openrouteservice # type: ignore[import-untyped] 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 utils @@ -600,6 +604,8 @@ def car_journeys_for_trip( def openrouteservice_fetcher(api_key: str) -> RouteFetcher: """Return a route fetcher backed by openrouteservice.""" + if openrouteservice is None: + raise ValueError("openrouteservice must be installed") client = openrouteservice.Client(key=api_key) def fetch(start: LonLat, end: LonLat) -> GeoJSON: @@ -639,15 +645,15 @@ def write_route_files( return written -def journey_key(item: StrDict) -> tuple[date, date, date, str]: +def journey_key(item: StrDict) -> tuple[date, datetime, datetime, str]: """Return a stable key for duplicate detection and sorting.""" route = item.get("route") if not isinstance(route, str): raise ValueError(f"car journey route must be a string: {item!r}") return ( utils.as_date(item["trip"]), - utils.as_date(item["depart"]), - utils.as_date(item["arrive"]), + utils.as_datetime(item["depart"]), + utils.as_datetime(item["arrive"]), trip_module.route_filename_without_extension(route), ) diff --git a/agenda/stats.py b/agenda/stats.py index 2e4dc67..91493c9 100644 --- a/agenda/stats.py +++ b/agenda/stats.py @@ -1,10 +1,15 @@ """Trip statistic functions.""" from collections import defaultdict -from typing import Counter, Mapping +import typing +from typing import TYPE_CHECKING, Counter, Mapping 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: @@ -18,6 +23,8 @@ def travel_legs(trip: Trip, stats: StrDict) -> None: stats["co2_by_transport_type"].setdefault(transport_type, 0) stats["co2_by_transport_type"][transport_type] += leg["co2_kg"] if leg["type"] == "flight": + from agenda.types import airport_label + stats.setdefault("flight_count", 0) stats.setdefault("airlines", Counter()) stats.setdefault("airports", Counter()) @@ -69,6 +76,8 @@ def calculate_overall_stats(yearly_stats: dict[int, StrDict]) -> StrDict: "stations": Counter(), "flight_count": 0, "train_count": 0, + "co2_kg": 0.0, + "co2_by_transport_type": {}, } for year_stats in yearly_stats.values(): @@ -80,6 +89,12 @@ def calculate_overall_stats(yearly_stats: dict[int, StrDict]) -> StrDict: overall["stations"] += year_stats["stations"] overall["flight_count"] += year_stats.get("flight_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 diff --git a/agenda/trip.py b/agenda/trip.py index c6573d3..6521ab7 100644 --- a/agenda/trip.py +++ b/agenda/trip.py @@ -17,6 +17,12 @@ from agenda import conference, ical, travel, trip_schengen from agenda.types import StrDict, Trip, TripElement 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): """Airline.""" @@ -191,13 +197,11 @@ def load_trains( if 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: - leg["co2_kg"] = leg["distance"] * 0.037 + leg["co2_kg"] = leg["distance"] * TRAIN_CO2_KG_PER_KM if all("distance" in leg 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"]) return trains @@ -220,9 +224,8 @@ def load_ferries( if 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: - item["co2_kg"] = item["distance"] * 0.02254 + item["co2_kg"] = item["distance"] * FERRY_CO2_KG_PER_KM geojson = from_terminal["routes"].get(item["to"]) if geojson: @@ -270,7 +273,12 @@ def load_coaches( ) -> list[StrDict]: """Load coaches.""" return load_road_transport( - "coach", "coaches", "coach_stations", data_dir, 0.027, route_distances + "coach", + "coaches", + "coach_stations", + data_dir, + COACH_CO2_KG_PER_KM, + route_distances, ) @@ -279,7 +287,7 @@ def load_buses( ) -> list[StrDict]: """Load buses.""" return load_road_transport( - "bus", "buses", "bus_stops", data_dir, 0.1, route_distances + "bus", "buses", "bus_stops", data_dir, BUS_CO2_KG_PER_KM, route_distances ) @@ -419,6 +427,8 @@ def load_cars(data_dir: str) -> list[StrDict]: if "distance" not in item: 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) from_label, to_label = car_route_labels(item) diff --git a/agenda/types.py b/agenda/types.py index 89d69b4..5450313 100644 --- a/agenda/types.py +++ b/agenda/types.py @@ -58,6 +58,12 @@ def airport_label(airport: StrDict) -> str: 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 class SchengenStay: """Represents a stay in the Schengen area.""" @@ -130,8 +136,13 @@ class Trip: if not (depart := (travel["depart"] and utils.as_date(travel["depart"]))): continue for when, from_or_to in ((self.start, "from"), (self.end, "to")): - if depart != when and travel[from_or_to] not in titles: - titles.append(travel[from_or_to]) + if depart == when: + continue + 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 @property diff --git a/agenda/utils.py b/agenda/utils.py index d8d30af..8c92790 100644 --- a/agenda/utils.py +++ b/agenda/utils.py @@ -6,7 +6,7 @@ from datetime import date, datetime, time, timedelta, timezone from time import time as unixtime from zoneinfo import ZoneInfo -from .types import StrDict +StrDict = dict[str, typing.Any] def as_date(d: datetime | date) -> date: diff --git a/scripts/import-car-journeys-from-timeline b/scripts/import-car-journeys-from-timeline new file mode 100644 index 0000000..aeb5daa --- /dev/null +++ b/scripts/import-car-journeys-from-timeline @@ -0,0 +1,15 @@ +#!/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()) diff --git a/templates/trip/stats.html b/templates/trip/stats.html index 38bf51b..52d3558 100644 --- a/templates/trip/stats.html +++ b/templates/trip/stats.html @@ -6,6 +6,14 @@ {% block title %}{{ heading }} - Edward Betts{% endblock %} +{% block style %} + +{% endblock %} + {% macro stat_list(id, label, counter, show_top=5) %}
{{ label }}: {{ counter | count }} @@ -24,6 +32,14 @@
{% 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 %}

Trip statistics

@@ -39,10 +55,21 @@ {% for transport_type, distance in distances_by_transport_type %}
{{ transport_type | title }}: {{ format_distance(distance) }}
{% endfor %} + {% if total_co2_kg %} +
CO₂: {{ format_co2(total_co2_kg) }}
+ {% endif %} + {% for transport_type, co2_kg in co2_by_transport_type %} +
{{ transport_type | title }} CO₂: {{ format_co2(co2_kg) }}
+ {% endfor %}
Flight segments: {{ overall_stats.flight_count }}
Train segments: {{ overall_stats.train_count }}
+ {% if co2_by_transport_type %} +
+ +
+ {% endif %}
@@ -79,12 +106,12 @@
Flight segments: {{ year_stats.flight_count or 0 }}
Train segments: {{ year_stats.train_count or 0 }}
{% if year_stats.co2_kg %} -
CO₂: - {% if year_stats.co2_kg >= 1000 %} - {{ "{:,.2f}".format(year_stats.co2_kg / 1000.0) }} tonnes - {% else %} - {{ "{:,.0f}".format(year_stats.co2_kg) }} kg - {% endif %} +
CO₂: {{ format_co2(year_stats.co2_kg) }}
+ {% for transport_type, co2_kg in year_stats.co2_by_transport_type.items() %} +
{{ transport_type | title }} CO₂: {{ format_co2(co2_kg) }}
+ {% endfor %} +
+
{% endif %}
@@ -119,3 +146,72 @@ {% endfor %} {% endblock %} + +{% block scripts %} + + +{% endblock %} diff --git a/templates/trip_page.html b/templates/trip_page.html index 7e873a7..10adf94 100644 --- a/templates/trip_page.html +++ b/templates/trip_page.html @@ -369,6 +369,8 @@ {% elif e.element_type in ("coach", "bus", "car") %} {% 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 %}
@@ -377,14 +379,14 @@ {% if item.operator %}{{ item.operator }}{% endif %}

- {% if item.depart.hour is defined and item.arrive.hour is defined %} - {{ item.depart.strftime("%H:%M") }} → {{ item.arrive.strftime("%H:%M") }} + {% if display_depart.hour is defined and display_arrive.hour is defined %} + {{ display_depart.strftime("%H:%M") }} → {{ display_arrive.strftime("%H:%M") }} {% endif %} {% if item.class %} {{ item.class }} {% endif %} - {% if item.depart.hour is defined and item.arrive.hour is defined %} - 🕒{{ trip_duration(item.depart, item.arrive) }} + {% if display_depart.hour is defined and display_arrive.hour is defined %} + 🕒{{ trip_duration(display_depart, display_arrive) }} {% endif %} {% if item.distance %} 🛤️ {{ "{:,.0f} km".format(item.distance) }} diff --git a/tests/test_car_journey_timeline.py b/tests/test_car_journey_timeline.py new file mode 100644 index 0000000..ad8b86d --- /dev/null +++ b/tests/test_car_journey_timeline.py @@ -0,0 +1,152 @@ +"""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" diff --git a/tests/test_trip.py b/tests/test_trip.py index efb3b9b..3301072 100644 --- a/tests/test_trip.py +++ b/tests/test_trip.py @@ -2,7 +2,7 @@ import json import pathlib -from datetime import date +from datetime import date, datetime, timezone import agenda.trip import pytest @@ -203,6 +203,9 @@ def test_load_cars_infers_route_labels_and_home_marker( assert cars[0]["to"] == "EMF" assert cars[0]["geojson_filename"] == "PCH_to_EMF" 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"] == { "name": "PCH", "type": "home", @@ -274,3 +277,49 @@ def test_get_trip_routes_includes_car_geojson() -> None: "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" diff --git a/tests/test_trip_page_route.py b/tests/test_trip_page_route.py index 488870d..6651bd9 100644 --- a/tests/test_trip_page_route.py +++ b/tests/test_trip_page_route.py @@ -1,6 +1,6 @@ """Regression tests for trip page route wiring and rendering.""" -from datetime import date, datetime +from datetime import date, datetime, timedelta, timezone import typing from unittest import mock @@ -307,3 +307,57 @@ def test_trip_page_uses_bed_icon_for_overnight_train_coach() -> None: assert response.status_code == 200 page = response.data.decode() 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 diff --git a/tests/test_trip_stats.py b/tests/test_trip_stats.py index 37d04b6..966cca9 100644 --- a/tests/test_trip_stats.py +++ b/tests/test_trip_stats.py @@ -3,6 +3,7 @@ from datetime import date import agenda +import pytest from agenda.stats import calculate_yearly_stats from agenda.types import Trip @@ -35,3 +36,22 @@ def test_new_country_respects_previously_visited() -> None: yearly_stats = calculate_yearly_stats(trips, {"CZ"}) 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, + } diff --git a/web_view.py b/web_view.py index 5388ada..2ddf9c4 100755 --- a/web_view.py +++ b/web_view.py @@ -3,6 +3,7 @@ """Web page to show upcoming events.""" import decimal +import functools import hashlib import importlib import inspect @@ -841,6 +842,16 @@ def sum_distances_by_transport_type(trips: list[Trip]) -> list[tuple[str, float] 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]: """Get Bristol home weather forecast, with date objects added for templates.""" from datetime import date as date_type @@ -911,6 +922,7 @@ def trip_past_list() -> str: total_distance=calc_total_distance(past), total_co2_kg=calc_total_co2_kg(past), distances_by_transport_type=sum_distances_by_transport_type(past), + co2_by_transport_type=sum_co2_by_transport_type(past), ) @@ -973,6 +985,7 @@ def trip_future_list() -> str: total_distance=calc_total_distance(current + future), total_co2_kg=calc_total_co2_kg(current + future), distances_by_transport_type=sum_distances_by_transport_type(current + future), + co2_by_transport_type=sum_co2_by_transport_type(current + future), ) @@ -1084,6 +1097,7 @@ def _timezone_from_coordinates(latitude: float, longitude: float) -> str | None: return None +@functools.lru_cache(maxsize=1) def _get_timezone_finder() -> typing.Any: """Get timezone finder instance if dependency is available.""" try: @@ -1098,6 +1112,94 @@ def _get_timezone_finder() -> typing.Any: 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]: """Build destination timezone metadata for the trip page.""" per_location: dict[tuple[str, str], list[str]] = defaultdict(list) @@ -1276,6 +1378,7 @@ def trip_page(start: str) -> str: trip, cache_only=True, ) + localize_trip_car_journey_display_times(trip, app.config["PERSONAL_DATA"]) return flask.render_template( "trip_page.html", @@ -1424,7 +1527,9 @@ def trip_stats() -> str: "trip/stats.html", count=len(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), + co2_by_transport_type=sum_co2_by_transport_type(trip_list), yearly_stats=yearly_stats, overall_stats=overall_stats, conferences=conferences,