152 lines
5 KiB
Python
152 lines
5 KiB
Python
"""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"
|