Introduction

AOGN is an Asynchronous OGN (Open Glider Network) client for modern Python. It connects to the APRS servers and receives the planes', gliders', receivers', etc. APRS messages, while still allowing you control of the program flow.

This can simplify programs since:

1. There are no callback or blocking functions (e.g. listen() or run_forever()).
2. One script/thread/process can still do other useful out-of-band tasks like computing aggregate statistics and making web requests.

To interpret the raw OGN messages, use a function like python-ogn-client'sogn.parser.parse.

Installation

pip install aogn

Usage

Basic example:

importasyncioimportloggingimportsysimportaognlogging.basicConfig(
stream=sys.stdout,
level=logging.DEBUG,
format='%(asctime)s %(levelname)s %(module)s %(message)s',
datefmt='%b %d %H:%M:%S',
)
asyncdefexample() ->None:
conn=aogn.Client(aprs_user='NO-CALL', )
try:
whileTrue:
# Get the APRS packet once available:raw_message=awaitconn.packet()
logging.debug(raw_message)
exceptKeyboardInterrupt:
logging.info('OGN Gateway stopped.')
awaitconn.disconnect()
if__name__=='__main__':
asyncio.run(example())

---

Concurrent example, with raw_message parsing:

importasyncioimportloggingimportsyslogging.basicConfig(
stream=sys.stdout,
level=logging.DEBUG,
format='%(asctime)s %(levelname)s %(module)s %(message)s',
datefmt='%b %d %H:%M:%S',
)
fromaognimportClientfromogn.parserimportparse, ParseErrordefprocess_beacon(raw_message):
beacon= {}
try:
beacon=parse(raw_message)
exceptParseErroraserr:
logging.warning(f'ParseError: {err}')
exceptNotImplementedErroraserr:
logging.error(f'NotImplementedError: {err}')
exceptAttributeErroraserr:
logging.error(f'raw_message: {raw_message}')
logging.error(f'beacon: {beacon}')
logging.error(err)
returnbeaconasyncdefexample() ->None:
conn=Client(aprs_user='NO-CALL', ) # aprs_filter='t/s')try:
whileTrue:
raw_message=awaitconn.packet()
ifraw_message:
beacon=process_beacon(raw_message)
exceptKeyboardInterrupt:
logging.info('OGN Gateway stopped.')
awaitconn.disconnect()
asyncdefanother_io_function() ->None:
importrandomwhileTrue:
sleep_duration=120*random.random()
logging.debug(f'Concurrently sleeping for {sleep_duration:.0f} seconds...')
awaitasyncio.sleep(sleep_duration)
asyncdefmain():
awaitasyncio.gather(example(), another_io_function())
if__name__=='__main__':
asyncio.run(main())