Introduce Scalar, SeriesLine, and SeriesPoint archetypes with their own visualizers

crates/re_space_view/src/data_query_blueprint.rs

@@ -10,7 +10,7 @@ use re_entity_db::{
 use re_log_types::{
     path::RuleEffect, DataRow, EntityPath, EntityPathFilter, EntityPathRule, RowId, StoreKind,
 };
-use re_types_core::{archetypes::Clear, ComponentName};
+use re_types_core::{archetypes::Clear, components::VisualizerOverrides, ComponentName};
 use re_viewer_context::{
     blueprint_timepoint_for_writes, DataQueryId, DataQueryResult, DataResult, DataResultHandle,
     DataResultNode, DataResultTree, IndicatedEntities, PerVisualizer, PropertyOverrides,
@@ -254,6 +254,9 @@ impl<'a> QueryExpressionEvaluator<'a> {
 
         // Only populate visualizers if this is a match
         // Note that allowed prefixes that aren't matches can still create groups
+        // TODO(jleibs): It would be nice to lookup the override queries here, but we don't have
+        // access to a query context. Also the entity-override-path-joining is expensive and we don't want
+        // to do it during heuristic evaluation.
         let visualizers: SmallVec<_> = if any_match {
             self.visualizable_entities_for_visualizer_systems
                 .iter()
@@ -467,6 +470,20 @@ impl DataQueryPropertyResolver<'_> {
                         .individual_override_root
                         .join(&node.data_result.entity_path);
 
+                    // If the user has overridden the visualizers, update which visualizers are used.
+                    {
+                        re_tracing::profile_scope!("Update visualizers from overrides");
+
+                        if let Some(viz_override) = ctx
+                            .blueprint
+                            .store()
+                            .query_latest_component::<VisualizerOverrides>(&override_path, query)
+                            .map(|c| c.value)
+                        {
+                            node.data_result.visualizers =
+                                viz_override.0.iter().map(|v| v.as_str().into()).collect();
+                        }
+                    }
                     let mut component_overrides: HashMap<ComponentName, (StoreKind, EntityPath)> =
                         Default::default();
 

crates/re_space_view_time_series/src/aggregation.rs

@@ -0,0 +1,202 @@
+use crate::{PlotPoint, PlotPointAttrs};
+
+/// Implements aggregation behavior corresponding to [`TimeSeriesAggregator::Average`].
+pub struct AverageAggregator;
+
+impl AverageAggregator {
+    #[inline]
+    pub fn aggregate(aggregation_factor: f64, points: &[PlotPoint]) -> Vec<PlotPoint> {
+        let min_time = points.first().map_or(i64::MIN, |p| p.time);
+        let max_time = points.last().map_or(i64::MAX, |p| p.time);
+
+        let mut aggregated =
+            Vec::with_capacity((points.len() as f64 / aggregation_factor) as usize);
+
+        // NOTE: `floor()` since we handle fractional tails separately.
+        let window_size = usize::max(1, aggregation_factor.floor() as usize);
+        let aggregation_factor_fract = aggregation_factor.fract();
+
+        let mut i = 0;
+        while i < points.len() {
+            let mut j = 0;
+
+            let mut ratio = 1.0;
+            let mut acc = points[i + j].clone();
+            j += 1;
+
+            while j < window_size
+                && i + j < points.len()
+                && are_aggregatable(&points[i], &points[i + j], window_size)
+            {
+                let point = &points[i + j];
+
+                acc.value += point.value;
+                acc.attrs.radius += point.attrs.radius;
+
+                ratio += 1.0;
+                j += 1;
+            }
+
+            // Do a weighted average for the fractional tail.
+            if aggregation_factor_fract > 0.0
+                && i + j < points.len()
+                && are_aggregatable(&points[i], &points[i + j], window_size)
+            {
+                let point = &points[i + j];
+
+                let w = aggregation_factor_fract;
+                acc.value += point.value * w;
+                acc.attrs.radius += (point.attrs.radius as f64 * w) as f32;
+
+                ratio += aggregation_factor_fract;
+                j += 1;
+            }
+
+            acc.value /= ratio;
+            acc.attrs.radius = (acc.attrs.radius as f64 / ratio) as _;
+
+            aggregated.push(acc);
+
+            i += j;
+        }
+
+        // Force align the start and end timestamps to prevent jarring visual glitches.
+        if let Some(p) = aggregated.first_mut() {
+            p.time = min_time;
+        }
+        if let Some(p) = aggregated.last_mut() {
+            p.time = max_time;
+        }
+
+        aggregated
+    }
+}
+
+/// Implements aggregation behaviors corresponding to [`TimeSeriesAggregator::Max`],
+/// [`TimeSeriesAggregator::Min`], [`TimeSeriesAggregator::MinMax`] and
+/// [`TimeSeriesAggregator::MinMaxAverage`], .
+pub enum MinMaxAggregator {
+    /// Keep only the maximum values in the range.
+    Max,
+
+    /// Keep only the minimum values in the range.
+    Min,
+
+    /// Keep both the minimum and maximum values in the range.
+    ///
+    /// This will yield two aggregated points instead of one, effectively creating a vertical line.
+    MinMax,
+
+    /// Find both the minimum and maximum values in the range, then use the average of those.
+    MinMaxAverage,
+}
+
+impl MinMaxAggregator {
+    #[inline]
+    pub fn aggregate(&self, aggregation_factor: f64, points: &[PlotPoint]) -> Vec<PlotPoint> {
+        let min_time = points.first().map_or(i64::MIN, |p| p.time);
+        let max_time = points.last().map_or(i64::MAX, |p| p.time);
+
+        let capacity = (points.len() as f64 / aggregation_factor) as usize;
+        let mut aggregated = match self {
+            MinMaxAggregator::MinMax => Vec::with_capacity(capacity * 2),
+            _ => Vec::with_capacity(capacity),
+        };
+
+        // NOTE: `round()` since this can only handle discrete window sizes.
+        let window_size = usize::max(1, aggregation_factor.round() as usize);
+
+        let mut i = 0;
+        while i < points.len() {
+            let mut j = 0;
+
+            let mut acc_min = points[i + j].clone();
+            let mut acc_max = points[i + j].clone();
+            j += 1;
+
+            while j < window_size
+                && i + j < points.len()
+                && are_aggregatable(&points[i], &points[i + j], window_size)
+            {
+                let point = &points[i + j];
+
+                match self {
+                    MinMaxAggregator::MinMax | MinMaxAggregator::MinMaxAverage => {
+                        acc_min.value = f64::min(acc_min.value, point.value);
+                        acc_min.attrs.radius = f32::min(acc_min.attrs.radius, point.attrs.radius);
+                        acc_max.value = f64::max(acc_max.value, point.value);
+                        acc_max.attrs.radius = f32::max(acc_max.attrs.radius, point.attrs.radius);
+                    }
+                    MinMaxAggregator::Min => {
+                        acc_min.value = f64::min(acc_min.value, point.value);
+                        acc_min.attrs.radius = f32::min(acc_min.attrs.radius, point.attrs.radius);
+                    }
+                    MinMaxAggregator::Max => {
+                        acc_max.value = f64::max(acc_max.value, point.value);
+                        acc_max.attrs.radius = f32::max(acc_max.attrs.radius, point.attrs.radius);
+                    }
+                }
+
+                j += 1;
+            }
+
+            match self {
+                MinMaxAggregator::MinMax => {
+                    aggregated.push(acc_min);
+                    // Don't push the same point twice.
+                    if j > 1 {
+                        aggregated.push(acc_max);
+                    }
+                }
+                MinMaxAggregator::MinMaxAverage => {
+                    // Don't average a single point with itself.
+                    if j > 1 {
+                        acc_min.value = (acc_min.value + acc_max.value) * 0.5;
+                        acc_min.attrs.radius = (acc_min.attrs.radius + acc_max.attrs.radius) * 0.5;
+                    }
+                    aggregated.push(acc_min);
+                }
+                MinMaxAggregator::Min => {
+                    aggregated.push(acc_min);
+                }
+                MinMaxAggregator::Max => {
+                    aggregated.push(acc_max);
+                }
+            }
+
+            i += j;
+        }
+
+        // Force align the start and end timestamps to prevent jarring visual glitches.
+        if let Some(p) = aggregated.first_mut() {
+            p.time = min_time;
+        }
+        if let Some(p) = aggregated.last_mut() {
+            p.time = max_time;
+        }
+
+        aggregated
+    }
+}
+
+/// Are two [`PlotPoint`]s safe to aggregate?
+fn are_aggregatable(point1: &PlotPoint, point2: &PlotPoint, window_size: usize) -> bool {
+    let PlotPoint {
+        time,
+        value: _,
+        attrs,
+    } = point1;
+    let PlotPointAttrs {
+        label,
+        color,
+        radius: _,
+        kind,
+    } = attrs;
+
+    // We cannot aggregate two points that don't live in the same aggregation window to start with.
+    // This is very common with e.g. sparse datasets.
+    time.abs_diff(point2.time) <= window_size as u64
+        && *label == point2.attrs.label
+        && *color == point2.attrs.color
+        && *kind == point2.attrs.kind
+}

crates/re_space_view_time_series/src/lib.rs

@@ -2,9 +2,16 @@
 //!
 //! A Space View that shows plots over Rerun timelines.
 
+mod aggregation;
+mod legacy_visualizer_system;
+mod line_visualizer_system;
+mod overrides;
+mod point_visualizer_system;
 mod space_view_class;
-mod visualizer_system;
+mod util;
 
+use re_log_types::EntityPath;
+use re_viewer_context::external::re_entity_db::TimeSeriesAggregator;
 pub use space_view_class::TimeSeriesSpaceView;
 
 /// Computes a deterministic, globally unique ID for the plot based on the ID of the space view
@@ -18,3 +25,63 @@ pub use space_view_class::TimeSeriesSpaceView;
 pub(crate) fn plot_id(space_view_id: re_viewer_context::SpaceViewId) -> egui::Id {
     egui::Id::new(("plot", space_view_id))
 }
+
+// ---
+
+#[derive(Clone, Debug)]
+pub struct PlotPointAttrs {
+    pub label: Option<String>,
+    pub color: egui::Color32,
+    pub radius: f32,
+    pub kind: PlotSeriesKind,
+}
+
+impl PartialEq for PlotPointAttrs {
+    fn eq(&self, rhs: &Self) -> bool {
+        let Self {
+            label,
+            color,
+            radius,
+            kind,
+        } = self;
+        label.eq(&rhs.label)
+            && color.eq(&rhs.color)
+            && radius.total_cmp(&rhs.radius).is_eq()
+            && kind.eq(&rhs.kind)
+    }
+}
+
+impl Eq for PlotPointAttrs {}
+
+#[derive(Clone, Debug)]
+struct PlotPoint {
+    time: i64,
+    value: f64,
+    attrs: PlotPointAttrs,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum PlotSeriesKind {
+    Continuous,
+    Scatter,
+    Clear,
+}
+
+#[derive(Clone, Debug)]
+pub struct PlotSeries {
+    pub label: String,
+    pub color: egui::Color32,
+    pub width: f32,
+    pub kind: PlotSeriesKind,
+    pub points: Vec<(i64, f64)>,
+    pub entity_path: EntityPath,
+
+    /// Earliest time an entity was recorded at on the current timeline.
+    pub min_time: i64,
+
+    /// What kind of aggregation was used to compute the graph?
+    pub aggregator: TimeSeriesAggregator,
+
+    /// `1.0` for raw data.
+    pub aggregation_factor: f64,
+}