Mathieu Moran
f5f20c315b
Trips starting/ending on speed-only rows violated the trips.*GPSX NOT NULL constraint. Use the first/last GPS-bearing row for start/end coordinates, and skip a trip that never got any GPS fix (no location to record).
196 lines
6.4 KiB
Python
196 lines
6.4 KiB
Python
import unittest
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
from modules import app_trip
|
|
|
|
|
|
class TripBuilderTest(unittest.TestCase):
|
|
def setUp(self):
|
|
# BUG-10 guard suppresses trips until the car ECU has been reached
|
|
# once. Trip-building tests simulate a connected car.
|
|
self._car_connected = patch(
|
|
"modules.app_obd.was_car_ever_connected", return_value=True
|
|
)
|
|
self._car_connected.start()
|
|
|
|
def tearDown(self):
|
|
self._car_connected.stop()
|
|
|
|
def rows_for_route(self, coordinates, speeds=None, start_id=1):
|
|
started = datetime(2026, 1, 1, 8)
|
|
speeds = speeds or [0] * len(coordinates)
|
|
return [
|
|
(
|
|
start_id + index,
|
|
started + timedelta(seconds=index * 5),
|
|
latitude,
|
|
longitude,
|
|
speeds[index],
|
|
)
|
|
for index, (latitude, longitude) in enumerate(coordinates)
|
|
]
|
|
|
|
def test_builds_gps_trip_after_sustained_motion(self):
|
|
rows = self.rows_for_route([
|
|
(45.5000, -73.5000),
|
|
(45.5001, -73.5000),
|
|
(45.5002, -73.5000),
|
|
(45.5003, -73.5000),
|
|
])
|
|
|
|
trips = app_trip.build_trip_summaries(rows)
|
|
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(trips[0]["start_tracking_id"], 1)
|
|
self.assertEqual(trips[0]["end_tracking_id"], 4)
|
|
self.assertGreater(trips[0]["distance_km"], 0.03)
|
|
self.assertGreater(trips[0]["average_speed_kmh"], 0)
|
|
self.assertGreater(trips[0]["max_speed_kmh"], 0)
|
|
|
|
def test_ignores_stationary_jitter_and_one_bad_coordinate(self):
|
|
rows = self.rows_for_route([
|
|
(45.500000, -73.500000),
|
|
(45.500010, -73.500000),
|
|
("\x0045.500020", -73.500000),
|
|
("not-a-coordinate", -73.500000),
|
|
(45.500010, -73.500000),
|
|
])
|
|
|
|
self.assertEqual(app_trip.build_trip_summaries(rows), [])
|
|
|
|
def test_inactivity_gap_splits_trips(self):
|
|
first = self.rows_for_route([
|
|
(45.5000, -73.5000),
|
|
(45.5001, -73.5000),
|
|
(45.5002, -73.5000),
|
|
(45.5003, -73.5000),
|
|
])
|
|
second = self.rows_for_route([
|
|
(45.5100, -73.5100),
|
|
(45.5101, -73.5100),
|
|
(45.5102, -73.5100),
|
|
(45.5103, -73.5100),
|
|
], start_id=5)
|
|
second = [
|
|
(row[0], row[1] + timedelta(minutes=10), row[2], row[3], row[4])
|
|
for row in second
|
|
]
|
|
|
|
trips = app_trip.build_trip_summaries(first + second)
|
|
|
|
self.assertEqual(len(trips), 2)
|
|
|
|
def test_obd_speed_detects_slow_trip(self):
|
|
rows = self.rows_for_route(
|
|
[(45.5, -73.5), (45.5, -73.5)],
|
|
speeds=[0, 5],
|
|
)
|
|
|
|
trips = app_trip.build_trip_summaries(rows)
|
|
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(trips[0]["max_speed_kmh"], 5)
|
|
|
|
def test_obd_trip_counts_sub_threshold_gps_distance(self):
|
|
rows = self.rows_for_route(
|
|
[(45.5 + index * 0.00006, -73.5) for index in range(10)],
|
|
speeds=[5] * 10,
|
|
)
|
|
|
|
trips = app_trip.build_trip_summaries(rows)
|
|
|
|
self.assertGreater(trips[0]["distance_km"], 0.05)
|
|
self.assertGreater(trips[0]["average_speed_kmh"], 0)
|
|
|
|
def test_combined_obd_and_gps_motion_creates_one_trip(self):
|
|
rows = self.rows_for_route(
|
|
[(45.5 + index * 0.0001, -73.5) for index in range(5)],
|
|
speeds=[20] * 5,
|
|
)
|
|
|
|
trips = app_trip.build_trip_summaries(rows)
|
|
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(trips[0]["start_tracking_id"], 1)
|
|
self.assertEqual(trips[0]["end_tracking_id"], 5)
|
|
|
|
def test_speed_only_rows_without_any_gps_are_skipped(self):
|
|
started = datetime(2026, 1, 1, 8)
|
|
rows = [
|
|
(index + 1, started + timedelta(seconds=index * 5), None, None, 30)
|
|
for index in range(4)
|
|
]
|
|
|
|
trips = app_trip.build_trip_summaries(rows)
|
|
|
|
# No GPS fix at all -> no location to record -> no trip (the speed-only
|
|
# rows remain in the DB but can't form a geolocated trip).
|
|
self.assertEqual(len(trips), 0)
|
|
|
|
def test_speed_only_start_uses_first_gps_fix_for_start_coords(self):
|
|
started = datetime(2026, 1, 1, 8)
|
|
rows = [
|
|
(1, started, None, None, 30),
|
|
(2, started + timedelta(seconds=5), None, None, 30),
|
|
(3, started + timedelta(seconds=10), 45.5000, -73.5000, 30),
|
|
(4, started + timedelta(seconds=15), 45.5001, -73.5000, 30),
|
|
(5, started + timedelta(seconds=20), 45.5002, -73.5000, 30),
|
|
]
|
|
|
|
trips = app_trip.build_trip_summaries(rows)
|
|
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(trips[0]["start_tracking_id"], 1)
|
|
self.assertEqual(trips[0]["end_tracking_id"], 5)
|
|
self.assertEqual(trips[0]["start_gpsx"], 45.5)
|
|
self.assertEqual(trips[0]["start_gpsy"], -73.5)
|
|
|
|
def test_clock_rollback_does_not_merge_motion_episodes(self):
|
|
first = self.rows_for_route(
|
|
[(45.5, -73.5), (45.5, -73.5)],
|
|
speeds=[0, 5],
|
|
)
|
|
second = self.rows_for_route(
|
|
[(45.51, -73.51), (45.51, -73.51)],
|
|
speeds=[0, 5],
|
|
start_id=3,
|
|
)
|
|
second = [
|
|
(row[0], row[1] - timedelta(hours=1), row[2], row[3], row[4])
|
|
for row in second
|
|
]
|
|
|
|
trips = app_trip.build_trip_summaries(first + second)
|
|
|
|
self.assertEqual(len(trips), 2)
|
|
|
|
def test_route_downsampling_keeps_endpoints(self):
|
|
rows = self.rows_for_route(
|
|
[(45.5 + index / 10000, -73.5) for index in range(20)]
|
|
)
|
|
|
|
points = app_trip.route_points(rows, max_points=5)
|
|
|
|
self.assertEqual(len(points), 5)
|
|
self.assertEqual(points[0]["id"], 1)
|
|
self.assertEqual(points[-1]["id"], 20)
|
|
self.assertEqual(points[0]["speed"], 0)
|
|
self.assertGreater(points[-1]["speed"], 0)
|
|
self.assertEqual(points[-1]["speed_source"], "gps")
|
|
|
|
def test_route_filters_implausible_gps_outlier(self):
|
|
rows = self.rows_for_route([
|
|
(45.5000, -73.5000),
|
|
(0, 0),
|
|
(45.5001, -73.5000),
|
|
])
|
|
|
|
points = app_trip.route_points(rows)
|
|
|
|
self.assertEqual([point["id"] for point in points], [1, 3])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|