Add timeline car import and CO2 trip stats
This commit is contained in:
parent
38dbb2b4e7
commit
3747e83ade
14 changed files with 1038 additions and 29 deletions
474
agenda/car_journey_timeline.py
Normal file
474
agenda/car_journey_timeline.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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),
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
15
scripts/import-car-journeys-from-timeline
Normal file
15
scripts/import-car-journeys-from-timeline
Normal file
|
|
@ -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())
|
||||
|
|
@ -6,6 +6,14 @@
|
|||
|
||||
{% block title %}{{ heading }} - Edward Betts{% endblock %}
|
||||
|
||||
{% block style %}
|
||||
<style>
|
||||
.co2-chart {
|
||||
height: 240px;
|
||||
}
|
||||
</style>
|
||||
{% endblock %}
|
||||
|
||||
{% macro stat_list(id, label, counter, show_top=5) %}
|
||||
<div class="mb-2">
|
||||
<strong>{{ label }}:</strong> {{ counter | count }}
|
||||
|
|
@ -24,6 +32,14 @@
|
|||
</div>
|
||||
{% 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 %}
|
||||
<div class="container-fluid">
|
||||
<h1>Trip statistics</h1>
|
||||
|
|
@ -39,10 +55,21 @@
|
|||
{% for transport_type, distance in distances_by_transport_type %}
|
||||
<div class="ms-3">{{ transport_type | title }}: {{ format_distance(distance) }}</div>
|
||||
{% 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 class="col-md-6">
|
||||
<div>Flight segments: {{ overall_stats.flight_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>
|
||||
|
||||
|
|
@ -79,12 +106,12 @@
|
|||
<div>Flight segments: {{ year_stats.flight_count or 0 }}</div>
|
||||
<div>Train segments: {{ year_stats.train_count or 0 }}</div>
|
||||
{% if year_stats.co2_kg %}
|
||||
<div>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 %}
|
||||
<div>CO₂: {{ format_co2(year_stats.co2_kg) }}</div>
|
||||
{% for transport_type, co2_kg in year_stats.co2_by_transport_type.items() %}
|
||||
<div class="ms-3">{{ transport_type | title }} CO₂: {{ format_co2(co2_kg) }}</div>
|
||||
{% endfor %}
|
||||
<div class="co2-chart mt-3">
|
||||
<canvas id="co2-chart-{{ year }}"></canvas>
|
||||
</div>
|
||||
{% endif %}
|
||||
</div>
|
||||
|
|
@ -119,3 +146,72 @@
|
|||
{% endfor %}
|
||||
</div>
|
||||
{% 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,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 %}
|
||||
<div class="trip-transport-card my-1">
|
||||
<div class="card-body">
|
||||
<h5 class="card-title">
|
||||
|
|
@ -377,14 +379,14 @@
|
|||
{% if item.operator %}<small class="text-muted fw-normal">{{ item.operator }}</small>{% endif %}
|
||||
</h5>
|
||||
<p class="card-text">
|
||||
{% 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 %}
|
||||
<span class="badge bg-info text-nowrap">{{ item.class }}</span>
|
||||
{% endif %}
|
||||
{% if item.depart.hour is defined and item.arrive.hour is defined %}
|
||||
<span class="text-muted">🕒{{ trip_duration(item.depart, item.arrive) }}</span>
|
||||
{% if display_depart.hour is defined and display_arrive.hour is defined %}
|
||||
<span class="text-muted">🕒{{ trip_duration(display_depart, display_arrive) }}</span>
|
||||
{% endif %}
|
||||
{% if item.distance %}
|
||||
<span class="text-muted">🛤️ {{ "{:,.0f} km".format(item.distance) }}</span>
|
||||
|
|
|
|||
152
tests/test_car_journey_timeline.py
Normal file
152
tests/test_car_journey_timeline.py
Normal file
|
|
@ -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"
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
|
|
|
|||
105
web_view.py
105
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,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue