yaris-obd/modules/app_gps.py

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

_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
    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"]
            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