quoije/yaris-obd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add trip history and route replay

Browse Source 
Co-authored-by: Copilot <[email protected]>
This commit is contained in:
Mathieu Moran
2026-08-09 19:48:06 -04:00
co-authored by Copilot
parent 3a981f5c6d
commit d3bc313e7f
13 changed files with 1158 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_app_trip.py
+152
View File
@@ -0,0 +1,152 @@
import unittest
from datetime import datetime, timedelta
from modules import app_trip
class TripBuilderTest(unittest.TestCase):
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_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()
tests/test_tracking_logger.py
+45
View File
@@ -0,0 +1,45 @@
import unittest
from modules import app_sql
class TrackingLoggerTest(unittest.TestCase):
def test_logs_first_fix(self):
self.assertTrue(
app_sql.should_log_tracking((45.5, -73.5), None, None, 0)
)
def test_does_not_persist_parked_five_second_samples(self):
self.assertFalse(
app_sql.should_log_tracking(
(45.500001, -73.5), 0, (45.5, -73.5), 5
)
)
def test_logs_fresh_obd_motion(self):
self.assertTrue(
app_sql.should_log_tracking(
(45.5, -73.5), 2, (45.5, -73.5), 5
)
)
def test_logs_accumulated_gps_motion_without_obd(self):
self.assertTrue(
app_sql.should_log_tracking(
(45.5001, -73.5), None, (45.5, -73.5), 10
)
)
def test_logs_stationary_heartbeat_after_one_minute(self):
self.assertTrue(
app_sql.should_log_tracking(
(45.5, -73.5),
0,
(45.5, -73.5),
app_sql.IDLE_LOG_INTERVAL,
)
)
if __name__ == "__main__":
unittest.main()
tests/test_trip_sql.py
+73
View File
@@ -0,0 +1,73 @@
import os
import tempfile
import unittest
from datetime import datetime, timedelta
from modules import app_sql
class TripSqlTest(unittest.TestCase):
def setUp(self):
handle, self.path = tempfile.mkstemp(suffix=".db")
os.close(handle)
self.original_path = app_sql.sql_path
app_sql._conn = None
app_sql._last_trip_refresh = 0
app_sql.sql_path = self.path
self.conn = app_sql.get_conn()
def tearDown(self):
if app_sql._conn is not None:
app_sql._conn.close()
app_sql._conn = None
app_sql._last_trip_refresh = 0
app_sql.sql_path = self.original_path
os.unlink(self.path)
def insert_motion(self, start_id, count, started, latitude):
rows = [
(
start_id + index,
"Yaris",
"test",
started + timedelta(seconds=index * 5),
latitude + index * 0.0001,
-73.5,
20,
None,
0,
)
for index in range(count)
]
self.conn.executemany(
"""INSERT INTO tracking (
id, VN, NAME, DATETIME, GPSX, GPSY, SPEED, ODOMETER, UPLOADED
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
rows,
)
self.conn.commit()
def test_incremental_rebuild_preserves_trip_before_overlap_boundary(self):
started = datetime(2026, 1, 1, 8)
self.insert_motion(1, 160, started, 45.5)
self.insert_motion(161, 10, started + timedelta(minutes=30), 45.6)
app_sql.refresh_trips(force=True)
original = self.conn.execute(
"""SELECT START_TRACKING_ID, END_TRACKING_ID, DISTANCE_KM
FROM trips ORDER BY START_TRACKING_ID"""
).fetchall()
self.assertEqual(len(original), 2)
self.insert_motion(171, 1, started + timedelta(minutes=31), 45.601)
app_sql._last_trip_refresh = 0
app_sql.refresh_trips()
rebuilt = self.conn.execute(
"""SELECT START_TRACKING_ID, END_TRACKING_ID, DISTANCE_KM
FROM trips ORDER BY START_TRACKING_ID"""
).fetchall()
self.assertEqual(rebuilt[0], original[0])
if __name__ == "__main__":
unittest.main()