openrvdas.logger.transforms.delta_transform

  1import logging
  2from typing import Union
  3
  4from logger.utils.das_record import DASRecord  # noqa: E402
  5from logger.transforms.transform import Transform
  6
  7KNOWN_FIELD_TYPES = ['polar']
  8
  9
 10################################################################################
 11def polar_diff(last_value, value):
 12    return ((value - last_value) + 180) % 360 - 180
 13
 14
 15################################################################################
 16class DeltaTransform(Transform):
 17    def __init__(self, rate=False, field_type=None, **kwargs):
 18        """Return a DASRecord (or dict, depending on input record type) with
 19        each field's delta in value from it's previous value. If a field
 20        is absent, it will be omitted. The first time a field appears, it
 21        will be omitted (as there is no previous value to delta from). If
 22        no deltas are available, None will be returned.
 23
 24        rate — If True, return the rate of change (delta/second). If a
 25               list of field names, return rate of change for field names
 26               in the list, simple delta for all others.  fields, or just
 27               return the delta.
 28
 29        field_type — if not None, should be a dict mapping field names to
 30               special field types, if any. Currently, only 'polar' is
 31               implemented.
 32        """
 33        super().__init__(**kwargs)  # processes 'quiet' and type hints
 34
 35        if type(rate) not in [bool, list]:
 36            raise ValueError('"rate" argument in DeltaTransform must be either '
 37                             'a list or Boolean. Found type %s' % type(rate))
 38        if field_type:
 39            if not type(field_type) is dict:
 40                raise ValueError('"field_type" argument in DeltaTransform must be '
 41                                 ' either None or a dict. Found "%s"' %
 42                                 type(field_type))
 43            # Check that any specified field types are ones we know about.
 44            for field_name, this_field_type in field_type.items():
 45                if this_field_type not in KNOWN_FIELD_TYPES:
 46                    raise ValueError('Unknown field_type specified for field %s: %s. '
 47                                     'Known field types are: %s' %
 48                                     (field_name, this_field_type, KNOWN_FIELD_TYPES))
 49        self.rate = rate
 50        self.field_type = field_type
 51
 52        # Dict of {field_name: (previous_timestamp, previous_value)} pairs
 53        self.last_value_dict = {}
 54
 55    ############################
 56    def transform(self, record: Union[DASRecord, dict]) -> Union[DASRecord, dict]:
 57
 58        # See if it's something we can process, and if not, try digesting
 59        if not self.can_process_record(record):  # BaseModule
 60            return self.digest_record(record)  # BaseModule
 61
 62        if type(record) is DASRecord:
 63            fields = record.fields
 64            timestamp = record.timestamp
 65
 66        elif type(record) is dict:
 67            fields = record.get('fields')
 68            timestamp = record.get('timestamp')
 69        else:
 70            logging.info('Record passed to DeltaTransform was neither a dict nor a '
 71                         'DASRecord. Type was %s: %s' % (type(record), str(record)[:80]))
 72            return None
 73
 74        if fields is None:
 75            logging.info('Record passed to DeltaTransform does not have "fields": %s', record)
 76            return None
 77
 78        if timestamp is None:
 79            logging.info('Record passed to DeltaTransform does not have "timestamp": %s', record)
 80            return None
 81
 82        delta_values = {}
 83
 84        for key, value in fields.items():
 85            # If we don't have a previous value for this field, store the
 86            # current one and move on to the next field.
 87            if key not in self.last_value_dict:
 88                self.last_value_dict[key] = (timestamp, value)
 89                continue
 90
 91            last_timestamp, last_value = self.last_value_dict.get(key, (None, None))
 92
 93            # Does this field have a special type?
 94            if type(self.field_type) is dict:
 95                this_field_type = self.field_type.get(key)
 96            else:
 97                this_field_type = self.field_type
 98
 99            # What do we do with this field_type? 'None' is a simple diff
100            if this_field_type == 'polar':
101                delta_values[key] = polar_diff(last_value, value)
102            elif this_field_type is None:
103                delta_values[key] = value - last_value
104            else:
105                raise ValueError('DeltaTransform configured with unrecognized '
106                                 'field type for %s: "%s"', key, this_field_type)
107
108            # Are we doing rate or simple diff for this field?
109            if self.rate is True or type(self.rate) is list and key in self.rate:
110                time_diff = timestamp - last_timestamp
111                # If rate, make sure it's a valid time difference. Bail if it isn't.
112                if time_diff <= 0:
113                    logging.warning('Invalid difference in successive timestamps for '
114                                    'field %s:  %g -> %g', key, last_timestamp, timestamp)
115                    return None
116                delta_values[key] = delta_values[key] / time_diff
117
118            # Finally, save the current values for next time
119            self.last_value_dict[key] = (timestamp, value)
120
121        # If, at the end of it all, we don't have any fields, return None
122        if not delta_values:
123            return None
124
125        # If they gave us a dict, return a dict; if they gave us a
126        # DASRecord, return a DASRecord.
127        if type(record) is dict:
128            return {'timestamp': timestamp, 'fields': delta_values}
129
130        return DASRecord(timestamp=timestamp, fields=delta_values)