Add grid_current and ev_charger_current methods to LogicalMeter

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py

@@ -15,11 +15,20 @@
 from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid import ComponentGraph
 from ...microgrid.component import ComponentMetricId
-from ._formula_engine import FormulaEngine, FormulaReceiver
+from ._formula_engine import (
+    FormulaEngine,
+    FormulaEngine3Phase,
+    FormulaReceiver,
+    FormulaReceiver3Phase,
+    _GenericEngine,
+    _GenericFormulaReceiver,
+)
 from ._formula_generators import (
     BatteryPowerFormula,
     BatterySoCFormula,
+    EVChargerCurrentFormula,
     FormulaGenerator,
+    GridCurrentFormula,
     GridPowerFormula,
     PVPowerFormula,
 )
@@ -110,7 +119,7 @@ def __init__(
         # meter to use when communicating with the resampling actor.
         self._namespace = f"logical-meter-{uuid.uuid4()}"
         self._component_graph = component_graph
-        self._engines: Dict[str, FormulaEngine] = {}
+        self._engines: Dict[str, FormulaEngine | FormulaEngine3Phase] = {}
         self._tasks: List[asyncio.Task[None]] = []
 
     async def _engine_from_formula_string(
@@ -163,8 +172,8 @@ async def start_formula(
     async def _get_formula_stream(
         self,
         channel_key: str,
-        generator: Type[FormulaGenerator],
-    ) -> FormulaReceiver:
+        generator: Type[FormulaGenerator[_GenericEngine]],
+    ) -> _GenericFormulaReceiver:
         if channel_key in self._engines:
             return self._engines[channel_key].new_receiver()
 
@@ -187,6 +196,34 @@ async def grid_power(self) -> FormulaReceiver:
         """
         return await self._get_formula_stream("grid_power", GridPowerFormula)
 
+    async def grid_current(self) -> FormulaReceiver3Phase:
+        """Fetch the grid power for the microgrid.
+
+        If a formula engine to calculate grid current is not already running, it
+        will be started.  Else, we'll just get a new receiver to the already
+        existing data stream.
+
+        Returns:
+            A *new* receiver that will stream grid_current values.
+
+        """
+        return await self._get_formula_stream("grid_current", GridCurrentFormula)
+
+    async def ev_charger_current(self) -> FormulaReceiver3Phase:
+        """Fetch the cumulative ev charger current for the microgrid.
+
+        If a formula engine to calculate ev charger current is not already
+        running, it will be started.  Else, we'll just get a new receiver to the
+        already existing data stream.
+
+        Returns:
+            A *new* receiver that will stream ev_charger_current values.
+
+        """
+        return await self._get_formula_stream(
+            "ev_charger_current", EVChargerCurrentFormula
+        )
+
     async def battery_power(self) -> FormulaReceiver:
         """Fetch the cumulative battery power in the microgrid.
 

tests/timeseries/test_logical_meter.py

@@ -362,3 +362,43 @@ async def test_formula_composition_missing_bat(self, mocker: MockerFixture) -> N
         await engine._stop()  # pylint: disable=protected-access
 
         assert count == 10
+
+    async def test_3_phase_formulas(self, mocker: MockerFixture) -> None:
+        """Test 3 phase formulas current formulas and their composition."""
+        mockgrid = await MockMicrogrid.new(mocker, grid_side_meter=False)
+        mockgrid.add_batteries(3)
+        mockgrid.add_ev_chargers(1)
+        request_sender, channel_registry = await mockgrid.start()
+        logical_meter = LogicalMeter(
+            channel_registry,
+            request_sender,
+            microgrid.get().component_graph,
+        )
+        grid_current_recv = await logical_meter.grid_current()
+        ev_current_recv = await logical_meter.ev_charger_current()
+
+        engine = (grid_current_recv.clone() - ev_current_recv.clone()).build(
+            "net_current"
+        )
+        net_current_recv = engine.new_receiver()
+        for _ in range(10):
+            grid_amps = await grid_current_recv.receive()
+            ev_amps = await ev_current_recv.receive()
+            net_amps = await net_current_recv.receive()
+
+            assert grid_amps.value_p1 is not None and grid_amps.value_p1 > 0.0
+            assert grid_amps.value_p2 is not None and grid_amps.value_p2 > 0.0
+            assert grid_amps.value_p3 is not None and grid_amps.value_p3 > 0.0
+            assert ev_amps.value_p1 is not None and ev_amps.value_p1 > 0.0
+            assert ev_amps.value_p2 is not None and ev_amps.value_p2 > 0.0
+            assert ev_amps.value_p3 is not None and ev_amps.value_p3 > 0.0
+            assert net_amps.value_p1 is not None and net_amps.value_p1 > 0.0
+            assert net_amps.value_p2 is not None and net_amps.value_p2 > 0.0
+            assert net_amps.value_p3 is not None and net_amps.value_p3 > 0.0
+
+            assert net_amps.value_p1 == grid_amps.value_p1 - ev_amps.value_p1
+            assert net_amps.value_p2 == grid_amps.value_p2 - ev_amps.value_p2
+            assert net_amps.value_p3 == grid_amps.value_p3 - ev_amps.value_p3
+
+        await mockgrid.cleanup()
+        await engine._stop()  # pylint: disable=protected-access