quoije/yaris-obd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
786d5b8f3ed9285a81cd4ada97d148a21fe35b3f
yaris-obd/modules/app_sql.py
T
Mathieu Moran e2f2314318 fix(phase26): persist DTC/MIL state + server speed_source (BUG-03, BUG-05) 
BUG-03: DTC/MIL codes were never persisted; lost on OBD disconnect or reboot.
- Add new dtc table schema (code, description, MIL, first/last seen, IS_SET)
  and a migration that rebuilds the original unused NAME/STATUS/DTC_ID table.
- Persist MIL + codes to SQLite on each successful OBD poll (_persist_dtc_state).
- /api/dtc falls back to the last-persisted snapshot when OBD is offline/stale.
- Fix get_overview_stats to count the real dtc table (was querying non-existent
  dtc_codes table).
- Add cleared/since tracking via IS_SET + FIRST_SEEN_AT/LAST_SEEN_AT.

BUG-05: server trip replay hardcoded speed_source:'gps'; mirror Pi's
'obd if speed>0 else gps' logic in server get_trip_route.

Adds tests/test_dtc_persistence.py (7 tests).
2026-08-11 00:38:05 -04:00

832 lines
30 KiB
Python
Raw Blame History

import sqlite3
import os
import time
from datetime import datetime
from sqlite3 import Error
from threading import Thread, Lock
from modules import app_car, app_account, app_gps
from modules import app_trip
# Resolve db/sql/ relative to the repo root regardless of the process's cwd
# (previously a relative "../db/sql/" path, which broke depending on how/from
# where app.py was launched).
_BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_SQL_DIR = os.path.join(_BASE_DIR, "db", "sql")
account = app_account.Account()
car = app_car.Car()
sCarModel = ''.join(letter for letter in car.model if letter.isalnum())
sql_file = account.name + "_" + sCarModel + ".db"
sql_path = os.path.join(_SQL_DIR, sql_file)
LOG_INTERVAL = 5 # seconds between trip-tracking snapshots
IDLE_LOG_INTERVAL = 60
GPS_LOG_DISTANCE_METERS = 8
TRIP_REFRESH_INTERVAL = 30
TRIP_REBUILD_OVERLAP_ROWS = 120
TRIP_BUILD_VERSION = 2
_conn = None
_conn_lock = Lock()
_log_thread = None
_last_trip_refresh = 0
SQL_CREATE_TRACKING_TABLE = """CREATE TABLE IF NOT EXISTS tracking (
id integer PRIMARY KEY,
VN text NOT NULL,
NAME text NOT NULL,
DATETIME datetime,
GPSX float,
GPSY float,
SPEED float,
ODOMETER float,
UPLOADED integer DEFAULT 0
);"""
SQL_CREATE_DTC_TABLE = """CREATE TABLE IF NOT EXISTS dtc (
id integer PRIMARY KEY,
code text NOT NULL UNIQUE,
description text,
MIL integer,
FIRST_SEEN_AT datetime,
LAST_SEEN_AT datetime,
IS_SET integer NOT NULL DEFAULT 1
);"""
SQL_CREATE_SETTINGS_TABLE = """CREATE TABLE IF NOT EXISTS settings (
key text PRIMARY KEY,
value text
);"""
SQL_CREATE_DASHCAM_UPLOADS_TABLE = """CREATE TABLE IF NOT EXISTS dashcam_uploads (
id integer PRIMARY KEY,
RELATIVE_PATH text NOT NULL,
SHA256 text NOT NULL,
UPLOADED_AT datetime,
UNIQUE(RELATIVE_PATH, SHA256)
);"""
# Cache of already-computed sha256 hashes, keyed by (path, size, mtime) so a
# file's hash never needs recomputing once copied from the dashcam (its size
# and mtime won't change afterward). Avoids re-hashing the entire, possibly
# 20GB+, local backlog on every background-upload loop iteration and every
# "pending files" list request from the phone-relay feature.
SQL_CREATE_DASHCAM_HASH_CACHE_TABLE = """CREATE TABLE IF NOT EXISTS dashcam_hash_cache (
RELATIVE_PATH text PRIMARY KEY,
SIZE integer NOT NULL,
MTIME real NOT NULL,
SHA256 text NOT NULL
);"""
SQL_CREATE_TRIPS_TABLE = """CREATE TABLE IF NOT EXISTS trips (
id integer PRIMARY KEY,
START_TRACKING_ID integer NOT NULL,
END_TRACKING_ID integer NOT NULL,
START_DATETIME datetime NOT NULL,
END_DATETIME datetime NOT NULL,
START_GPSX float NOT NULL,
START_GPSY float NOT NULL,
END_GPSX float NOT NULL,
END_GPSY float NOT NULL,
DISTANCE_KM float NOT NULL,
DURATION_SECONDS integer NOT NULL,
POINT_COUNT integer NOT NULL,
AVERAGE_SPEED_KMH float NOT NULL DEFAULT 0,
MAX_SPEED_KMH float NOT NULL DEFAULT 0,
START_ODOMETER float,
END_ODOMETER float
);"""
SQL_CREATE_TRIP_BUILD_STATE_TABLE = """CREATE TABLE IF NOT EXISTS trip_build_state (
id integer PRIMARY KEY CHECK (id = 1),
LAST_TRACKING_ID integer NOT NULL,
BUILD_VERSION integer NOT NULL DEFAULT 1
);"""
SQL_CREATE_ALERT_STATE_TABLE = """CREATE TABLE IF NOT EXISTS alert_state (
ALERT_KEY text PRIMARY KEY,
ACTIVE integer NOT NULL DEFAULT 0,
ACTIVE_SINCE real,
LAST_ATTEMPT_AT real,
LAST_SENT_AT real,
RECOVERY_PENDING integer NOT NULL DEFAULT 0,
LAST_RECOVERY_AT real,
LAST_ERROR text,
LAST_VALUE text
);"""
def create_conn(db_file):
"""Open a SQLite connection to db_file, creating its parent folder if needed.
Returns the open connection (or None on failure) - previously this closed
the connection immediately in a `finally` block and returned nothing,
so every caller always got a dead `None` connection.
"""
os.makedirs(os.path.dirname(db_file), exist_ok=True)
conn = None
try:
conn = sqlite3.connect(db_file, check_same_thread=False)
print("SQLite library version:", sqlite3.sqlite_version)
except Error as e:
print(e)
return conn
def create_table(conn, create_table_sql):
""" create a table from the create_table_sql statement
:param conn: Connection object
:param create_table_sql: a CREATE TABLE statement
:return:
"""
try:
c = conn.cursor()
c.execute(create_table_sql)
except Error as e:
print(e)
def get_conn():
"""Return the shared SQLite connection, opening + initializing tables on first use."""
global _conn
if _conn is None:
_conn = create_conn(sql_path)
if _conn is not None:
create_table(_conn, SQL_CREATE_TRACKING_TABLE)
create_table(_conn, SQL_CREATE_DTC_TABLE)
create_table(_conn, SQL_CREATE_SETTINGS_TABLE)
create_table(_conn, SQL_CREATE_DASHCAM_UPLOADS_TABLE)
create_table(_conn, SQL_CREATE_DASHCAM_HASH_CACHE_TABLE)
create_table(_conn, SQL_CREATE_TRIPS_TABLE)
create_table(_conn, SQL_CREATE_TRIP_BUILD_STATE_TABLE)
create_table(_conn, SQL_CREATE_ALERT_STATE_TABLE)
_migrate_tracking_table(_conn)
_migrate_trips_table(_conn)
_migrate_trip_build_state_table(_conn)
_migrate_dtc_table(_conn)
else:
print("Error! cannot create the database connection.")
return _conn
def _migrate_tracking_table(conn):
"""Add the UPLOADED column to a pre-existing tracking table that predates
the remote-upload feature (SQLite has no "ADD COLUMN IF NOT EXISTS")."""
existing_columns = {row[1] for row in conn.execute("PRAGMA table_info(tracking)").fetchall()}
if "UPLOADED" not in existing_columns:
conn.execute("ALTER TABLE tracking ADD COLUMN UPLOADED integer DEFAULT 0")
conn.commit()
def _migrate_trips_table(conn):
existing_columns = {
row[1] for row in conn.execute("PRAGMA table_info(trips)").fetchall()
}
additions = {
"AVERAGE_SPEED_KMH": "float NOT NULL DEFAULT 0",
"MAX_SPEED_KMH": "float NOT NULL DEFAULT 0",
"START_ODOMETER": "float",
"END_ODOMETER": "float",
}
for name, definition in additions.items():
if name not in existing_columns:
conn.execute(f"ALTER TABLE trips ADD COLUMN {name} {definition}")
conn.commit()
def _migrate_trip_build_state_table(conn):
existing_columns = {
row[1] for row in conn.execute("PRAGMA table_info(trip_build_state)").fetchall()
}
if "BUILD_VERSION" not in existing_columns:
conn.execute(
"ALTER TABLE trip_build_state "
"ADD COLUMN BUILD_VERSION integer NOT NULL DEFAULT 1"
)
conn.commit()
def _migrate_dtc_table(conn):
"""Rebuild the dtc table if it predates the persisted-DTC schema.
The original dtc table (NAME/STATUS/DTC_ID) was never written to by any
code path, so dropping and recreating it with the new schema is safe.
"""
existing_columns = {
row[1] for row in conn.execute("PRAGMA table_info(dtc)").fetchall()
}
if existing_columns and "code" not in existing_columns:
conn.execute("DROP TABLE IF EXISTS dtc")
create_table(conn, SQL_CREATE_DTC_TABLE)
conn.commit()
def save_dtc_state(mil_on, dtc_codes, recorded_at=None):
"""Persist a DTC snapshot (MIL + currently-set codes) to SQLite.
Last-known state can then survive an OBD disconnect or reboot. Codes that
were previously set but are no longer detected are marked cleared
(IS_SET = 0) and LAST_SEEN_AT is bumped, while FIRST_SEEN_AT preserves when
each code was first observed.
"""
conn = get_conn()
if conn is None:
return
now_str = (recorded_at or datetime.now()).isoformat()
current_codes = {c["code"] for c in dtc_codes if c.get("code")}
with _conn_lock:
try:
conn.execute(
"INSERT INTO settings (key, value) VALUES ('dtc_mil', ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
("1" if mil_on else "0",),
)
conn.execute(
"INSERT INTO settings (key, value) VALUES ('dtc_last_polled_at', ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(now_str,),
)
if current_codes:
placeholders = ",".join("?" for _ in current_codes)
conn.execute(
f"UPDATE dtc SET IS_SET = 0, LAST_SEEN_AT = ? "
f"WHERE code NOT IN ({placeholders})",
(now_str,) + tuple(current_codes),
)
else:
conn.execute(
"UPDATE dtc SET IS_SET = 0, LAST_SEEN_AT = ? WHERE IS_SET = 1",
(now_str,),
)
for code in sorted(current_codes):
description = next(
(c["description"] for c in dtc_codes if c.get("code") == code),
None,
)
conn.execute(
"""INSERT INTO dtc (code, description, MIL, FIRST_SEEN_AT,
LAST_SEEN_AT, IS_SET)
VALUES (?, ?, ?, ?, ?, 1)
ON CONFLICT(code) DO UPDATE SET
description = excluded.description,
MIL = excluded.MIL,
LAST_SEEN_AT = excluded.LAST_SEEN_AT,
IS_SET = 1""",
(code, description, 1 if mil_on else 0, now_str, now_str),
)
conn.commit()
except sqlite3.Error as e:
print("Failed to persist DTC state:", e)
def get_persisted_dtc_state():
"""Return the last-persisted MIL + DTC codes for display when OBD is offline."""
conn = get_conn()
if conn is None:
return {"mil": None, "codes": []}
with _conn_lock:
mil_row = conn.execute(
"SELECT value FROM settings WHERE key = 'dtc_mil'"
).fetchone()
codes = conn.execute(
"SELECT code, description FROM dtc WHERE IS_SET = 1 ORDER BY code ASC"
).fetchall()
mil = None
if mil_row and mil_row[0] not in (None, ""):
mil = mil_row[0] == "1"
return {
"mil": mil,
"codes": [{"code": r[0], "description": r[1]} for r in codes],
}
def count_active_dtcs():
"""Number of currently-set DTC codes persisted in SQLite (cleared excluded)."""
conn = get_conn()
if conn is None:
return 0
with _conn_lock:
row = conn.execute(
"SELECT COUNT(*) FROM dtc WHERE IS_SET = 1"
).fetchone()
return row[0] if row else 0
def insert_tracking(gps_x, gps_y, speed, odometer=None):
"""Insert a single trip-tracking snapshot row (GPS position + speed)."""
conn = get_conn()
if conn is None:
return False
sql = """INSERT INTO tracking (VN, NAME, DATETIME, GPSX, GPSY, SPEED, ODOMETER)
VALUES (?, ?, ?, ?, ?, ?, ?)"""
with _conn_lock:
try:
latitude = float(str(gps_x).replace("\x00", "").strip())
longitude = float(str(gps_y).replace("\x00", "").strip())
conn.execute(
sql,
(car.model, account.name, datetime.now(), latitude, longitude, speed, odometer),
)
conn.commit()
return True
except (Error, TypeError, ValueError) as e:
print("Failed to insert tracking row:", e)
return False
def should_log_tracking(position, speed, last_position, seconds_since_log):
if last_position is None:
return True
if speed is not None and speed > 1:
return True
distance_meters = app_trip.haversine_km(last_position, position) * 1000
return (
distance_meters >= GPS_LOG_DISTANCE_METERS
or seconds_since_log >= IDLE_LOG_INTERVAL
)
def refresh_trips(force=False):
"""Rebuild the derived trip index when new raw tracking rows are available."""
global _last_trip_refresh
conn = get_conn()
if conn is None:
return 0
with _conn_lock:
last_tracking_id = conn.execute(
"SELECT COALESCE(MAX(id), 0) FROM tracking"
).fetchone()[0]
state = conn.execute(
"SELECT LAST_TRACKING_ID, BUILD_VERSION "
"FROM trip_build_state WHERE id = 1"
).fetchone()
recently_refreshed = (
_last_trip_refresh
and time.monotonic() - _last_trip_refresh < TRIP_REFRESH_INTERVAL
)
if (
not force
and state is not None
and state[1] == TRIP_BUILD_VERSION
and (state[0] == last_tracking_id or recently_refreshed)
):
return conn.execute("SELECT COUNT(*) FROM trips").fetchone()[0]
rebuild_start_id = 1
if not force and state is not None and state[1] == TRIP_BUILD_VERSION:
rebuild_start_id = max(1, state[0] - TRIP_REBUILD_OVERLAP_ROWS)
intersecting_trip = conn.execute(
"""SELECT MIN(START_TRACKING_ID) FROM trips
WHERE END_TRACKING_ID >= ?""",
(rebuild_start_id,),
).fetchone()
if intersecting_trip[0] is not None:
rebuild_start_id = min(rebuild_start_id, intersecting_trip[0])
rows = conn.execute(
"""SELECT id, DATETIME, GPSX, GPSY, SPEED, ODOMETER
FROM tracking WHERE id >= ? ORDER BY id ASC""",
(rebuild_start_id,),
).fetchall()
trips = app_trip.build_trip_summaries(rows)
with _conn_lock:
try:
conn.execute(
"DELETE FROM trips WHERE END_TRACKING_ID >= ?",
(rebuild_start_id,),
)
conn.executemany(
"""INSERT INTO trips (
id, START_TRACKING_ID, END_TRACKING_ID,
START_DATETIME, END_DATETIME,
START_GPSX, START_GPSY, END_GPSX, END_GPSY,
DISTANCE_KM, DURATION_SECONDS, POINT_COUNT,
AVERAGE_SPEED_KMH, MAX_SPEED_KMH,
START_ODOMETER, END_ODOMETER
) VALUES (
:id, :start_tracking_id, :end_tracking_id,
:start_datetime, :end_datetime,
:start_gpsx, :start_gpsy, :end_gpsx, :end_gpsy,
:distance_km, :duration_seconds, :point_count,
:average_speed_kmh, :max_speed_kmh,
:start_odometer, :end_odometer
)""",
trips,
)
conn.execute(
"""INSERT INTO trip_build_state (
id, LAST_TRACKING_ID, BUILD_VERSION
) VALUES (1, ?, ?)
ON CONFLICT(id) DO UPDATE
SET LAST_TRACKING_ID = excluded.LAST_TRACKING_ID,
BUILD_VERSION = excluded.BUILD_VERSION""",
(last_tracking_id, TRIP_BUILD_VERSION),
)
conn.commit()
_last_trip_refresh = time.monotonic()
except Error:
conn.rollback()
raise
with _conn_lock:
return conn.execute("SELECT COUNT(*) FROM trips").fetchone()[0]
def get_overview_stats():
conn = get_conn()
with _conn_lock:
row = conn.execute("""
SELECT COUNT(*), COALESCE(SUM(DISTANCE_KM),0),
COALESCE(SUM(DURATION_SECONDS),0), COALESCE(MAX(MAX_SPEED_KMH),0)
FROM trips
""").fetchone()
odo_row = conn.execute("""
SELECT MAX(ODOMETER) FROM tracking WHERE ODOMETER IS NOT NULL AND ODOMETER > 0
""").fetchone()
dtc_row = conn.execute(
"SELECT COUNT(*) FROM dtc WHERE IS_SET = 1"
).fetchone()
from modules import app_video as _av
try:
video_count = sum(1 for _ in _av.list_local_videos().get("files", []))
except Exception:
video_count = 0
return {
"trip_count": row[0],
"total_km": row[1],
"total_duration_s": row[2],
"all_time_max_speed": row[3],
"active_dtc": dtc_row[0] if dtc_row else 0,
"video_count": video_count,
"odometer_km": odo_row[0] if odo_row and odo_row[0] else None,
}
def list_trips():
refresh_trips()
conn = get_conn()
if conn is None:
return []
with _conn_lock:
rows = conn.execute(
"""SELECT id, START_DATETIME, END_DATETIME,
START_GPSX, START_GPSY, END_GPSX, END_GPSY,
DISTANCE_KM, DURATION_SECONDS, POINT_COUNT,
AVERAGE_SPEED_KMH, MAX_SPEED_KMH,
START_ODOMETER, END_ODOMETER
FROM trips ORDER BY START_DATETIME ASC"""
).fetchall()
# If real OBD odometer is available use it; otherwise fall back to cumulative distance
result = []
cumulative_km = 0.0
for row in rows:
dist = row[7] or 0
start_odo = row[12] if row[12] is not None else round(cumulative_km, 1)
end_odo = row[13] if row[13] is not None else round(cumulative_km + dist, 1)
cumulative_km += dist
result.append({
"id": row[0],
"start_datetime": row[1],
"end_datetime": row[2],
"start": {"latitude": row[3], "longitude": row[4]},
"end": {"latitude": row[5], "longitude": row[6]},
"distance_km": row[7],
"duration_seconds": row[8],
"point_count": row[9],
"average_speed_kmh": row[10],
"max_speed_kmh": row[11],
"odo_start_km": start_odo,
"odo_end_km": end_odo,
})
result.reverse() # newest first for display
return result
def get_trip_route(trip_id, max_points=2000):
refresh_trips()
conn = get_conn()
if conn is None:
return None
with _conn_lock:
trip = conn.execute(
"""SELECT id, START_DATETIME, END_DATETIME, DISTANCE_KM,
DURATION_SECONDS, POINT_COUNT,
AVERAGE_SPEED_KMH, MAX_SPEED_KMH,
START_TRACKING_ID, END_TRACKING_ID
FROM trips WHERE id = ?""",
(trip_id,),
).fetchone()
if trip is None:
return None
rows = conn.execute(
"""SELECT id, DATETIME, GPSX, GPSY, SPEED
FROM tracking WHERE id BETWEEN ? AND ?
ORDER BY DATETIME ASC, id ASC""",
(trip[8], trip[9]),
).fetchall()
return {
"trip": {
"id": trip[0],
"start_datetime": trip[1],
"end_datetime": trip[2],
"distance_km": trip[3],
"duration_seconds": trip[4],
"point_count": trip[5],
"average_speed_kmh": trip[6],
"max_speed_kmh": trip[7],
},
"points": app_trip.route_points(rows, max_points=max_points),
}
def get_setting(key, default=None):
"""Read a persisted UI/app setting (e.g. the dashcam-upload toggle)."""
conn = get_conn()
if conn is None:
return default
with _conn_lock:
cur = conn.execute("SELECT value FROM settings WHERE key = ?", (key,))
row = cur.fetchone()
return row[0] if row is not None else default
def set_setting(key, value):
conn = get_conn()
if conn is None:
return
with _conn_lock:
conn.execute(
"INSERT INTO settings (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(key, value),
)
conn.commit()
def set_settings(settings):
"""Persist a validated group of settings atomically."""
conn = get_conn()
if conn is None:
raise RuntimeError("Settings database is not available.")
with _conn_lock:
try:
conn.executemany(
"INSERT INTO settings (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
settings.items(),
)
conn.commit()
except sqlite3.Error:
conn.rollback()
raise
def get_alert_states():
conn = get_conn()
if conn is None:
return {}
with _conn_lock:
rows = conn.execute(
"""SELECT ALERT_KEY, ACTIVE, ACTIVE_SINCE, LAST_ATTEMPT_AT,
LAST_SENT_AT, RECOVERY_PENDING, LAST_RECOVERY_AT,
LAST_ERROR, LAST_VALUE
FROM alert_state"""
).fetchall()
return {
row[0]: {
"active": bool(row[1]),
"active_since": row[2],
"last_attempt_at": row[3],
"last_sent_at": row[4],
"recovery_pending": bool(row[5]),
"last_recovery_at": row[6],
"last_error": row[7],
"last_value": row[8],
}
for row in rows
}
def save_alert_state(alert_key, state):
conn = get_conn()
if conn is None:
raise RuntimeError("SQLite connection unavailable")
with _conn_lock:
conn.execute(
"""INSERT INTO alert_state (
ALERT_KEY, ACTIVE, ACTIVE_SINCE, LAST_ATTEMPT_AT,
LAST_SENT_AT, RECOVERY_PENDING, LAST_RECOVERY_AT,
LAST_ERROR, LAST_VALUE
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(ALERT_KEY) DO UPDATE SET
ACTIVE = excluded.ACTIVE,
ACTIVE_SINCE = excluded.ACTIVE_SINCE,
LAST_ATTEMPT_AT = excluded.LAST_ATTEMPT_AT,
LAST_SENT_AT = excluded.LAST_SENT_AT,
RECOVERY_PENDING = excluded.RECOVERY_PENDING,
LAST_RECOVERY_AT = excluded.LAST_RECOVERY_AT,
LAST_ERROR = excluded.LAST_ERROR,
LAST_VALUE = excluded.LAST_VALUE""",
(
alert_key,
1 if state["active"] else 0,
state["active_since"],
state["last_attempt_at"],
state["last_sent_at"],
1 if state["recovery_pending"] else 0,
state["last_recovery_at"],
state["last_error"],
state["last_value"],
),
)
conn.commit()
def get_unsent_tracking_rows(limit=100):
"""Return up to `limit` tracking rows not yet uploaded to the remote server,
oldest first, as a list of dicts ready to be sent as JSON."""
conn = get_conn()
if conn is None:
return []
with _conn_lock:
cur = conn.execute(
"""SELECT id, VN, NAME, DATETIME, GPSX, GPSY, SPEED, ODOMETER
FROM tracking WHERE UPLOADED = 0 ORDER BY id ASC LIMIT ?""",
(limit,),
)
rows = cur.fetchall()
return [
{
"id": r[0], "vn": r[1], "name": r[2], "datetime": r[3],
"gpsx": r[4], "gpsy": r[5], "speed": r[6], "odometer": r[7],
}
for r in rows
]
def count_unsent_tracking_rows():
conn = get_conn()
if conn is None:
return 0
with _conn_lock:
cur = conn.execute("SELECT COUNT(*) FROM tracking WHERE UPLOADED = 0")
return cur.fetchone()[0]
def mark_tracking_rows_uploaded(ids):
"""Flip UPLOADED=1 for the given row ids after a successful server ack."""
if not ids:
return
conn = get_conn()
if conn is None:
return
with _conn_lock:
conn.executemany("UPDATE tracking SET UPLOADED = 1 WHERE id = ?", [(i,) for i in ids])
conn.commit()
def dashcam_file_marked_uploaded(relative_path, sha256):
"""Whether this exact (path, content) was already confirmed uploaded."""
conn = get_conn()
if conn is None:
return False
with _conn_lock:
cur = conn.execute(
"SELECT 1 FROM dashcam_uploads WHERE RELATIVE_PATH = ? AND SHA256 = ?",
(relative_path, sha256),
)
return cur.fetchone() is not None
def dashcam_file_uploaded_for_stat(relative_path, size, mtime):
"""Whether the current local file version is confirmed uploaded.
Uses the existing hash cache so list/retention scans never rehash large
video files merely to display their status.
"""
conn = get_conn()
if conn is None:
return False
with _conn_lock:
cur = conn.execute(
"""SELECT 1
FROM dashcam_hash_cache AS cache
JOIN dashcam_uploads AS uploads
ON uploads.RELATIVE_PATH = cache.RELATIVE_PATH
AND uploads.SHA256 = cache.SHA256
WHERE cache.RELATIVE_PATH = ?
AND cache.SIZE = ?
AND cache.MTIME = ?""",
(relative_path, size, mtime),
)
return cur.fetchone() is not None
def uploaded_dashcam_file_stats():
"""Return exact uploaded file versions as (path, size, mtime) tuples."""
conn = get_conn()
if conn is None:
return set()
with _conn_lock:
cur = conn.execute(
"""SELECT cache.RELATIVE_PATH, cache.SIZE, cache.MTIME
FROM dashcam_hash_cache AS cache
JOIN dashcam_uploads AS uploads
ON uploads.RELATIVE_PATH = cache.RELATIVE_PATH
AND uploads.SHA256 = cache.SHA256"""
)
return set(cur.fetchall())
def mark_dashcam_file_uploaded(relative_path, sha256):
conn = get_conn()
if conn is None:
return
with _conn_lock:
conn.execute(
"INSERT OR IGNORE INTO dashcam_uploads (RELATIVE_PATH, SHA256, UPLOADED_AT) VALUES (?, ?, ?)",
(relative_path, sha256, datetime.now()),
)
conn.commit()
def get_cached_dashcam_hash(relative_path, size, mtime):
"""Return a previously-computed sha256 for this exact (path, size, mtime),
or None if not cached / the file has changed since it was last hashed."""
conn = get_conn()
if conn is None:
return None
with _conn_lock:
cur = conn.execute(
"SELECT SHA256 FROM dashcam_hash_cache WHERE RELATIVE_PATH = ? AND SIZE = ? AND MTIME = ?",
(relative_path, size, mtime),
)
row = cur.fetchone()
return row[0] if row is not None else None
def set_cached_dashcam_hash(relative_path, size, mtime, sha256):
conn = get_conn()
if conn is None:
return
with _conn_lock:
conn.execute(
"INSERT INTO dashcam_hash_cache (RELATIVE_PATH, SIZE, MTIME, SHA256) VALUES (?, ?, ?, ?) "
"ON CONFLICT(RELATIVE_PATH) DO UPDATE SET SIZE = excluded.SIZE, MTIME = excluded.MTIME, SHA256 = excluded.SHA256",
(relative_path, size, mtime, sha256),
)
conn.commit()
def _log_loop():
# Deferred import to avoid a circular import at module load time
# (app_obd.py imports app_sql, and app_sql needs app_obd.speed here).
from modules import app_obd
last_position = None
last_logged_at = None
while True:
gps_x = app_gps.LocationX
gps_y = app_gps.LocationY
speed = app_obd.get_fresh_speed()
if gps_x is not None and gps_y is not None:
try:
position = (
float(str(gps_x).replace("\x00", "").strip()),
float(str(gps_y).replace("\x00", "").strip()),
)
except (TypeError, ValueError) as error:
print("Skipping invalid GPS tracking snapshot:", error)
else:
now = time.monotonic()
seconds_since_log = (
now - last_logged_at
if last_logged_at is not None
else IDLE_LOG_INTERVAL
)
if should_log_tracking(
position, speed, last_position, seconds_since_log
) and insert_tracking(position[0], position[1], speed, app_obd.odometer):
last_position = position
last_logged_at = now
time.sleep(LOG_INTERVAL)
def start_logging():
"""Start the background thread that periodically snapshots GPS+speed into SQLite."""
global _log_thread
if _log_thread is None or not _log_thread.is_alive():
get_conn() # make sure the db file/tables exist before the first insert
_log_thread = Thread(target=_log_loop)
_log_thread.daemon = True
_log_thread.start()
print(f"SQL tracking-logger thread started, alive: {_log_thread.is_alive()}")
else:
print(f"SQL tracking-logger thread already running, alive: {_log_thread.is_alive()}")
if __name__ == '__main__':
get_conn()
Reference in New Issue View Git Blame Copy Permalink