Implement CoalesceFormula

And introduce support for voltage and frequency metrics using
`CoalesceFormula`s.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>

Author: shsms

src/logical_meter/formula.rs

@@ -4,8 +4,10 @@
 //! Formula module for the logical meter.
 
 mod aggregation_formula;
+mod coalesce_formula;
 pub(crate) mod graph_formula_provider;
 pub use aggregation_formula::AggregationFormula;
+pub use coalesce_formula::CoalesceFormula;
 
 use crate::{Error, Sample};
 use tokio::sync::broadcast;

src/logical_meter/formula/coalesce_formula.rs

@@ -0,0 +1,57 @@
+// License: MIT
+// Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+//! An coalesce formula.
+
+use super::Formula;
+use crate::{
+    Error, Sample, logical_meter::logical_meter_actor, proto::common::v1::metrics::Metric,
+};
+use tokio::sync::{broadcast, mpsc, oneshot};
+
+#[derive(Clone)]
+pub struct CoalesceFormula {
+    formula: frequenz_microgrid_component_graph::CoalesceFormula,
+    metric: Metric,
+    instructions_tx: mpsc::Sender<logical_meter_actor::Instruction>,
+}
+
+impl std::fmt::Display for CoalesceFormula {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.formula.fmt(f)
+    }
+}
+
+impl CoalesceFormula {
+    pub(crate) fn new(
+        formula: frequenz_microgrid_component_graph::CoalesceFormula,
+        metric: Metric,
+        instructions_tx: mpsc::Sender<logical_meter_actor::Instruction>,
+    ) -> Self {
+        Self {
+            formula,
+            metric,
+            instructions_tx,
+        }
+    }
+}
+
+impl Formula for CoalesceFormula {
+    async fn subscribe(&self) -> Result<broadcast::Receiver<Sample>, Error> {
+        let (tx, rx) = oneshot::channel();
+
+        self.instructions_tx
+            .send(logical_meter_actor::Instruction::SubscribeFormula {
+                formula: self.formula.to_string(),
+                metric: self.metric,
+                response_tx: tx,
+            })
+            .await
+            .map_err(|e| Error::connection_failure(format!("Could not send instruction: {e}")))?;
+        let receiver = rx.await.map_err(|e| {
+            Error::connection_failure(format!("Could not receive instruction: {e}"))
+        })?;
+
+        Ok(receiver)
+    }
+}

src/logical_meter/formula/graph_formula_provider.rs

@@ -11,7 +11,7 @@ use frequenz_microgrid_component_graph::ComponentGraph;
 use std::collections::BTreeSet;
 use tokio::sync::mpsc;
 
-use super::AggregationFormula;
+use super::{AggregationFormula, CoalesceFormula};
 
 macro_rules! graph_formula_provider {
     ($(($fnname:ident $(, $idsparam:ident)?)),+ $(,)?) => {$(
@@ -82,3 +82,11 @@ impl GraphFormulaProvider for AggregationFormula {
         (ev_charger, ev_charger_formula, ev_charger_ids),
     );
 }
+
+impl GraphFormulaProvider for CoalesceFormula {
+    impl_graph_formula_provider!(
+        (grid, grid_coalesce_formula),
+        (battery, battery_ac_coalesce_formula, battery_ids),
+        (pv, pv_ac_coalesce_formula, pv_inverter_ids),
+    );
+}

src/logical_meter/metric.rs

@@ -41,4 +41,14 @@ define_metric! {
     {name: AcCurrentPhase1, formula: AggregationFormula},
     {name: AcCurrentPhase2, formula: AggregationFormula},
     {name: AcCurrentPhase3, formula: AggregationFormula},
+
+    {name: AcVoltage,             formula: CoalesceFormula},
+    {name: AcVoltagePhase1N,      formula: CoalesceFormula},
+    {name: AcVoltagePhase2N,      formula: CoalesceFormula},
+    {name: AcVoltagePhase3N,      formula: CoalesceFormula},
+    {name: AcVoltagePhase1Phase2, formula: CoalesceFormula},
+    {name: AcVoltagePhase2Phase3, formula: CoalesceFormula},
+    {name: AcVoltagePhase3Phase1, formula: CoalesceFormula},
+
+    {name: AcFrequency, formula: CoalesceFormula},
 }