Background thread: read JSON objects from gpsd and parse position.
(self)
| 467 | 'fix': self.fix_quality, |
| 468 | 'satellites': self.satellites, |
| 469 | 'hdop': self.hdop, |
| 470 | 'timestamp': self.last_update |
| 471 | } |
| 472 | |
| 473 | def _load_last_known(self): |
| 474 | """Load the persisted last-known position at startup (best-effort).""" |
| 475 | try: |
| 476 | with open(self.state_file) as f: |
| 477 | d = json.load(f) |
| 478 | self.last_known = { |
| 479 | 'lat': float(d['lat']), 'lon': float(d['lon']), |
| 480 | 'alt': d.get('alt'), 't': d.get('t'), |
| 481 | } |
| 482 | logger.info("GPS last-known position loaded: " |
| 483 | f"{self.last_known['lat']:.4f}, {self.last_known['lon']:.4f}") |
| 484 | except FileNotFoundError: |
| 485 | pass |
| 486 | except Exception as e: |
| 487 | logger.debug(f"last-known GPS load failed: {e}") |
| 488 | |
| 489 | def _record_position(self, now): |
| 490 | """Update (and throttle-persist) last-known position. Call under _lock |
| 491 | at each confirmed-fix point, with self.latitude/longitude already set.""" |
| 492 | if self.latitude is None or self.longitude is None: |
| 493 | return |
| 494 | self.last_known = {'lat': self.latitude, 'lon': self.longitude, |
| 495 | 'alt': self.altitude, 't': now} |
| 496 | if self.state_file and (now - self._last_persist) > 60: |
| 497 | self._last_persist = now |
| 498 | try: |
| 499 | os.makedirs(os.path.dirname(self.state_file), exist_ok=True) |
| 500 | tmp = self.state_file + '.tmp' |
| 501 | with open(tmp, 'w') as f: |
| 502 | json.dump(self.last_known, f) |
| 503 | os.replace(tmp, self.state_file) # atomic |
| 504 | except Exception as e: |
| 505 | logger.debug(f"last-known GPS persist failed: {e}") |
| 506 | |
| 507 | # Talker code -> human constellation name, for the sky view. Shared by the |
| 508 | # gpsd (SKY) and serial (GSV) paths, both of which key by these codes. |
| 509 | _TALKER_NAMES = { |
nothing calls this directly
no test coverage detected