use std::collections::VecDeque;
use std::sync::Arc;
use parking_lot::Mutex;
use web_time::{SystemTime, UNIX_EPOCH};
use chrono::{Local, TimeZone};
use eframe::egui::{self, Align2, Color32, FontId, Pos2, Rect, Stroke};
use pv::{Listener, Pv, PvState};
use crate::axis::{self, Scale};
use crate::plot::{self, text_size, vertical_text};
use crate::screen::Node;
use crate::style::{self, color, named};
use crate::widgets::{alarm_border, fill};
const MAX_SAMPLES: usize = 100_000;
const MAX_AGE: f64 = 86_400.0;
#[derive(Default)]
pub struct Samples {
points: VecDeque<(f64, f64)>,
pushed: u64,
}
impl Samples {
pub fn push(&mut self, t: f64, v: f64) {
if v.is_nan() && self.points.back().is_none_or(|p| p.1.is_nan()) {
return;
}
let t = self.points.back().map_or(t, |p| t.max(p.0));
self.points.push_back((t, v));
self.pushed += 1;
while self.points.len() > MAX_SAMPLES
|| self.points.front().is_some_and(|p| p.0 < t - MAX_AGE)
{
self.points.pop_front();
}
}
pub fn since(&self, from: f64) -> impl Iterator<Item = (f64, f64)> + '_ {
let i = self.points.partition_point(|p| p.0 < from);
self.points.range(i.saturating_sub(1)..).copied()
}
#[cfg(test)]
fn len(&self) -> usize {
self.points.len()
}
}
#[derive(Default)]
struct Thinned {
dt: f64,
from: f64,
seen: u64,
done: VecDeque<(f64, f64)>,
open: Option<Column>,
}
#[derive(Clone, Copy)]
struct Column {
index: i64,
first: (f64, f64),
low: (f64, f64),
high: (f64, f64),
last: (f64, f64),
}
impl Column {
fn points(&self) -> impl Iterator<Item = (f64, f64)> {
let mut pts = [self.first, self.low, self.high, self.last];
pts.sort_by(|a, b| a.0.total_cmp(&b.0));
let mut n = 0;
for i in 0..pts.len() {
if n == 0 || pts[i] != pts[n - 1] {
pts[n] = pts[i];
n += 1;
}
}
pts.into_iter().take(n)
}
}
impl Thinned {
fn new(dt: f64, from: f64) -> Self {
Self {
dt,
from,
..Default::default()
}
}
fn close(&mut self) {
if let Some(c) = self.open.take() {
self.done.extend(c.points());
}
}
fn add(&mut self, (t, v): (f64, f64)) {
if v.is_nan() {
self.close();
self.done.push_back((t, v));
return;
}
let index = (t / self.dt).floor() as i64;
match &mut self.open {
Some(c) if c.index == index => {
if v < c.low.1 {
c.low = (t, v);
}
if v > c.high.1 {
c.high = (t, v);
}
c.last = (t, v);
}
_ => {
self.close();
self.open = Some(Column {
index,
first: (t, v),
low: (t, v),
high: (t, v),
last: (t, v),
});
}
}
}
fn update(&mut self, samples: &Samples, from: f64, dt: f64) {
let new = samples.pushed - self.seen;
if self.dt != dt || from < self.from || new > samples.points.len() as u64 {
*self = Self::new(dt, from);
for p in samples.since(from) {
self.add(p);
}
} else {
let len = samples.points.len();
for p in samples.points.range(len - new as usize..) {
self.add(*p);
}
}
self.seen = samples.pushed;
let i = self.done.partition_point(|p| p.0 < from);
if i > 1 {
self.done.drain(..i - 1);
}
}
fn points(&self) -> impl Iterator<Item = (f64, f64)> + '_ {
self.done
.iter()
.copied()
.chain(self.open.iter().flat_map(Column::points))
}
}
pub const MIN_SCROLL_REPAINT: f64 = 0.25;
fn scroll_repaint(span: f64, width: f32) -> std::time::Duration {
let per_pixel = span / width.max(1.0) as f64;
std::time::Duration::from_secs_f64(per_pixel.clamp(MIN_SCROLL_REPAINT, 1.0))
}
pub fn now() -> f64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0.0, |d| d.as_secs_f64())
}
fn sample(s: &PvState) -> Option<f64> {
if s.connected {
s.value.as_f64()
} else {
Some(f64::NAN)
}
}
pub struct Series {
pub pv: Pv,
samples: Arc<Mutex<Samples>>,
thinned: Mutex<Thinned>,
_listener: Listener,
}
impl Series {
pub fn new(pv: Pv) -> Self {
let samples = Arc::new(Mutex::new(Samples::default()));
let into = samples.clone();
let listener: Listener = Arc::new(move |s: &PvState| {
if let Some(v) = sample(s) {
into.lock().push(now(), v);
}
});
pv.listen(&listener);
Self {
pv,
samples,
thinned: Mutex::default(),
_listener: listener,
}
}
fn points(&self, from: f64, to: f64, dt: f64) -> Vec<(f64, f64)> {
let mut pts: Vec<(f64, f64)> = {
let mut thinned = self.thinned.lock();
thinned.update(&self.samples.lock(), from, dt);
thinned.points().collect()
};
if let Some(&(t, v)) = pts.last()
&& v.is_finite()
&& t < to
&& self.pv.state().connected
{
pts.push((to, v));
}
pts
}
}
pub struct Line<'a> {
pub series: &'a Series,
pub color: Color32,
pub width: f32,
}
pub struct Look {
pub log: bool,
pub range: Option<(f64, f64)>,
pub title: String,
pub grid: bool,
pub fg: Color32,
pub back: Color32,
pub font: FontId,
pub title_font: FontId,
}
pub fn draw_plot(
painter: &egui::Painter,
rect: Rect,
lines: &[Line],
span: f64,
now: f64,
look: &Look,
z: f32,
) {
let painter = &painter.with_clip_rect(painter.clip_rect().intersect(rect));
fill(painter, rect, look.back);
let (pad, tick) = (4.0 * z, 5.0 * z);
let from = now - span;
let dt = span / rect.width().max(1.0) as f64;
let data: Vec<Vec<(f64, f64)>> = lines
.iter()
.map(|l| l.series.points(from, now, dt))
.collect();
let ok = |v: f64| v.is_finite() && (!look.log || v > 0.0);
let in_view = data
.iter()
.flat_map(|pts| pts.iter().filter(|p| p.0 >= from).map(|p| p.1));
let (lo, hi) = look
.range
.or_else(|| plot::autoscale(look.log, in_view, true))
.unwrap_or(if look.log { (1.0, 10.0) } else { (0.0, 1.0) });
let label_h = text_size(painter, "0", &look.font).y;
let bottom = rect.bottom() - pad - label_h - tick;
let top = rect.top() + pad + label_h / 2.0;
let ys = Scale {
lo,
hi,
log: look.log,
from: bottom,
to: top,
};
let (y_ticks, y_step) = ys.ticks(label_h * 2.5);
let y_labels: Vec<(f64, String)> = y_ticks
.iter()
.map(|v| (*v, axis::label(*v, y_step)))
.collect();
let label_w = y_labels
.iter()
.map(|(_, l)| text_size(painter, l, &look.font).x)
.fold(0.0, f32::max);
let title_w = if look.title.is_empty() {
0.0
} else {
text_size(painter, &look.title, &look.title_font).y + pad
};
let time_w = text_size(painter, "00:00:00", &look.font).x;
let left = rect.left() + pad + title_w + label_w + pad / 2.0 + tick;
let right = rect.right() - pad - time_w / 2.0;
let plot = Rect::from_min_max(Pos2::new(left, top), Pos2::new(right, bottom));
if plot.width() < 4.0 || plot.height() < 4.0 {
return;
}
let xs = Scale {
lo: from,
hi: now,
log: false,
from: plot.left(),
to: plot.right(),
};
let max_ticks = (plot.width() / (time_w + 20.0 * z)).floor().max(2.0) as usize;
let (t_ticks, t_step) = time_ticks(from, now, max_ticks, local_offset(now));
if look.grid {
let grid = Stroke::new(1.0, named("Grid").gamma_multiply(0.4));
for t in &t_ticks {
let x = xs.pos(*t);
painter.line_segment(
[Pos2::new(x, plot.top()), Pos2::new(x, plot.bottom())],
grid,
);
}
for (v, _) in &y_labels {
let y = ys.pos(*v);
painter.line_segment(
[Pos2::new(plot.left(), y), Pos2::new(plot.right(), y)],
grid,
);
}
}
let clipped = painter.with_clip_rect(painter.clip_rect().intersect(plot));
let bounds = plot.expand(plot.width().max(plot.height()) * 4.0);
for (line, pts) in lines.iter().zip(&data) {
let screen = |(t, v): (f64, f64)| {
ok(v).then(|| {
Pos2::new(
xs.pos(t).clamp(bounds.left(), bounds.right()),
ys.pos(v).clamp(bounds.top(), bounds.bottom()),
)
})
};
let pts: Vec<Option<Pos2>> = pts.iter().map(|p| screen(*p)).collect();
for run in pts.split(Option::is_none).filter(|r| !r.is_empty()) {
let mut steps: Vec<Pos2> = Vec::with_capacity(run.len() * 2);
for p in run.iter().flatten() {
if let Some(last) = steps.last().copied() {
steps.push(Pos2::new(p.x, last.y));
}
steps.push(*p);
}
plot::line(
&clipped,
plot::decimate(steps),
Stroke::new(line.width, line.color),
0,
);
}
}
let stroke = Stroke::new(z.max(1.0), look.fg);
painter.rect_stroke(plot, 0.0, stroke, egui::StrokeKind::Middle);
for t in &t_ticks {
let x = xs.pos(*t);
painter.line_segment(
[
Pos2::new(x, plot.bottom()),
Pos2::new(x, plot.bottom() + tick),
],
stroke,
);
painter.text(
Pos2::new(x, plot.bottom() + tick),
Align2::CENTER_TOP,
time_label(*t, t_step),
look.font.clone(),
look.fg,
);
}
for (v, label) in &y_labels {
let y = ys.pos(*v);
painter.line_segment(
[Pos2::new(plot.left() - tick, y), Pos2::new(plot.left(), y)],
stroke,
);
painter.text(
Pos2::new(plot.left() - tick - pad / 2.0, y),
Align2::RIGHT_CENTER,
label,
look.font.clone(),
look.fg,
);
}
if !look.title.is_empty() {
let h = title_w - pad;
vertical_text(
painter,
Pos2::new(rect.left() + pad + h / 2.0, plot.center().y),
&look.title,
look.title_font.clone(),
look.fg,
);
}
}
const TIME_STEPS: &[f64] = &[
1.0, 2.0, 5.0, 10.0, 15.0, 30.0, 60.0, 120.0, 300.0, 600.0, 900.0, 1800.0, 3600.0, 7200.0,
10_800.0, 21_600.0, 43_200.0, 86_400.0,
];
fn time_ticks(from: f64, to: f64, max_ticks: usize, offset: f64) -> (Vec<f64>, f64) {
let span = to - from;
if span <= 0.0 || !span.is_finite() {
return (Vec::new(), 0.0);
}
let step = TIME_STEPS
.iter()
.copied()
.find(|s| span / s <= max_ticks.max(1) as f64)
.unwrap_or_else(|| 86_400.0 * (span / 86_400.0 / max_ticks.max(1) as f64).ceil());
let first = ((from + offset) / step).ceil() * step - offset;
let ticks = (0..)
.map(|i| first + i as f64 * step)
.take_while(|t| *t <= to)
.take(1000)
.collect();
(ticks, step)
}
fn local_offset(t: f64) -> f64 {
Local
.timestamp_opt(t as i64, 0)
.single()
.map_or(0.0, |d| d.offset().local_minus_utc() as f64)
}
fn time_label(t: f64, step: f64) -> String {
let Some(d) = Local.timestamp_opt(t.round() as i64, 0).single() else {
return String::new();
};
let format = match step {
s if s < 60.0 => "%H:%M:%S",
s if s < 86_400.0 => "%H:%M",
_ => "%d %b",
};
d.format(format).to_string()
}
pub fn parse_span(text: &str) -> Option<f64> {
let mut total = 0.0;
let mut words = text.split_whitespace().peekable();
let mut any = false;
while let Some(word) = words.next() {
let split = word
.find(|c: char| !(c.is_ascii_digit() || c == '.'))
.unwrap_or(word.len());
let (num, unit) = word.split_at(split);
let n: f64 = num.parse().ok()?;
let unit = if unit.is_empty() {
words.next().unwrap_or("s")
} else {
unit
};
let secs = match unit.to_ascii_lowercase().trim_end_matches('s') {
"" | "sec" | "second" => 1.0,
"m" | "min" | "minute" => 60.0,
"h" | "hour" => 3600.0,
"d" | "day" => 86_400.0,
"w" | "week" => 7.0 * 86_400.0,
_ => return None,
};
total += n * secs;
any = true;
}
(any && total > 0.0).then_some(total)
}
pub fn draw_stripchart(painter: &egui::Painter, node: &Node, rect: Rect, z: f32) {
let w = &node.w;
let back = w
.color("background_color")
.map(|c| color(&c))
.unwrap_or(Color32::WHITE);
fill(painter, rect, back);
let fg = w
.color("foreground_color")
.map(|c| color(&c))
.unwrap_or(Color32::BLACK);
let pad = 4.0 * z;
let top = plot::title(painter, node, rect, fg, z);
let traces = plot::traces(node, z);
let shown: Vec<(&plot::Trace, &Series)> = traces
.iter()
.zip(&node.series)
.filter(|(t, _)| t.visible)
.filter_map(|(t, s)| Some((t, s.as_ref()?)))
.collect();
let legend_font = style::font_id(&w.font("legend_font").unwrap_or_default(), z);
let legend_h = if w.bool("show_legend").unwrap_or(true) && !shown.is_empty() {
text_size(painter, "Xg", &legend_font).y + pad
} else {
0.0
};
if legend_h > 0.0 {
let named: Vec<&plot::Trace> = shown.iter().map(|(t, _)| *t).collect();
plot::legend(painter, &named, rect, legend_h, pad, &legend_font, fg, z);
}
let axis = plot::axis_cfg(
w.get("y_axes").and_then(|e| e.child("y_axis")),
"Value",
node,
z,
);
let span = w
.text("time_range")
.map(|t| node.expand(t))
.and_then(|t| parse_span(&t))
.unwrap_or(60.0);
let look = Look {
log: axis.log,
range: (!axis.autoscale).then_some(axis.range),
title: if axis.visible {
axis.title.clone()
} else {
String::new()
},
grid: axis.grid,
fg: axis.color,
back,
font: axis.scale_font,
title_font: axis.title_font,
};
let lines: Vec<Line> = shown
.iter()
.map(|(t, s)| Line {
series: s,
color: t.color,
width: t.width,
})
.collect();
let area = Rect::from_min_max(
Pos2::new(rect.left(), top),
Pos2::new(rect.right(), rect.bottom() - legend_h),
);
draw_plot(painter, area, &lines, span, now(), &look, z);
painter
.ctx()
.request_repaint_after(scroll_repaint(span, area.width()));
if let Some(s) = plot::worst_state(node) {
alarm_border(painter, w, &s, rect, z);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn samples_keep_gaps_once_and_stay_in_order() {
let mut s = Samples::default();
s.push(1.0, f64::NAN);
assert_eq!(s.len(), 0, "a trend starts with a value");
s.push(1.0, 5.0);
s.push(2.0, f64::NAN);
s.push(3.0, f64::NAN);
s.push(1.5, 6.0);
let pts: Vec<(f64, f64)> = s.since(0.0).collect();
assert_eq!(pts.len(), 3);
assert!(pts[1].1.is_nan());
assert_eq!(pts[2], (2.0, 6.0));
}
#[test]
fn samples_are_bounded_by_count_and_age() {
let mut s = Samples::default();
for i in 0..MAX_SAMPLES + 10 {
s.push(i as f64 * 0.001, i as f64);
}
assert_eq!(s.len(), MAX_SAMPLES);
assert_eq!(s.since(0.0).next().map(|p| p.1), Some(10.0));
s.push(MAX_AGE + 200.0, 1.0);
assert_eq!(s.len(), 1);
}
#[test]
fn since_includes_the_sample_that_holds_at_the_start() {
let mut s = Samples::default();
for t in [10.0, 20.0, 30.0, 40.0] {
s.push(t, t);
}
let ts: Vec<f64> = s.since(25.0).map(|p| p.0).collect();
assert_eq!(ts, [20.0, 30.0, 40.0]);
assert_eq!(s.since(50.0).count(), 1);
assert_eq!(s.since(0.0).count(), 4);
}
#[test]
fn series_sample_every_update() {
let pvs = pv::PvManager::new();
let pv = pvs.subscribe("loc://trend_test(1)");
let series = Series::new(pv.clone());
pv.write("2");
pv.write("3");
let samples: Vec<f64> = series.samples.lock().since(0.0).map(|p| p.1).collect();
assert_eq!(samples, [1.0, 2.0, 3.0]);
let values: Vec<f64> = series
.points(0.0, now() + 1.0, 1.0)
.iter()
.map(|p| p.1)
.collect();
assert_eq!(values.first(), Some(&1.0));
assert_eq!(values[values.len() - 2..], [3.0, 3.0]);
}
fn bits(pts: &[(f64, f64)]) -> Vec<(u64, u64)> {
pts.iter().map(|p| (p.0.to_bits(), p.1.to_bits())).collect()
}
fn thinned(s: &Samples, from: f64, dt: f64) -> Vec<(f64, f64)> {
let mut t = Thinned::default();
t.update(s, from, dt);
t.points().collect()
}
#[test]
fn thinning_keeps_each_columns_ends_and_extremes() {
let mut s = Samples::default();
for i in 0..86_400 {
let t = i as f64;
s.push(
t,
(t * 0.01).sin() * 10.0 + if i == 5000 { 100.0 } else { 0.0 },
);
}
let dt = 86.4;
let pts = thinned(&s, 0.0, dt);
assert!(pts.len() <= 4 * 1001, "{} points", pts.len());
assert!(pts.windows(2).all(|w| w[0].0 <= w[1].0), "in time order");
assert!(pts.contains(&(5000.0, s.since(5000.0).nth(1).unwrap().1)));
assert_eq!(pts.first(), s.since(0.0).next().as_ref());
assert_eq!(pts.last().map(|p| p.0), Some(86_399.0));
let max = |p: &mut dyn Iterator<Item = (f64, f64)>| p.map(|p| p.1).fold(f64::MIN, f64::max);
assert_eq!(max(&mut pts.iter().copied()), max(&mut s.since(0.0)));
}
#[test]
fn thinning_takes_in_only_new_samples_and_keeps_gaps() {
let mut s = Samples::default();
let mut t = Thinned::default();
let push = |s: &mut Samples, t0: f64, n: usize| {
for i in 0..n {
let x = t0 + i as f64 * 0.05;
s.push(x, (x * 3.0).cos());
}
};
push(&mut s, 0.0, 500);
t.update(&s, 0.0, 0.5);
s.push(25.0, f64::NAN);
push(&mut s, 26.0, 500);
t.update(&s, 0.0, 0.5);
assert_eq!(t.seen, s.pushed);
let all: Vec<(f64, f64)> = t.points().collect();
assert_eq!(bits(&all), bits(&thinned(&s, 0.0, 0.5)));
assert_eq!(all.iter().filter(|p| p.1.is_nan()).count(), 1, "the gap");
t.update(&s, 40.0, 0.5);
let later: Vec<(f64, f64)> = t.points().collect();
assert!(later[0].0 < 40.0 && later[1].0 >= 40.0);
t.update(&s, 40.0, 0.25);
let rebuilt: Vec<(f64, f64)> = t.points().collect();
assert_eq!(bits(&rebuilt), bits(&thinned(&s, 40.0, 0.25)));
}
#[test]
fn plots_scroll_a_pixel_at_a_time_but_not_too_often() {
let secs = |d: std::time::Duration| d.as_secs_f64();
assert_eq!(secs(scroll_repaint(3600.0, 600.0)), 1.0);
assert_eq!(secs(scroll_repaint(300.0, 600.0)), 0.5);
assert_eq!(secs(scroll_repaint(60.0, 600.0)), MIN_SCROLL_REPAINT);
assert_eq!(secs(scroll_repaint(60.0, 0.0)), 1.0);
}
#[test]
fn disconnected_pvs_leave_a_gap() {
let off = PvState::default();
assert!(sample(&off).unwrap().is_nan());
let on = PvState {
connected: true,
value: pv::Value::Double(2.0),
..Default::default()
};
assert_eq!(sample(&on), Some(2.0));
let text = PvState {
connected: true,
value: pv::Value::Str("idle".into()),
..Default::default()
};
assert_eq!(sample(&text), None);
}
#[test]
fn time_ticks_fall_on_whole_local_steps() {
let (t, step) = time_ticks(1000.0, 1600.0, 6, 0.0);
assert_eq!(step, 120.0);
assert_eq!(t, [1080.0, 1200.0, 1320.0, 1440.0, 1560.0]);
let (t, step) = time_ticks(0.0, 4.0 * 3600.0, 4, 1800.0);
assert_eq!(step, 3600.0);
assert_eq!(t[0], 1800.0);
let (_, step) = time_ticks(0.0, 30.0 * 86_400.0, 5, 0.0);
assert_eq!(step, 6.0 * 86_400.0);
assert!(time_ticks(5.0, 5.0, 5, 0.0).0.is_empty());
}
#[test]
fn time_ranges_parse() {
assert_eq!(parse_span("1 minute"), Some(60.0));
assert_eq!(parse_span("10 min"), Some(600.0));
assert_eq!(parse_span("1 hour 30 minutes"), Some(5400.0));
assert_eq!(parse_span("30s"), Some(30.0));
assert_eq!(parse_span("2 days"), Some(172_800.0));
assert_eq!(parse_span("soon"), None);
assert_eq!(parse_span(""), None);
}
}