mirror
/
ogn-python
kopia lustrzana https://github.com/glidernet/ogn-python
1
0
Forkuj 0
Kod Zgłoszenia Projekty Wydania Wiki Aktywność

ogn.gateway: Remove implicit sqlalchemy dependency 

The callback gets the beacon data by a python dict instead of a sqlalchemy class
pull/43/head
Fabian P. Schmidt 2016-02-18 22:28:03 +01:00
rodzic 6a31a648cb
commit d69476158b
7 zmienionych plików z 121 dodań i 123 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

21
README.md
Wyświetl plik

@ -27,16 +27,23 @@ lets you process the incoming data.
Example:
```python
#!/usr/bin/env python3
from ogn.model import AircraftBeacon, ReceiverBeacon
from ogn.gateway.client import ognGateway
from ogn.parser.parse import parse_aprs, parse_ogn_beacon
from ogn.parser.exceptions import ParseError
def process_beacon(beacon):
if type(beacon) is AircraftBeacon:
print('Received aircraft beacon from {}'.format(beacon.name))
elif type(beacon) is ReceiverBeacon:
print('Received receiver beacon from {}'.format(beacon.name))
def process_beacon(raw_message):
if raw_message[0] == '#':
print('Server Status: {}'.format(raw_message))
return
try:
message = parse_aprs(raw_message)
message.update(parse_ogn_beacon(message['comment']))
print('Received {beacon_type} from {name}'.format(**message))
except ParseError as e:
print('Error, {}'.format(e.message))
if __name__ == '__main__':

35
ogn/aprs_parser.py
Wyświetl plik

@ -1,35 +0,0 @@
from datetime import datetime
from .model import Beacon, AircraftBeacon, ReceiverBeacon
from ogn.parser.exceptions import AprsParseError
def parse_aprs(packet, reference_date=None):
if reference_date is None:
reference_date = datetime.utcnow()
if not isinstance(packet, str):
raise TypeError("Expected packet to be str, got %s" % type(packet))
elif packet == "":
raise AprsParseError("(empty string)")
elif packet[0] == "#":
return None
beacon = Beacon()
beacon.parse(packet, reference_date)
# symboltable / symbolcodes used by OGN:
# I&: used as receiver
# /X: helicopter_rotorcraft
# /': glider_or_motorglider
# \^: powered_aircraft
# /g: para_glider
# /O: ?
# /^: ?
# \n: ?
# /z: ?
# /o: ?
if beacon.symboltable == "I" and beacon.symbolcode == "&":
return ReceiverBeacon(beacon)
else:
return AircraftBeacon(beacon)

23
ogn/gateway/client.py
Wyświetl plik

@ -3,8 +3,6 @@ import logging
from time import time
from ogn.gateway import settings
from ogn.aprs_parser import parse_aprs
from ogn.parser.exceptions import AprsParseError, OgnParseError, AmbigousTimeError
def create_aprs_login(user_name, pass_code, app_name, app_version, aprs_filter=None):
@ -48,8 +46,6 @@ class ognGateway:
self.logger.error('Socket close error', exc_info=True)
def run(self, callback, autoreconnect=False):
self.process_beacon = callback
while True:
try:
keepalive_time = time()
@ -68,7 +64,7 @@ class ognGateway:
self.logger.warning('Read returns zero length string. Failure. Orderly closeout')
break
self.proceed_line(packet_str)
callback(packet_str)
except BrokenPipeError:
self.logger.error('BrokenPipeError', exc_info=True)
except socket.error:
@ -78,20 +74,3 @@ class ognGateway:
self.connect()
else:
return
def proceed_line(self, line):
try:
beacon = parse_aprs(line)
self.logger.debug('Received beacon: {}'.format(beacon))
except AprsParseError:
self.logger.error('AprsParseError while parsing line: {}'.format(line), exc_info=True)
return
except OgnParseError:
self.logger.error('OgnParseError while parsing line: {}'.format(line), exc_info=True)
return
except AmbigousTimeError as e:
self.logger.error('Drop packet, {:.0f}s from past: {}'.format(e.timedelta.total_seconds(), line))
return
if beacon is not None:
self.process_beacon(beacon)

6
ogn/gateway/manage.py
Wyświetl plik

@ -1,7 +1,7 @@
import logging
from ogn.gateway.client import ognGateway
from ogn.commands.dbutils import session
from ogn.gateway.process import process_beacon
from manager import Manager
@ -33,10 +33,6 @@ def run(aprs_user='anon-dev', logfile='main.log', loglevel='INFO'):
gateway = ognGateway(aprs_user)
gateway.connect()
def process_beacon(beacon):
session.add(beacon)
session.commit()
try:
gateway.run(callback=process_beacon, autoreconnect=True)
except KeyboardInterrupt:

38
ogn/gateway/process.py 100644
Wyświetl plik

@ -0,0 +1,38 @@
import logging
from ogn.commands.dbutils import session
from ogn.model import AircraftBeacon, ReceiverBeacon
from ogn.parser.parse import parse_aprs, parse_ogn_receiver_beacon, parse_ogn_aircraft_beacon
from ogn.parser.exceptions import AprsParseError, OgnParseError, AmbigousTimeError
logger = logging.getLogger(__name__)
def process_beacon(raw_message):
if raw_message[0] == '#':
return
try:
message = parse_aprs(raw_message)
# symboltable / symbolcodes used by OGN:
# I&: used as receiver
# /X: helicopter_rotorcraft
# /': glider_or_motorglider
# \^: powered_aircraft
# /g: para_glider
# /O: ?
# /^: ?
# \n: ?
# /z: ?
# /o: ?
if message['symboltable'] == "I" and message['symbolcode'] == '&':
message.update(parse_ogn_receiver_beacon(message['comment']))
beacon = ReceiverBeacon(**message)
else:
message.update(parse_ogn_aircraft_beacon(message['comment']))
beacon = AircraftBeacon(**message)
session.add(beacon)
session.commit()
logger.debug('Received message: {}'.format(raw_message))
except (AprsParseError, OgnParseError, AmbigousTimeError) as e:
logger.error('Received message: {}'.format(raw_message))
logger.error('Drop packet, {}'.format(e.message))

67
tests/parser/test_parse.py 100644
Wyświetl plik

@ -0,0 +1,67 @@
import unittest
import unittest.mock as mock
from datetime import datetime
from time import sleep
from ogn.parser.parse import parse_aprs, parse_ogn_beacon
from ogn.parser.exceptions import AprsParseError, OgnParseError
class TestStringMethods(unittest.TestCase):
def test_valid_beacons(self):
with open('tests/valid_beacons.txt') as f:
for line in f:
if not line[0] == '#':
aprs = parse_aprs(line, datetime(2015, 4, 10, 17, 0))
parse_ogn_beacon(aprs['comment'])
def test_fail_none(self):
with self.assertRaises(TypeError):
parse_aprs(None)
def test_fail_empty(self):
with self.assertRaises(AprsParseError):
parse_aprs("")
def test_fail_bad_string(self):
with self.assertRaises(AprsParseError):
parse_aprs("Lachens>APRS,TCPIwontbeavalidstring")
def test_incomplete_device_string(self):
with self.assertRaises(OgnParseError):
aprs = parse_aprs("ICA4B0E3A>APRS,qAS,Letzi:/072319h4711.75N\\00802.59E^327/149/A=006498 id154B0E3A -395",
datetime(2015, 4, 10, 7, 24))
parse_ogn_beacon(aprs['comment'])
def test_incomplete_receiver_string(self):
with self.assertRaises(OgnParseError):
aprs = parse_aprs("Lachens>APRS,TCPIP*,qAC,GLIDERN2:/165334h4344.70NI00639.19E&/A=005435 v0.2.1 CPU:0.3 RAM:1764.4/21",
datetime(2015, 4, 10, 16, 54))
parse_ogn_beacon(aprs['comment'])
@mock.patch('ogn.parser.parse.createTimestamp')
def test_default_reference_date(self, createTimestamp_mock):
valid_aprs_string = "Lachens>APRS,TCPIP*,qAC,GLIDERN2:/165334h4344.70NI00639.19E&/A=005435 v0.2.1 CPU:0.3 RAM:1764.4/21"
parse_aprs(valid_aprs_string)
call_args_before = createTimestamp_mock.call_args
sleep(1)
parse_aprs(valid_aprs_string)
call_args_seconds_later = createTimestamp_mock.call_args
self.assertNotEqual(call_args_before, call_args_seconds_later)
def test_copy_constructor(self):
valid_aprs_string = "FLRDDA5BA>APRS,qAS,LFMX:/160829h4415.41N/00600.03E'342/049/A=005524 id0ADDA5BA -454fpm -1.1rot 8.8dB 0e +51.2kHz gps4x5"
aprs = parse_aprs(valid_aprs_string, reference_date=datetime(2015, 1, 1, 16, 8, 29))
aircraft_beacon = parse_ogn_beacon(aprs['comment'])
self.assertEqual(aprs['name'], 'FLRDDA5BA')
self.assertEqual(aircraft_beacon['address'], 'DDA5BA')
if __name__ == '__main__':
unittest.main()

54
tests/test_aprs_parser.py
Wyświetl plik

@ -1,54 +0,0 @@
import unittest
import unittest.mock as mock
from datetime import datetime
from time import sleep
from ogn.aprs_parser import parse_aprs
from ogn.parser.exceptions import AprsParseError, OgnParseError
class TestStringMethods(unittest.TestCase):
def test_valid_beacons(self):
with open('tests/valid_beacons.txt') as f:
for line in f:
parse_aprs(line, datetime(2015, 4, 10, 17, 0))
def test_fail_none(self):
with self.assertRaises(TypeError):
parse_aprs(None)
def test_fail_empty(self):
with self.assertRaises(AprsParseError):
parse_aprs("")
def test_fail_bad_string(self):
with self.assertRaises(AprsParseError):
parse_aprs("Lachens>APRS,TCPIwontbeavalidstring")
def test_incomplete_device_string(self):
with self.assertRaises(OgnParseError):
parse_aprs("ICA4B0E3A>APRS,qAS,Letzi:/072319h4711.75N\\00802.59E^327/149/A=006498 id154B0E3A -395",
datetime(2015, 4, 10, 7, 24))
def test_incomplete_receiver_string(self):
with self.assertRaises(OgnParseError):
parse_aprs("Lachens>APRS,TCPIP*,qAC,GLIDERN2:/165334h4344.70NI00639.19E&/A=005435 v0.2.1 CPU:0.3 RAM:1764.4/21",
datetime(2015, 4, 10, 16, 54))
@mock.patch('ogn.aprs_parser.Beacon')
def test_default_reference_date(self, beacon_mock):
instance = beacon_mock.return_value
valid_aprs_string = "Lachens>APRS,TCPIP*,qAC,GLIDERN2:/165334h4344.70NI00639.19E&/A=005435 v0.2.1 CPU:0.3 RAM:1764.4/21"
parse_aprs(valid_aprs_string)
call_args = instance.parse.call_args
sleep(1)
parse_aprs(valid_aprs_string)
call_args_one_second_later = instance.parse.call_args
self.assertNotEqual(call_args, call_args_one_second_later)
if __name__ == '__main__':
unittest.main()