openrvdas.logger.transforms.interpolation_transform
Compute interpolations of input data.
1#!/usr/bin/env python3 2"""Compute interpolations of input data. 3""" 4 5import logging 6import bisect 7 8from math import degrees, radians, sin, cos, atan2 9from statistics import mean 10from typing import Union, Any 11 12 13from logger.utils.das_record import DASRecord # noqa: E402 14from logger.transforms.derived_data_transform import DerivedDataTransform # noqa: E402 15 16 17################################################################################ 18class InterpolationTransform(DerivedDataTransform): 19 """Transform that computes interpolations of the specified variables. 20 """ 21 22 def __init__(self, field_spec, interval, window, 23 data_id=None, metadata_interval=None, **kwargs): 24 """ 25 ``` 26 field_spec - a dict of interpolated variables that are to be created, 27 where the key is the new variable's name, and the value is a dict 28 specifying the source field name and the algorithm that is to be 29 used to do the interpolation. E.g.: 30 31 { 32 'AvgCNAVCourseTrue': { 33 'source': 'CNAVCourseTrue', 34 'algorithm': { 35 'type': 'boxcar_average', 36 'window': 30 37 }, 38 }, 39 'AvgCNAVGPSDay': { 40 'source': 'CNAVGPSDay', 41 'algorithm': { 'type': 'nearest' }, 42 }, 43 ... 44 } 45 46 To simplify templating, can also accept a spec of the form 47 of a list: 48 49 [ 50 { sources: [MwxAirTemp, RTMPTemp, ...], 51 algorithm: boxcar_average, 52 window: 10, 53 result_prefix: Avg 54 }, 55 { sources: [PortTrueWindDir, StbdTrueWindDir], 56 algorithm: polar_average, 57 window: 10, 58 result_prefix: Avg 59 } 60 ] 61 62 interval - At what intervals (in seconds) should the interpolation 63 be computed? 64 65 window - Time window (in seconds) of data we should maintain 66 around the computation we're going to make. 67 68 data_id - What data id to assign to the output 69 70 metadata_interval - how many seconds between when we attach field metadata 71 to a record we send out. 72 ``` 73 74 """ 75 super().__init__(**kwargs) # processes 'quiet' and type hints 76 77 self.field_spec = {} 78 self.source_fields = set() 79 if isinstance(field_spec, dict): 80 for result_field, entry in field_spec.items(): 81 if 'source' in entry and 'algorithm' in entry: 82 self.field_spec[result_field] = entry 83 self.source_fields.add(entry.get('source')) 84 else: 85 logging.warning('InterpolationTransform field definition for %s ' 86 'must specify both "source" and "algorithm": %s', 87 result_field, entry) 88 89 # Alternate way of setting up a field spec that makes it easier to templatize. 90 # We'll expand the list of specs into a traditional field_spec 91 elif isinstance(field_spec, list): 92 for spec_instance in field_spec: 93 # Each spec_instance should be a dict of fields:, algorithm:, 94 # output_field_prefix: and window: 95 if not isinstance(spec_instance, dict): 96 raise ValueError('InterpolationTransform: if field_spec is list, must be ' 97 f'a list of dicts; found list of {type(spec_instance)}') 98 sources = spec_instance.get('sources') 99 100 if not isinstance(sources, list): 101 raise ValueError('InterpolationTransform: sources for field spec must be ' 102 f'a list ; found {type(spec_instance)}') 103 104 # Expand source list into a traditional field_spec 105 algorithm = spec_instance.get('algorithm') 106 window = spec_instance.get('window') 107 result_prefix = spec_instance.get('result_prefix') 108 for source in sources: 109 entry = {'source': source, 'algorithm': {'type': algorithm, 'window': window}} 110 result_field = result_prefix + source 111 self.field_spec[result_field] = entry 112 113 # Finally, stash sources so we know what to look for 114 self.source_fields.update(sources) 115 116 else: 117 raise ValueError('InterpolationTransform: field_spec must be either list ' 118 f'or dict. Found {type(field_spec)}') 119 120 self.interval = interval 121 self.window = window 122 self.data_id = data_id 123 self.metadata_interval = metadata_interval 124 125 # A dict of the cached values we're hanging onto - use sorted 126 # lists of (timestamp, value) pairs 127 self.cached_values = {f: [] for f in self.source_fields} 128 129 # The next timestamp we'd like to emit. Is set the first time we 130 # call transform(). 131 self.next_timestamp = 0 132 self.earliest_timestamp = float('inf') # Track earliest timestamp we've seen 133 self.latest_timestamp = 0 134 self.last_metadata_send = 0 # last time we've sent metadata 135 136 ############################ 137 def fields(self): 138 """Which fields are we interested in to produce transformed data?""" 139 return list(self.source_fields) 140 141 ############################ 142 def _metadata(self): 143 """Return a dict of metadata for our derived fields.""" 144 metadata_fields = { 145 'field': { 146 'description': 147 'Interpolated values of %s via %s' % 148 (entry['source'], entry['algorithm']), 149 'device': 'InterpolationTransform', 150 'device_type': 'DerivedDataTransform', 151 'device_type_field': result_field 152 } 153 for result_field, entry in self.field_spec.items() 154 } 155 return metadata_fields 156 157 ############################ 158 def _add_record(self, record): 159 """Cached the values contained in a new record, maintaining timestamp order.""" 160 if type(record) not in [DASRecord, dict]: 161 logging.error('InterpolationTransform records must be dict or ' 162 'DASRecord. Received type %s: %s', type(record), record) 163 return 0 164 165 if type(record) is DASRecord: 166 timestamp = record.timestamp 167 fields = record.fields 168 else: 169 timestamp = record.get('timestamp', 0) 170 fields = record.get('fields') 171 172 if not fields: 173 logging.info('InterpolationTransform: record has no fields: %s', record) 174 return timestamp 175 176 # First, copy the new data into our cache. NOTE: It's a judgment 177 # call whether it's more efficient to iterate over the fields 178 # we're looking for or the fields in the record. 179 for field, new_value in fields.items(): 180 if field not in self.source_fields: 181 continue 182 183 # Examine the value we've gotten. If list, we assume it's [(ts, 184 # value), (ts, value),...] 185 if type(new_value) is list: 186 for ts, val in new_value.items(): 187 self._insert_sorted(field, ts, val) 188 logging.debug(f'adding {ts}: {field} - {val}') 189 190 # If not list, assume DASRecord or simple field dict; add tuple 191 elif timestamp: 192 self._insert_sorted(field, timestamp, new_value) 193 logging.debug(f'adding {timestamp}: {field} - {new_value}') 194 else: 195 logging.warning('Interpolation found no timestamp in ' 196 'record: %s', record) 197 198 # Update our tracking of the earliest and latest timestamps we've seen 199 self.earliest_timestamp = min(self.earliest_timestamp, timestamp) 200 self.latest_timestamp = max(self.latest_timestamp, timestamp) 201 202 # Return the timestamp of the record 203 return timestamp 204 205 ############################ 206 def _insert_sorted(self, field: str, timestamp: float, value: Any) -> None: 207 """Insert a (timestamp, value) pair into the sorted cached values for a field.""" 208 cache = self.cached_values[field] 209 210 # Use binary search to find insertion position to maintain sorted order 211 timestamps = [ts for ts, _ in cache] 212 pos = bisect.bisect_left(timestamps, timestamp) 213 214 # Insert at correct position to maintain sort order 215 cache.insert(pos, (timestamp, value)) 216 217 ############################ 218 def _clean_cache(self): 219 """Remove values from cache that are too old to be useful.""" 220 for field in self.source_fields: 221 # Iterate forward through field cache until we find a timestamp 222 # that is recent enough to keep 223 cache = self.cached_values[field] 224 lower_limit = self.next_timestamp - self.window / 2 225 keep_index = 0 226 while keep_index < len(cache) and cache[keep_index][0] < lower_limit: 227 keep_index += 1 228 229 # Throw away everything before that index 230 self.cached_values[field] = cache[keep_index:] 231 232 ############################ 233 def transform(self, record: Union[DASRecord, dict]): 234 """Incorporate any useable fields in this record, and if it gives 235 us any new interpolated values, aggregate and return them as a list of 236 dicts of the form: 237 238 [ 239 {'timestamp': timestamp, 240 'fields': { 241 fieldname: value, 242 fieldname: value, 243 ... 244 } 245 }, 246 {'timestamp': timestamp, 247 'fields': ... 248 } 249 ] 250 251 If there are insufficient data in the window to compute any 252 interpolation, return an empty list. 253 """ 254 # See if it's something we can process, and if not, try digesting 255 if not self.can_process_record(record): # inherited from BaseModule() 256 return self.digest_record(record) # inherited from BaseModule() 257 258 # Add the record 259 self._add_record(record) 260 261 # First time through, our 'next_timestamp' will be zero. Set it to 262 # a good starting place. 263 if not self.next_timestamp: 264 self.next_timestamp = self.earliest_timestamp 265 266 # What fields do we have data for in our cache? 267 non_empty = {} 268 for dest, spec in self.field_spec.items(): 269 source = spec.get('source') 270 if source: 271 values = self.cached_values.get(source, []) 272 if len(values): 273 non_empty[dest] = [source, len(values)] 274 275 # Iterate through all timestamps up to the edge of what we can fit 276 # in our window without running into the edge of our latest timestamp. 277 results = [] 278 logging.debug(f'latest timestamp: {self.latest_timestamp}, next: {self.next_timestamp}') 279 while self.next_timestamp < self.latest_timestamp - self.window / 2: 280 # Clean out old data 281 self._clean_cache() 282 283 result = {} 284 for result_field, entry in self.field_spec.items(): 285 source = entry.get('source') 286 source_values = self.cached_values.get(source) 287 algorithm = entry.get('algorithm') 288 # logging.warning('%s->%s: %d values', 289 # source, result_field, len(source_values)) 290 value = interpolate(algorithm, source_values, 291 self.next_timestamp, 292 self.latest_timestamp) 293 if value is not None: 294 result[result_field] = value 295 if result: 296 result_record = {'timestamp': self.next_timestamp, 'fields': result} 297 if self.data_id: 298 result_record['data_id'] = self.data_id 299 results.append(result_record) 300 self.next_timestamp += self.interval 301 302 return results 303 304 305############################ 306def interpolate(algorithm, values, timestamp, now): 307 """An omnibus routine for taking a list of timestamped values, a 308 specification of an averaging algorithm, and returning a value 309 computed at the specified timestamp. Returns None if there aren't 310 enough data to compute a value. 311 312 algorithm The name of the algorithm to be used 313 314 values A list of [(timestamp, value),...] pairs 315 316 timestamp The timestamp for which interpolation should be computed 317 318 now Timestamp now. This should be used to determine whether 319 we're far enough beyond our timestamp to compute a value. 320 """ 321 if not type(algorithm) is dict: 322 logging.warning('Function interpolate() handed non-dict algorithm ' 323 'specification: %s', algorithm) 324 return None 325 if not values: 326 logging.debug('Function interpolate() handed empty values list') 327 return None 328 329 ################## 330 # Select algorithm 331 alg_type = algorithm.get('type') 332 333 # boxcar_average: all values within symmetric interval window get 334 # same weight. 335 if alg_type == 'boxcar_average': 336 window = algorithm.get('window', 10) # How far back/forward to average 337 lower_limit = timestamp - window / 2 338 upper_limit = timestamp + window / 2 339 vals_to_average = [val for ts, val in values 340 if ts >= lower_limit and ts <= upper_limit] 341 if not vals_to_average: 342 return None 343 344 try: 345 return mean(vals_to_average) 346 except TypeError: 347 logging.error('Non-numeric value in interpolation list: %s', vals_to_average) 348 return None 349 350 # nearest: return value of nearest timestamp. Note that we assume 351 # timestamps are in order, so once distance starts going up, we're 352 # done. 353 if alg_type == 'nearest': 354 best_distance = float('inf') 355 value = None 356 for i in range(len(values)): 357 ts, ts_value = values[i] 358 distance = abs(ts - timestamp) 359 if distance <= best_distance: 360 best_distance = distance 361 value = ts_value 362 else: 363 break 364 return value 365 366 # polar_average: interpret as an angle in degrees. Convert to points 367 # on a unit circle and return the angle of their centroid from the origin. 368 if alg_type == 'polar_average': 369 window = algorithm.get('window', 10) # How far back/forward to average 370 lower_limit = timestamp - window / 2 371 upper_limit = timestamp + window / 2 372 vals_to_average = [val for ts, val in values 373 if ts >= lower_limit and ts <= upper_limit] 374 if not vals_to_average: 375 return None 376 377 try: 378 val_radians = [radians(val) for val in vals_to_average] 379 x_mean = mean([sin(val) for val in val_radians]) 380 y_mean = mean([cos(val) for val in val_radians]) 381 angle = degrees(atan2(x_mean, y_mean)) 382 if angle < 0: 383 angle += 360 384 return angle 385 except TypeError: 386 logging.error('Non-numeric value in interpolation list: %s', vals_to_average) 387 return None 388 389 # Not an algorithm we recognize 390 else: 391 logging.warning('Function interpolate() received unrecognized algorithm ' 392 'type: %s', alg_type) 393 return None
class
InterpolationTransform(logger.transforms.derived_data_transform.DerivedDataTransform):
19class InterpolationTransform(DerivedDataTransform): 20 """Transform that computes interpolations of the specified variables. 21 """ 22 23 def __init__(self, field_spec, interval, window, 24 data_id=None, metadata_interval=None, **kwargs): 25 """ 26 ``` 27 field_spec - a dict of interpolated variables that are to be created, 28 where the key is the new variable's name, and the value is a dict 29 specifying the source field name and the algorithm that is to be 30 used to do the interpolation. E.g.: 31 32 { 33 'AvgCNAVCourseTrue': { 34 'source': 'CNAVCourseTrue', 35 'algorithm': { 36 'type': 'boxcar_average', 37 'window': 30 38 }, 39 }, 40 'AvgCNAVGPSDay': { 41 'source': 'CNAVGPSDay', 42 'algorithm': { 'type': 'nearest' }, 43 }, 44 ... 45 } 46 47 To simplify templating, can also accept a spec of the form 48 of a list: 49 50 [ 51 { sources: [MwxAirTemp, RTMPTemp, ...], 52 algorithm: boxcar_average, 53 window: 10, 54 result_prefix: Avg 55 }, 56 { sources: [PortTrueWindDir, StbdTrueWindDir], 57 algorithm: polar_average, 58 window: 10, 59 result_prefix: Avg 60 } 61 ] 62 63 interval - At what intervals (in seconds) should the interpolation 64 be computed? 65 66 window - Time window (in seconds) of data we should maintain 67 around the computation we're going to make. 68 69 data_id - What data id to assign to the output 70 71 metadata_interval - how many seconds between when we attach field metadata 72 to a record we send out. 73 ``` 74 75 """ 76 super().__init__(**kwargs) # processes 'quiet' and type hints 77 78 self.field_spec = {} 79 self.source_fields = set() 80 if isinstance(field_spec, dict): 81 for result_field, entry in field_spec.items(): 82 if 'source' in entry and 'algorithm' in entry: 83 self.field_spec[result_field] = entry 84 self.source_fields.add(entry.get('source')) 85 else: 86 logging.warning('InterpolationTransform field definition for %s ' 87 'must specify both "source" and "algorithm": %s', 88 result_field, entry) 89 90 # Alternate way of setting up a field spec that makes it easier to templatize. 91 # We'll expand the list of specs into a traditional field_spec 92 elif isinstance(field_spec, list): 93 for spec_instance in field_spec: 94 # Each spec_instance should be a dict of fields:, algorithm:, 95 # output_field_prefix: and window: 96 if not isinstance(spec_instance, dict): 97 raise ValueError('InterpolationTransform: if field_spec is list, must be ' 98 f'a list of dicts; found list of {type(spec_instance)}') 99 sources = spec_instance.get('sources') 100 101 if not isinstance(sources, list): 102 raise ValueError('InterpolationTransform: sources for field spec must be ' 103 f'a list ; found {type(spec_instance)}') 104 105 # Expand source list into a traditional field_spec 106 algorithm = spec_instance.get('algorithm') 107 window = spec_instance.get('window') 108 result_prefix = spec_instance.get('result_prefix') 109 for source in sources: 110 entry = {'source': source, 'algorithm': {'type': algorithm, 'window': window}} 111 result_field = result_prefix + source 112 self.field_spec[result_field] = entry 113 114 # Finally, stash sources so we know what to look for 115 self.source_fields.update(sources) 116 117 else: 118 raise ValueError('InterpolationTransform: field_spec must be either list ' 119 f'or dict. Found {type(field_spec)}') 120 121 self.interval = interval 122 self.window = window 123 self.data_id = data_id 124 self.metadata_interval = metadata_interval 125 126 # A dict of the cached values we're hanging onto - use sorted 127 # lists of (timestamp, value) pairs 128 self.cached_values = {f: [] for f in self.source_fields} 129 130 # The next timestamp we'd like to emit. Is set the first time we 131 # call transform(). 132 self.next_timestamp = 0 133 self.earliest_timestamp = float('inf') # Track earliest timestamp we've seen 134 self.latest_timestamp = 0 135 self.last_metadata_send = 0 # last time we've sent metadata 136 137 ############################ 138 def fields(self): 139 """Which fields are we interested in to produce transformed data?""" 140 return list(self.source_fields) 141 142 ############################ 143 def _metadata(self): 144 """Return a dict of metadata for our derived fields.""" 145 metadata_fields = { 146 'field': { 147 'description': 148 'Interpolated values of %s via %s' % 149 (entry['source'], entry['algorithm']), 150 'device': 'InterpolationTransform', 151 'device_type': 'DerivedDataTransform', 152 'device_type_field': result_field 153 } 154 for result_field, entry in self.field_spec.items() 155 } 156 return metadata_fields 157 158 ############################ 159 def _add_record(self, record): 160 """Cached the values contained in a new record, maintaining timestamp order.""" 161 if type(record) not in [DASRecord, dict]: 162 logging.error('InterpolationTransform records must be dict or ' 163 'DASRecord. Received type %s: %s', type(record), record) 164 return 0 165 166 if type(record) is DASRecord: 167 timestamp = record.timestamp 168 fields = record.fields 169 else: 170 timestamp = record.get('timestamp', 0) 171 fields = record.get('fields') 172 173 if not fields: 174 logging.info('InterpolationTransform: record has no fields: %s', record) 175 return timestamp 176 177 # First, copy the new data into our cache. NOTE: It's a judgment 178 # call whether it's more efficient to iterate over the fields 179 # we're looking for or the fields in the record. 180 for field, new_value in fields.items(): 181 if field not in self.source_fields: 182 continue 183 184 # Examine the value we've gotten. If list, we assume it's [(ts, 185 # value), (ts, value),...] 186 if type(new_value) is list: 187 for ts, val in new_value.items(): 188 self._insert_sorted(field, ts, val) 189 logging.debug(f'adding {ts}: {field} - {val}') 190 191 # If not list, assume DASRecord or simple field dict; add tuple 192 elif timestamp: 193 self._insert_sorted(field, timestamp, new_value) 194 logging.debug(f'adding {timestamp}: {field} - {new_value}') 195 else: 196 logging.warning('Interpolation found no timestamp in ' 197 'record: %s', record) 198 199 # Update our tracking of the earliest and latest timestamps we've seen 200 self.earliest_timestamp = min(self.earliest_timestamp, timestamp) 201 self.latest_timestamp = max(self.latest_timestamp, timestamp) 202 203 # Return the timestamp of the record 204 return timestamp 205 206 ############################ 207 def _insert_sorted(self, field: str, timestamp: float, value: Any) -> None: 208 """Insert a (timestamp, value) pair into the sorted cached values for a field.""" 209 cache = self.cached_values[field] 210 211 # Use binary search to find insertion position to maintain sorted order 212 timestamps = [ts for ts, _ in cache] 213 pos = bisect.bisect_left(timestamps, timestamp) 214 215 # Insert at correct position to maintain sort order 216 cache.insert(pos, (timestamp, value)) 217 218 ############################ 219 def _clean_cache(self): 220 """Remove values from cache that are too old to be useful.""" 221 for field in self.source_fields: 222 # Iterate forward through field cache until we find a timestamp 223 # that is recent enough to keep 224 cache = self.cached_values[field] 225 lower_limit = self.next_timestamp - self.window / 2 226 keep_index = 0 227 while keep_index < len(cache) and cache[keep_index][0] < lower_limit: 228 keep_index += 1 229 230 # Throw away everything before that index 231 self.cached_values[field] = cache[keep_index:] 232 233 ############################ 234 def transform(self, record: Union[DASRecord, dict]): 235 """Incorporate any useable fields in this record, and if it gives 236 us any new interpolated values, aggregate and return them as a list of 237 dicts of the form: 238 239 [ 240 {'timestamp': timestamp, 241 'fields': { 242 fieldname: value, 243 fieldname: value, 244 ... 245 } 246 }, 247 {'timestamp': timestamp, 248 'fields': ... 249 } 250 ] 251 252 If there are insufficient data in the window to compute any 253 interpolation, return an empty list. 254 """ 255 # See if it's something we can process, and if not, try digesting 256 if not self.can_process_record(record): # inherited from BaseModule() 257 return self.digest_record(record) # inherited from BaseModule() 258 259 # Add the record 260 self._add_record(record) 261 262 # First time through, our 'next_timestamp' will be zero. Set it to 263 # a good starting place. 264 if not self.next_timestamp: 265 self.next_timestamp = self.earliest_timestamp 266 267 # What fields do we have data for in our cache? 268 non_empty = {} 269 for dest, spec in self.field_spec.items(): 270 source = spec.get('source') 271 if source: 272 values = self.cached_values.get(source, []) 273 if len(values): 274 non_empty[dest] = [source, len(values)] 275 276 # Iterate through all timestamps up to the edge of what we can fit 277 # in our window without running into the edge of our latest timestamp. 278 results = [] 279 logging.debug(f'latest timestamp: {self.latest_timestamp}, next: {self.next_timestamp}') 280 while self.next_timestamp < self.latest_timestamp - self.window / 2: 281 # Clean out old data 282 self._clean_cache() 283 284 result = {} 285 for result_field, entry in self.field_spec.items(): 286 source = entry.get('source') 287 source_values = self.cached_values.get(source) 288 algorithm = entry.get('algorithm') 289 # logging.warning('%s->%s: %d values', 290 # source, result_field, len(source_values)) 291 value = interpolate(algorithm, source_values, 292 self.next_timestamp, 293 self.latest_timestamp) 294 if value is not None: 295 result[result_field] = value 296 if result: 297 result_record = {'timestamp': self.next_timestamp, 'fields': result} 298 if self.data_id: 299 result_record['data_id'] = self.data_id 300 results.append(result_record) 301 self.next_timestamp += self.interval 302 303 return results
Transform that computes interpolations of the specified variables.
InterpolationTransform( field_spec, interval, window, data_id=None, metadata_interval=None, **kwargs)
23 def __init__(self, field_spec, interval, window, 24 data_id=None, metadata_interval=None, **kwargs): 25 """ 26 ``` 27 field_spec - a dict of interpolated variables that are to be created, 28 where the key is the new variable's name, and the value is a dict 29 specifying the source field name and the algorithm that is to be 30 used to do the interpolation. E.g.: 31 32 { 33 'AvgCNAVCourseTrue': { 34 'source': 'CNAVCourseTrue', 35 'algorithm': { 36 'type': 'boxcar_average', 37 'window': 30 38 }, 39 }, 40 'AvgCNAVGPSDay': { 41 'source': 'CNAVGPSDay', 42 'algorithm': { 'type': 'nearest' }, 43 }, 44 ... 45 } 46 47 To simplify templating, can also accept a spec of the form 48 of a list: 49 50 [ 51 { sources: [MwxAirTemp, RTMPTemp, ...], 52 algorithm: boxcar_average, 53 window: 10, 54 result_prefix: Avg 55 }, 56 { sources: [PortTrueWindDir, StbdTrueWindDir], 57 algorithm: polar_average, 58 window: 10, 59 result_prefix: Avg 60 } 61 ] 62 63 interval - At what intervals (in seconds) should the interpolation 64 be computed? 65 66 window - Time window (in seconds) of data we should maintain 67 around the computation we're going to make. 68 69 data_id - What data id to assign to the output 70 71 metadata_interval - how many seconds between when we attach field metadata 72 to a record we send out. 73 ``` 74 75 """ 76 super().__init__(**kwargs) # processes 'quiet' and type hints 77 78 self.field_spec = {} 79 self.source_fields = set() 80 if isinstance(field_spec, dict): 81 for result_field, entry in field_spec.items(): 82 if 'source' in entry and 'algorithm' in entry: 83 self.field_spec[result_field] = entry 84 self.source_fields.add(entry.get('source')) 85 else: 86 logging.warning('InterpolationTransform field definition for %s ' 87 'must specify both "source" and "algorithm": %s', 88 result_field, entry) 89 90 # Alternate way of setting up a field spec that makes it easier to templatize. 91 # We'll expand the list of specs into a traditional field_spec 92 elif isinstance(field_spec, list): 93 for spec_instance in field_spec: 94 # Each spec_instance should be a dict of fields:, algorithm:, 95 # output_field_prefix: and window: 96 if not isinstance(spec_instance, dict): 97 raise ValueError('InterpolationTransform: if field_spec is list, must be ' 98 f'a list of dicts; found list of {type(spec_instance)}') 99 sources = spec_instance.get('sources') 100 101 if not isinstance(sources, list): 102 raise ValueError('InterpolationTransform: sources for field spec must be ' 103 f'a list ; found {type(spec_instance)}') 104 105 # Expand source list into a traditional field_spec 106 algorithm = spec_instance.get('algorithm') 107 window = spec_instance.get('window') 108 result_prefix = spec_instance.get('result_prefix') 109 for source in sources: 110 entry = {'source': source, 'algorithm': {'type': algorithm, 'window': window}} 111 result_field = result_prefix + source 112 self.field_spec[result_field] = entry 113 114 # Finally, stash sources so we know what to look for 115 self.source_fields.update(sources) 116 117 else: 118 raise ValueError('InterpolationTransform: field_spec must be either list ' 119 f'or dict. Found {type(field_spec)}') 120 121 self.interval = interval 122 self.window = window 123 self.data_id = data_id 124 self.metadata_interval = metadata_interval 125 126 # A dict of the cached values we're hanging onto - use sorted 127 # lists of (timestamp, value) pairs 128 self.cached_values = {f: [] for f in self.source_fields} 129 130 # The next timestamp we'd like to emit. Is set the first time we 131 # call transform(). 132 self.next_timestamp = 0 133 self.earliest_timestamp = float('inf') # Track earliest timestamp we've seen 134 self.latest_timestamp = 0 135 self.last_metadata_send = 0 # last time we've sent metadata
field_spec - a dict of interpolated variables that are to be created,
where the key is the new variable's name, and the value is a dict
specifying the source field name and the algorithm that is to be
used to do the interpolation. E.g.:
{
'AvgCNAVCourseTrue': {
'source': 'CNAVCourseTrue',
'algorithm': {
'type': 'boxcar_average',
'window': 30
},
},
'AvgCNAVGPSDay': {
'source': 'CNAVGPSDay',
'algorithm': { 'type': 'nearest' },
},
...
}
To simplify templating, can also accept a spec of the form
of a list:
[
{ sources: [MwxAirTemp, RTMPTemp, ...],
algorithm: boxcar_average,
window: 10,
result_prefix: Avg
},
{ sources: [PortTrueWindDir, StbdTrueWindDir],
algorithm: polar_average,
window: 10,
result_prefix: Avg
}
]
interval - At what intervals (in seconds) should the interpolation
be computed?
window - Time window (in seconds) of data we should maintain
around the computation we're going to make.
data_id - What data id to assign to the output
metadata_interval - how many seconds between when we attach field metadata
to a record we send out.
def
fields(self):
138 def fields(self): 139 """Which fields are we interested in to produce transformed data?""" 140 return list(self.source_fields)
Which fields are we interested in to produce transformed data?
def
transform(self, record: Union[logger.utils.das_record.DASRecord, dict]):
234 def transform(self, record: Union[DASRecord, dict]): 235 """Incorporate any useable fields in this record, and if it gives 236 us any new interpolated values, aggregate and return them as a list of 237 dicts of the form: 238 239 [ 240 {'timestamp': timestamp, 241 'fields': { 242 fieldname: value, 243 fieldname: value, 244 ... 245 } 246 }, 247 {'timestamp': timestamp, 248 'fields': ... 249 } 250 ] 251 252 If there are insufficient data in the window to compute any 253 interpolation, return an empty list. 254 """ 255 # See if it's something we can process, and if not, try digesting 256 if not self.can_process_record(record): # inherited from BaseModule() 257 return self.digest_record(record) # inherited from BaseModule() 258 259 # Add the record 260 self._add_record(record) 261 262 # First time through, our 'next_timestamp' will be zero. Set it to 263 # a good starting place. 264 if not self.next_timestamp: 265 self.next_timestamp = self.earliest_timestamp 266 267 # What fields do we have data for in our cache? 268 non_empty = {} 269 for dest, spec in self.field_spec.items(): 270 source = spec.get('source') 271 if source: 272 values = self.cached_values.get(source, []) 273 if len(values): 274 non_empty[dest] = [source, len(values)] 275 276 # Iterate through all timestamps up to the edge of what we can fit 277 # in our window without running into the edge of our latest timestamp. 278 results = [] 279 logging.debug(f'latest timestamp: {self.latest_timestamp}, next: {self.next_timestamp}') 280 while self.next_timestamp < self.latest_timestamp - self.window / 2: 281 # Clean out old data 282 self._clean_cache() 283 284 result = {} 285 for result_field, entry in self.field_spec.items(): 286 source = entry.get('source') 287 source_values = self.cached_values.get(source) 288 algorithm = entry.get('algorithm') 289 # logging.warning('%s->%s: %d values', 290 # source, result_field, len(source_values)) 291 value = interpolate(algorithm, source_values, 292 self.next_timestamp, 293 self.latest_timestamp) 294 if value is not None: 295 result[result_field] = value 296 if result: 297 result_record = {'timestamp': self.next_timestamp, 'fields': result} 298 if self.data_id: 299 result_record['data_id'] = self.data_id 300 results.append(result_record) 301 self.next_timestamp += self.interval 302 303 return results
Incorporate any useable fields in this record, and if it gives us any new interpolated values, aggregate and return them as a list of dicts of the form:
[ {'timestamp': timestamp, 'fields': { fieldname: value, fieldname: value, ... } }, {'timestamp': timestamp, 'fields': ... } ]
If there are insufficient data in the window to compute any interpolation, return an empty list.
def
interpolate(algorithm, values, timestamp, now):
307def interpolate(algorithm, values, timestamp, now): 308 """An omnibus routine for taking a list of timestamped values, a 309 specification of an averaging algorithm, and returning a value 310 computed at the specified timestamp. Returns None if there aren't 311 enough data to compute a value. 312 313 algorithm The name of the algorithm to be used 314 315 values A list of [(timestamp, value),...] pairs 316 317 timestamp The timestamp for which interpolation should be computed 318 319 now Timestamp now. This should be used to determine whether 320 we're far enough beyond our timestamp to compute a value. 321 """ 322 if not type(algorithm) is dict: 323 logging.warning('Function interpolate() handed non-dict algorithm ' 324 'specification: %s', algorithm) 325 return None 326 if not values: 327 logging.debug('Function interpolate() handed empty values list') 328 return None 329 330 ################## 331 # Select algorithm 332 alg_type = algorithm.get('type') 333 334 # boxcar_average: all values within symmetric interval window get 335 # same weight. 336 if alg_type == 'boxcar_average': 337 window = algorithm.get('window', 10) # How far back/forward to average 338 lower_limit = timestamp - window / 2 339 upper_limit = timestamp + window / 2 340 vals_to_average = [val for ts, val in values 341 if ts >= lower_limit and ts <= upper_limit] 342 if not vals_to_average: 343 return None 344 345 try: 346 return mean(vals_to_average) 347 except TypeError: 348 logging.error('Non-numeric value in interpolation list: %s', vals_to_average) 349 return None 350 351 # nearest: return value of nearest timestamp. Note that we assume 352 # timestamps are in order, so once distance starts going up, we're 353 # done. 354 if alg_type == 'nearest': 355 best_distance = float('inf') 356 value = None 357 for i in range(len(values)): 358 ts, ts_value = values[i] 359 distance = abs(ts - timestamp) 360 if distance <= best_distance: 361 best_distance = distance 362 value = ts_value 363 else: 364 break 365 return value 366 367 # polar_average: interpret as an angle in degrees. Convert to points 368 # on a unit circle and return the angle of their centroid from the origin. 369 if alg_type == 'polar_average': 370 window = algorithm.get('window', 10) # How far back/forward to average 371 lower_limit = timestamp - window / 2 372 upper_limit = timestamp + window / 2 373 vals_to_average = [val for ts, val in values 374 if ts >= lower_limit and ts <= upper_limit] 375 if not vals_to_average: 376 return None 377 378 try: 379 val_radians = [radians(val) for val in vals_to_average] 380 x_mean = mean([sin(val) for val in val_radians]) 381 y_mean = mean([cos(val) for val in val_radians]) 382 angle = degrees(atan2(x_mean, y_mean)) 383 if angle < 0: 384 angle += 360 385 return angle 386 except TypeError: 387 logging.error('Non-numeric value in interpolation list: %s', vals_to_average) 388 return None 389 390 # Not an algorithm we recognize 391 else: 392 logging.warning('Function interpolate() received unrecognized algorithm ' 393 'type: %s', alg_type) 394 return None
An omnibus routine for taking a list of timestamped values, a specification of an averaging algorithm, and returning a value computed at the specified timestamp. Returns None if there aren't enough data to compute a value.
algorithm The name of the algorithm to be used
values A list of [(timestamp, value),...] pairs
timestamp The timestamp for which interpolation should be computed
now Timestamp now. This should be used to determine whether we're far enough beyond our timestamp to compute a value.