///|
/// Time-window planning utilities for batch and streaming consumers.
pub struct TimeWindow {
start : Int64
end : Int64
ordinal : Int
}
///|
pub fn TimeWindow::new(start : Int64, end : Int64, ordinal : Int) -> TimeWindow {
{ start, end, ordinal }
}
///|
pub fn TimeWindow::duration(self : TimeWindow) -> Int64 {
self.end - self.start
}
///|
pub fn TimeWindow::contains(self : TimeWindow, timestamp : Int64) -> Bool {
timestamp >= self.start && timestamp < self.end
}
///|
pub fn TimeWindow::is_empty(self : TimeWindow) -> Bool {
self.end <= self.start
}
///|
pub fn make_time_windows(
start : Int64,
end : Int64,
width : Int64,
step? : Int64 = 0L,
) -> Array[TimeWindow] {
let result : Array[TimeWindow] = []
if width <= 0L || end <= start {
return result
}
let stride = if step <= 0L { width } else { step }
let mut cursor = start
let mut ordinal = 0
while cursor < end {
let boundary = if cursor + width > end { end } else { cursor + width }
result.push(TimeWindow::new(cursor, boundary, ordinal))
cursor += stride
ordinal += 1
}
result
}
///|
pub fn locate_time_window(
windows : Array[TimeWindow],
timestamp : Int64,
) -> Int {
for window in windows {
if window.contains(timestamp) {
return window.ordinal
}
}
-1
}
///|
pub fn bucket_timestamp(
timestamp : Int64,
origin : Int64,
width : Int64,
) -> Int64 {
if width <= 0L {
origin
} else {
origin + (timestamp - origin) / width * width
}
}
///|
pub fn window_counts(
points : Array[SignalPoint],
windows : Array[TimeWindow],
) -> Array[Int] {
let result = Array::make(windows.length(), 0)
for point in points {
let ordinal = locate_time_window(windows, point.timestamp)
if ordinal >= 0 && ordinal < result.length() {
result[ordinal] += 1
}
}
result
}
///|
pub fn window_means(
points : Array[SignalPoint],
windows : Array[TimeWindow],
) -> Array[Double] {
let values : Array[Array[Double]] = []
for _ in windows {
values.push([])
}
for point in points {
let ordinal = locate_time_window(windows, point.timestamp)
if ordinal >= 0 && ordinal < values.length() {
values[ordinal].push(point.value)
}
}
let result : Array[Double] = []
for bucket in values {
result.push(mean(bucket))
}
result
}
///|
pub fn window_summaries(
points : Array[SignalPoint],
windows : Array[TimeWindow],
) -> Array[StatsSummary] {
let values : Array[Array[Double]] = []
for _ in windows {
values.push([])
}
for point in points {
let ordinal = locate_time_window(windows, point.timestamp)
if ordinal >= 0 && ordinal < values.length() {
values[ordinal].push(point.value)
}
}
let result : Array[StatsSummary] = []
for bucket in values {
let accumulator = OnlineMoments::new()
for value in bucket {
accumulator.push(value)
}
result.push(accumulator.summary(median=median(bucket)))
}
result
}
///|
pub fn time_window_csv(
windows : Array[TimeWindow],
counts : Array[Int],
) -> String {
let mut output = "ordinal,start,end,count\n"
let length = if windows.length() < counts.length() {
windows.length()
} else {
counts.length()
}
for i in 0..