openrvdas.logger.transforms.subsample_transform

  1#!/usr/bin/env python3
  2"""Compute subsamples of input data.
  3"""
  4
  5import logging
  6import time
  7
  8from logger.utils.das_record import DASRecord  # noqa: E402
  9from logger.utils.subsample import subsample  # noqa: E402
 10from logger.transforms.derived_data_transform import DerivedDataTransform  # noqa: E402
 11
 12
 13################################################################################
 14class SubsampleTransform(DerivedDataTransform):
 15    """Transform that computes subsamples of the specified variables.
 16    """
 17
 18    def __init__(self, field_spec, back_seconds=60*60,
 19                 metadata_interval=None, **kwargs):
 20        """
 21        ```
 22        field_spec - a dict that contains the fields to be subsampled,
 23                     the algorithms and parameters to be used, and the
 24                     names of the output values to be produced.
 25
 26          e.g.  {
 27                  field_1:{
 28                    output: smoothed_field_1,
 29                    subsample:{
 30                      'type':'boxcar_average', 'window': 30, 'interval': 30
 31                    }
 32                  },
 33                  field_2:{
 34                    output: smoothed_field_2,
 35                    subsample:{
 36                      'type':'boxcar_average', 'window': 15, 'interval': 5
 37                    }
 38                  }
 39                }
 40
 41        back_seconds - the number of seconds of data to cache for use by sampler
 42
 43        metadata_interval - how many seconds between when we attach field metadata
 44                     to a record we send out.
 45        ```
 46        """
 47        super().__init__(**kwargs)  # processes 'quiet' and type hints
 48
 49        logging.warning('SubsampleTransform is deprecated in favor of InterpolationTransform')
 50
 51        self.field_spec = field_spec
 52        self.back_seconds = back_seconds
 53        self.field_list = list(field_spec.keys())
 54
 55        # A dict of the cached values we're hanging onto
 56        self.cached_values = {f: [] for f in self.field_list}
 57
 58        # Last timestamp that's been emitted for each field
 59        self.last_timestamp = {f: 0 for f in self.field_list}
 60
 61        self.metadata_interval = metadata_interval
 62        self.last_metadata_send = 0
 63
 64    ############################
 65    def fields(self):
 66        """Which fields are we interested in to produce transformed data?"""
 67        return self.field_list
 68
 69    ############################
 70    def _metadata(self):
 71        """Return a dict of metadata for our derived fields."""
 72        metadata_fields = {
 73            self.field_spec[field]['output']: {
 74                'description':
 75                'Subsampled values of %s via %s' %
 76                (field, self.field_spec[field].get('subsample',
 77                                                   'unspecified algorithm')),
 78                'device': 'SubsampleTransform',
 79                'device_type': 'DerivedDataTransform',
 80                'device_type_field': field
 81            }
 82            for field in self.field_list
 83        }
 84        return metadata_fields
 85
 86    ############################
 87    def _add_record(self, record):
 88        """Cached the values contained in a new record."""
 89        if type(record) not in [DASRecord, dict]:
 90            logging.error('SubsampleTransform records must be dict or '
 91                          'DASRecord. Received type %s: %s', type(record), record)
 92            return
 93
 94        if type(record) is DASRecord:
 95            timestamp = record.timestamp
 96            fields = record.fields
 97        else:
 98            timestamp = record.get('timestamp')
 99            fields = record.get('fields')
100
101        if not fields:
102            logging.error('SubsampleTransform: no fields found in record: %s', record)
103            return
104
105        # First, copy the new data into our cache
106        for field in self.field_list:
107            new_vals = fields.get(field)
108            if not new_vals:
109                continue
110            # If list, we assume it's [(ts, value), (ts, value),...]
111            if type(new_vals) is list:
112                self.cached_values[field].extend(new_vals)
113                # If not list, assume DASRecord or simple field dict; add tuple
114            elif timestamp:
115                self.cached_values[field].append((timestamp, new_vals))
116            else:
117                logging.error('SubsampleTransform found no timestamp in '
118                              'record: %s', record)
119
120    ############################
121    def _clean_cache(self):
122        """Which fields are we interested in to produce transformed data?"""
123        now = time.time()
124        for field in self.field_list:
125            # Iterate forward through field cache until we find a timestamp
126            # that is recent enough to keep
127            cache = self.cached_values[field]
128            for keep_index in range(len(cache)):
129                if cache[keep_index][0] > now - self.back_seconds:
130                    # Throw away everything before that index
131                    self.cached_values[field] = cache[keep_index:]
132                    break
133
134    ############################
135    def transform(self, record):
136        """Incorporate any useable fields in this record, and if it gives
137        us any new subsampled values, aggregate and return them.
138        """
139
140        # If we've got a list, hope it's a list of records. Recurse,
141        # calling transform() on each of the list elements in order and
142        # return the resulting list.
143        if type(record) is list:
144            results = []
145            for single_record in record:
146                results.append(self.transform(single_record))
147            return results
148
149        # Clean out old data
150        self._add_record(record)
151
152        # Clean out old data
153        self._clean_cache()
154
155        now = time.time()
156
157        result_fields = {}
158        for field in self.field_list:
159            if not self.cached_values[field]:
160                continue
161
162            output_field = self.field_spec[field].get('output')
163            if not output_field:
164                logging.warning('No "output" spec found for field %s', field)
165                continue
166            algorithm = self.field_spec[field].get('subsample')
167            if not algorithm:
168                logging.warning('No "subsample" spec found for field %s', field)
169                continue
170            field_result = subsample(algorithm, self.cached_values[field],
171                                     self.last_timestamp[field], now)
172            if field_result:
173                result_fields[output_field] = field_result
174                self.last_timestamp[field] = field_result[-1][0]
175
176        if not result_fields:
177            return None
178
179        # Form the response, adding in metadata if so specified and it's
180        # been long enough since we last sent it.
181        result = {'fields': result_fields}
182        if self.metadata_interval and \
183           now - self.metadata_interval > self.last_metadata_send:
184            result['metadata'] = {'fields': self._metadata()}
185            self.last_metadata_send = now
186
187        return result