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]>
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
from modules.app_events import socketio
|
|
from threading import Thread, Lock
|
|
import serial
|
|
import time
|
|
|
|
# SIM868 GNSS/GSM/GPRS HAT is wired to the Pi's GPIO UART (jumper set to "B: Pi-SIM868"),
|
|
# exposed as /dev/serial0 once the Pi's onboard Bluetooth has been disabled/freed.
|
|
SERIAL_PORT = "/dev/serial0"
|
|
BAUDRATE = 115200
|
|
POLL_INTERVAL = 2 # seconds between AT+CGNSINF polls
|
|
|
|
LocationX = None
|
|
LocationY = None
|
|
|
|
_serial_lock = Lock()
|
|
_serial = None
|
|
_gps_thread = None
|
|
|
|
|
|
class GPS:
|
|
def __init__(self):
|
|
self.LocationX = None
|
|
self.LocationY = None
|
|
|
|
|
|
def _get_serial():
|
|
global _serial
|
|
if _serial is None or not _serial.is_open:
|
|
_serial = serial.Serial(SERIAL_PORT, BAUDRATE, timeout=1)
|
|
time.sleep(0.3)
|
|
_serial.reset_input_buffer()
|
|
return _serial
|
|
|
|
|
|
def _send_at(cmd, wait=0.5):
|
|
"""Send an AT command to the SIM868 module and return the raw response."""
|
|
ser = _get_serial()
|
|
ser.write((cmd + "\r\n").encode())
|
|
time.sleep(wait)
|
|
return ser.read(1000).decode(errors="replace")
|
|
|
|
|
|
def _ensure_gnss_powered():
|
|
resp = _send_at("AT+CGNSPWR?")
|
|
if "+CGNSPWR: 0" in resp:
|
|
print("GNSS was off, powering it on...")
|
|
_send_at("AT+CGNSPWR=1", wait=1)
|
|
|
|
|
|
def _parse_cgnsinf(resp):
|
|
"""Parse a +CGNSINF response line into a dict, or None if there's no fix yet.
|
|
|
|
Format: +CGNSINF: run,fix,utc,lat,lon,alt,speed,course,fixmode,reserved1,
|
|
hdop,pdop,vdop,reserved2,gps_sat_view,gnss_sat_used,gnss_sat_view,cn0,hpa,vpa
|
|
"""
|
|
for line in resp.splitlines():
|
|
line = line.strip()
|
|
if line.startswith("+CGNSINF:"):
|
|
fields = line.split(":", 1)[1].strip().split(",")
|
|
if len(fields) < 8 or fields[1] != "1":
|
|
return None
|
|
lat, lon = fields[3], fields[4]
|
|
if not lat or not lon:
|
|
return None
|
|
return {
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"alt": fields[5],
|
|
"speed": fields[6],
|
|
"course": fields[7],
|
|
}
|
|
return None
|
|
|
|
|
|
def _poll_loop():
|
|
global LocationX, LocationY
|
|
try:
|
|
with _serial_lock:
|
|
_ensure_gnss_powered()
|
|
except Exception as e:
|
|
print("Failed to initialize GNSS:", str(e))
|
|
|
|
while True:
|
|
try:
|
|
with _serial_lock:
|
|
resp = _send_at("AT+CGNSINF")
|
|
fix = _parse_cgnsinf(resp)
|
|
except Exception as e:
|
|
print("GPS read error:", str(e))
|
|
fix = None
|
|
|
|
if fix:
|
|
LocationX = fix["lat"]
|
|
LocationY = fix["lon"]
|
|
socketio.emit("gps_location", [LocationX, LocationY])
|
|
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
|
|
def start_gps():
|
|
"""Start the background thread that continuously polls the SIM868 for a GNSS fix."""
|
|
global _gps_thread
|
|
if _gps_thread is None or not _gps_thread.is_alive():
|
|
_gps_thread = Thread(target=_poll_loop)
|
|
_gps_thread.daemon = True
|
|
_gps_thread.start()
|
|
print(f"GPS thread started, alive: {_gps_thread.is_alive()}")
|
|
else:
|
|
print(f"GPS thread already running, alive: {_gps_thread.is_alive()}")
|
|
return GPS()
|
|
|
|
|
|
@socketio.on("gps")
|
|
def get_gps():
|
|
"""Return the last known GPS location and emit it over the 'gps_location' event."""
|
|
Location = [LocationX, LocationY]
|
|
socketio.emit("gps_location", Location)
|
|
print("gps_location emitted:", Location)
|
|
return Location
|