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

Refactoring

pull/77/head
Konstantin Gründger 2020-05-21 23:14:31 +02:00
rodzic 4d2557863f
commit 4968b9adf7
10 zmienionych plików z 82 dodań i 102 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

12
ogn/parser/aprs_comment/flarm_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_FLARM_POSITION_COMMENT
from ogn.parser.utils import FPM_TO_MS, HPM_TO_DEGS
@ -9,13 +7,13 @@ from .base import BaseParser
class FlarmParser(BaseParser):
def __init__(self):
self.beacon_type = 'flarm'
self.position_pattern = re.compile(PATTERN_FLARM_POSITION_COMMENT)
self.position_pattern = PATTERN_FLARM_POSITION_COMMENT
def parse_position(self, aprs_comment):
ac_match = self.position_pattern.match(aprs_comment)
match = self.position_pattern.match(aprs_comment)
def if_present(arg, func):
result = ac_match.group(arg)
result = match.group(arg)
return (func(result)) if result else None
return {'address_type': if_present('details', lambda x: int(x, 16) & 0b00000011),
@ -28,8 +26,8 @@ class FlarmParser(BaseParser):
'error_count': if_present('error_count', int),
'frequency_offset': if_present('frequency_offset', float),
'gps_quality': if_present('gps_quality', lambda _x: {
'horizontal': int(ac_match.group('gps_quality_horizontal')),
'vertical': int(ac_match.group('gps_quality_vertical'))}),
'horizontal': int(match.group('gps_quality_horizontal')),
'vertical': int(match.group('gps_quality_vertical'))}),
'software_version': if_present('software_version', float),
'hardware_version': if_present('hardware_version', lambda x: int(x, 16)),
'real_address': if_present('real_address', lambda x: x),

14
ogn/parser/aprs_comment/inreach_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_INREACH_POSITION_COMMENT
from .base import BaseParser
@ -8,11 +6,11 @@ from .base import BaseParser
class InreachParser(BaseParser):
def __init__(self):
self.beacon_type = 'inreach'
self.position_pattern = re.compile(PATTERN_INREACH_POSITION_COMMENT)
self.position_pattern = PATTERN_INREACH_POSITION_COMMENT
def parse_position(self, aprs_comment):
ac_match = self.position_pattern.match(aprs_comment)
return {'address': ac_match.group('id'),
'model': ac_match.group('model') if ac_match.group('model') else None,
'status': ac_match.group('status') == 'True' if ac_match.group('status') else None,
'pilot_name': ac_match.group('pilot_name') if ac_match.group('pilot_name') else None}
match = self.position_pattern.match(aprs_comment)
return {'address': match.group('id'),
'model': match.group('model') if match.group('model') else None,
'status': match.group('status') == 'True' if match.group('status') else None,
'pilot_name': match.group('pilot_name') if match.group('pilot_name') else None}

12
ogn/parser/aprs_comment/lt24_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.utils import FPM_TO_MS
from ogn.parser.pattern import PATTERN_LT24_POSITION_COMMENT
@ -9,10 +7,10 @@ from .base import BaseParser
class LT24Parser(BaseParser):
def __init__(self):
self.beacon_type = 'lt24'
self.position_pattern = re.compile(PATTERN_LT24_POSITION_COMMENT)
self.position_pattern = PATTERN_LT24_POSITION_COMMENT
def parse_position(self, aprs_comment):
ac_match = self.position_pattern.match(aprs_comment)
return {'address': ac_match.group('id'),
'climb_rate': int(ac_match.group('climb_rate')) * FPM_TO_MS if ac_match.group('climb_rate') else None,
'source': ac_match.group('source') if ac_match.group('source') else None}
match = self.position_pattern.match(aprs_comment)
return {'address': match.group('id'),
'climb_rate': int(match.group('climb_rate')) * FPM_TO_MS if match.group('climb_rate') else None,
'source': match.group('source') if match.group('source') else None}

4
ogn/parser/aprs_comment/naviter_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_NAVITER_POSITION_COMMENT
from ogn.parser.utils import FPM_TO_MS, HPM_TO_DEGS
@ -9,7 +7,7 @@ from .base import BaseParser
class NaviterParser(BaseParser):
def __init__(self):
self.beacon_type = 'naviter'
self.position_pattern = re.compile(PATTERN_NAVITER_POSITION_COMMENT)
self.position_pattern = PATTERN_NAVITER_POSITION_COMMENT
def parse_position(self, aprs_comment):
match = self.position_pattern.match(aprs_comment)

104
ogn/parser/aprs_comment/ogn_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.utils import FPM_TO_MS, HPM_TO_DEGS
from ogn.parser.pattern import PATTERN_RECEIVER_BEACON, PATTERN_AIRCRAFT_BEACON
@ -9,71 +7,71 @@ from .base import BaseParser
class OgnParser(BaseParser):
def __init__(self):
self.beacon_type = None
self.aircraft_pattern = re.compile(PATTERN_AIRCRAFT_BEACON)
self.receiver_pattern = re.compile(PATTERN_RECEIVER_BEACON)
self.aircraft_pattern = PATTERN_AIRCRAFT_BEACON
self.receiver_pattern = PATTERN_RECEIVER_BEACON
def parse(self, aprs_comment, aprs_type):
if not aprs_comment:
return {'beacon_type': 'aprs_receiver'}
ac_data = self.parse_aircraft_beacon(aprs_comment)
if ac_data:
ac_data.update({'beacon_type': 'aprs_aircraft'})
return ac_data
ab_data = self.parse_aircraft_beacon(aprs_comment)
if ab_data:
ab_data.update({'beacon_type': 'aprs_aircraft'})
return ab_data
rc_data = self.parse_receiver_beacon(aprs_comment)
if rc_data:
rc_data.update({'beacon_type': 'aprs_receiver'})
return rc_data
rb_data = self.parse_receiver_beacon(aprs_comment)
if rb_data:
rb_data.update({'beacon_type': 'aprs_receiver'})
return rb_data
else:
return {'user_comment': aprs_comment,
'beacon_type': 'aprs_receiver'}
def parse_aircraft_beacon(self, aprs_comment):
ac_match = self.aircraft_pattern.match(aprs_comment)
if ac_match:
return {'address_type': int(ac_match.group('details'), 16) & 0b00000011,
'aircraft_type': (int(ac_match.group('details'), 16) & 0b01111100) >> 2,
'stealth': (int(ac_match.group('details'), 16) & 0b10000000) >> 7 == 1,
'address': ac_match.group('address'),
'climb_rate': int(ac_match.group('climb_rate')) * FPM_TO_MS if ac_match.group('climb_rate') else None,
'turn_rate': float(ac_match.group('turn_rate')) * HPM_TO_DEGS if ac_match.group('turn_rate') else None,
'flightlevel': float(ac_match.group('flight_level')) if ac_match.group('flight_level') else None,
'signal_quality': float(ac_match.group('signal_quality')) if ac_match.group('signal_quality') else None,
'error_count': int(ac_match.group('errors')) if ac_match.group('errors') else None,
'frequency_offset': float(ac_match.group('frequency_offset')) if ac_match.group('frequency_offset') else None,
'gps_quality': {'horizontal': int(ac_match.group('gps_quality_horizontal')),
'vertical': int(ac_match.group('gps_quality_vertical'))} if ac_match.group('gps_quality') else None,
'software_version': float(ac_match.group('flarm_software_version')) if ac_match.group('flarm_software_version') else None,
'hardware_version': int(ac_match.group('flarm_hardware_version'), 16) if ac_match.group('flarm_hardware_version') else None,
'real_address': ac_match.group('flarm_id') if ac_match.group('flarm_id') else None,
'signal_power': float(ac_match.group('signal_power')) if ac_match.group('signal_power') else None,
'proximity': [hear[4:] for hear in ac_match.group('proximity').split(" ")] if ac_match.group('proximity') else None}
ab_match = self.aircraft_pattern.match(aprs_comment)
if ab_match:
return {'address_type': int(ab_match.group('details'), 16) & 0b00000011,
'aircraft_type': (int(ab_match.group('details'), 16) & 0b01111100) >> 2,
'stealth': (int(ab_match.group('details'), 16) & 0b10000000) >> 7 == 1,
'address': ab_match.group('address'),
'climb_rate': int(ab_match.group('climb_rate')) * FPM_TO_MS if ab_match.group('climb_rate') else None,
'turn_rate': float(ab_match.group('turn_rate')) * HPM_TO_DEGS if ab_match.group('turn_rate') else None,
'flightlevel': float(ab_match.group('flight_level')) if ab_match.group('flight_level') else None,
'signal_quality': float(ab_match.group('signal_quality')) if ab_match.group('signal_quality') else None,
'error_count': int(ab_match.group('errors')) if ab_match.group('errors') else None,
'frequency_offset': float(ab_match.group('frequency_offset')) if ab_match.group('frequency_offset') else None,
'gps_quality': {'horizontal': int(ab_match.group('gps_quality_horizontal')),
'vertical': int(ab_match.group('gps_quality_vertical'))} if ab_match.group('gps_quality') else None,
'software_version': float(ab_match.group('flarm_software_version')) if ab_match.group('flarm_software_version') else None,
'hardware_version': int(ab_match.group('flarm_hardware_version'), 16) if ab_match.group('flarm_hardware_version') else None,
'real_address': ab_match.group('flarm_id') if ab_match.group('flarm_id') else None,
'signal_power': float(ab_match.group('signal_power')) if ab_match.group('signal_power') else None,
'proximity': [hear[4:] for hear in ab_match.group('proximity').split(" ")] if ab_match.group('proximity') else None}
else:
return None
def parse_receiver_beacon(self, aprs_comment):
rec_match = self.receiver_pattern.match(aprs_comment)
if rec_match:
return {'version': rec_match.group('version'),
'platform': rec_match.group('platform'),
'cpu_load': float(rec_match.group('cpu_load')),
'free_ram': float(rec_match.group('ram_free')),
'total_ram': float(rec_match.group('ram_total')),
'ntp_error': float(rec_match.group('ntp_offset')),
'rt_crystal_correction': float(rec_match.group('ntp_correction')),
'voltage': float(rec_match.group('voltage')) if rec_match.group('voltage') else None,
'amperage': float(rec_match.group('amperage')) if rec_match.group('amperage') else None,
'cpu_temp': float(rec_match.group('cpu_temperature')) if rec_match.group('cpu_temperature') else None,
'senders_visible': int(rec_match.group('visible_senders')) if rec_match.group('visible_senders') else None,
'senders_total': int(rec_match.group('senders')) if rec_match.group('senders') else None,
'rec_crystal_correction': int(rec_match.group('rf_correction_manual')) if rec_match.group('rf_correction_manual') else None,
'rec_crystal_correction_fine': float(rec_match.group('rf_correction_automatic')) if rec_match.group('rf_correction_automatic') else None,
'rec_input_noise': float(rec_match.group('signal_quality')) if rec_match.group('signal_quality') else None,
'senders_signal': float(rec_match.group('senders_signal_quality')) if rec_match.group('senders_signal_quality') else None,
'senders_messages': float(rec_match.group('senders_messages')) if rec_match.group('senders_messages') else None,
'good_senders_signal': float(rec_match.group('good_senders_signal_quality')) if rec_match.group('good_senders_signal_quality') else None,
'good_senders': float(rec_match.group('good_senders')) if rec_match.group('good_senders') else None,
'good_and_bad_senders': float(rec_match.group('good_and_bad_senders')) if rec_match.group('good_and_bad_senders') else None}
rb_match = self.receiver_pattern.match(aprs_comment)
if rb_match:
return {'version': rb_match.group('version'),
'platform': rb_match.group('platform'),
'cpu_load': float(rb_match.group('cpu_load')),
'free_ram': float(rb_match.group('ram_free')),
'total_ram': float(rb_match.group('ram_total')),
'ntp_error': float(rb_match.group('ntp_offset')),
'rt_crystal_correction': float(rb_match.group('ntp_correction')),
'voltage': float(rb_match.group('voltage')) if rb_match.group('voltage') else None,
'amperage': float(rb_match.group('amperage')) if rb_match.group('amperage') else None,
'cpu_temp': float(rb_match.group('cpu_temperature')) if rb_match.group('cpu_temperature') else None,
'senders_visible': int(rb_match.group('visible_senders')) if rb_match.group('visible_senders') else None,
'senders_total': int(rb_match.group('senders')) if rb_match.group('senders') else None,
'rec_crystal_correction': int(rb_match.group('rf_correction_manual')) if rb_match.group('rf_correction_manual') else None,
'rec_crystal_correction_fine': float(rb_match.group('rf_correction_automatic')) if rb_match.group('rf_correction_automatic') else None,
'rec_input_noise': float(rb_match.group('signal_quality')) if rb_match.group('signal_quality') else None,
'senders_signal': float(rb_match.group('senders_signal_quality')) if rb_match.group('senders_signal_quality') else None,
'senders_messages': float(rb_match.group('senders_messages')) if rb_match.group('senders_messages') else None,
'good_senders_signal': float(rb_match.group('good_senders_signal_quality')) if rb_match.group('good_senders_signal_quality') else None,
'good_senders': float(rb_match.group('good_senders')) if rb_match.group('good_senders') else None,
'good_and_bad_senders': float(rb_match.group('good_and_bad_senders')) if rb_match.group('good_and_bad_senders') else None}
else:
return None

6
ogn/parser/aprs_comment/receiver_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_RECEIVER_POSITION_COMMENT, PATTERN_RECEIVER_STATUS_COMMENT
from .base import BaseParser
@ -8,8 +6,8 @@ from .base import BaseParser
class ReceiverParser(BaseParser):
def __init__(self):
self.beacon_type = 'receiver'
self.position_pattern = re.compile(PATTERN_RECEIVER_POSITION_COMMENT)
self.status_pattern = re.compile(PATTERN_RECEIVER_STATUS_COMMENT)
self.position_pattern = PATTERN_RECEIVER_POSITION_COMMENT
self.status_pattern = PATTERN_RECEIVER_STATUS_COMMENT
def parse_position(self, aprs_comment):
if aprs_comment is None:

10
ogn/parser/aprs_comment/skylines_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.utils import FPM_TO_MS
from ogn.parser.pattern import PATTERN_SKYLINES_POSITION_COMMENT
@ -9,9 +7,9 @@ from .base import BaseParser
class SkylinesParser(BaseParser):
def __init__(self):
self.beacon_type = 'skylines'
self.position_pattern = re.compile(PATTERN_SKYLINES_POSITION_COMMENT)
self.position_pattern = PATTERN_SKYLINES_POSITION_COMMENT
def parse_position(self, aprs_comment):
ac_match = self.position_pattern.match(aprs_comment)
return {'address': ac_match.group('id'),
'climb_rate': int(ac_match.group('climb_rate')) * FPM_TO_MS if ac_match.group('climb_rate') else None}
match = self.position_pattern.match(aprs_comment)
return {'address': match.group('id'),
'climb_rate': int(match.group('climb_rate')) * FPM_TO_MS if match.group('climb_rate') else None}

4
ogn/parser/aprs_comment/spider_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_SPIDER_POSITION_COMMENT
from .base import BaseParser
@ -8,7 +6,7 @@ from .base import BaseParser
class SpiderParser(BaseParser):
def __init__(self):
self.beacon_type = 'spider'
self.position_pattern = re.compile(PATTERN_SPIDER_POSITION_COMMENT)
self.position_pattern = PATTERN_SPIDER_POSITION_COMMENT
def parse_position(self, aprs_comment):
ac_match = self.position_pattern.match(aprs_comment)

12
ogn/parser/aprs_comment/spot_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_SPOT_POSITION_COMMENT
from .base import BaseParser
@ -8,10 +6,10 @@ from .base import BaseParser
class SpotParser(BaseParser):
def __init__(self):
self.beacon_type = 'spot'
self.position_pattern = re.compile(PATTERN_SPOT_POSITION_COMMENT)
self.position_pattern = PATTERN_SPOT_POSITION_COMMENT
def parse_position(self, aprs_comment):
ac_match = self.position_pattern.match(aprs_comment)
return {'address': ac_match.group('id'),
'model': ac_match.group('model') if ac_match.group('model') else None,
'status': ac_match.group('status') if ac_match.group('status') else None}
match = self.position_pattern.match(aprs_comment)
return {'address': match.group('id'),
'model': match.group('model') if match.group('model') else None,
'status': match.group('status') if match.group('status') else None}

6
ogn/parser/aprs_comment/tracker_parser.py
Wyświetl plik

@ -1,5 +1,3 @@
import re
from ogn.parser.pattern import PATTERN_TRACKER_POSITION_COMMENT, PATTERN_TRACKER_STATUS_COMMENT
from ogn.parser.utils import FPM_TO_MS, HPM_TO_DEGS
@ -9,8 +7,8 @@ from .base import BaseParser
class TrackerParser(BaseParser):
def __init__(self):
self.beacon_type = 'tracker'
self.position_pattern = re.compile(PATTERN_TRACKER_POSITION_COMMENT)
self.status_pattern = re.compile(PATTERN_TRACKER_STATUS_COMMENT)
self.position_pattern = PATTERN_TRACKER_POSITION_COMMENT
self.status_pattern = PATTERN_TRACKER_STATUS_COMMENT
def parse_position(self, aprs_comment):
match = self.position_pattern.match(aprs_comment)