openrvdas.logger.utils.das_record

  1#!/usr/bin/env python3
  2
  3import json
  4import pprint
  5import logging
  6
  7from logger.utils.timestamp import timestamp as timestamp_method  # noqa: E402
  8
  9
 10class DASRecord:
 11    """DASRecord is a structured representation of the field names and
 12    values (and metadata) contained in a sensor record.
 13    """
 14    ############################
 15
 16    def __init__(self, json_str=None, data_id=None, message_type=None,
 17                 timestamp=0, fields=None, metadata=None):
 18        """
 19        If a json string is passed, it is parsed into a dictionary and its
 20        values for timestamp, fields and metadata are copied in. Otherwise,
 21        the DASRecord object is initialized with the passed-in values for
 22        instrument, timestamp, fields (a dictionary of fieldname-value pairs)
 23        and metadata.
 24
 25        If timestamp is not specified, the instance will use the current time.
 26        """
 27        if json_str:
 28            parsed = json.loads(json_str)
 29            self.data_id = parsed.get('data_id')
 30            self.message_type = parsed.get('message_type')
 31            self.timestamp = parsed.get('timestamp')
 32            self.fields = parsed.get('fields', {})
 33            self.metadata = parsed.get('metadata', {})
 34        else:
 35            # self.source =
 36            self.data_id = data_id
 37            self.message_type = message_type
 38            self.timestamp = timestamp or timestamp_method()
 39            if fields is None:
 40                self.fields = {}
 41            else:
 42                self.fields = fields
 43            if metadata is None:
 44                self.metadata = {}
 45            else:
 46                self.metadata = metadata
 47
 48    ############################
 49    def as_json(self, pretty=False):
 50        """Return DASRecord as a JSON string."""
 51        json_dict = {
 52            'data_id': self.data_id,
 53            'message_type': self.message_type,
 54            'timestamp': self.timestamp,
 55            'fields': self.fields,
 56            'metadata': self.metadata
 57        }
 58        if pretty:
 59            return json.dumps(json_dict, sort_keys=True, indent=4)
 60        else:
 61            return json.dumps(json_dict)
 62
 63    ############################
 64    def __str__(self):
 65        das_dict = {
 66            'data_id': self.data_id,
 67            'message_type': self.message_type,
 68            'timestamp': self.timestamp,
 69            'fields': self.fields,
 70            'metadata': self.metadata
 71        }
 72        return pprint.pformat(das_dict)
 73
 74    ############################
 75    def __eq__(self, other):
 76        return (other and
 77                self.data_id == other.data_id and
 78                self.message_type == other.message_type and
 79                self.timestamp == other.timestamp and
 80                self.fields == other.fields and
 81                self.metadata == other.metadata)
 82
 83    ############################
 84    def __setitem__(self, field, value):
 85        self.fields[field] = value
 86
 87    ############################
 88    def __getitem__(self, field):
 89        try:
 90            return self.fields[field]
 91        except KeyError:
 92            logging.error(f'No field "{field}" found in DASRecord {self}')
 93            raise
 94
 95    ############################
 96    def __delitem__(self, field):
 97        try:
 98            del self.fields[field]
 99        except KeyError:
100            logging.error(f'Attempt to delete non-existent field "{field}" in DASRecord: {self}')
101
102    ############################
103    def get(self, field, default=None):
104        return self.fields.get(field, default)
105
106
107def to_das_record_list(record, data_id=None):
108    """Utility function to normalize different types of records into a
109    list of DASRecords.
110
111    Take input in one of these three formats:
112       - DASRecord
113       - a single record dict with keys 'timestamp' and 'fields'
114       - a field dict of format
115         ``` {field_name: [(timestamp, value), (timestamp, value),...],
116              field_name: [(timestamp, value), (timestamp, value),...],
117             }
118         ```
119    and convert it into a list of zero or more DASRecords.
120    """
121    # What type of record is this?
122    if not record:
123        return []
124
125    # If it's a list, assume it's already a list of DASRecords
126    if isinstance(record, list):
127        return record
128
129    # If it's a single DASRecord, it's easy
130    if isinstance(record, DASRecord):
131        return [record]
132
133    # At this point, if it's not a dict, we don't know *what* it is
134    if not isinstance(record, dict):
135        logging.error('Unknown type of input passed to to_das_record_list: %s: %s',
136                      type(record), record)
137        return []
138
139    # If it's a single timestamp dict, it's easy
140    elif 'timestamp' in record and 'fields' in record:
141        return [DASRecord(data_id=data_id,
142                          timestamp=record['timestamp'],
143                          fields=record['fields'],
144                          metadata=record.get('metadata'))]
145
146    # If here, we believe we've received a field dict, in which each
147    # field may have multiple [timestamp, value] pairs. First thing we
148    # do is reformat the data into a map of
149    #        {timestamp: {field:value, field:value},...}}
150    try:
151        by_timestamp = {}
152        for field, ts_value_list in record.items():
153            if not isinstance(ts_value_list, list):
154                logging.warning('Expected field_name: [(timestamp, value),...] pairs, '
155                                'found %s: %s', field, ts_value_list)
156                continue
157            for (timestamp, value) in ts_value_list:
158                if timestamp not in by_timestamp:
159                    by_timestamp[timestamp] = {}
160                by_timestamp[timestamp][field] = value
161
162        # Now copy the entries into an ordered-by-timestamp list.
163        results = [DASRecord(data_id=data_id, timestamp=ts, fields=by_timestamp[ts])
164                   for ts in sorted(by_timestamp)]
165        return results
166    except ValueError:
167        logging.error('Badly-structured field dictionary: %s: %s',
168                      field, pprint.pformat(ts_value_list))
169        return []
170
171
172def collect_metadata_for_fields(field_names, timestamp, metadata,
173                                metadata_interval, metadata_last_sent):
174    """
175    Collect metadata for fields that are due to be sent based on the interval.
176
177    This function is shared between RecordParser and RegexParser to avoid
178    code duplication.
179
180    Args:
181        field_names: Iterable of field names to check for metadata.
182        timestamp: Current record timestamp (numeric).
183        metadata: Dict mapping field names to their metadata dicts.
184        metadata_interval: Minimum seconds between metadata sends per field.
185        metadata_last_sent: Dict tracking last send time per field (modified in place).
186
187    Returns:
188        Dict with 'fields' key containing metadata to inject, or None if no
189        metadata is due to be sent.
190    """
191    if not metadata or not metadata_interval:
192        return None
193
194    metadata_fields = {}
195    for field_name in field_names:
196        last_sent = metadata_last_sent.get(field_name, 0)
197        if timestamp - last_sent > metadata_interval:
198            field_metadata = metadata.get(field_name)
199            if field_metadata:
200                metadata_fields[field_name] = field_metadata
201                metadata_last_sent[field_name] = timestamp
202
203    return {'fields': metadata_fields} if metadata_fields else None