openrvdas.logger.utils.nmea_timestamp
Extract timestamps from NMEA sentences.
Supports all common NMEA 0183 sentence types that include timestamps, both standard and proprietary. Tracks the most recent NMEA-derived timestamp so that non-timestamped sentences (e.g. VTG, HDT) can reuse it, subject to a configurable staleness timeout.
Supported standard sentences: GGA, GLL, RMC, ZDA, GBS, GST, GNS, BWC, TLL, TTM
Supported proprietary sentences: PASHR, PGRMF, PTNL GGK/PJK, PUBX 00/04, PSIMSNS, PSIMSSB, PSXN 26
1#!/usr/bin/env python3 2"""Extract timestamps from NMEA sentences. 3 4Supports all common NMEA 0183 sentence types that include timestamps, 5both standard and proprietary. Tracks the most recent NMEA-derived 6timestamp so that non-timestamped sentences (e.g. VTG, HDT) can reuse 7it, subject to a configurable staleness timeout. 8 9Supported standard sentences: 10 GGA, GLL, RMC, ZDA, GBS, GST, GNS, BWC, TLL, TTM 11 12Supported proprietary sentences: 13 PASHR, PGRMF, PTNL GGK/PJK, PUBX 00/04, PSIMSNS, PSIMSSB, PSXN 26 14""" 15 16import logging 17import re 18import time 19from datetime import datetime, timezone 20 21from logger.utils import timestamp as ts_util 22 23# Captures the full sentence identifier between $/! and the first comma. 24# Works for both standard (e.g. "GPGGA") and proprietary (e.g. "PSIMSNS"). 25NMEA_PREFIX_RE = re.compile(r'^[$!](\w+),') 26 27# NMEA time field: hhmmss or hhmmss.ss(s). 28# No end-of-string anchor so it tolerates residual checksum characters. 29NMEA_TIME_RE = re.compile(r'^(\d{2})(\d{2})(\d{2}(?:\.\d+)?)') 30 31# --------------------------------------------------------------------------- 32# Sentence lookup tables 33# 34# Standard sentences are identified by the last 3 characters of a 5-char 35# talker+type prefix (e.g. "GPGGA" -> "GGA"). Proprietary sentences are 36# identified by their full prefix or by prefix + sub-type field. 37# 38# Values are the comma-split field index of the time (hhmmss.ss) field. 39# --------------------------------------------------------------------------- 40 41# Standard NMEA sentences: 3-char type -> time field index 42_STD_TIME = { 43 'GGA': 1, 'GLL': 5, 'RMC': 1, 'ZDA': 1, 44 'GBS': 1, 'GST': 1, 'GNS': 1, 'BWC': 1, 45 'TLL': 7, 'TTM': 14, 46} 47 48# Standard sentences carrying a ddmmyy date: type -> date field index 49_STD_DATE = {'RMC': 9} 50 51# ZDA carries day/month/year in separate fields 52_ZDA_DATE_FIELDS = (2, 3, 4) # (day_idx, month_idx, year_idx) 53 54# 5-char proprietary sentences (matched by full prefix without $) 55_PROP5_TIME = {'PASHR': 1, 'PGRMF': 4} 56_PROP5_DATE = {'PGRMF': 3} 57 58# Sub-typed proprietary: (prefix, first-data-field) -> time field index 59_SUB_TIME = { 60 ('PTNL', 'GGK'): 2, ('PTNL', 'PJK'): 2, 61 ('PUBX', '00'): 2, ('PUBX', '04'): 2, 62} 63_SUB_DATE = { 64 ('PTNL', 'GGK'): 3, ('PTNL', 'PJK'): 3, 65 ('PUBX', '04'): 3, 66} 67 68# Long-name proprietary (>5 chars): full prefix -> time field index 69_LONG_TIME = {'PSIMSNS': 1, 'PSIMSSB': 6} 70 71 72def _strip_checksum(field): 73 """Remove ``*checksum`` suffix from a field value.""" 74 idx = field.find('*') 75 return field[:idx] if idx >= 0 else field 76 77 78class NMEATimestampExtractor: 79 """Parse NMEA sentences to extract embedded timestamps.""" 80 81 def __init__(self, timeout=1, quiet=False, time_drift_threshold=0.1): 82 self.timeout = timeout 83 self.quiet = quiet 84 self.time_drift_threshold = time_drift_threshold 85 self.last_nmea_date = None # datetime.date from RMC/ZDA/etc. 86 self.last_nmea_datetime = None # full datetime object 87 self.last_nmea_system_time = 0 # time.time() of last extraction 88 self.seen_timestamp = False 89 90 def get_timestamp(self, record, time_format=ts_util.TIME_FORMAT, 91 time_zone=ts_util.timezone.utc): 92 """Return a formatted timestamp string extracted from *record*, 93 or fall back to the last observed NMEA timestamp. Returns None 94 when no NMEA timestamp is available (caller should use system time). 95 """ 96 if not isinstance(record, str): 97 return None 98 99 nmea_dt = self._parse_nmea_datetime(record) 100 if nmea_dt is not None: 101 now = time.time() 102 self.last_nmea_datetime = nmea_dt 103 self.last_nmea_system_time = now 104 self.seen_timestamp = True 105 if not self.quiet and self.time_drift_threshold is not None: 106 nmea_epoch = nmea_dt.timestamp() 107 drift = abs(nmea_epoch - now) 108 if drift > self.time_drift_threshold: 109 logging.warning( 110 'NMEATimestampExtractor: NMEA time differs from ' 111 'system time by %.3fs (threshold=%.3fs)', 112 drift, self.time_drift_threshold) 113 return nmea_dt.strftime(time_format) 114 115 # Fall back to last observed timestamp 116 if not self.seen_timestamp: 117 return None 118 119 elapsed = time.time() - self.last_nmea_system_time 120 if elapsed > self.timeout: 121 if not self.quiet: 122 logging.warning( 123 'NMEATimestampExtractor: last NMEA timestamp is %.1fs ' 124 'old (timeout=%s); falling back to system time', 125 elapsed, self.timeout) 126 return None 127 128 return self.last_nmea_datetime.strftime(time_format) 129 130 # ------------------------------------------------------------------ 131 # Internal helpers 132 # ------------------------------------------------------------------ 133 134 def _parse_nmea_datetime(self, record): 135 """Try to extract a datetime from an NMEA sentence. 136 Returns a datetime object or None. 137 """ 138 m = NMEA_PREFIX_RE.match(record) 139 if not m: 140 return None 141 142 prefix = m.group(1).upper() 143 fields = record.split(',') 144 145 # 1. Standard 5-char prefix (2-char talker + 3-char type) 146 # and 5-char proprietary (e.g. PASHR, PGRMF) 147 if len(prefix) == 5: 148 stype = prefix[2:] 149 if stype in _STD_TIME: 150 return self._extract_standard(stype, fields) 151 if prefix in _PROP5_TIME: 152 return self._extract_prop5(prefix, fields) 153 154 # 2. Sub-typed proprietary (prefix + first data field) 155 if len(fields) > 1: 156 sub = _strip_checksum(fields[1]).strip().upper() 157 subkey = (prefix, sub) 158 if subkey == ('PSXN', '26'): 159 return self._extract_psxn26(fields) 160 if subkey in _SUB_TIME: 161 return self._extract_subtype(subkey, fields) 162 163 # 3. Long-name proprietary (>5 chars) 164 if prefix in _LONG_TIME: 165 return self._extract_long(prefix, fields) 166 167 return None 168 169 # -- Extraction helpers ------------------------------------------------ 170 171 def _extract_standard(self, stype, fields): 172 """Extract datetime from a standard NMEA sentence.""" 173 time_idx = _STD_TIME[stype] 174 date = None 175 if stype == 'ZDA': 176 day_i, month_i, year_i = _ZDA_DATE_FIELDS 177 date = self._parse_zda_date( 178 self._get_field(fields, day_i), 179 self._get_field(fields, month_i), 180 self._get_field(fields, year_i)) 181 elif stype in _STD_DATE: 182 date = self._parse_ddmmyy( 183 self._get_field(fields, _STD_DATE[stype])) 184 return self._datetime_from_time_field( 185 self._get_field(fields, time_idx), date=date) 186 187 def _extract_prop5(self, prefix, fields): 188 """Extract datetime from a 5-char proprietary sentence.""" 189 time_idx = _PROP5_TIME[prefix] 190 date = None 191 if prefix in _PROP5_DATE: 192 date = self._parse_ddmmyy( 193 self._get_field(fields, _PROP5_DATE[prefix])) 194 return self._datetime_from_time_field( 195 self._get_field(fields, time_idx), date=date) 196 197 def _extract_subtype(self, subkey, fields): 198 """Extract datetime from a sub-typed proprietary sentence.""" 199 time_idx = _SUB_TIME[subkey] 200 date = None 201 if subkey in _SUB_DATE: 202 date = self._parse_ddmmyy( 203 self._get_field(fields, _SUB_DATE[subkey])) 204 return self._datetime_from_time_field( 205 self._get_field(fields, time_idx), date=date) 206 207 def _extract_long(self, prefix, fields): 208 """Extract datetime from a long-name proprietary sentence.""" 209 time_idx = _LONG_TIME[prefix] 210 return self._datetime_from_time_field( 211 self._get_field(fields, time_idx)) 212 213 def _extract_psxn26(self, fields): 214 """Extract datetime from PSXN,26 (separate Y/M/D/H/M/S fields).""" 215 try: 216 year = int(self._get_field(fields, 2)) 217 month = int(self._get_field(fields, 3)) 218 day = int(self._get_field(fields, 4)) 219 hour = int(self._get_field(fields, 5)) 220 minute = int(self._get_field(fields, 6)) 221 sec_str = self._get_field(fields, 7) 222 second_frac = float(sec_str) 223 second = int(second_frac) 224 microsecond = int((second_frac - second) * 1_000_000) 225 date = datetime(year, month, day, tzinfo=timezone.utc).date() 226 self.last_nmea_date = date 227 return datetime( 228 year=year, month=month, day=day, 229 hour=hour, minute=minute, second=second, 230 microsecond=microsecond, tzinfo=timezone.utc) 231 except (ValueError, IndexError): 232 return None 233 234 # -- Field access and parsing ------------------------------------------ 235 236 @staticmethod 237 def _get_field(fields, idx): 238 """Safely get a comma-split field, stripping any checksum suffix.""" 239 if idx < len(fields): 240 return _strip_checksum(fields[idx].strip()) 241 return '' 242 243 def _datetime_from_time_field(self, time_field, date=None): 244 """Parse an NMEA time field (hhmmss.ss) and combine with a date. 245 If *date* is provided, also update ``self.last_nmea_date``. 246 """ 247 m = NMEA_TIME_RE.match(time_field) 248 if not m: 249 return None 250 251 hour = int(m.group(1)) 252 minute = int(m.group(2)) 253 second_frac = float(m.group(3)) 254 second = int(second_frac) 255 microsecond = int((second_frac - second) * 1_000_000) 256 257 if date is not None: 258 self.last_nmea_date = date 259 use_date = self.last_nmea_date 260 if use_date is None: 261 use_date = datetime.now(timezone.utc).date() 262 263 return datetime( 264 year=use_date.year, month=use_date.month, day=use_date.day, 265 hour=hour, minute=minute, second=second, 266 microsecond=microsecond, tzinfo=timezone.utc, 267 ) 268 269 @staticmethod 270 def _parse_ddmmyy(date_field): 271 """Parse a ``ddmmyy`` date string (used by RMC, PGRMF, etc.).""" 272 if not date_field or len(date_field) < 6: 273 return None 274 try: 275 day = int(date_field[0:2]) 276 month = int(date_field[2:4]) 277 year = int(date_field[4:6]) 278 year += 2000 if year < 80 else 1900 279 return datetime(year, month, day, tzinfo=timezone.utc).date() 280 except (ValueError, IndexError): 281 return None 282 283 @staticmethod 284 def _parse_zda_date(day_field, month_field, year_field): 285 """Parse day, month, year fields from a ZDA sentence.""" 286 try: 287 day = int(day_field) 288 month = int(month_field) 289 year = int(year_field) 290 return datetime(year, month, day, tzinfo=timezone.utc).date() 291 except (ValueError, IndexError): 292 return None
NMEA_PREFIX_RE =
re.compile('^[$!](\\w+),')
NMEA_TIME_RE =
re.compile('^(\\d{2})(\\d{2})(\\d{2}(?:\\.\\d+)?)')
class
NMEATimestampExtractor:
79class NMEATimestampExtractor: 80 """Parse NMEA sentences to extract embedded timestamps.""" 81 82 def __init__(self, timeout=1, quiet=False, time_drift_threshold=0.1): 83 self.timeout = timeout 84 self.quiet = quiet 85 self.time_drift_threshold = time_drift_threshold 86 self.last_nmea_date = None # datetime.date from RMC/ZDA/etc. 87 self.last_nmea_datetime = None # full datetime object 88 self.last_nmea_system_time = 0 # time.time() of last extraction 89 self.seen_timestamp = False 90 91 def get_timestamp(self, record, time_format=ts_util.TIME_FORMAT, 92 time_zone=ts_util.timezone.utc): 93 """Return a formatted timestamp string extracted from *record*, 94 or fall back to the last observed NMEA timestamp. Returns None 95 when no NMEA timestamp is available (caller should use system time). 96 """ 97 if not isinstance(record, str): 98 return None 99 100 nmea_dt = self._parse_nmea_datetime(record) 101 if nmea_dt is not None: 102 now = time.time() 103 self.last_nmea_datetime = nmea_dt 104 self.last_nmea_system_time = now 105 self.seen_timestamp = True 106 if not self.quiet and self.time_drift_threshold is not None: 107 nmea_epoch = nmea_dt.timestamp() 108 drift = abs(nmea_epoch - now) 109 if drift > self.time_drift_threshold: 110 logging.warning( 111 'NMEATimestampExtractor: NMEA time differs from ' 112 'system time by %.3fs (threshold=%.3fs)', 113 drift, self.time_drift_threshold) 114 return nmea_dt.strftime(time_format) 115 116 # Fall back to last observed timestamp 117 if not self.seen_timestamp: 118 return None 119 120 elapsed = time.time() - self.last_nmea_system_time 121 if elapsed > self.timeout: 122 if not self.quiet: 123 logging.warning( 124 'NMEATimestampExtractor: last NMEA timestamp is %.1fs ' 125 'old (timeout=%s); falling back to system time', 126 elapsed, self.timeout) 127 return None 128 129 return self.last_nmea_datetime.strftime(time_format) 130 131 # ------------------------------------------------------------------ 132 # Internal helpers 133 # ------------------------------------------------------------------ 134 135 def _parse_nmea_datetime(self, record): 136 """Try to extract a datetime from an NMEA sentence. 137 Returns a datetime object or None. 138 """ 139 m = NMEA_PREFIX_RE.match(record) 140 if not m: 141 return None 142 143 prefix = m.group(1).upper() 144 fields = record.split(',') 145 146 # 1. Standard 5-char prefix (2-char talker + 3-char type) 147 # and 5-char proprietary (e.g. PASHR, PGRMF) 148 if len(prefix) == 5: 149 stype = prefix[2:] 150 if stype in _STD_TIME: 151 return self._extract_standard(stype, fields) 152 if prefix in _PROP5_TIME: 153 return self._extract_prop5(prefix, fields) 154 155 # 2. Sub-typed proprietary (prefix + first data field) 156 if len(fields) > 1: 157 sub = _strip_checksum(fields[1]).strip().upper() 158 subkey = (prefix, sub) 159 if subkey == ('PSXN', '26'): 160 return self._extract_psxn26(fields) 161 if subkey in _SUB_TIME: 162 return self._extract_subtype(subkey, fields) 163 164 # 3. Long-name proprietary (>5 chars) 165 if prefix in _LONG_TIME: 166 return self._extract_long(prefix, fields) 167 168 return None 169 170 # -- Extraction helpers ------------------------------------------------ 171 172 def _extract_standard(self, stype, fields): 173 """Extract datetime from a standard NMEA sentence.""" 174 time_idx = _STD_TIME[stype] 175 date = None 176 if stype == 'ZDA': 177 day_i, month_i, year_i = _ZDA_DATE_FIELDS 178 date = self._parse_zda_date( 179 self._get_field(fields, day_i), 180 self._get_field(fields, month_i), 181 self._get_field(fields, year_i)) 182 elif stype in _STD_DATE: 183 date = self._parse_ddmmyy( 184 self._get_field(fields, _STD_DATE[stype])) 185 return self._datetime_from_time_field( 186 self._get_field(fields, time_idx), date=date) 187 188 def _extract_prop5(self, prefix, fields): 189 """Extract datetime from a 5-char proprietary sentence.""" 190 time_idx = _PROP5_TIME[prefix] 191 date = None 192 if prefix in _PROP5_DATE: 193 date = self._parse_ddmmyy( 194 self._get_field(fields, _PROP5_DATE[prefix])) 195 return self._datetime_from_time_field( 196 self._get_field(fields, time_idx), date=date) 197 198 def _extract_subtype(self, subkey, fields): 199 """Extract datetime from a sub-typed proprietary sentence.""" 200 time_idx = _SUB_TIME[subkey] 201 date = None 202 if subkey in _SUB_DATE: 203 date = self._parse_ddmmyy( 204 self._get_field(fields, _SUB_DATE[subkey])) 205 return self._datetime_from_time_field( 206 self._get_field(fields, time_idx), date=date) 207 208 def _extract_long(self, prefix, fields): 209 """Extract datetime from a long-name proprietary sentence.""" 210 time_idx = _LONG_TIME[prefix] 211 return self._datetime_from_time_field( 212 self._get_field(fields, time_idx)) 213 214 def _extract_psxn26(self, fields): 215 """Extract datetime from PSXN,26 (separate Y/M/D/H/M/S fields).""" 216 try: 217 year = int(self._get_field(fields, 2)) 218 month = int(self._get_field(fields, 3)) 219 day = int(self._get_field(fields, 4)) 220 hour = int(self._get_field(fields, 5)) 221 minute = int(self._get_field(fields, 6)) 222 sec_str = self._get_field(fields, 7) 223 second_frac = float(sec_str) 224 second = int(second_frac) 225 microsecond = int((second_frac - second) * 1_000_000) 226 date = datetime(year, month, day, tzinfo=timezone.utc).date() 227 self.last_nmea_date = date 228 return datetime( 229 year=year, month=month, day=day, 230 hour=hour, minute=minute, second=second, 231 microsecond=microsecond, tzinfo=timezone.utc) 232 except (ValueError, IndexError): 233 return None 234 235 # -- Field access and parsing ------------------------------------------ 236 237 @staticmethod 238 def _get_field(fields, idx): 239 """Safely get a comma-split field, stripping any checksum suffix.""" 240 if idx < len(fields): 241 return _strip_checksum(fields[idx].strip()) 242 return '' 243 244 def _datetime_from_time_field(self, time_field, date=None): 245 """Parse an NMEA time field (hhmmss.ss) and combine with a date. 246 If *date* is provided, also update ``self.last_nmea_date``. 247 """ 248 m = NMEA_TIME_RE.match(time_field) 249 if not m: 250 return None 251 252 hour = int(m.group(1)) 253 minute = int(m.group(2)) 254 second_frac = float(m.group(3)) 255 second = int(second_frac) 256 microsecond = int((second_frac - second) * 1_000_000) 257 258 if date is not None: 259 self.last_nmea_date = date 260 use_date = self.last_nmea_date 261 if use_date is None: 262 use_date = datetime.now(timezone.utc).date() 263 264 return datetime( 265 year=use_date.year, month=use_date.month, day=use_date.day, 266 hour=hour, minute=minute, second=second, 267 microsecond=microsecond, tzinfo=timezone.utc, 268 ) 269 270 @staticmethod 271 def _parse_ddmmyy(date_field): 272 """Parse a ``ddmmyy`` date string (used by RMC, PGRMF, etc.).""" 273 if not date_field or len(date_field) < 6: 274 return None 275 try: 276 day = int(date_field[0:2]) 277 month = int(date_field[2:4]) 278 year = int(date_field[4:6]) 279 year += 2000 if year < 80 else 1900 280 return datetime(year, month, day, tzinfo=timezone.utc).date() 281 except (ValueError, IndexError): 282 return None 283 284 @staticmethod 285 def _parse_zda_date(day_field, month_field, year_field): 286 """Parse day, month, year fields from a ZDA sentence.""" 287 try: 288 day = int(day_field) 289 month = int(month_field) 290 year = int(year_field) 291 return datetime(year, month, day, tzinfo=timezone.utc).date() 292 except (ValueError, IndexError): 293 return None
Parse NMEA sentences to extract embedded timestamps.
NMEATimestampExtractor(timeout=1, quiet=False, time_drift_threshold=0.1)
82 def __init__(self, timeout=1, quiet=False, time_drift_threshold=0.1): 83 self.timeout = timeout 84 self.quiet = quiet 85 self.time_drift_threshold = time_drift_threshold 86 self.last_nmea_date = None # datetime.date from RMC/ZDA/etc. 87 self.last_nmea_datetime = None # full datetime object 88 self.last_nmea_system_time = 0 # time.time() of last extraction 89 self.seen_timestamp = False
def
get_timestamp( self, record, time_format='%Y-%m-%dT%H:%M:%S.%fZ', time_zone=datetime.timezone.utc):
91 def get_timestamp(self, record, time_format=ts_util.TIME_FORMAT, 92 time_zone=ts_util.timezone.utc): 93 """Return a formatted timestamp string extracted from *record*, 94 or fall back to the last observed NMEA timestamp. Returns None 95 when no NMEA timestamp is available (caller should use system time). 96 """ 97 if not isinstance(record, str): 98 return None 99 100 nmea_dt = self._parse_nmea_datetime(record) 101 if nmea_dt is not None: 102 now = time.time() 103 self.last_nmea_datetime = nmea_dt 104 self.last_nmea_system_time = now 105 self.seen_timestamp = True 106 if not self.quiet and self.time_drift_threshold is not None: 107 nmea_epoch = nmea_dt.timestamp() 108 drift = abs(nmea_epoch - now) 109 if drift > self.time_drift_threshold: 110 logging.warning( 111 'NMEATimestampExtractor: NMEA time differs from ' 112 'system time by %.3fs (threshold=%.3fs)', 113 drift, self.time_drift_threshold) 114 return nmea_dt.strftime(time_format) 115 116 # Fall back to last observed timestamp 117 if not self.seen_timestamp: 118 return None 119 120 elapsed = time.time() - self.last_nmea_system_time 121 if elapsed > self.timeout: 122 if not self.quiet: 123 logging.warning( 124 'NMEATimestampExtractor: last NMEA timestamp is %.1fs ' 125 'old (timeout=%s); falling back to system time', 126 elapsed, self.timeout) 127 return None 128 129 return self.last_nmea_datetime.strftime(time_format)
Return a formatted timestamp string extracted from record, or fall back to the last observed NMEA timestamp. Returns None when no NMEA timestamp is available (caller should use system time).