openrvdas.logger.utils.regex_parser

  1#!/usr/bin/env python3
  2
  3"""Tools for parsing NMEA and other text records using regex.
  4"""
  5import datetime
  6import logging
  7import re
  8import pprint
  9import time
 10
 11
 12
 13# Append openrvdas root to syspath prior to importing openrvdas modules
 14
 15from logger.utils.das_record import DASRecord  # noqa: E402
 16from logger.utils.read_config import load_definitions  # noqa: E402
 17from logger.utils.das_record import collect_metadata_for_fields  # noqa: E402
 18
 19# Import ConvertFieldsTransform, but handle gracefully if unavailable
 20try:
 21    from logger.transforms.convert_fields_transform import ConvertFieldsTransform
 22except ImportError:
 23    ConvertFieldsTransform = None
 24
 25# Note: this is a "permissive" regex. It looks for data_id and timestamp prior to field_string,
 26# But still parses field_string if they are absent
 27# Works for both "data_id timestamp fields" but also parses "fields" if
 28# data_id or timestamp are missing
 29
 30DEFAULT_RECORD_FORMAT = r"^(?:(?P<data_id>\w+)\s+(?P<timestamp>[0-9TZ:\-\.]*)\s+)?(?P<field_string>(.|\r|\n)*)"  # noqa: E501
 31
 32DEFAULT_DEFINITION_PATH = 'local/devices/*.yaml,contrib/devices/*.yaml'
 33
 34
 35################################################################################
 36class RegexParser:
 37    ############################
 38    def __init__(self,
 39                 record_format=None,
 40                 field_patterns=None,
 41                 data_id=None,
 42                 definition_path=None,
 43                 metadata=None,
 44                 metadata_interval=None,
 45                 quiet=False):
 46        r"""Create a parser that will parse field values out of a text record
 47        and return a DASRecord object.
 48        ```
 49        record_format - string for re.match() to use to break out data_id
 50            and timestamp from the rest of the message. By default this will
 51            look for 'data_id timestamp field_string', where 'field_string'
 52            is a str containing the fields to be parsed.
 53
 54        field_patterns
 55            If not None, either
 56            - a list of regex patterns to be tried
 57            - a dict of message_type:regex patterns to be tried. When one
 58              matches, the record's message_type is set accordingly.
 59            If None and definition_path is provided, patterns are loaded from
 60            device definition files.
 61
 62        data_id
 63            If specified, this string is used as the data_id for all records,
 64            overriding any data_id extracted from the source record.
 65
 66        definition_path
 67            Wildcarded path matching YAML definitions for devices. Used only
 68            if 'field_patterns' is None. Defaults to DEFAULT_DEFINITION_PATH.
 69            Comma-separated globs are supported.
 70
 71        metadata
 72            If provided, a dict mapping field names to their metadata dicts.
 73            If None and definition_path is used, metadata is compiled from
 74            device definitions.
 75
 76        metadata_interval
 77            If not None, include the description, units and other metadata
 78            pertaining to each field in the returned record if those data
 79            haven't been returned in the last metadata_interval seconds.
 80
 81        quiet - if not False, don't complain when unable to parse a record.
 82        ```
 83        """
 84        self.quiet = quiet
 85        self.record_format = record_format or DEFAULT_RECORD_FORMAT
 86        self.compiled_record_format = re.compile(self.record_format)
 87        self.data_id = data_id  # Store the data_id override
 88
 89        # Check for conflict
 90        if field_patterns and definition_path:
 91            raise ValueError('RegexParser: Both field_patterns and definition_path '
 92                             'specified. Please specify only one.')
 93
 94        # Device-aware parsing state
 95        self.devices = {}
 96        self.device_types = {}
 97        self.type_converters = {}
 98
 99        # Metadata state
100        self.metadata = metadata or {}
101        self.metadata_interval = metadata_interval
102        self.metadata_last_sent = {}
103
104        # If field patterns not provided, look them up in definitions
105        if field_patterns is None and definition_path is not None:
106            field_patterns = self._load_definitions(definition_path)
107
108            # If metadata not explicitly provided, compile it from definitions
109            if not metadata and metadata_interval:
110                self._compile_metadata()
111
112        self.field_patterns = field_patterns
113
114        # If we've been explicitly given the field_patterns we're to use for
115        # parsing, compile them now.
116        if field_patterns:
117            if isinstance(field_patterns, list):
118                self.compiled_field_patterns = [
119                    re.compile(pattern)
120                    for pattern in field_patterns
121                ]
122            elif isinstance(field_patterns, dict):
123                self.compiled_field_patterns = {
124                    message_type: re.compile(pattern)
125                    for (message_type, pattern) in field_patterns.items()
126                }
127            else:
128                raise ValueError('field_patterns must either be a list of patterns or '
129                                 'dict of message_type:pattern pairs. Found type '
130                                 f'{type(field_patterns)}')
131        else:
132            self.compiled_field_patterns = None
133
134    ############################
135    def _load_definitions(self, definition_path):
136        """Load device definitions and return aggregated field patterns.
137        Populates self.devices and self.device_types.
138        """
139        field_patterns = {}
140
141        # Use shared utility to load definitions
142        definitions = load_definitions(definition_path)
143
144        # Store devices
145        self.devices = definitions.get('devices', {})
146
147        # Process device_types: store them, extract patterns, create converters
148        for dt_name, dt_def in definitions.get('device_types', {}).items():
149            self.device_types[dt_name] = dt_def
150
151            # Aggregate formats (regexes)
152            dt_formats = dt_def.get('format', {})
153            if isinstance(dt_formats, dict):
154                field_patterns.update(dt_formats)
155
156            # Create cached component converter for this type
157            dt_fields = dt_def.get('fields', {})
158            if dt_fields and ConvertFieldsTransform:
159                self.type_converters[dt_name] = ConvertFieldsTransform(
160                    fields=dt_fields,
161                    quiet=self.quiet
162                )
163
164        return field_patterns
165
166    ############################
167    def _compile_metadata(self):
168        """
169        Compile metadata from device definitions if available.
170        Logic adapted from RecordParser.
171        """
172        # It's a map from variable name to the device and device type it
173        # came from, along with device type variable and its units and
174        # description, if provided in the device type definition.
175        for device, device_def in self.devices.items():
176            device_type_name = device_def.get('device_type')
177            if not device_type_name:
178                continue
179
180            device_type_def = self.device_types.get(device_type_name)
181            if not device_type_def:
182                continue
183
184            device_type_fields = device_type_def.get('fields')
185            if not device_type_fields:
186                continue
187
188            fields = device_def.get('fields')
189            if not fields:
190                continue
191
192            # e.g. device_type_field = GPSTime, device_field = S330GPSTime
193            for device_type_field, device_field in fields.items():
194                # e.g. GPSTime: {'units':..., 'description':...}
195                field_desc = device_type_fields.get(device_type_field)
196
197                # field_desc might be a string (type) or dict (metadata) or None
198                if not field_desc or not isinstance(field_desc, dict):
199                    continue
200
201                self.metadata[device_field] = {
202                    'device': device,
203                    'device_type': device_type_name,
204                    'device_type_field': device_type_field,
205                }
206                # Copy relevant keys like units, description
207                for k in ['units', 'description']:
208                    if k in field_desc:
209                        self.metadata[device_field][k] = field_desc[k]
210
211    ############################
212    def parse_record(self, record):
213        """Parse an id-prefixed text record into a DASRecord.
214        """
215        if not record:
216            return None
217        if not isinstance(record, str):
218            logging.info('Record is not a string: "%s"', record)
219            return None
220        try:
221            parsed_record = self.compiled_record_format.match(record).groupdict()
222        except (ValueError, AttributeError):
223            if not self.quiet:
224                logging.warning('Unable to parse record into "%s"', self.record_format)
225                logging.warning('Record: %s', record)
226            return None
227
228        if parsed_record is None:
229            return None
230
231        # Logic to determine data_id:
232        # 1. If self.data_id is set (in __init__), use it (Override).
233        # 2. Else, look for 'data_id' extracted from the record via regex.
234        # 3. If that fails, default to 'unknown'.
235        if self.data_id:
236            data_id = self.data_id
237        else:
238            data_id = parsed_record.get('data_id', None)
239            if not data_id:
240                if not self.quiet:
241                    logging.warning('No data_id found in record and none specified. '
242                                    'Defaulting to "unknown".')
243                data_id = 'unknown'
244
245        # Convert timestamp to numeric, if it's there.
246        # Initialize to None first to avoid UnboundLocalError if 'timestamp'
247        # is not in the regex groups.
248        timestamp = None
249        timestamp_text = parsed_record.get('timestamp', None)
250
251        if timestamp_text is not None:
252            timestamp = self.convert_timestamp(timestamp_text)
253
254        # If no timestamp found, DASRecord will default to time.time()
255        # if passed None.
256        if timestamp is None:
257            timestamp = time.time()
258
259        # Extract the field string we're going to parse;
260        # remove trailing whitespace.
261        field_string = parsed_record.get('field_string', None)
262        if field_string is not None:
263            field_string = field_string.rstrip()
264
265        message_type = None
266        fields = {}
267        if field_string:
268            # If we've been given a set of field_patterns to apply,
269            # use the first that matches.
270            # Shortcut that lets us iterate through a list or a dict with the same
271            # invocation. With a list, it returns (None, value); with a dict it
272            # returns (key, value).
273            def iterate_patterns(obj):
274                return (obj.items() if isinstance(obj, dict) else ((None, v) for v in obj))
275
276            if self.field_patterns:
277                for message_type, pattern in iterate_patterns(self.compiled_field_patterns):
278                    try:
279                        try_parse = pattern.match(field_string)
280                        # Did we find a parse that matched?
281                        # If so, return its fields
282                        if try_parse:
283                            fields = try_parse.groupdict()
284                            break
285                    except Exception as e:
286                        logging.error(e)
287
288        logging.debug('Created parsed fields: %s', pprint.pformat(fields))
289
290        # Create the initial DASRecord
291        try:
292            das_record = DASRecord(data_id=data_id, timestamp=timestamp,
293                                   message_type=message_type,
294                                   fields=fields)
295        except KeyError:
296            return None
297
298        # Device-Specific Processing
299        # Try to match data_id to a known device
300        if data_id in self.devices:
301            device_def = self.devices[data_id]
302            device_type = device_def.get('device_type')
303
304            # A. Type Conversion (delegated to cached ConvertFieldsTransform)
305            if device_type in self.type_converters:
306                converter = self.type_converters[device_type]
307                das_record = converter.transform(das_record)
308                if not das_record:
309                    return None
310
311            # B. Field Renaming / Filtering
312            # Only retain fields that are in the device's 'fields' map
313            device_fields_map = device_def.get('fields', {})
314            if device_fields_map:
315                new_fields = {}
316                for original_name, mapped_name in device_fields_map.items():
317                    if original_name in das_record.fields:
318                        # Use the mapped name (value)
319                        new_fields[mapped_name] = das_record.fields[original_name]
320
321                das_record.fields = new_fields
322
323        # Metadata Injection
324        # If we have parsed fields, see if we also have metadata. Are we
325        # supposed to occasionally send it for our variables? Is it time
326        # to send it again?
327        metadata_to_inject = collect_metadata_for_fields(
328            das_record.fields,
329            das_record.timestamp or 0,
330            self.metadata,
331            self.metadata_interval,
332            self.metadata_last_sent
333        )
334        if metadata_to_inject:
335            if das_record.metadata is None:
336                das_record.metadata = {}
337            das_record.metadata['fields'] = metadata_to_inject['fields']
338
339        return das_record
340
341    ############################
342    def convert_timestamp(self, datetime_text):
343        """Validates a datetime string and converts to numeric.
344        """
345
346        DEFAULT_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
347
348        try:
349            datetime_ti = datetime.datetime.strptime(
350                datetime_text, DEFAULT_FORMAT)
351        except ValueError:
352            logging.debug("Incorrect datetime format.")
353            return None
354
355        if datetime_ti:
356            # Explicitly set UTC timezone because the format expects 'Z'
357            # .replace(tzinfo=...) ensures .timestamp() treats it as UTC
358            # regardless of the local system clock.
359            timestamp = datetime_ti.replace(tzinfo=datetime.timezone.utc).timestamp()
360            return timestamp