diff --git a/agenda/car_journey_timeline.py b/agenda/car_journey_timeline.py deleted file mode 100644 index d3518ee..0000000 --- a/agenda/car_journey_timeline.py +++ /dev/null @@ -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 diff --git a/agenda/car_journey_yaml.py b/agenda/car_journey_yaml.py index e8cb1f9..cbf360a 100644 --- a/agenda/car_journey_yaml.py +++ b/agenda/car_journey_yaml.py @@ -9,16 +9,12 @@ import re import typing import unicodedata from dataclasses import dataclass -from datetime import date, datetime +from datetime import date 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 @@ -604,8 +600,6 @@ 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: @@ -645,15 +639,15 @@ def write_route_files( 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.""" 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_datetime(item["depart"]), - utils.as_datetime(item["arrive"]), + utils.as_date(item["depart"]), + utils.as_date(item["arrive"]), trip_module.route_filename_without_extension(route), ) diff --git a/agenda/stats.py b/agenda/stats.py index 91493c9..2e4dc67 100644 --- a/agenda/stats.py +++ b/agenda/stats.py @@ -1,15 +1,10 @@ """Trip statistic functions.""" from collections import defaultdict -import typing -from typing import TYPE_CHECKING, Counter, Mapping +from typing import Counter, Mapping import agenda - -if TYPE_CHECKING: - from agenda.types import Trip - -StrDict = dict[str, typing.Any] +from agenda.types import StrDict, Trip, airport_label 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"][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()) @@ -76,8 +69,6 @@ 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(): @@ -89,12 +80,6 @@ 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 6521ab7..c6573d3 100644 --- a/agenda/trip.py +++ b/agenda/trip.py @@ -17,12 +17,6 @@ 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.""" @@ -197,11 +191,13 @@ 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"] * TRAIN_CO2_KG_PER_KM + leg["co2_kg"] = leg["distance"] * 0.037 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 @@ -224,8 +220,9 @@ 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"] * FERRY_CO2_KG_PER_KM + item["co2_kg"] = item["distance"] * 0.02254 geojson = from_terminal["routes"].get(item["to"]) if geojson: @@ -273,12 +270,7 @@ def load_coaches( ) -> list[StrDict]: """Load coaches.""" return load_road_transport( - "coach", - "coaches", - "coach_stations", - data_dir, - COACH_CO2_KG_PER_KM, - route_distances, + "coach", "coaches", "coach_stations", data_dir, 0.027, route_distances ) @@ -287,7 +279,7 @@ def load_buses( ) -> list[StrDict]: """Load buses.""" 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: 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 5450313..89d69b4 100644 --- a/agenda/types.py +++ b/agenda/types.py @@ -58,12 +58,6 @@ 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.""" @@ -136,13 +130,8 @@ 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: - 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) + if depart != when and travel[from_or_to] not in titles: + titles.append(travel[from_or_to]) return titles @property diff --git a/agenda/utils.py b/agenda/utils.py index 8c92790..d8d30af 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 -StrDict = dict[str, typing.Any] +from .types import StrDict def as_date(d: datetime | date) -> date: diff --git a/scripts/import-car-journeys-from-timeline b/scripts/import-car-journeys-from-timeline deleted file mode 100644 index aeb5daa..0000000 --- a/scripts/import-car-journeys-from-timeline +++ /dev/null @@ -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()) diff --git a/templates/conference_list.html b/templates/conference_list.html index 70e5f10..e20b154 100644 --- a/templates/conference_list.html +++ b/templates/conference_list.html @@ -212,28 +212,6 @@ tr.conf-hl > td {