148 lines
4.0 KiB
Python
148 lines
4.0 KiB
Python
# Copyright 2021-2023 Teemu Ikonen
|
|
# SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
import os.path
|
|
import socket
|
|
from gi.repository import GLib
|
|
|
|
|
|
class NmeaSourceNotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
class ModemError(Exception):
|
|
pass
|
|
|
|
|
|
class ModemLockedError(ModemError):
|
|
pass
|
|
|
|
|
|
class ModemNoNMEAError(ModemError):
|
|
pass
|
|
|
|
|
|
class NmeaSource:
|
|
def __init__(self, update_callback, refresh_rate=1, save_filename=None,
|
|
**kwargs):
|
|
self.update_callback = update_callback
|
|
self.refresh_rate = refresh_rate
|
|
self.save_file = None if save_filename is None \
|
|
else open(save_filename, 'w')
|
|
self.initialized = False
|
|
self.manufacturer = None
|
|
self.model = None
|
|
self.revision = None
|
|
|
|
def __del__(self):
|
|
if self.save_file is not None:
|
|
self.save_file.close()
|
|
|
|
def initialize(self):
|
|
pass
|
|
|
|
def _really_get(self):
|
|
pass
|
|
|
|
def _maybe_save(self, nmeas):
|
|
if self.save_file is not None:
|
|
self.save_file.write(nmeas)
|
|
self.save_file.write("\n\n")
|
|
self.save_file.flush()
|
|
|
|
def get(self):
|
|
nmeas = self._really_get()
|
|
self._maybe_save(nmeas)
|
|
return nmeas
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class UnixSocketNmeaSource(NmeaSource):
|
|
def __init__(self,
|
|
update_callback,
|
|
socket_file_path=None,
|
|
**kwargs):
|
|
super().__init__(update_callback, **kwargs)
|
|
self.socket_file_path = socket_file_path
|
|
|
|
def on_read_data_available(self, io_channel, condition, **unused):
|
|
self.update_callback()
|
|
|
|
def initialize(self):
|
|
if (self.socket_file_path is None
|
|
or not os.path.exists(self.socket_file_path)):
|
|
raise FileNotFoundError(f"Could not open socket {self.socket_file_path}")
|
|
|
|
self.s = socket.socket(socket.AF_UNIX,
|
|
socket.SOCK_NONBLOCK | socket.SOCK_STREAM)
|
|
try:
|
|
self.s.connect(self.socket_file_path)
|
|
except Exception as e:
|
|
raise e
|
|
|
|
self.channel = GLib.IOChannel.unix_new(self.s.fileno())
|
|
self.channel.add_watch(GLib.IO_IN, self.on_read_data_available)
|
|
|
|
self.old_nmea_buf = ''
|
|
self.initialized = True
|
|
|
|
def _really_get(self):
|
|
if not self.initialized:
|
|
self.initialize()
|
|
|
|
buf = ''
|
|
do_read = True
|
|
|
|
while do_read:
|
|
read_buf = self.channel.readline()
|
|
buf += read_buf
|
|
if read_buf == '':
|
|
do_read = False
|
|
|
|
return buf
|
|
|
|
|
|
class GnssShareNmeaSource(UnixSocketNmeaSource):
|
|
def __init__(self, update_callback, **kwargs):
|
|
super().__init__(update_callback,
|
|
socket_file_path='/var/run/gnss-share.sock',
|
|
**kwargs)
|
|
|
|
|
|
class ReplayNmeaSource(NmeaSource):
|
|
def __init__(self, update_callback, replay_filename=None, refresh_rate=1,
|
|
**kwargs):
|
|
super().__init__(update_callback, refresh_rate=refresh_rate, **kwargs)
|
|
self.nmeas = []
|
|
self.counter = 0
|
|
if replay_filename is None:
|
|
raise NmeaSourceNotFoundError("NMEA file name not given")
|
|
self.replay_filename = replay_filename
|
|
self.manufacturer = "Satellite"
|
|
self.model = "ReplayNmeaSource"
|
|
|
|
def initialize(self):
|
|
try:
|
|
with open(self.replay_filename, 'r') as ff:
|
|
self.nmeas = ff.read().split("\n\n")
|
|
except Exception:
|
|
raise NmeaSourceNotFoundError(
|
|
f"Could not read NMEA file '{self.replay_filename}'")
|
|
self.counter = 0
|
|
GLib.timeout_add(self.refresh_rate * 1000, self.callback, None)
|
|
|
|
def callback(self, x):
|
|
self.update_callback()
|
|
return True
|
|
|
|
def _really_get(self):
|
|
if not self.nmeas:
|
|
return None
|
|
nmea = self.nmeas[self.counter]
|
|
self.counter += 1
|
|
if self.counter >= len(self.nmeas):
|
|
self.counter = 0
|
|
return nmea
|