openrvdas.logger.utils.subsample

 1#!/usr/bin/env python3
 2"""
 3"""
 4import logging
 5
 6
 7def subsample(algorithm, values, latest_timestamp, now):
 8    """An omnibus routine for taking a list of timestamped values, a
 9    specification of an averaging algorithm, and returning a list of
10    zero, one or more timestamped "averaged" output values.
11
12    algorithm    The name of the algorithm to be used
13
14    values       List of values to be averaged in format of
15                 [(timestamp, value), (timestamp, value),...]
16
17    latest_timestamp
18                 Timestamp of the last value that was output
19
20    now          Timestamp now
21    """
22    if not isinstance(algorithm, dict):
23        logging.warning('Function subsample() handed non-dict algorithm '
24                        'specification: %s', algorithm)
25        return None
26    if not values:
27        logging.info('Function subsample() handed empty values list')
28        return None
29
30    alg_type = algorithm.get('type')
31
32    ##################
33    # Select algorithm
34
35    # boxcar_average: all values within symmetric interval window get
36    # same weight.
37    if alg_type == 'boxcar_average':
38        interval = algorithm.get('interval', 10)  # How often to output
39        window = algorithm.get('window', 10)     # How far back to average
40
41        # Which timestamps are we going to emit as averages? Start at
42        # 'interval' seconds after the last timestamp we emitted and end
43        # at 'window/2' seconds before now, because we want to have a full
44        # 'window' seconds available for most recent point.
45        ts = max(latest_timestamp + interval, values[0][0] + window / 2)
46        ts_list = []
47        while ts <= now - window / 2:
48            ts_list.append(ts)
49            ts += interval
50
51        if not ts_list:
52            logging.debug('No timestamps to emit this time')
53            return None
54
55        # Start and end intervals for data for each timestamp
56        ts_start = {ts: (ts - window / 2) for ts in ts_list}
57        ts_end = {ts: (ts + window / 2) for ts in ts_list}
58        ts_data = {ts: [] for ts in ts_list}
59
60        # Iterate through values backwards until we're outside the window
61        # of the first output ts.
62        earliest_ts_of_interest = ts_list[0] - window / 2
63        for v_index in range(len(values) - 1, -1, -1):
64            (value_ts, value) = values[v_index]
65            if value_ts < earliest_ts_of_interest:
66                break
67
68            # Does this ts,value pair belong in any of our averages?
69            for ts in ts_list:
70                if value_ts > ts_start[ts] and value_ts < ts_end[ts]:
71                    ts_data[ts].append(value)
72
73        # Assemble averages for all timestamps we're going to emit
74        results = []
75        for ts in ts_list:
76            if type(ts) not in [int, float, bool]:
77                logging.warning('Trying to subsample non-numeric value "%s"', ts)
78                continue
79            if len(ts_data[ts]):
80                try:
81                    results.append((ts, sum(ts_data[ts]) / len(ts_data[ts])))
82                except TypeError:
83                    logging.warning('Non-numeric input in subsample: %s, in %s, list: %s',
84                                    ts_data[ts], ts_data, ts_list)
85
86        return results
87
88    else:
89        logging.warning('Function subsample() received unrecognized algorithm '
90                        'type: %s', alg_type)
91        return None