Customizable resampling functions (#29)

This PR adds configurable resampling behavior to the logical meter,
allowing callers to change the default resampling function and override
it per metric or per (component_id, metric) pair.

**Changes:**
- Add `LogicalMeterConfig` builder-style API for default resampling
function and per-metric/per-component overrides.
- Update `LogicalMeterActor` to select the resampling function using
override precedence (component-specific → metric-only → default →
average).
- Update examples/tests and bump `frequenz-resampling` dependency to
`0.2`.

Author: shsms

Cargo.toml

@@ -15,7 +15,7 @@ async-trait = "0.1.89"
 chrono = "0.4"
 frequenz-microgrid-component-graph = "0.2.0"
 frequenz-microgrid-formula-engine = "0.1.0"
-frequenz-resampling = "0.1.0"
+frequenz-resampling = "0.2"
 futures = "0.3.31"
 prost = "0.14"
 tokio = { version = "1.48", features = ["rt", "rt-multi-thread"] }

RELEASE_NOTES.md

@@ -2,16 +2,16 @@
 
 ## Summary
 
-This release makes incremental improvements to the client and the quantity types.
+<!-- Here goes a general summary of what this release is about -->
 
 ## Upgrading
 
-- The `MicrogridClientHandle::list_electrical_components` method now expects `ElectricalComponentCategory` enum values instead of `i32`, to filter by component category.
+- `LogicalMeterConfig` instances can't be created directly anymore, and need to be created using the `LogicalMeterConfig::new` method.  This helps avoid future breaking changes, as we add more config parameters.
 
 ## New Features
 
-- The new `MicrogridClientHandle::augment_electrical_component_bounds` method can be used to augment the bounds for specific metrics of electrical components.
+- It is now possible to change the default resampling function, and to override the resampling function for specific metrics.
 
-- All methods on `Quantity` types are now `const`.
+## Bug Fixes
 
-- `Quantity` types have two new methods `min` and `max`, similar to the `min` and `max` methods on fundamental numerical types.
+<!-- Here goes notable bug fixes that are worth a special mention or explanation -->

examples/logical_meter.rs

@@ -21,9 +21,7 @@ async fn main() -> Result<(), Error> {
     let client = MicrogridClientHandle::try_new("http://[::1]:8800").await?;
     let mut logical_meter = LogicalMeterHandle::try_new(
         client,
-        LogicalMeterConfig {
-            resampling_interval: TimeDelta::try_seconds(1).unwrap(),
-        },
+        LogicalMeterConfig::new(TimeDelta::try_seconds(1).unwrap()),
     )
     .await?;
 

src/logical_meter/config.rs

@@ -3,9 +3,56 @@
 
 //! This module defines the configuration for the logical meter.
 
+use crate::Sample;
+use crate::proto::common::metrics::Metric;
 use chrono::TimeDelta;
+use frequenz_resampling::ResamplingFunction;
+use std::collections::HashMap;
 
 pub struct LogicalMeterConfig {
     /// The resampling interval for the logical meter.
-    pub resampling_interval: TimeDelta,
+    pub(crate) resampling_interval: TimeDelta,
+    /// Resampler function.
+    pub(crate) resampling_function: Option<ResamplingFunction<f32, Sample<f32>>>,
+    /// Resampler overrides.
+    pub(crate) resampling_overrides: HashMap<Metric, ResamplingFunction<f32, Sample<f32>>>,
+}
+
+impl LogicalMeterConfig {
+    /// Creates a new `LogicalMeterConfig` with the given resampling interval.
+    pub fn new(resampling_interval: TimeDelta) -> Self {
+        Self {
+            resampling_interval,
+            resampling_function: None,
+            resampling_overrides: HashMap::new(),
+        }
+    }
+
+    /// Sets the default resampling function.
+    ///
+    /// This function will be used for all metrics that do not have a specific
+    /// override set.
+    ///
+    /// If no default resampling function is set, the logical meter will default
+    /// to using the `Average` resampling function.
+    pub fn with_default_resampling_function(
+        mut self,
+        function: ResamplingFunction<f32, Sample<f32>>,
+    ) -> Self {
+        self.resampling_function = Some(function);
+        self
+    }
+
+    /// Sets a resampling function override for a specific metric.
+    ///
+    /// If this function is called multiple times for the same metric, the last
+    /// function provided will be used.
+    pub fn override_resampling_function<M: crate::metric::Metric>(
+        mut self,
+        function: ResamplingFunction<f32, Sample<f32>>,
+    ) -> Self {
+        self.resampling_overrides.insert(M::METRIC, function);
+
+        self
+    }
 }

src/logical_meter/logical_meter_actor.rs

@@ -373,7 +373,16 @@ impl LogicalMeterActor {
                 metric,
                 resampler: frequenz_resampling::Resampler::new(
                     self.config.resampling_interval,
-                    ResamplingFunction::Average,
+                    self.config
+                        // Look for a specific metric override first
+                        .resampling_overrides
+                        .get(&metric)
+                        .cloned()
+                        // Then look for a configured default
+                        .or_else(|| self.config.resampling_function.clone())
+                        // Finally, default to average if no default is
+                        // configured
+                        .unwrap_or(ResamplingFunction::Average),
                     3,
                     self.resampler_ts,
                     false,

src/logical_meter/logical_meter_handle.rs

@@ -186,6 +186,7 @@ impl LogicalMeterHandle {
 #[cfg(test)]
 mod tests {
     use chrono::TimeDelta;
+    use frequenz_resampling::ResamplingFunction;
     use tokio_stream::{StreamExt, wrappers::BroadcastStream};
 
     use crate::{
@@ -198,7 +199,7 @@ mod tests {
         quantity::Quantity,
     };
 
-    async fn new_logical_meter_handle() -> LogicalMeterHandle {
+    async fn new_logical_meter_handle(config: Option<LogicalMeterConfig>) -> LogicalMeterHandle {
         let api_client = MockMicrogridApiClient::new(
             // Grid connection point
             MockComponent::grid(1).with_children(vec![
@@ -251,17 +252,15 @@ mod tests {
 
         LogicalMeterHandle::try_new(
             MicrogridClientHandle::new_from_client(api_client),
-            LogicalMeterConfig {
-                resampling_interval: TimeDelta::try_seconds(1).unwrap(),
-            },
+            config.unwrap_or_else(|| LogicalMeterConfig::new(TimeDelta::try_seconds(1).unwrap())),
         )
         .await
         .unwrap()
     }
 
     #[tokio::test]
     async fn test_formula_display() {
-        let mut lm = new_logical_meter_handle().await;
+        let mut lm = new_logical_meter_handle(None).await;
 
         let formula = lm.grid(crate::metric::AcPowerActive).unwrap();
         assert_eq!(formula.to_string(), "METRIC_AC_POWER_ACTIVE::(#2)");
@@ -342,7 +341,7 @@ mod tests {
 
     #[tokio::test(start_paused = true)]
     async fn test_grid_power_formula() {
-        let formula = new_logical_meter_handle()
+        let formula = new_logical_meter_handle(None)
             .await
             .grid(crate::metric::AcPowerActive)
             .unwrap();
@@ -352,6 +351,7 @@ mod tests {
         check_samples(
             samples,
             |q| q.as_watts(),
+            TimeDelta::try_seconds(1).unwrap(),
             vec![
                 Some(5.8),
                 Some(6.0),
@@ -369,7 +369,7 @@ mod tests {
 
     #[tokio::test(start_paused = true)]
     async fn test_pv_reactive_power_formula() {
-        let formula = new_logical_meter_handle()
+        let formula = new_logical_meter_handle(None)
             .await
             .pv(None, crate::metric::AcPowerReactive)
             .unwrap();
@@ -379,6 +379,7 @@ mod tests {
         check_samples(
             samples,
             |q| q.as_volt_amperes_reactive(),
+            TimeDelta::try_seconds(1).unwrap(),
             vec![
                 Some(-1.4),
                 Some(-0.5),
@@ -396,7 +397,7 @@ mod tests {
 
     #[tokio::test(start_paused = true)]
     async fn test_battery_voltage_formula() {
-        let formula = new_logical_meter_handle()
+        let formula = new_logical_meter_handle(None)
             .await
             .battery(None, crate::metric::AcVoltage)
             .unwrap();
@@ -405,6 +406,7 @@ mod tests {
         check_samples(
             samples,
             |q| q.as_volts(),
+            TimeDelta::try_seconds(1).unwrap(),
             vec![
                 Some(398.0),
                 Some(397.67),
@@ -420,9 +422,60 @@ mod tests {
         )
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn test_resampling_functions() {
+        let lm_config = Some(
+            LogicalMeterConfig::new(TimeDelta::try_milliseconds(200).unwrap())
+                .with_default_resampling_function(ResamplingFunction::Count)
+                .override_resampling_function::<crate::metric::AcVoltage>(ResamplingFunction::Last),
+        );
+        let mut lm = new_logical_meter_handle(lm_config).await;
+        let bat_volt_formula = lm.battery(None, crate::metric::AcVoltage).unwrap();
+
+        let samples = fetch_samples(bat_volt_formula, 10).await;
+        check_samples(
+            samples,
+            |q| q.as_volts(),
+            TimeDelta::try_milliseconds(200).unwrap(),
+            vec![
+                Some(400.0),
+                Some(400.0),
+                Some(398.0),
+                Some(396.0),
+                Some(396.0),
+                Some(396.0),
+                Some(396.0),
+                Some(396.0),
+                None,
+                None,
+            ],
+        );
+
+        let cons_pow_formula = lm.consumer(crate::metric::AcPowerActive).unwrap();
+
+        let samples = fetch_samples(cons_pow_formula, 10).await;
+        check_samples(
+            samples,
+            |q| q.as_watts(),
+            TimeDelta::try_milliseconds(200).unwrap(),
+            vec![
+                Some(1.0),
+                Some(2.0),
+                Some(3.0),
+                Some(3.0),
+                Some(3.0),
+                Some(3.0),
+                Some(2.0),
+                Some(1.0),
+                Some(0.0),
+                Some(0.0),
+            ],
+        );
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_consumer_current_formula() {
-        let formula = new_logical_meter_handle()
+        let formula = new_logical_meter_handle(None)
             .await
             .consumer(crate::metric::AcCurrent)
             .unwrap();
@@ -431,6 +484,7 @@ mod tests {
         check_samples(
             samples,
             |q| q.as_amperes(),
+            TimeDelta::try_seconds(1).unwrap(),
             vec![
                 Some(15.0),
                 Some(14.75),
@@ -460,27 +514,30 @@ mod tests {
     fn check_samples<Q: Quantity>(
         samples: Vec<Sample<Q>>,
         extractor: impl Fn(Q) -> f32,
+        expected_interval: TimeDelta,
         expected_values: Vec<Option<f32>>,
     ) {
         let values = samples
             .iter()
             .map(|res| res.value().map(|v| extractor(v)))
             .collect::<Vec<_>>();
 
-        let one_second = TimeDelta::try_seconds(1).unwrap();
-
         samples.as_slice().windows(2).for_each(|w| {
-            assert_eq!(w[1].timestamp() - w[0].timestamp(), one_second);
+            assert_eq!(
+                w[1].timestamp() - w[0].timestamp(),
+                expected_interval,
+                "Samples are not spaced at the expected interval"
+            );
         });
 
-        for (v, ev) in values.iter().zip(expected_values.iter()) {
+        for (id, (v, ev)) in values.iter().zip(expected_values.iter()).enumerate() {
             match (v, ev) {
                 (Some(v), Some(ev)) => assert!(
                     (v - ev).abs() < 0.01,
-                    "expected value {ev:?}, got value {v:?}"
+                    "Item {id} - expected value {ev:?}, got value {v:?}"
                 ),
                 (None, None) => {}
-                _ => panic!("expected value {ev:?}, got value {v:?}"),
+                _ => panic!("Item {id} - expected value {ev:?}, got value {v:?}"),
             }
         }
     }