52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
from datetime import datetime
|
|
from datetime import timezone
|
|
|
|
|
|
class TimeSeriesAccumulator:
|
|
def __init__(self, start_time, end_time, n_bins):
|
|
self.start_time = start_time
|
|
self.start_timestamp = start_time.replace(
|
|
tzinfo=timezone.utc).timestamp()
|
|
self.end_time = end_time
|
|
self.end_timestamp = end_time.replace(tzinfo=timezone.utc).timestamp()
|
|
self.n_bins = n_bins
|
|
self.bin_time = (self.end_timestamp -
|
|
self.start_timestamp)/self.n_bins
|
|
self.data = [[] for _ in range(n_bins)]
|
|
|
|
def add(self, time, value):
|
|
timestamp = time.replace(tzinfo=timezone.utc).timestamp()
|
|
if time < self.start_time or time > self.end_time:
|
|
return
|
|
idx = int((timestamp - self.start_timestamp)/self.bin_time)
|
|
self.data[idx].append(value)
|
|
|
|
def get_f(self, f, initial=0):
|
|
prev = initial
|
|
result = []
|
|
for i in self.data:
|
|
new = f(i)
|
|
if new == numpy.nan:
|
|
result.append(prev)
|
|
else:
|
|
result.append(new)
|
|
prev = new
|
|
return result
|
|
|
|
|
|
return [f(i) for i in self.data]
|
|
|
|
|
|
def get_ts(self):
|
|
return [datetime.utcfromtimestamp(self.start_timestamp+(i+1)*self.bin_time) for i in range(self.n_bins)]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
tsa = TimeSeriesAccumulator(start_time=datetime(
|
|
2020, 7, 27, 23, 0, 0), end_time=datetime(2020, 7, 28, 0, 0, 0), n_bins=2)
|
|
tsa.add(datetime(2020, 7, 27, 23, 15, 0), 1)
|
|
tsa.add(datetime(2020, 7, 27, 23, 45, 0), 2)
|
|
tsa.add(datetime(2020, 7, 27, 22, 45, 0), 3)
|
|
print(tsa.get_ts())
|
|
print(tsa.get_f(lambda x: sum(x)))
|