202 lines
6.6 KiB
Python
202 lines
6.6 KiB
Python
from modules.app_events import socketio
|
|
from threading import Thread, Lock
|
|
import serial
|
|
import time
|
|
|
|
try:
|
|
import RPi.GPIO as GPIO
|
|
except (ImportError, RuntimeError):
|
|
# Not running on a Pi (e.g. local dev machine) - PWRKEY auto power-on is
|
|
# simply unavailable there; _pulse_pwrkey() below degrades gracefully.
|
|
GPIO = None
|
|
|
|
# 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
|
|
|
|
# This HAT revision has the "P4" GPIO power-on mod wire connected to the SIM868's
|
|
# PWRKEY line (confirmed 2026-08-09 by pulsing it and seeing the module respond to
|
|
# AT afterward) - physical header pin 7 (BCM GPIO4), same pin Waveshare's own
|
|
# GSM_PWRKEY.py reference script uses. This lets us auto power-on the module chip
|
|
# on every app start instead of requiring someone to press its PWRKEY button by
|
|
# hand after every Pi power cycle.
|
|
PWRKEY_PIN = 7 # BOARD numbering
|
|
PWRKEY_HOLD_SECONDS = 4 # matches Waveshare's reference script
|
|
|
|
LocationX = None
|
|
LocationY = None
|
|
last_fix_at = 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 _module_responds():
|
|
"""Check whether the SIM868 module chip is powered on and answering AT commands."""
|
|
try:
|
|
return "OK" in _send_at("AT", wait=0.3)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _pulse_pwrkey():
|
|
"""Pulse the PWRKEY line low then high, same sequence as Waveshare's own
|
|
GSM_PWRKEY.py reference script.
|
|
|
|
PWRKEY is a toggle, not a simple "power on" switch: pressing it while the
|
|
module is off powers it on, but pressing it again while already on powers
|
|
it back off. Callers must only invoke this after confirming via
|
|
`_module_responds()` that the module is NOT currently responding - never
|
|
unconditionally on every app start, or a normal restart while the module
|
|
is already up would accidentally shut it back down.
|
|
"""
|
|
if GPIO is None:
|
|
print("RPi.GPIO not available, cannot auto-power-on the SIM868 module - "
|
|
"press its PWRKEY button by hand.")
|
|
return
|
|
GPIO.setmode(GPIO.BOARD)
|
|
GPIO.setup(PWRKEY_PIN, GPIO.OUT)
|
|
GPIO.output(PWRKEY_PIN, GPIO.LOW)
|
|
time.sleep(PWRKEY_HOLD_SECONDS)
|
|
GPIO.output(PWRKEY_PIN, GPIO.HIGH)
|
|
GPIO.cleanup(PWRKEY_PIN)
|
|
|
|
|
|
def _ensure_module_powered():
|
|
"""Power on the SIM868 module chip via its PWRKEY line if it isn't already on.
|
|
|
|
The HAT's 5V rail can be live (board powered) while the module chip itself
|
|
stays fully off until PWRKEY is pulsed - this is what previously required a
|
|
manual button press after every Pi power cycle. Returns True once the
|
|
module is confirmed responding to AT commands, False if it still isn't
|
|
after attempting the power-on pulse (e.g. GPIO unavailable, or this board
|
|
doesn't have the P4 mod wire).
|
|
"""
|
|
if _module_responds():
|
|
return True
|
|
|
|
print("SIM868 module not responding to AT, pulsing PWRKEY to power it on...")
|
|
_pulse_pwrkey()
|
|
time.sleep(3) # give the module a few seconds to boot after PWRKEY release
|
|
|
|
for _ in range(5):
|
|
if _module_responds():
|
|
print("SIM868 module powered on and responding.")
|
|
return True
|
|
time.sleep(1)
|
|
|
|
print("SIM868 module still not responding after PWRKEY pulse - check the "
|
|
"P4 wiring or press the physical PWRKEY button.")
|
|
return False
|
|
|
|
|
|
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, last_fix_at
|
|
try:
|
|
with _serial_lock:
|
|
if _ensure_module_powered():
|
|
_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"]
|
|
last_fix_at = time.monotonic()
|
|
socketio.emit("gps_location", [LocationX, LocationY])
|
|
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
|
|
def get_fix_age():
|
|
return None if last_fix_at is None else time.monotonic() - last_fix_at
|
|
|
|
|
|
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
|