Verified Commit 0423b83a authored by Clayton Craft's avatar Clayton Craft
Browse files

Initial commit of application

parent d8e24de5
An app for sharing GNSS location data, with support multiple clients and
loading/saving AGPS data.
This is meant to replace things like `gpsd`, and `gps-share`, and work together
with `geoclue`* or other clients that support fetching NMEA location data over
*To work with `geoclue`, [these patches are required.]( A future version of
`geoclue` should work out of the box after pointing it to the socket created by
this app.
# Configuration
`gnss_share.conf` can be used to change the listening socket, group owner for
socket, and other options. The application looks for this file in either the
current working directory, or in `/etc/gnss_share.conf`.
See this file for descriptions of supported options.
# Installation
### Dependencies:
- [Trio](
- [pynmea2](
This project uses meson to "build" and install things:
$ meson _build
$ meson install -C _build
To run locally from the source repo:
$ python3 -m gnss_share
# Development
### New GNSS device support
Support for additional gnss devices can be added by implementing a new 'driver', see `` for an example.
At a minimum, `gnss_share` expects that the following methods are implemented in the driver:
- context manager support (`__aenter__`, `__aexit__`)
- `load(directory: str) -> None`
- `store(directory: str) -> None`
- `open() -> None`
- `close() -> None`
- `readline() -> bytes`
- `reset() -> None`
- `settime() -> None`
This application uses the Python Trio library for async coroutines, so blocking tasks in the driver must be made async.
# Socket to sent NMEA location to
# Group to set as owner for the socket
# GPS device driver to use
# Supported values: stm
# Path to GPS device to use
# Supported values: stm
# Directory to load/store almanac and ephemeris data
# Copyright(c) 2021 by craftyguy "Clayton Craft" <>
# Distributed under GPLv3+ (see COPYING) WITHOUT ANY WARRANTY.
import sys
if __name__ == '__main__':
from gnss_share import main
#!/usr/bin/env python3
# Copyright(c) 2021 by craftyguy "Clayton Craft" <>
# Distributed under GPLv3+ (see COPYING) WITHOUT ANY WARRANTY.
import sys
pkgdatadir = '@pkgdatadir@'
sys.path.insert(1, pkgdatadir)
if __name__ == '__main__':
from gnss_share import main
# Copyright(c) 2021 by craftyguy "Clayton Craft" <>
# Distributed under GPLv3+ (see COPYING) WITHOUT ANY WARRANTY.
import logging
class LoggedException(Exception):
""" Exception class that sends the message to the logger """
def __init__(self, *args):
self.__log = logging.getLogger()
#!/usr/bin/env python3
# Copyright(c) 2021 by craftyguy "Clayton Craft" <>
# Distributed under GPLv3+ (see COPYING) WITHOUT ANY WARRANTY.
import argparse
import configparser
import grp
import logging
import os
import signal
import stat
import trio
from logging.handlers import SysLogHandler
from .logger import LoggedException
from .stm_agps import STM_AGPS
# List of NMEA prefixes to send to clients.
# This list is the same one that geoclue monitors.
b"$GAGGA", # Galieo
b"$GBGGA", # BeiDou
b"$BDGGA", # BeiDou
b"$GNGGA", # GNSS (combined)
# new root logger
logger = logging.getLogger("gnss_share")
class GnssShare:
drivers = {
'stm': STM_AGPS
def __init__(self, config):
self.__log = logging.getLogger(__name__)
# References to open connections are stored here
self._open_connections = []
# Reference to open device/driver
self._active_driver = None
# Holds the last NMEA location sentence retrieved from the device
self._location = b""
self._socket_path = config['gnss_share'].get('socket')
self._socket_owner_group = config['gnss_share'].get('group')
self._device_path = config['gnss_share'].get('device_path')
self._agps_dir = config['gnss_share'].get('agps_directory')
driver = config['gnss_share'].get('device_driver')
if driver not in self.drivers:
raise LoggedException(f"Driver {driver} is unknown!")
if not os.path.exists(self._device_path):
raise LoggedException("Serial device not found: "
os.makedirs(self._agps_dir, exist_ok=True)
self._driver = self.drivers[driver]
async def load(self, task_status=trio.TASK_STATUS_IGNORED):
""" Wrapper for running driver.load() """"running load()")
async with self._driver(self._device_path) as driver:
await driver.load(self._agps_dir)
# signal to nursery that this task is done (if not called from nursery,
# this is a noop)
async def store(self):
""" Wrapper for running """"running store()")
async with self._driver(self._device_path) as driver:
async def _get_location(self):
""" Returns last received location from gnss driver, or None """
if not self._active_driver:
self.__log.warn("Tried to read from inactive device!")
return b""
return self._location
async def _handle_socket_connection(self, conn):
""" Handler for client connections over a socket """"Got new connection from client on "
while True:
location = await self._get_location()
if location:
await conn.send_all(location)
# Send data to clients at this rate.
# TODO: is this too fast or too slow?
await trio.sleep(1)
except trio.BrokenResourceError:"A socket client disconnected")
await conn.aclose()"Number of connected clients: "
async def _start_sock_server(self):
""" Task to start a socket server and listen for new connections """
if os.path.exists(self._socket_path):
if (os.path.islink(self._socket_path)
or os.path.isfile(self._socket_path)):
raise LoggedException("Unable to use existing file as socket: "
self._sock = trio.socket.socket(trio.socket.AF_UNIX,
await self._sock.bind(self._socket_path)
# set socket group owner and make R/W
os.chown(self._socket_path, os.getuid(),
os.chmod(self._socket_path, stat.S_IRWXG | stat.S_IRWXU)
await trio.serve_listeners(self._handle_socket_connection,
async def _run_loop(self):
Main loop, for managing the gnss driver and getting data for clients
while True:
if len(self._open_connections) > 0:
if self._active_driver is None:
self._active_driver = self._driver(self._device_path)
line = await self._active_driver.readline()
prefix = line.split(b',')[0]
if prefix in GNSS_PREFIXES:
self._location = line
# polling loop delay when clients are connected
# TODO: is this adequate? Any faster and CPU utilization
# climbs...
await trio.sleep(.2)
if self._active_driver is not None:"No more clients connected, closing "
await self._active_driver.close()
self._active_driver = None
# polling loop delay when no clients are connected
# TODO: is this adequate?
await trio.sleep(2)
async def _signal_receiver(self):
Catch specific signals, and raise a SystemExit exception to stop
the parent nursery
with trio.open_signal_receiver(signal.SIGTERM,
signal.SIGINT) as signal_aiter:
async for sig in signal_aiter:"Caught signal: {sig}")
if sig == signal.SIGINT:
self.__log.warn("App exit requested.. hit C-c to "
"quit now.")
raise SystemExit
async def run(self):
async with trio.open_nursery() as nursery:
# start socket server as soon as possible so it can accept
# connections
# load() should complete before the main loop, which reads from
# the gnss device. So it's started right now, and blocks
await nursery.start(self.load)
# signal reciever started here, since calling store() before
# load() doesn't make sense
# main loop
except (SystemExit):
# Only store() on SystemExit, since it is thrown when certain
# signals are received. Everything else might lead to
# inconsistent/incomplete agps data
def main(version=None):
parser = argparse.ArgumentParser(description='Manager for GNSS devices')
parser.add_argument('--version', action='version',
version=f"version: {version}")
parser.add_argument('--syslog', action='store_true',
help=('Write output to syslog instead of stdout'))
parser.add_argument('-v', '--verbose', action='store_true',
help=("Enable verbose output"))
group = parser.add_mutually_exclusive_group()
parser.add_argument('-c', '--config', default='/etc/gnss_share.conf',
help=('Configuration file to use '
'(default: %(default)s).'))
group.add_argument('-s', '--store', nargs='?',
help=('Dump almanac and ephemeris data. A directory '
'can be specified (default: %(const)s).'))
group.add_argument('-l', '--load', nargs='?',
help=('Load almanac and ephemeris data. A directory '
'can be specified (default: %(const)s). '
'Note: this expects almanac data to be in a '
' file "almanac.txt", and ephemeris in a file '
args = parser.parse_args()
if args.syslog:
# syslog socket on Linux
syslog_handler = SysLogHandler('/dev/log')
# syslog uses first word as tag
logging.Formatter('%(name)s %(levelname)s: %(message)s'))
if args.verbose:
config_file = None
for file in [args.config, './gnss_share.conf']:
if os.path.exists(file):
config_file = file
if not config_file:
raise LoggedException("Unable to find config file at: \n"
"./gnss_share.conf")"Using configuration file: {config_file}")
config = configparser.ConfigParser()
gnss_share = GnssShare(config)
if args.load:
if __name__ == "__main__":
# Copyright(c) 2021 by Angus Ainslee
# Copyright(c) 2021 by Purism SPC
# Copyright(c) 2021 by craftyguy "Clayton Craft" <>
# Distributed under GPLv3+ (see COPYING) WITHOUT ANY WARRANTY.
import logging
import os
import pynmea2
import trio
from datetime import datetime
from .logger import LoggedException
class STM_AGPS:
def __init__(self, serial_port):
self.__log = logging.getLogger(__name__)
if not os.path.exists(serial_port):
raise LoggedException("Serial port does not exist: "
self._ser_port = serial_port
self._location = b""
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_value, traceback):
await self.close()
async def close(self):
await self._ser.aclose()
async def open(self):
self._ser = await trio.open_file(self._ser_port,
"w+b", buffering=0)
except Exception as e:
raise LoggedException(e)
async def readline(self):
return await self._ser.readline()
async def _serial_write_cmd(self, cmd, expect=None):
# number of times to poll serial output for ACK after sending command
polling_loops = 50"cmd: {cmd}")
await self._ser.write(str(cmd).encode("ascii"))
await self._ser.write(b'\r\n')
if expect:
for i in range(polling_loops):
line = await self._ser.readline()
line = line[:-1]"read: {line}")
if expect.encode("ascii") in line:"found: {expect}: {line}")
return True, line"not found: {expect}: {line}")
return False, line
# wait for cmd completion
for i in range(polling_loops):
line = await self._ser.readline()
line = line[:-1]"read: {line}")
if str(cmd).encode("ascii") in line:
return True, line
return False, line
async def _store_to_file(self, cmd, ack, file):
msg = pynmea2.GGA('P', cmd, ())
result, line = await self._serial_write_cmd(msg, ack)
if not result:
raise LoggedException("Unable to get data from device")
async with await trio.open_file(file, 'wb') as f:
while True:
if cmd.encode() in line:
if line.startswith(ack.encode()):
await f.write(line + b'\n')
line = await self._ser.readline()
line = line[:-1]
async def reset(self):
msg = pynmea2.GGA('P', 'STMGPSRESET', ())
await self._serial_write_cmd(msg)
await trio.sleep(1)
async def store(self, dir):
almanac_path = os.path.join(dir, 'almanac.txt')
ephemeris_path = os.path.join(dir, 'ephemeris.txt')
# reset device in case it is stuck
await self.reset()
await self._store_almanac(almanac_path)
await self._store_ephemeris(ephemeris_path)
async def load(self, dir):
almanac_path = os.path.join(dir, 'almanac.txt')
ephemeris_path = os.path.join(dir, 'ephemeris.txt')
for file in [almanac_path, ephemeris_path]:
if not os.path.exists(almanac_path):
self.__log.warn(f"AGPS file not found: {file}")
self.__log.warn("*NOT* loading AGPS data")
# reset device in case it is stuck, and set time
await self.reset()
await self.set_time()
await self._load_almanac(almanac_path)
await self._load_ephemeris(ephemeris_path)
async def _store_ephemeris(self, file='ephemeris.txt'):
await self._store_to_file('STMDUMPEPHEMS', '$PSTMEPHEM', file)
async def _store_almanac(self, file='almanac.txt'):
await self._store_to_file('STMDUMPALMANAC', '$PSTMALMANAC', file)
async def _load_ephemeris(self, file='ephemeris.txt'):
await self._store_to_file('STMDUMPEPHEMS', '$PSTMEPHEM', file)
async def _load_almanac(self, file='almanac.txt'):
await self._store_to_file('STMDUMPALMANAC', '$PSTMALMANAC', file)
async def set_time(self):
await self.reset()
now = datetime.utcnow()
# INITTIME expects values to be 2 or 4 digits long
msg = pynmea2.GGA('P', 'STMINITTIME', (
ret, line = await self._serial_write_cmd(msg, "STMINITTIMEOK")
if not ret:
raise LoggedException(f"ERROR: {line}")
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment