openrvdas.logger.utils.record_parser

  1#!/usr/bin/env python3
  2
  3"""Tools for parsing NMEA and other text records.
  4
  5By default, will load device and device_type definitions from files in
  6logger/devices/*.yaml and contrib/devices/*.yaml. Please see documentation
  7in contrib/devices/README.md for a description of the format these
  8definitions should take.
  9"""
 10import datetime
 11import json
 12import logging
 13import pprint
 14
 15try:
 16    import parse
 17    PARSE_INSTALLED = True
 18except ImportError:
 19    PARSE_INSTALLED = False
 20
 21# Append openrvdas root to syspath prior to importing openrvdas modules
 22from logger.utils.das_record import DASRecord  # noqa: E402
 23from logger.utils.read_config import load_definitions  # noqa: E402
 24from logger.utils.das_record import collect_metadata_for_fields  # noqa: E402
 25
 26# Dict of format types that extend the default formats recognized by the
 27# parse module.
 28from logger.utils.record_parser_formats import extra_format_types  # noqa: E402
 29
 30DEFAULT_DEFINITION_PATH = 'logger/devices/*.yaml,contrib/devices/*.yaml'
 31DEFAULT_RECORD_FORMAT = '{data_id:w} {timestamp:ti} {field_string}'
 32
 33
 34class RecordParser:
 35    ############################
 36    def __init__(self, record_format=None,
 37                 field_patterns=None, metadata=None,
 38                 definition_path=DEFAULT_DEFINITION_PATH,
 39                 return_das_record=False, return_json=False,
 40                 metadata_interval=None, strip_unprintable=False,
 41                 quiet=False, prepend_data_id=False, delimiter=':'):
 42        """Create a parser that will parse field values out of a text record
 43        and return either a Python dict of data_id, timestamp and fields,
 44        a JSON encoding of that dict, or a binary DASRecord.
 45        ```
 46        record_format - string for parse.parse() to use to break out data_id
 47            and timestamp from the rest of the message. By default this will
 48            look for 'data_id timestamp field_string', where 'field_string'
 49            is a str containing the fields to be parsed.
 50
 51        field_patterns
 52            If not None, a list of parse patterns to be tried instead
 53            of looking for device definitions along the definition path,
 54            or a dict of message_type:[parse_pattern, parse_pattern].
 55
 56        metadata
 57            If field_patterns is not None, the metadata to send along with
 58            data records.
 59
 60        definition_path - a comma-separated set of file globs in which to look
 61            for device and device_type definitions with which to parse message.
 62
 63        return_json - return the parsed fields as a JSON encoded dict
 64
 65        return_das_record - return the parsed fields as a DASRecord object
 66
 67        metadata_interval - if not None, include the description, units
 68            and other metadata pertaining to each field in the returned
 69            record if those data haven't been returned in the last
 70            metadata_interval seconds.
 71
 72        strip_unprintable
 73                Strip out and ignore any leading or trailing non-printable binary
 74                characters in the string to be parsed.
 75
 76        quiet - if not False, don't complain when unable to parse a record.
 77
 78        prepend_data_id - If true prepend the instrument data_id to field_names
 79            in the record.
 80
 81        delimiter
 82            The string to insert between data_id and field_name when prepend_data_id is true.
 83            Defaults to ':'.
 84            Not used if prepend_data_id is false.
 85        ```
 86        """
 87        if not PARSE_INSTALLED:
 88            raise ImportError('RecordParser requires Python "parse" module; '
 89                              'please run "pip install parse"')
 90
 91        self.strip_unprintable = strip_unprintable
 92        self.quiet = quiet
 93        self.field_patterns = field_patterns
 94        self.metadata = metadata or {}
 95        self.record_format = record_format or DEFAULT_RECORD_FORMAT
 96        self.compiled_record_format = parse.compile(format=self.record_format,
 97                                                    extra_types=extra_format_types)
 98        self.return_das_record = return_das_record
 99        self.return_json = return_json
100        if return_das_record and return_json:
101            raise ValueError('Only one of return_json and return_das_record '
102                             'may be true.')
103
104        self.metadata_interval = metadata_interval
105        self.metadata_last_sent = {}
106        self.prepend_data_id = prepend_data_id
107        self.delimiter = delimiter
108
109        # If we've been explicitly given the field_patterns we're to use for
110        # parsing, compile them now. Patterns may either be a list of strings,
111        # a dict of strings or a dict of lists of strings.
112        if field_patterns:
113            self.compiled_field_patterns = self._compile_formats_from_patterns(field_patterns)
114            self.metadata = metadata
115
116        # If we've not been given field_patterns to use for parsing, read in all
117        # the devices and device types to compile them.
118        else:
119            # Fill in the devices and device_types - NOTE: we won't be using
120            # these if 'field_patterns' is provided as an argument.
121            definitions = load_definitions(definition_path)
122            self.devices = definitions.get('devices', {})
123            self.device_types = definitions.get('device_types', {})
124
125            # Some limited error checking: make sure that all devices have a
126            # defined device_type.
127            for device, device_def in self.devices.items():
128                device_type = device_def.get('device_type')
129                if not device_type:
130                    raise ValueError('Device definition for "%s" has no declaration of '
131                                     'its device_type.' % device)
132                if device_type not in self.device_types:
133                    raise ValueError('Device type "%s" (declared in definition of "%s") '
134                                     'is undefined.' % (device_type, device))
135
136            # Compile format definitions so that we can run them more
137            # quickly. If format is a single string, normalize it into a list
138            # to simplify later code.
139            for device_type, device_type_def in self.device_types.items():
140                format = device_type_def.get('format')
141                if format is None:
142                    raise ValueError('Device type %s has no format definition'
143                                     % device_type)
144                compiled_format = self._compile_formats_from_patterns(format)
145                self.device_types[device_type]['compiled_format'] = compiled_format
146
147            # Metadata: If we haven't been handed a dict of metadata, compile it from
148            # the devices we've read.
149            #
150            # It's a map from variable name to the device and device type it
151            # came from, along with device type variable and its units and
152            # description, if provided in the device type
153            # definition. Compiling this information is kind of excruciating
154            # and voluminous.
155            if not metadata and metadata_interval is not None:
156                for device, device_def in self.devices.items():  # e.g. s330
157                    device_type_name = device_def.get('device_type')  # Seapath330
158                    if not device_type_name:
159                        raise ValueError('Device definition for "%s" has no declaration of '
160                                         'its device_type.' % device)
161                    device_type_def = self.device_types.get(device_type_name)
162                    if not device_type_def:
163                        raise ValueError('Device type "%s" (declared in definition of "%s")'
164                                         ' is undefined.' % (device_type_name, device))
165                    device_type_fields = device_type_def.get('fields')
166                    if not device_type_fields:
167                        raise ValueError('Device type "%s" has no fields?'
168                                         % device_type_name)
169
170                    fields = device_def.get('fields')
171                    if not fields:
172                        raise ValueError('Device "%s" has no fields?!?' % device)
173
174                    # e.g. device_type_field = GPSTime, device_field = S330GPSTime
175                    for device_type_field, device_field in fields.items():
176                        # e.g. GPSTime: {'units':..., 'description':...}
177                        field_desc = device_type_fields.get(device_type_field)
178                        if not field_desc:
179                            logging.warning('Device type "%s" has no field corresponding to '
180                                            'device field "%s"' % (device_type_name,
181                                                                   device_type_field))
182                            continue
183                        self.metadata[device_field] = {
184                            'device': device,
185                            'device_type': device_type_name,
186                            'device_type_field': device_type_field,
187                        }
188                        self.metadata[device_field].update(field_desc)
189
190    ############################
191    def parse_record(self, record):
192        """Parse an id-prefixed text record into a Python dict of data_id,
193        timestamp and fields.
194        """
195        if not record:
196            return None
197        if not isinstance(record, str):
198            if not self.quiet:
199                logging.info('Record is not a string: "%s"', record)
200            return None
201        try:
202            # Break record into (by default) data_id, timestamp and field_string
203            parsed_record = self.compiled_record_format.parse(record).named
204        except (ValueError, AttributeError):
205            if not self.quiet:
206                logging.warning('Unable to parse record into "%s"', self.record_format)
207                logging.warning('Record: %s', record)
208            return None
209
210        data_id = parsed_record.get('data_id', 'no_data_id')
211
212        # Convert timestamp to numeric, if it's there
213        timestamp = parsed_record.get('timestamp')
214        if timestamp is not None and isinstance(timestamp, datetime.datetime):
215            timestamp = timestamp.timestamp()
216            parsed_record['timestamp'] = timestamp
217
218        # Extract the field string we're going to parse; remove trailing
219        # whitespace.
220        field_string = parsed_record.get('field_string')
221
222        # If we don't have fields, there's nothing to parse
223        if field_string is None:
224            if not self.quiet:
225                logging.warning('No field_string found in record "%s"', record)
226            return None
227
228        if self.strip_unprintable:
229            field_string = ''.join([c for c in field_string if c.isprintable()])
230        field_string = field_string.strip()
231        if not field_string:
232            if not self.quiet:
233                logging.warning('No field_string found in record "%s"', record)
234            return None
235
236        # If we've been given a set of field_patterns: If they're a dict, see if there's
237        # a key that matches our data_id. If so, only look at those patterns. Otherwise
238        # try them all and use the first that matches.
239        if self.field_patterns:
240            # If field_patterns is a dict, see if our data_id matches any of the keys
241            if isinstance(self.field_patterns, dict):
242                patterns = self.compiled_field_patterns.get(data_id)
243            else:
244                patterns = self.compiled_field_patterns
245
246            # If no patterns to try, go home emptyhanded
247            if not patterns:
248                if not self.quiet:
249                    logging.warning(f'No parse field patterns matched data_id "{data_id}"')
250                return None
251            fields, message_type = self._parse_field_string(field_string, patterns)
252
253        # If we were given no explicit field_patterns to use, we need to
254        # count on the record having a data_id that lets us figure out
255        # which device, and therefore which field_patterns to try.
256        else:
257            fields, message_type = self.parse_for_data_id(data_id, field_string)
258
259        # We should now have a dictionary of fields. If not, go home
260        if fields is None:
261            if not self.quiet:
262                logging.warning('No formats matched field_string "%s"', field_string)
263            return None
264
265        # Some folks want the data_id prepended
266        if self.prepend_data_id:
267            # This conditional dictates whether fields are stored with just the
268            # field_name key, or <data_id><delimiter><field_name>
269            # Doing some work directly on the fields dict, so we'll take a copy
270            # to loop over.
271            fields_copy = fields.copy()
272            # Reset the fields dict
273            fields = {}
274            for field in fields_copy:
275                # Determine the new "field_name"
276                key = '' + data_id + self.delimiter + field
277                # Set the value
278                fields[key] = fields_copy[field]
279
280        # Remove raw 'field_string' and add parsed 'fields' to parsed_record
281        del parsed_record['field_string']
282        parsed_record['fields'] = fields
283        if message_type:
284            parsed_record['message_type'] = message_type
285
286        # Metadata Injection - use shared utility
287        metadata = collect_metadata_for_fields(
288            fields, timestamp, self.metadata,
289            self.metadata_interval, self.metadata_last_sent
290        )
291        if metadata:
292            parsed_record['metadata'] = metadata
293
294        logging.debug('Created parsed record: %s', pprint.pformat(parsed_record))
295
296        # What are we going to do with the result we've created?
297        if self.return_das_record:
298            try:
299                return DASRecord(data_id=data_id, timestamp=timestamp,
300                                 message_type=message_type, fields=fields,
301                                 metadata=metadata)
302            except KeyError:
303                return None
304
305        elif self.return_json:
306            return json.dumps(parsed_record)
307        else:
308            return parsed_record
309
310    ############################
311    def _parse_field_string(self, field_string, compiled_field_patterns):
312        # Default if we don't match anything
313        fields = message_type = None
314
315        # If our pattern(s) are just a single compiled parser, try parsing and
316        # return with no message type.
317        if isinstance(compiled_field_patterns, parse.Parser):
318            result = compiled_field_patterns.parse(field_string)
319            if result:
320                fields = result.named
321
322        # Else, if it's a list, try it out on all the elements.
323        elif isinstance(compiled_field_patterns, list):
324            for pattern in compiled_field_patterns:
325                fields, message_type = self._parse_field_string(field_string, pattern)
326                if fields is not None:
327                    break
328
329        # If it's a dict, try out on all values, using the key as message type.
330        # It's syntactically possible for the internal set of patterns to have
331        # their own message types. Not sure why someone would ever create patterns
332        # that did this, but if they do, let that override our base one.
333        elif isinstance(compiled_field_patterns, dict):
334            for message_type, pattern in compiled_field_patterns.items():
335                fields, int_message_type = self._parse_field_string(field_string, pattern)
336                message_type = int_message_type or message_type
337                if fields is not None:
338                    break
339        else:
340            raise ValueError('Unexpected pattern type in parser: %s'
341                             % type(compiled_field_patterns))
342
343        return fields, message_type
344
345    ############################
346    def parse_for_data_id(self, data_id, field_string):
347        """Look up the device and device type for a data_id. Parse the field_string
348        according to those formats. If successful, return a tuple of
349        (field_dict, message_type), where field_dict is a dict of
350        {field_name: field_value}. Return ({}, None) if unable to match a format pattern.
351        """
352        failure_values = (None, None)
353        if not self.devices:
354            logging.warning('RecordParser has no device definitions; unable to parse!')
355            return failure_values
356
357        # Get device and device_type definitions for data_id
358        device = self.devices.get(data_id)
359        if not device:
360            if not self.quiet:
361                logging.warning('Unrecognized data id "%s", field string: %s',
362                                data_id, field_string)
363                logging.warning('Known data ids are: "%s"',
364                                ', '.join(self.devices.keys()))
365            return failure_values
366
367        device_type = device.get('device_type')
368        if not device_type:
369            if not self.quiet:
370                logging.error('Internal error: No "device_type" for device %s!', device)
371            return failure_values
372
373        device_definition = self.device_types.get(device_type)
374        if not device_definition:
375            if not self.quiet:
376                logging.error('No definition found for device_type "%s"', device_type)
377            return failure_values
378
379        compiled_format_patterns = device_definition.get('compiled_format')
380        parsed_fields, message_type = \
381            self._parse_field_string(field_string, compiled_format_patterns)
382
383        # Did we get anything?
384        if parsed_fields is None:
385            if not self.quiet:
386                logging.warning('No formats matched field_string "%s"', field_string)
387            return failure_values
388
389        logging.debug('Got fields: %s', pprint.pformat(parsed_fields))
390
391        # Finally, convert field values to variable names specific to device
392        device_fields = device.get('fields')
393        if not device_fields:
394            if not self.quiet:
395                logging.error('No "fields" definition found for device %s', data_id)
396            return failure_values
397
398        # Assign field values to the appropriate named variable.
399        fields = {}
400        for field_name, value in parsed_fields.items():
401            variable_name = device_fields.get(field_name)
402            # None means we're not supposed to report it.
403            if variable_name is None:
404                continue
405            # None means we didn't have a value for this field; omit it.
406            if value is None:
407                continue
408            # If it's a datetime, convert to numeric timestamp
409            if isinstance(value, datetime.datetime):
410                value = value.timestamp()
411            fields[variable_name] = value
412
413        logging.debug('Got fields: %s', pprint.pformat(fields))
414        return fields, message_type
415
416    ############################
417    def _compile_formats_from_patterns(self, field_patterns):
418        """Return a list/dict of patterns compiled from the
419        str/list/dict of passed field_patterns.
420        """
421        if isinstance(field_patterns, str):
422            return [parse.compile(format=field_patterns,
423                                  extra_types=extra_format_types)]
424        elif isinstance(field_patterns, list):
425            return [parse.compile(format=p, extra_types=extra_format_types)
426                    for p in field_patterns]
427        elif isinstance(field_patterns, dict):
428            compiled_field_patterns = {}
429            for message_type, message_pattern in field_patterns.items():
430                compiled_patterns = self._compile_formats_from_patterns(message_pattern)
431                compiled_field_patterns[message_type] = compiled_patterns
432            return compiled_field_patterns
433
434        else:
435            raise ValueError('Passed field_patterns must be str, list or dict. Found %s: %s'
436                             % (type(field_patterns), str(field_patterns)))