Evaluate composed formulas in the logical-meter actor

README.md

@@ -58,6 +58,18 @@ cargo add --dev frequenz-microgrid --features test-utils
 - `MicrogridClientHandle`: cloneable low-level gRPC handle with per-stream automatic reconnect.
 - Typed quantities — `Power`, `Current`, `Voltage`, `ReactivePower`, `Energy`, `Frequency`, `Percentage` — with unit conversions explicit at every API surface.
 
+Formulas compose without subscribing. `+`, `-`, `* f32`, `/ f32`, `* Percentage`, `coalesce`, `min`, `max` and `avg` build one expression. The logical meter evaluates it once per resampling tick, so all operands, of any metric, are read from the same tick's resampled values.
+
+```rust , ignore
+use frequenz_microgrid::{metric, quantity::Power};
+
+let logical_meter = microgrid.logical_meter();
+let net = logical_meter.grid::<metric::AcPowerActive>()?
+    - logical_meter.pv::<metric::AcPowerActive>(None)?
+    + Power::from_kilowatts(100.0);
+let mut rx = net.subscribe().await?;
+```
+
 See the [API documentation](https://docs.rs/frequenz-microgrid) for the full surface.
 
 ## Configuring the underlying graph

RELEASE_NOTES.md

@@ -2,13 +2,25 @@
 
 ## Summary
 
-<!-- Here goes a general summary of what this release is about -->
+The logical meter now evaluates formulas once per resampling tick, against one snapshot of component data. Composed formulas no longer subscribe to each operand separately.
 
 ## Upgrading
 
+- `Formula<Q>` is now a struct with one type parameter instead of the `Formula<QOut, QIn1, QIn2>` enum.
+  - The `FormulaSubscriber` trait is gone: `subscribe()` is a method on `Formula`, so remove `use frequenz_microgrid::FormulaSubscriber`. Use `Formula<Q>` where you used `FormulaSubscriber` bounds or trait objects. A type that implemented `FormulaSubscriber` can no longer be used in a formula: subscribe to it on its own and combine the streams yourself.
+  - The enum variants are gone, `Formula::Subscriber`, `Formula::Multiply` and `Formula::Divide` included. Build formulas with the operators and methods instead: a formula can be multiplied or divided by an `f32`, or multiplied by a `Percentage`.
+  - `coalesce`, `min`, `max` and `avg` no longer return `Result`: drop the `?`.
+- `FormulaOperand<Q>`, the operand of `+`, `-`, `coalesce`, `min`, `max` and `avg`, is now public. It holds a formula or a constant of the same quantity.
+  - An external `broadcast::Receiver` can no longer be used as an operand. Subscribe to the formula and combine the streams yourself.
+  - `avg` takes `Vec<impl Into<FormulaOperand<Q>>>`, so an empty list needs its type spelled out, e.g. `Vec::<Formula<Power>>::new()`.
+- A formula that combines formulas from different logical meters can still be built, but its `subscribe()` now fails with `ErrorKind::FormulaEngineError`. Each `Microgrid::try_new` and each `LogicalMeterHandle::try_new` call creates its own logical meter. Clones of a handle share it, and so does a `Microgrid` built from that handle with `new_from_handles`. Subscribe to each formula on its own and combine the streams yourself.
 - `Metric::FormulaType` is removed. No public API returned it; use the `Formula<M::QuantityType>` that the `LogicalMeterHandle` methods return.
+- `Formula`'s `Display` output changed.
+  - The `METRIC_X::(...)` wrapper around each formula from the component graph is gone, and every component carries its metric instead, e.g. `#2:AC_POWER_ACTIVE`.
+  - Parentheses appear only where precedence or left associativity needs them.
+  - Constants render as plain numbers in the quantity's base unit: `5 W` is now `5`, `Power::from_kilowatts(100.0)` is `100000`, and `0.0` is `0`.
 - `ErrorKind::DroppedUnusedFormulas` is removed. No public API returned it, so remove any match arm or comparison on it.
-- In a formula from a `LogicalMeterHandle` method, a NaN or infinite component value or intermediate result is now treated as missing data: `COALESCE` moves past it to its next operand, and `+`, `-`, `MIN` and `MAX` give `None`.
+- A NaN or infinite component value, constant or intermediate result is now treated as missing data. `COALESCE` moves past it to its next operand, `AVG` skips it, and the other operators give `None`.
 - A component-graph formula with more than about 1000 `+` and `-` operators is now rejected: the `LogicalMeterHandle` method that builds it returns `ErrorKind::FormulaEngineError`. This can happen at large sites, for example:
   - for the consumer formula with phantom loads included, at a few hundred metered branches;
   - for a battery or PV formula over a meter with more than about 510 inverters.
@@ -20,7 +32,9 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+- All operands of a composed formula, including operands of different metrics, are evaluated against the same tick's snapshot instead of being lined up by timestamp.
+- `Formula<Q> * Percentage` scales a formula by a percentage.
+- `Formula<Q>` implements `Clone`, and `avg` accepts constants as well as formulas.
 
 ## Bug Fixes
 

examples/logical_meter.rs

@@ -72,8 +72,8 @@ async fn main() -> Result<(), Error> {
         logical_meter.grid::<metric::AcVoltagePhase1N>()?.avg(vec![
             logical_meter.grid::<metric::AcVoltagePhase2N>()?,
             logical_meter.grid::<metric::AcVoltagePhase3N>()?,
-        ])? * 3.0_f32.sqrt(),
-    )?;
+        ]) * 3.0_f32.sqrt(),
+    );
 
     tracing::info!("formula_grid_voltage: {}", formula_grid_voltage);
     let mut grid_voltage_rx = formula_grid_voltage.subscribe().await?;

src/lib.rs

@@ -29,7 +29,7 @@ mod sample;
 pub use sample::Sample;
 
 mod logical_meter;
-pub use logical_meter::{Formula, FormulaSubscriber, LogicalMeterConfig, LogicalMeterHandle};
+pub use logical_meter::{Formula, FormulaOperand, LogicalMeterConfig, LogicalMeterHandle};
 
 // Re-exported so callers can build the config accepted by
 // [`LogicalMeterConfig::with_component_graph_config`] without depending on the

src/logical_meter.rs

@@ -5,7 +5,7 @@
 
 mod config;
 pub(crate) mod formula;
-pub use formula::{Formula, FormulaSubscriber};
+pub use formula::{Formula, FormulaOperand};
 
 mod logical_meter_actor;
 mod logical_meter_handle;

src/logical_meter/formula.rs

@@ -1,31 +1,240 @@
 // License: MIT
 // Copyright © 2025 Frequenz Energy-as-a-Service GmbH
 
-//! Formula module for the logical meter.
+//! Composable formulas over component metrics.
+//!
+//! A [`Formula`] is a typed wrapper around a formula-engine expression whose
+//! component leaves are [`Key`]s. Composing formulas only builds the
+//! expression; [`Formula::subscribe`] hands it to the logical meter,
+//! which evaluates it once per resampling tick.
 
-use async_trait::async_trait;
-mod async_formula;
-pub(crate) mod graph_formula;
-pub use async_formula::Formula;
+use std::marker::PhantomData;
 
 use chrono::{DateTime, Utc};
+use frequenz_microgrid_formula_engine as engine;
+use tokio::sync::{broadcast, mpsc};
 
 use crate::{
-    Error,
-    Sample,
-    logical_meter::logical_meter_actor::FormulaSink,
-    quantity::Quantity, //
+    Error, Sample,
+    client::proto::common::metrics::Metric as MetricPb,
+    logical_meter::logical_meter_actor::{FormulaSink, Instruction},
+    quantity::{Percentage, Quantity},
 };
-use tokio::sync::broadcast;
 
 /// Capacity of the per-subscriber broadcast channel carrying a formula's
 /// samples.
 const FORMULA_STREAM_CHANNEL_CAPACITY: usize = 100;
 
-#[async_trait]
-pub trait FormulaSubscriber: std::fmt::Display + Sync + Send {
-    type QuantityType: Quantity;
-    async fn subscribe(&self) -> Result<broadcast::Receiver<Sample<Self::QuantityType>>, Error>;
+/// A component leaf of a formula expression: one metric of one component.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub(crate) struct Key {
+    /// The metric read from the component.
+    pub metric: MetricPb,
+    /// The component's id.
+    pub component_id: u64,
+}
+
+impl std::fmt::Display for Key {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let metric = self.metric.as_str_name();
+        let metric = metric.strip_prefix("METRIC_").unwrap_or(metric);
+        write!(f, "{}:{}", self.component_id, metric)
+    }
+}
+
+/// A formula over component metrics, evaluated by the logical meter once per
+/// resampling tick.
+///
+/// Formulas compose with `+`, `-`, `* f32`, `/ f32`, `* Percentage`,
+/// [`coalesce`](Self::coalesce), [`min`](Self::min), [`max`](Self::max) and
+/// [`avg`](Self::avg). Composition never fails and never subscribes; only
+/// [`subscribe`](Self::subscribe) does.
+///
+/// All formulas in one expression must come from the same logical meter, that
+/// is one [`LogicalMeterHandle`](crate::LogicalMeterHandle) or its clones;
+/// otherwise [`subscribe`](Self::subscribe) fails.
+#[derive(Clone)]
+pub struct Formula<Q: Quantity> {
+    engine_formula: engine::Formula<f32, Key>,
+    /// The logical meter that evaluates this formula; `None` when its operands
+    /// come from different logical meters.
+    instructions_tx: Option<mpsc::Sender<Instruction>>,
+    _quantity: PhantomData<Q>,
+}
+
+/// Something a formula can be combined with: another formula of the same
+/// quantity, or a constant.
+pub enum FormulaOperand<Q: Quantity> {
+    /// Another formula.
+    Formula(Formula<Q>),
+    /// A constant value.
+    Constant(Q),
+}
+
+impl<Q: Quantity> From<Formula<Q>> for FormulaOperand<Q> {
+    fn from(formula: Formula<Q>) -> Self {
+        FormulaOperand::Formula(formula)
+    }
+}
+
+impl<Q: Quantity> From<Q> for FormulaOperand<Q> {
+    fn from(value: Q) -> Self {
+        FormulaOperand::Constant(value)
+    }
+}
+
+impl<Q: Quantity> Formula<Q> {
+    pub(crate) fn new(
+        engine_formula: engine::Formula<f32, Key>,
+        instructions_tx: mpsc::Sender<Instruction>,
+    ) -> Self {
+        Self {
+            engine_formula,
+            instructions_tx: Some(instructions_tx),
+            _quantity: PhantomData,
+        }
+    }
+
+    /// The expression this formula evaluates.
+    #[cfg(test)]
+    pub(crate) fn engine_formula(&self) -> &engine::Formula<f32, Key> {
+        &self.engine_formula
+    }
+
+    fn map(self, f: impl FnOnce(engine::Formula<f32, Key>) -> engine::Formula<f32, Key>) -> Self {
+        Self {
+            engine_formula: f(self.engine_formula),
+            ..self
+        }
+    }
+
+    /// Returns the operand's expression. A formula operand from another logical
+    /// meter, or one that already mixes meters, clears this formula's logical
+    /// meter, so `subscribe()` fails.
+    fn take_operand(&mut self, operand: impl Into<FormulaOperand<Q>>) -> engine::Formula<f32, Key> {
+        match operand.into() {
+            FormulaOperand::Formula(formula) => {
+                let same_meter = self
+                    .instructions_tx
+                    .as_ref()
+                    .zip(formula.instructions_tx.as_ref())
+                    .is_some_and(|(ours, theirs)| ours.same_channel(theirs));
+                if !same_meter {
+                    self.instructions_tx = None;
+                }
+                formula.engine_formula
+            }
+            FormulaOperand::Constant(value) => engine::Formula::Constant(Some(value.base_value())),
+        }
+    }
+
+    fn combine(
+        mut self,
+        other: impl Into<FormulaOperand<Q>>,
+        build: impl FnOnce(
+            engine::Formula<f32, Key>,
+            engine::Formula<f32, Key>,
+        ) -> engine::Formula<f32, Key>,
+    ) -> Self {
+        let rhs = self.take_operand(other);
+        self.map(|lhs| build(lhs, rhs))
+    }
+
+    /// `COALESCE(self, other)`: the first operand with a value.
+    pub fn coalesce(self, other: impl Into<FormulaOperand<Q>>) -> Self {
+        self.combine(other, engine::Formula::coalesce)
+    }
+
+    /// `MIN(self, other)`. An operand with no value makes the sample `None`.
+    pub fn min(self, other: impl Into<FormulaOperand<Q>>) -> Self {
+        self.combine(other, engine::Formula::min)
+    }
+
+    /// `MAX(self, other)`. An operand with no value makes the sample `None`.
+    pub fn max(self, other: impl Into<FormulaOperand<Q>>) -> Self {
+        self.combine(other, engine::Formula::max)
+    }
+
+    /// `AVG(self, others...)`. Operands with no value are skipped; the sample
+    /// is `None` when no operand has a value.
+    pub fn avg(mut self, others: Vec<impl Into<FormulaOperand<Q>>>) -> Self {
+        let others: Vec<_> = others
+            .into_iter()
+            .map(|other| self.take_operand(other))
+            .collect();
+        self.map(|lhs| lhs.avg(others))
+    }
+}
+
+impl<Q: Quantity + 'static> Formula<Q> {
+    /// Starts streaming this formula's samples.
+    ///
+    /// Each call gets its own channel. Formulas with the same expression share
+    /// one evaluation in the logical meter.
+    ///
+    /// Fails if the formula combines formulas from different logical meters.
+    pub async fn subscribe(&self) -> Result<broadcast::Receiver<Sample<Q>>, Error> {
+        let Some(instructions_tx) = &self.instructions_tx else {
+            return Err(Error::formula_engine_error(
+                "A formula cannot combine formulas from different logical meters",
+            ));
+        };
+        let (tx, rx) = broadcast::channel(FORMULA_STREAM_CHANNEL_CAPACITY);
+        instructions_tx
+            .send(Instruction::SubscribeFormula {
+                engine_formula: self.engine_formula.clone(),
+                sink: Box::new(QuantitySink { tx }),
+            })
+            .await
+            .map_err(|e| Error::connection_failure(format!("Could not send instruction: {e}")))?;
+        Ok(rx)
+    }
+}
+
+impl<Q: Quantity, R: Into<FormulaOperand<Q>>> std::ops::Add<R> for Formula<Q> {
+    type Output = Self;
+
+    fn add(self, rhs: R) -> Self {
+        self.combine(rhs, |lhs, rhs| lhs + rhs)
+    }
+}
+
+impl<Q: Quantity, R: Into<FormulaOperand<Q>>> std::ops::Sub<R> for Formula<Q> {
+    type Output = Self;
+
+    fn sub(self, rhs: R) -> Self {
+        self.combine(rhs, |lhs, rhs| lhs - rhs)
+    }
+}
+
+impl<Q: Quantity> std::ops::Mul<f32> for Formula<Q> {
+    type Output = Self;
+
+    fn mul(self, rhs: f32) -> Self {
+        self.map(|lhs| lhs * engine::Formula::Constant(Some(rhs)))
+    }
+}
+
+impl<Q: Quantity> std::ops::Div<f32> for Formula<Q> {
+    type Output = Self;
+
+    fn div(self, rhs: f32) -> Self {
+        self.map(|lhs| lhs / engine::Formula::Constant(Some(rhs)))
+    }
+}
+
+impl<Q: Quantity> std::ops::Mul<Percentage> for Formula<Q> {
+    type Output = Self;
+
+    fn mul(self, rhs: Percentage) -> Self {
+        self * rhs.as_fraction()
+    }
+}
+
+impl<Q: Quantity> std::fmt::Display for Formula<Q> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.engine_formula.fmt(f)
+    }
 }
 
 /// Delivers evaluated values to one subscriber as typed samples.