Fix GPS location updates and status reporting

This commit is contained in:
Mitchell Moss
2026-04-29 09:54:22 -04:00
parent bf44efbfd9
commit da8f83964a
2 changed files with 61 additions and 2 deletions
+9 -1
View File
@@ -359,6 +359,8 @@ class NMEAParser:
status = (self._field(fields, 2) or "").upper()
self.fix["status"] = "valid" if status == "A" else "invalid" if status == "V" else status
self.fix["valid"] = status == "A" or bool(self.fix.get("quality"))
if status == "A" and self.fix.get("quality") is None:
self.fix["quality_label"] = "RMC valid"
latitude = _parse_lat_lon(self._field(fields, 3), self._field(fields, 4))
longitude = _parse_lat_lon(self._field(fields, 5), self._field(fields, 6))
@@ -535,6 +537,8 @@ class NMEAParser:
with self._lock:
now = time.time()
age = now - self.last_update if self.last_update else None
if age is not None and age < 0:
age = 0.0
stale = age is None or age > self.stale_after_seconds
fix_valid = bool(self.fix.get("valid")) and not stale
if self.last_error:
@@ -615,7 +619,10 @@ class GPSService:
self._clock_setter = clock_setter or _set_system_clock_from_datetime
self._time_provider = time_provider or time.time
self.location_update_enabled = bool(
gps_config.get("update_repeater_location_from_fix", True)
gps_config.get(
"update_repeater_location_from_fix",
gps_config.get("use_gps_for_repeater_location", True),
)
)
self.location_update_interval_seconds = max(
1.0, float(gps_config.get("location_update_interval_seconds", 600.0))
@@ -1073,6 +1080,7 @@ class GPSService:
offset_seconds=offset_seconds,
success=True,
)
self.parser.last_update = self._time_provider()
logger.info(
"System clock synchronized from GPS time %s (offset %.3fs)",
gps_time.isoformat(),
+52 -1
View File
@@ -113,6 +113,37 @@ def test_gps_service_file_source_reads_nmea_lines(tmp_path):
assert snapshot["satellites"]["used_count"] == 5
def test_rmc_only_fix_has_non_conflicting_quality_label():
parser = NMEAParser()
assert parser.ingest_sentence(
_sentence("GPRMC,010203,A,4250.123,N,07106.456,W,000.0,180.0,230426,,")
)
snapshot = parser.snapshot()
assert snapshot["status"]["state"] == "valid_fix"
assert snapshot["fix"]["valid"] is True
assert snapshot["fix"]["quality"] is None
assert snapshot["fix"]["quality_label"] == "RMC valid"
assert snapshot["position"]["latitude"] == 42.83538333
assert snapshot["position"]["longitude"] == -71.1076
def test_snapshot_clamps_negative_age_after_system_clock_step():
parser = NMEAParser()
assert parser.ingest_sentence(
_sentence("GPRMC,010203,A,4250.123,N,07106.456,W,000.0,180.0,230426,,")
)
parser.last_update = time.time() + 120
snapshot = parser.snapshot()
assert snapshot["status"]["state"] == "valid_fix"
assert snapshot["status"]["age_seconds"] == 0.0
def test_gps_service_uses_manual_location_until_gps_fix():
service = GPSService(
{
@@ -347,7 +378,6 @@ def test_gps_service_reflects_runtime_manual_location_updates():
assert snapshot["gps_position"]["longitude"] == -71.1076
assert snapshot["position_meta"]["source"] == "manual_config"
def test_repeater_location_uses_config_when_gps_opt_in_disabled():
service = GPSService(
{
@@ -484,3 +514,24 @@ def test_gps_service_does_not_update_repeater_location_without_valid_fix():
assert location_updates == []
assert service.get_snapshot()["location_update"]["state"] == "waiting_for_fix"
def test_gps_service_honors_legacy_use_gps_for_repeater_location_flag():
location_updates = []
service = GPSService(
{
"gps": {
"enabled": True,
"time_sync_enabled": False,
"use_gps_for_repeater_location": False,
}
},
location_update_callback=lambda payload: location_updates.append(payload) or True,
)
assert service.ingest_sentence(
_sentence("GPRMC,010203,A,4250.123,N,07106.456,W,000.0,180.0,230426,,")
)
assert location_updates == []
assert service.get_snapshot()["location_update"]["state"] == "disabled"