Mathieu MoranandCopilot
261ef3f4db
Squashed rewrite of the project revival work (replaces prior history on this branch to remove personal data that had been accidentally committed - real LAN IP, GPS coordinates, home WiFi SSID, and phone model). See PLAN.md for full phase-by-phase details. Summary: - Phase 2: adapt app_obd.py for a USB OBD-II adapter (was Bluetooth-only), auto-detect /dev/ttyUSB*, fix several pre-existing crash bugs. - Phase 3/4: make the web UI work fully offline (vendored Socket.IO/Leaflet/ Chart.js/FontAwesome, offline map tile downloader script, live GPS marker tracking fix). - Phase 5: systemd deployment/autostart on the Pi. - Phase 6: WiFi hotspot mode via NetworkManager (auto-switch between home WiFi and a local AP for in-car web UI access), mDNS hostname + port 80 redirect so the UI is reachable at a plain hostname with no port number. - Phase 7: fix app_db.py/app_sql.py bugs, wire up local SQLite trip logging. - Phase 8: brand-agnostic USB dashcam file retrieval (lsblk transport-based detection, works with any dashcam brand/folder layout). - Rewritten readme.md with a proper project overview. Co-authored-by: Copilot <[email protected]>
142 lines
4.5 KiB
Python
142 lines
4.5 KiB
Python
from modules import app_gps, app_sql, app_events
|
|
import glob, obd, os, time, threading
|
|
from datetime import datetime
|
|
from threading import Event, Thread
|
|
|
|
status = None
|
|
speed = 0
|
|
rpm = None
|
|
|
|
class OBD:
|
|
def __init__(self):
|
|
self.status = status
|
|
self.rpm = rpm
|
|
self.speed = speed
|
|
|
|
obd_status = ["Disconnected", "Connected", "Connection...", "wtf"]
|
|
obd_speed_lock = threading.Lock()
|
|
obd_rpm_lock = threading.Lock()
|
|
thread_speed = None
|
|
thread_rpm = None
|
|
stop_speed = Event()
|
|
stop_rpm = Event()
|
|
|
|
connection = None
|
|
|
|
|
|
def _find_obd_port():
|
|
"""Locate the USB-serial OBD-II adapter.
|
|
|
|
Honors an OBD_PORT env var override; otherwise picks the first
|
|
/dev/ttyUSB* device found (the PL2303 USB adapter shows up here).
|
|
Returns None if nothing is found, letting python-obd fall back to its
|
|
own auto-scan (which also covers /dev/rfcomm* for the old Bluetooth setup).
|
|
"""
|
|
env_port = os.environ.get("OBD_PORT")
|
|
if env_port:
|
|
return env_port
|
|
ports = sorted(glob.glob("/dev/ttyUSB*"))
|
|
return ports[0] if ports else None
|
|
|
|
|
|
def connect_obd():
|
|
"""(Re)connect to the OBD-II adapter. Safe to call repeatedly."""
|
|
global connection
|
|
port = _find_obd_port()
|
|
try:
|
|
print(obd_status[2])
|
|
connection = obd.OBD(portstr=port) if port else obd.OBD()
|
|
except Exception as e:
|
|
print("An error occurred:", str(e))
|
|
connection = None
|
|
return connection
|
|
|
|
|
|
connect_obd()
|
|
|
|
|
|
def get_speedtime():
|
|
global speed
|
|
time_now = str(datetime.now().strftime("%H:%M:%S"))
|
|
return [time_now, speed]
|
|
|
|
|
|
def obd_connection():
|
|
global thread_speed, thread_rpm
|
|
|
|
if connection is None or not connection.is_connected():
|
|
connect_obd()
|
|
|
|
if connection is not None and connection.is_connected():
|
|
print(obd_status[1])
|
|
app_events.socketio.emit('obd_status', obd_status[1])
|
|
print("OBD connection established.")
|
|
|
|
if thread_speed is None or not thread_speed.is_alive():
|
|
stop_speed.clear()
|
|
thread_speed = Thread(target=obd_speed)
|
|
thread_speed.daemon = True
|
|
thread_speed.start()
|
|
print(f'thread_speed alive? {thread_speed.is_alive()}')
|
|
|
|
if thread_rpm is None or not thread_rpm.is_alive():
|
|
stop_rpm.clear()
|
|
thread_rpm = Thread(target=obd_rpm)
|
|
thread_rpm.daemon = True
|
|
thread_rpm.start()
|
|
print(f'thread_rpm alive? {thread_rpm.is_alive()}')
|
|
else:
|
|
print("OBD connection failed.")
|
|
print(obd_status[0])
|
|
|
|
app_events.socketio.emit('obd_status', obd_status[0])
|
|
app_events.socketio.emit('obd_speed', "-")
|
|
app_events.socketio.emit('obd_error', "-")
|
|
app_events.socketio.emit('obd_rpm', "-")
|
|
|
|
def obd_speed():
|
|
global speed
|
|
with obd_speed_lock:
|
|
cmd = obd.commands.SPEED # select an OBD command (sensor)
|
|
while not stop_speed.is_set():
|
|
if connection is None or not connection.is_connected():
|
|
print("OBD disconnected, stopping speed polling.")
|
|
break
|
|
|
|
response = connection.query(cmd) # send the command and parse the response
|
|
if response.is_null():
|
|
print("No data received. #" + str(get_speedtime()))
|
|
print(app_gps.get_gps())
|
|
else:
|
|
speed = round(response.value.to("kph").magnitude, 1)
|
|
print("Speed:", speed, "kph")
|
|
|
|
app_events.socketio.emit('obd_speed', get_speedtime())
|
|
time.sleep(2)
|
|
|
|
def obd_rpm():
|
|
global rpm
|
|
with obd_rpm_lock:
|
|
cmd = obd.commands.RPM # select an OBD command (sensor)
|
|
while not stop_rpm.is_set():
|
|
if connection is None or not connection.is_connected():
|
|
print("OBD disconnected, stopping rpm polling.")
|
|
break
|
|
|
|
response = connection.query(cmd) # send the command and parse the response
|
|
if response.is_null():
|
|
print("No RPM data received. #" + str(get_speedtime()))
|
|
app_events.socketio.emit('obd_rpm', "-")
|
|
else:
|
|
rpm = round(response.value.magnitude, 0)
|
|
print("RPM:", rpm)
|
|
app_events.socketio.emit('obd_rpm', rpm)
|
|
|
|
time.sleep(1)
|
|
|
|
def kill_obd(obd_type):
|
|
if obd_type == "speed":
|
|
stop_speed.set()
|
|
elif obd_type == "rpm":
|
|
stop_rpm.set()
|