diff --git a/ogn/parser/parse.py b/ogn/parser/parse.py index 74b011f..7deb7e4 100644 --- a/ogn/parser/parse.py +++ b/ogn/parser/parse.py @@ -1,4 +1,5 @@ from datetime import datetime, timezone +import re from ogn.parser.utils import FPM_TO_MS, HPM_TO_DEGS, createTimestamp, KNOTS_TO_MS, KPH_TO_MS, FEETS_TO_METER, INCH_TO_MM, fahrenheit_to_celsius, CheapRuler, normalized_quality from ogn.parser.exceptions import AprsParseError @@ -10,6 +11,7 @@ dstcall_beacontype_mapping = { 'OGCAPT': 'capturs', + 'OGNDSC': 'dsc', 'OGNFNT': 'fanet', 'OGFLR': 'flarm', 'OGFLR6': 'flarm', @@ -146,6 +148,16 @@ def parse(aprs_message, reference_timestamp=None, calculate_relations=False, use else: raise ValueError("Raised unreachable exception") + if message.get('beacon_type') == 'dsc' and message['aprs_type'] == 'position': + # DSC proposal: provenance of the measurement timestamp, not a UTC accuracy claim. + codes = re.findall(r'(?:^|\s)dscT(\S*)(?=\s|$)', message.get('user_comment', '')) + if len(codes) == 1 and codes[0] in ('O', 'G', 'N', 'R'): + message['dsc_time_source'] = { + 'O': 'device_fix', 'G': 'android_gnss_clock', + 'N': 'android_network_clock', 'R': 'other_reconstruction', + }[codes[0]] + message['dsc_time_reconstructed'] = codes[0] != 'O' + if message['aprs_type'].startswith('position') and calculate_relations is True: positions[message['name']] = (message['longitude'], message['latitude']) if message['receiver_name'] in positions: diff --git a/tests/parser/parse/test_dsc.py b/tests/parser/parse/test_dsc.py new file mode 100644 index 0000000..c8e18ed --- /dev/null +++ b/tests/parser/parse/test_dsc.py @@ -0,0 +1,32 @@ +# Synthetic offline fixtures only: ABC123 and DSCDEMO are not allocated. +from datetime import datetime, timezone +import pytest +from ogn.parser import parse + +BASE = "OGNABC123>OGNDSC,qAR,DSCDEMO:/123456h4530.00N/00915.00E'360/021/A=000394 !W00! id37ABC123 +197fpm" + +@pytest.mark.parametrize('code,source,reconstructed', [('O','device_fix',False),('G','android_gnss_clock',True),('N','android_network_clock',True),('R','other_reconstruction',True)]) +def test_dsc_timestamp_provenance(code, source, reconstructed): + value=parse(BASE+' dscT'+code, datetime(2026,10,26,12,34,56,tzinfo=timezone.utc), use_server_timestamp=False) + assert value['beacon_type']=='dsc' + assert value['dsc_time_source']==source + assert value['dsc_time_reconstructed'] is reconstructed + assert value['address']=='ABC123' + assert value['timestamp'].isoformat()=='2026-10-26T12:34:56+00:00' + assert value['receiver_name']=='DSCDEMO' + +@pytest.mark.parametrize('suffix',['',' dscTX',' dscTG dscTN',' xdscTG',' dscTGjunk',' dscTG dscTX',' dscTG dscTG',' dscT',' dscTG dscT']) +def test_absent_invalid_or_ambiguous_quality_not_invented(suffix): + assert 'dsc_time_source' not in parse(BASE+suffix) + +def test_dsc_field_does_not_change_other_feeders(): + value=parse(BASE.replace('OGNDSC','OGNTRK')+' dscTG') + assert value['beacon_type']=='tracker' + assert 'dsc_time_source' not in value + +@pytest.mark.parametrize('path',['qOR','qAR,DSCDEMO','qAS,DSCDEMO']) +def test_optional_altitude_and_motion(path): + value=parse("OGNABC123>OGNDSC,"+path+":/123456h4530.00N/00915.00E' !W00! id37ABC123 dscTG") + assert value['beacon_type']=='dsc' + assert 'altitude' not in value + assert 'ground_speed' not in value