openrvdas.logger.transforms.modify_value_transform

  1#!/usr/bin/env python3
  2"""Modify values according to simple formula
  3"""
  4
  5import copy
  6import logging
  7import time
  8from typing import Union
  9
 10from logger.utils.das_record import DASRecord  # noqa: E402
 11from logger.transforms.transform import Transform  # noqa: E402
 12
 13
 14################################################################################
 15class ModifyValueTransform(Transform):
 16    """Modify the value of specified fields according to simple formulae.
 17    """
 18    def __init__(self, fields, data_id=None, delete_unmatched=False, quiet=False,
 19                 metadata_interval=None, **kwargs):
 20        """
 21        fields
 22           A dict of fields to match. Key of each is the field to match. Values are
 23           a dict of what to do with the field value. E.g.:
 24
 25            fields:
 26              FieldName:
 27                mult_factor:  1.5   # default 1.0
 28                add_factor: 3.44  # default 0.0
 29                output_name: CorrectedFieldName  # default is FieldName
 30                metadata: Field name with linear foobar correction applied
 31                delete_original: true  # default false
 32              FieldName2:
 33                mapping_function: my_magic_function  # God knows how we'd implement this, but...
 34                output_name: CorrectedFieldName2
 35              ....
 36
 37            Currently implemented manipulations for a field are:
 38              mult_factor (default=1), add_factor (default=0)
 39                result = mult_factor * value + add_factor
 40
 41              output_name (default is original field name)
 42                Add the result to the record as a new field
 43
 44              delete_original (default=False)
 45                If true, and if output_name is specified, delete original field from record
 46
 47              metadata (default=None)
 48                If specified, any metadata associated with the new value
 49
 50        data_id (default None)
 51          If not None, the data_id to substitute into the record. Otherwise use
 52          data_id found in the original record, if available.
 53
 54        delete_unmatched (default=False)
 55          If true, delete any unmatched fields from the record
 56
 57        quiet (default=False)
 58          If True, don't warn if can't convert, or if overwriting existing values
 59
 60        metadata_interval (default=None)
 61          If not None, how frequently, in seconds to attach field metadata to records
 62          (NOTE: need to address which fields' metadata is sent along - all specified,
 63          or only fields that have appeared since last send, or...?)
 64        """
 65        super().__init__(**kwargs)  # processes 'quiet' and type hints
 66
 67        self.fields = fields
 68        self.data_id = data_id
 69        self.delete_unmatched = delete_unmatched
 70        self.metadata_interval = metadata_interval or 0
 71
 72        self._validate_fields()
 73        self.last_metadata_send = 0
 74        self.metadata = {
 75            field: spec.get('metadata')
 76            for field, spec in fields.items() if spec.get('metadata')
 77        }
 78
 79    ############################
 80    def _validate_fields(self):
 81        """Validate the configuration of fields."""
 82        allowed_keys = {'mult_factor', 'add_factor', 'output_name', 'metadata', 'delete_original'}
 83
 84        for field, spec in self.fields.items():
 85            extraneous_keys = set(spec) - allowed_keys
 86            if extraneous_keys:
 87                raise ValueError(f'Unexpected keys in field "{field}": {extraneous_keys}. '
 88                                 f'Allowed keys are {allowed_keys}')
 89
 90            if not isinstance(spec.get('mult_factor', 1), (int, float)):
 91                raise ValueError(f'Invalid "mult_factor" for field "{field}". Must be numeric.')
 92            if not isinstance(spec.get('add_factor', 0), (int, float)):
 93                raise ValueError(f'Invalid "add_factor" for field "{field}". Must be numeric.')
 94
 95    ############################
 96    def transform(self, record: Union[DASRecord, dict]):
 97        """
 98        Transform a record or list of records.
 99
100        Args:
101            record (DASRecord or list): The input record(s).
102
103        Returns:
104            Transformed record(s) or None if the input is invalid.
105        """
106        # See if it's something we can process, and if not, try digesting
107        if not self.can_process_record(record):  # inherited from BaseModule()
108            return self.digest_record(record)  # inherited from BaseModule()
109
110        # If we've got a dict, convert it to a DASRecord for uniform handling
111        if isinstance(record, dict):
112            record = DASRecord(record, data_id=self.data_id)
113
114        # Make a copy of the original record we're going to munge
115        result = copy.deepcopy(record)
116
117        # More efficient, but doesn't allow use to do delete_unmatched:
118        # for field in record.fields.keys() & self.fields.keys():
119        for field in record.fields:
120            field_spec = self.fields.get(field)
121
122            # If there isn't a rule for this field
123            if field_spec is None:
124                if self.delete_unmatched:
125                    del result[field]
126                continue
127
128            # Get the field value, make sure we can work with it
129            value = record.get(field)
130            try:
131                value = float(value)
132            except ValueError:
133                if not self.quiet:
134                    logging.warning(f'ModifyValueTransform could not convert field {field} value '
135                                    f'"{value}" to float for modification. Type: {type(value)}')
136                continue
137
138            # Do the actual computation
139            value *= field_spec.get('mult_factor', 1.0)
140            value += field_spec.get('add_factor', 0.0)
141
142            # Where are we going to write the value? Check if it already exists. If there is no
143            # target field name, use original field name.
144            target_field = field_spec.get('output_name')
145            if target_field and target_field in record.fields and not self.quiet:
146                logging.warning(f'ModifyValueTransform overwriting existing field: {target_field}')
147            if not target_field:
148                target_field = field
149
150            # Are we getting rid of the original field? If so, do that now, to avoid
151            # the semantic question of deleting original value when we're writing back
152            # to original field
153            if field_spec.get('delete_original'):
154                del result[field]
155
156            result[target_field] = value
157
158        # We're now done going through the record fields. If it's time to send metadata,
159        # add it to any existing metadata for record, overwriting existing fields.
160        if self._should_attach_metadata(record):
161            result.metadata.update(self.metadata)
162
163        return result
164
165    ############################
166    def _should_attach_metadata(self, record):
167        """Determine if metadata should be attached to the record."""
168        now = record.timestamp or time.time()
169        if self.metadata_interval and now > self.last_metadata_send + self.metadata_interval:
170            self.last_metadata_send = now
171            return True
172        return False