Implement an async composable Formula type (#12)

Author: shsms

Cargo.toml

@@ -8,6 +8,7 @@ name = "frequenz_microgrid"
 path = "src/lib.rs"
 
 [dependencies]
+async-trait = "0.1.89"
 chrono = "0.4"
 frequenz-microgrid-component-graph = { git = "https://github.com/frequenz-floss/frequenz-microgrid-component-graph-rs.git", rev = "0b28f99" }
 frequenz-microgrid-formula-engine = { git = "https://github.com/frequenz-floss/frequenz-microgrid-formula-engine-rs.git", rev = "463414a" }
@@ -23,5 +24,5 @@ tracing-subscriber = { version = "0.3" }
 tracing-subscriber = { version = "0.3", features = ["std", "env-filter"] }
 
 [build-dependencies]
+prost-build = { version = "0.13", features = ["cleanup-markdown"] }
 tonic-build = "0.13"
-prost-build = "0.13"

examples/logical_meter.rs

@@ -3,7 +3,7 @@
 
 use chrono::TimeDelta;
 use frequenz_microgrid::{
-    Error, Formula, LogicalMeterConfig, LogicalMeterHandle, MicrogridClientHandle, metric,
+    Error, LogicalMeterConfig, LogicalMeterHandle, MicrogridClientHandle, metric, quantity::Power,
 };
 use tracing_subscriber::{
     EnvFilter,
@@ -14,7 +14,7 @@ use tracing_subscriber::{
 #[tokio::main]
 async fn main() -> Result<(), Error> {
     tracing_subscriber::registry()
-        .with(EnvFilter::new("info,frequenz_microgrid=debug"))
+        .with(EnvFilter::new("info,frequenz_microgrid=warn"))
         .with(fmt::layer().with_file(true).with_line_number(true))
         .init();
 
@@ -29,49 +29,113 @@ async fn main() -> Result<(), Error> {
 
     // Create a formula that calculates `grid_power - battery_power`.
     let formula_grid = logical_meter.grid(metric::AcPowerActive)?;
-    let formula_battery = logical_meter.battery(None, metric::AcPowerActive)?;
+    let formula_pv = logical_meter.pv(None, metric::AcPowerActive)?;
     let formula_consumer = logical_meter.consumer(metric::AcPowerActive)?;
 
-    let formula = (logical_meter.grid(metric::AcPowerActive)?
-        - logical_meter.battery(None, metric::AcPowerActive)?
-        + logical_meter.consumer(metric::AcPowerActive)?)?;
+    let formula = logical_meter.grid(metric::AcPowerActive)?
+        - logical_meter.pv(None, metric::AcPowerActive)?
+        + logical_meter.consumer(metric::AcPowerActive)?
+        + Power::from_kilowatts(100.0);
 
     let mut rx = formula.subscribe().await?;
     let mut grid_rx = formula_grid.subscribe().await?;
-    let mut battery_rx = formula_battery.subscribe().await?;
+    let mut pv_rx = formula_pv.subscribe().await?;
     let mut consumer_rx = formula_consumer.subscribe().await?;
 
     for _ in 0..3 {
         let sample = rx.recv().await.unwrap();
         let grid_sample = grid_rx.recv().await.unwrap();
-        let battery_sample = battery_rx.recv().await.unwrap();
+        let pv_sample = pv_rx.recv().await.unwrap();
         let consumer_sample = consumer_rx.recv().await.unwrap();
         tracing::info!(
-            "grid({}) - battery({}) + consumer({}) = {}",
-            grid_sample.value().unwrap(),
-            battery_sample.value().unwrap(),
-            consumer_sample.value().unwrap(),
-            sample.value().unwrap()
+            "grid({}) - pv({}) + consumer({}) + 100kW = {}",
+            grid_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            pv_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            consumer_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string())
         );
     }
-    let formula_grid_voltage = logical_meter
-        .battery(None, metric::AcVoltagePhase1N)?
-        .coalesce(logical_meter.pv(None, metric::AcVoltagePhase1N)?)?;
+
+    let formula_grid_voltage = logical_meter.grid(metric::AcVoltage)?.coalesce(
+        logical_meter.grid(metric::AcVoltagePhase1N)?.avg(vec![
+            logical_meter.grid(metric::AcVoltagePhase2N)?,
+            logical_meter.grid(metric::AcVoltagePhase3N)?,
+        ])? * 3.0_f32.sqrt(),
+    )?;
 
     tracing::info!("formula_grid_voltage: {}", formula_grid_voltage);
     let mut grid_voltage_rx = formula_grid_voltage.subscribe().await?;
+
     for _ in 0..3 {
         let sample = grid_voltage_rx.recv().await.unwrap();
-        tracing::info!("grid voltage: {}", sample.value().unwrap());
+        tracing::info!(
+            "grid voltage: {}",
+            sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string())
+        );
     }
 
     drop(rx);
     drop(grid_rx);
-    drop(battery_rx);
+    drop(pv_rx);
     drop(consumer_rx);
 
+    let mut p1 = logical_meter
+        .grid(metric::AcVoltagePhase1N)?
+        .subscribe()
+        .await?;
+    let mut p2 = logical_meter
+        .grid(metric::AcVoltagePhase2N)?
+        .subscribe()
+        .await?;
+    let mut p3 = logical_meter
+        .grid(metric::AcVoltagePhase3N)?
+        .subscribe()
+        .await?;
+    let mut three_phase = logical_meter.grid(metric::AcVoltage)?.subscribe().await?;
+
     loop {
         let sample = grid_voltage_rx.recv().await.unwrap();
-        tracing::info!("grid voltage: {}", sample.value().unwrap());
+        let p1_sample = p1.recv().await.unwrap();
+        let p2_sample = p2.recv().await.unwrap();
+        let p3_sample = p3.recv().await.unwrap();
+        let three_phase_sample = three_phase.recv().await.unwrap();
+        tracing::info!(
+            "grid voltage: Coalesce({}, Avg({}, {}, {}) * sqrt(3)) = {}",
+            three_phase_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            p1_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            p2_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            p3_sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string()),
+            sample
+                .value()
+                .map(|v| v.to_string())
+                .unwrap_or_else(|| "None".to_string())
+        );
     }
 }

src/lib.rs

@@ -9,7 +9,7 @@ pub use client::MicrogridClientHandle;
 mod error;
 pub use error::{Error, ErrorKind};
 
-mod quantity;
+pub mod quantity;
 
 mod proto;
 
@@ -18,5 +18,5 @@ pub use sample::Sample;
 
 mod logical_meter;
 pub use logical_meter::{
-    AggregationFormula, Formula, LogicalMeterConfig, LogicalMeterHandle, metric,
+    Formula, FormulaSubscriber, LogicalMeterConfig, LogicalMeterHandle, metric,
 };

src/logical_meter.rs

@@ -5,10 +5,11 @@
 
 mod config;
 mod formula;
+pub use formula::{Formula, FormulaSubscriber};
+
 mod logical_meter_actor;
 mod logical_meter_handle;
 pub use logical_meter_handle::LogicalMeterHandle;
 pub mod metric;
 
 pub use config::LogicalMeterConfig;
-pub use formula::{AggregationFormula, Formula};

src/logical_meter/formula.rs

@@ -3,45 +3,40 @@
 
 //! Formula module for the logical meter.
 
-use frequenz_microgrid_component_graph::Formula as _;
-mod aggregation_formula;
-mod coalesce_formula;
+use async_trait::async_trait;
+pub(crate) mod aggregation_formula;
+mod async_formula;
+pub(crate) mod coalesce_formula;
 pub(crate) mod graph_formula_provider;
-pub use aggregation_formula::AggregationFormula;
-pub use coalesce_formula::CoalesceFormula;
+pub use async_formula::Formula;
 
-use crate::{Error, Sample, metric::Metric, quantity::Quantity};
+use crate::{
+    Error, Sample, logical_meter::formula::async_formula::FormulaOperand, metric::Metric,
+    quantity::Quantity,
+};
 use tokio::sync::{broadcast, mpsc};
 
 use super::logical_meter_actor;
 
 /// Connects logical meter formulas to the component graph formulas.
-pub(crate) trait GraphFormulaProvider: std::fmt::Display {
+pub(crate) trait GraphFormulaConnector: std::fmt::Display {
     type GraphFormulaType: frequenz_microgrid_component_graph::Formula;
 }
 
-/// Defines a formula that can be subscribed to for receiving samples.
-pub(crate) trait FormulaSubscriber: std::fmt::Display {
-    type MetricType: Metric;
-
-    fn subscribe(
-        &self,
-    ) -> impl Future<
-        Output = Result<
-            broadcast::Receiver<Sample<<Self::MetricType as Metric>::QuantityType>>,
-            Error,
-        >,
-    > + Send;
+#[async_trait]
+pub trait FormulaSubscriber: std::fmt::Display + Sync + Send {
+    type QuantityType: Quantity;
+    async fn subscribe(&self) -> Result<broadcast::Receiver<Sample<Self::QuantityType>>, Error>;
 }
 
 /// Parameters for creating a logical meter formula.
-pub(super) struct FormulaParams<F: GraphFormulaProvider, M: Metric> {
+pub(super) struct FormulaParams<F: GraphFormulaConnector, M: Metric> {
     pub(super) formula: F::GraphFormulaType,
     pub(super) metric: M,
     pub(super) instructions_tx: mpsc::Sender<logical_meter_actor::Instruction>,
 }
 
-impl<F: GraphFormulaProvider, M: Metric> FormulaParams<F, M> {
+impl<F: GraphFormulaConnector, M: Metric> FormulaParams<F, M> {
     pub(super) fn new(
         formula: F::GraphFormulaType,
         metric: M,
@@ -55,54 +50,108 @@ impl<F: GraphFormulaProvider, M: Metric> FormulaParams<F, M> {
     }
 }
 
-/// A trait that defines generic formula operations.
-pub trait Formula<Q: Quantity>: std::fmt::Display + Sized {
-    fn coalesce(self, other: Self) -> Result<Self, Error>;
-    fn min(self, other: Self) -> Result<Self, Error>;
-    fn max(self, other: Self) -> Result<Self, Error>;
-    fn subscribe(
-        &self,
-    ) -> impl Future<Output = Result<broadcast::Receiver<Sample<Q>>, Error>> + Send;
+// TODO: extend previous Coalesce instead of creating a new one, etc.
+impl<Q> Formula<Q>
+where
+    Q: Quantity + 'static,
+{
+    pub fn coalesce(self, other: Formula<Q>) -> Result<Formula<Q>, Error> {
+        match self {
+            Formula::Coalesce(mut items) => {
+                items.push(other.into());
+                Ok(Formula::Coalesce(items))
+            }
+            _ => Ok(Formula::Coalesce(vec![
+                FormulaOperand::Formula(Box::new(Formula::<Q, Q, f32>::Subscriber(Box::new(self)))),
+                other.into(),
+            ])),
+        }
+    }
+
+    pub fn min(self, other: Formula<Q>) -> Result<Formula<Q>, Error> {
+        match self {
+            Formula::Min(mut items) => {
+                items.push(other.into());
+                Ok(Formula::Min(items))
+            }
+            _ => Ok(Formula::Min(vec![
+                FormulaOperand::Formula(Box::new(Formula::<Q, Q, f32>::Subscriber(Box::new(self)))),
+                other.into(),
+            ])),
+        }
+    }
+
+    pub fn max(self, other: Formula<Q>) -> Result<Formula<Q>, Error> {
+        match self {
+            Formula::Max(mut items) => {
+                items.push(other.into());
+                Ok(Formula::Max(items))
+            }
+            _ => Ok(Formula::Max(vec![
+                FormulaOperand::Formula(Box::new(Formula::<Q, Q, f32>::Subscriber(Box::new(self)))),
+                other.into(),
+            ])),
+        }
+    }
+
+    pub fn avg(self, others: Vec<Formula<Q>>) -> Result<Formula<Q>, Error> {
+        let mut exprs: Vec<FormulaOperand<Q>> =
+            vec![FormulaOperand::Formula(Box::new(
+                Formula::<Q, Q, f32>::Subscriber(Box::new(self)),
+            ))];
+        for other in others {
+            exprs.push(other.into());
+        }
+        Ok(Formula::Avg(exprs))
+    }
+
+    pub async fn subscribe(&self) -> Result<broadcast::Receiver<Sample<Q>>, Error> {
+        <Self as FormulaSubscriber>::subscribe(self).await
+    }
 }
 
-impl<T, Q, M> Formula<Q> for T
+impl<Q, F> std::ops::Add<F> for Formula<Q>
 where
-    T: FormulaSubscriber<MetricType = M>
-        + GraphFormulaProvider
-        + From<FormulaParams<T, M>>
-        + Into<FormulaParams<T, M>>
-        + std::fmt::Display,
-    Q: Quantity,
-    M: Metric<QuantityType = Q>,
+    F: Into<FormulaOperand<Q>>,
+    Q: Quantity + 'static,
 {
-    fn coalesce(self, other: Self) -> Result<Self, Error> {
-        let mut params_self: FormulaParams<T, M> = self.into();
-        let params_other: FormulaParams<T, M> = other.into();
+    type Output = Formula<Q>;
 
-        params_self.formula = params_self.formula.coalesce(params_other.formula);
-        Ok(params_self.into())
+    fn add(self, other: F) -> Self::Output {
+        Formula::Add(vec![FormulaOperand::Formula(Box::new(self)), other.into()])
     }
+}
 
-    fn min(self, other: Self) -> Result<Self, Error> {
-        let mut params_self: FormulaParams<T, M> = self.into();
-        let params_other: FormulaParams<T, M> = other.into();
+impl<Q, F> std::ops::Sub<F> for Formula<Q>
+where
+    F: Into<FormulaOperand<Q>>,
+    Q: Quantity + 'static,
+{
+    type Output = Formula<Q>;
 
-        params_self.formula = params_self.formula.min(params_other.formula);
-        Ok(params_self.into())
+    fn sub(self, other: F) -> Self::Output {
+        Formula::Subtract(vec![FormulaOperand::Formula(Box::new(self)), other.into()])
     }
+}
 
-    fn max(self, other: Self) -> Result<Self, Error> {
-        let mut params_self: FormulaParams<T, M> = self.into();
-        let params_other: FormulaParams<T, M> = other.into();
+impl<Q> std::ops::Mul<f32> for Formula<Q, Q, f32>
+where
+    Q: Quantity + 'static,
+{
+    type Output = Formula<Q, Q, f32>;
 
-        params_self.formula = params_self.formula.max(params_other.formula);
-        Ok(params_self.into())
+    fn mul(self, other: f32) -> Self::Output {
+        Formula::<Q, Q, f32>::Multiply(FormulaOperand::<Q>::Formula(Box::new(self)), other.into())
     }
+}
+
+impl<Q> std::ops::Div<f32> for Formula<Q, Q, f32>
+where
+    Q: Quantity + 'static,
+{
+    type Output = Formula<Q, Q, f32>;
 
-    fn subscribe(
-        &self,
-    ) -> impl Future<Output = Result<broadcast::Receiver<Sample<M::QuantityType>>, Error>> + Send
-    {
-        <T as FormulaSubscriber>::subscribe(self)
+    fn div(self, rhs: f32) -> Self::Output {
+        Formula::<Q, Q, f32>::Divide(FormulaOperand::<Q>::Formula(Box::new(self)), rhs.into())
     }
 }