import serial import time from datetime import datetime, timezone import threading from logger import gps_log, main_log from config import GPS_PORT, GPS_BAUD, PINS import lgpio class GPSHandler: def __init__(self, rtc_handler=None): self.ser = None self.last_latlon = None self.last_gps_time = None self.connected = False self.position_reported = False self.gps_signal_good = False self.last_valid_data = None self.rtc_handler = rtc_handler self.gps_synced_with_rtc = False self.chip = lgpio.gpiochip_open(0) lgpio.gpio_claim_input(self.chip, PINS["PPS"]) def connect(self): while not self.connected: try: self.ser = serial.Serial(GPS_PORT, GPS_BAUD, timeout=1) timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Connected to {GPS_PORT} @ {GPS_BAUD} baud") self.connected = True self.position_reported = False self.gps_synced_with_rtc = False except serial.SerialException as e: timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Connection failed: {e}, retrying...") time.sleep(2) def start_reading(self): threading.Thread(target=self._read_loop, daemon=True).start() def _read_loop(self): while True: try: line = self.ser.readline().decode(errors="ignore").strip() # Process position data if line.startswith("$GPGGA") or line.startswith("$GPRMC"): latlon = self.parse_latlon(line) if latlon: self.last_latlon = latlon self.last_valid_data = time.time() self.gps_signal_good = True gps_log(f"Lat: {latlon[0]:.6f}, Lon: {latlon[1]:.6f}") if not self.position_reported: timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Position acquired - Lat: {latlon[0]:.6f}, Lon: {latlon[1]:.6f}") self.position_reported = True # Process time data from RMC sentences if line.startswith("$GPRMC"): gps_time = self.parse_gps_time(line) if gps_time: self.last_gps_time = gps_time # Sync RTC with GPS time if we have an RTC handler and haven't synced yet if self.rtc_handler and not self.gps_synced_with_rtc: if self.rtc_handler.sync_with_gps(gps_time): self.gps_synced_with_rtc = True # Check for GPS signal timeout (no valid data for 10 seconds) if self.last_valid_data and (time.time() - self.last_valid_data > 10): if self.gps_signal_good: # Only log once when signal is lost timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Signal lost - no valid data for 10+ seconds") main_log("GPS signal lost - switching to RTC timing") self.gps_signal_good = False except Exception as e: timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Error in read loop: {e}") time.sleep(1) def parse_gps_time(self, nmea_line): """Parse GPS time from RMC sentence""" try: parts = nmea_line.split(",") if len(parts) < 10 or parts[2] != 'A': # 'A' means valid fix return None time_str = parts[1] # HHMMSS.sss date_str = parts[9] # DDMMYY if len(time_str) < 6 or len(date_str) < 6: return None # Parse time hours = int(time_str[:2]) minutes = int(time_str[2:4]) seconds = int(float(time_str[4:])) # Parse date day = int(date_str[:2]) month = int(date_str[2:4]) year = 2000 + int(date_str[4:6]) # Assume 21st century return datetime(year, month, day, hours, minutes, seconds, tzinfo=timezone.utc) except (ValueError, IndexError): return None def get_position_for_transmission(self): """Get current position and print to REPL for transmission start""" if self.last_latlon: timestamp = datetime.now().strftime("%H:%M:%S") signal_status = "GPS" if self.gps_signal_good else "GPS(stale)" print(f"[{timestamp}] [{signal_status}] Transmission position - Lat: {self.last_latlon[0]:.6f}, Lon: {self.last_latlon[1]:.6f}") return self.last_latlon return None def parse_latlon(self, nmea_line): parts = nmea_line.split(",") if len(parts) < 6: return None try: # For GPRMC, check if fix is valid if nmea_line.startswith("$GPRMC") and parts[2] != 'A': return None # For GPGGA, check if we have enough satellites if nmea_line.startswith("$GPGGA") and (not parts[7] or int(parts[7]) < 4): return None lat = float(parts[2][:2]) + float(parts[2][2:]) / 60.0 if parts[3] == "S": lat = -lat lon = float(parts[4][:3]) + float(parts[4][3:]) / 60.0 if parts[5] == "W": lon = -lon return lat, lon except (ValueError, IndexError): return None def has_good_signal(self): """Check if GPS signal is currently good""" return self.gps_signal_good and self.last_valid_data and (time.time() - self.last_valid_data < 5) def wait_for_next_minute(self): """ Uses PPS pin to align start exactly to the top of the next UTC minute. Falls back to RTC if GPS PPS is not available. """ if self.has_good_signal(): timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Waiting for PPS pulse...") main_log("Using GPS PPS for minute synchronization") try: # Wait for PPS pulse timeout = time.time() + 5 # 5 second timeout while lgpio.gpio_read(self.chip, PINS["PPS"]) == 1 and time.time() < timeout: pass while lgpio.gpio_read(self.chip, PINS["PPS"]) == 0 and time.time() < timeout: pass if time.time() >= timeout: raise Exception("PPS timeout") # Calculate seconds to wait based on GPS time if available if self.last_gps_time: now = self.last_gps_time else: now = datetime.utcnow() seconds_to_wait = 60 - now.second timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Waiting {seconds_to_wait} seconds to align to minute...") for _ in range(seconds_to_wait): timeout = time.time() + 2 # 2 second timeout per pulse while lgpio.gpio_read(self.chip, PINS["PPS"]) == 1 and time.time() < timeout: pass while lgpio.gpio_read(self.chip, PINS["PPS"]) == 0 and time.time() < timeout: pass if time.time() >= timeout: raise Exception("PPS timeout during minute alignment") except Exception as e: timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] PPS failed ({e}), falling back to RTC") main_log(f"GPS PPS failed ({e}), switching to RTC timing") if self.rtc_handler: self.rtc_handler.wait_for_next_minute_rtc() else: timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [ERROR] No RTC available for fallback timing!") else: # GPS signal is poor, use RTC timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [GPS] Poor GPS signal, using RTC for timing") if self.rtc_handler: self.rtc_handler.wait_for_next_minute_rtc() else: timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] [ERROR] No RTC available for fallback timing!")