Add tests for voltage streaming

Signed-off-by: Daniel Zullo <[email protected]>

Author: daniel-zullo-frequenz

tests/timeseries/mock_resampler.py

@@ -81,12 +81,15 @@ def frequency_senders(
             [NON_EXISTING_COMPONENT_ID]
         )[0]
 
-        def current_senders(ids: list[int]) -> list[list[Sender[Sample[Quantity]]]]:
+        def multi_phase_senders(
+            ids: list[int],
+            metrics: tuple[ComponentMetricId, ComponentMetricId, ComponentMetricId],
+        ) -> list[list[Sender[Sample[Quantity]]]]:
             senders: list[list[Sender[Sample[Quantity]]]] = []
             for comp_id in ids:
-                p1_name = f"{comp_id}:{ComponentMetricId.CURRENT_PHASE_1}"
-                p2_name = f"{comp_id}:{ComponentMetricId.CURRENT_PHASE_2}"
-                p3_name = f"{comp_id}:{ComponentMetricId.CURRENT_PHASE_3}"
+                p1_name = f"{comp_id}:{metrics[0]}"
+                p2_name = f"{comp_id}:{metrics[1]}"
+                p3_name = f"{comp_id}:{metrics[2]}"
 
                 senders.append(
                     [
@@ -109,12 +112,34 @@ def current_senders(ids: list[int]) -> list[list[Sender[Sample[Quantity]]]]:
                 ]
             return senders
 
+        def current_senders(ids: list[int]) -> list[list[Sender[Sample[Quantity]]]]:
+            return multi_phase_senders(
+                ids,
+                (
+                    ComponentMetricId.CURRENT_PHASE_1,
+                    ComponentMetricId.CURRENT_PHASE_2,
+                    ComponentMetricId.CURRENT_PHASE_3,
+                ),
+            )
+
+        def voltage_senders(ids: list[int]) -> list[list[Sender[Sample[Quantity]]]]:
+            return multi_phase_senders(
+                ids,
+                (
+                    ComponentMetricId.VOLTAGE_PHASE_1,
+                    ComponentMetricId.VOLTAGE_PHASE_2,
+                    ComponentMetricId.VOLTAGE_PHASE_3,
+                ),
+            )
+
         self._bat_inverter_current_senders = current_senders(bat_inverter_ids)
         self._pv_inverter_current_senders = current_senders(pv_inverter_ids)
         self._ev_current_senders = current_senders(evc_ids)
         self._chp_current_senders = current_senders(chp_ids)
         self._meter_current_senders = current_senders(meter_ids)
 
+        self._meter_voltage_senders = voltage_senders(meter_ids)
+
         self._next_ts = datetime.now()
 
         mocker.patch(
@@ -242,3 +267,12 @@ async def send_meter_current(self, values: list[list[float | None]]) -> None:
             for phase, value in enumerate(meter_values):
                 sample = Sample(self._next_ts, None if not value else Quantity(value))
                 await chan[phase].send(sample)
+
+    async def send_meter_voltage(self, values: list[list[float | None]]) -> None:
+        """Send the given values as resampler output for meter voltage."""
+        assert len(values) == len(self._meter_voltage_senders)
+        for chan, meter_values in zip(self._meter_voltage_senders, values):
+            assert len(meter_values) == 3  # 3 values for phases
+            for phase, value in enumerate(meter_values):
+                sample = Sample(self._next_ts, None if not value else Quantity(value))
+                await chan[phase].send(sample)

tests/timeseries/test_voltage_streaming.py

@@ -0,0 +1,124 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for fetching and streaming the phase-to-neutral voltage."""
+
+
+import asyncio
+
+from pytest_mock import MockerFixture
+
+from frequenz.sdk import microgrid
+
+from .mock_microgrid import MockMicrogrid
+
+# pylint: disable=protected-access
+
+
+async def test_voltage_1(mocker: MockerFixture) -> None:
+    """Test the phase-to-neutral voltage with a grid side meter."""
+    mockgrid = MockMicrogrid(grid_meter=True)
+    mockgrid.add_batteries(1, no_meter=True)
+    mockgrid.add_batteries(1, no_meter=False)
+    await mockgrid.start(mocker)
+
+    voltage = microgrid.voltage()
+    voltage_recv = voltage.new_receiver()
+
+    assert voltage._task is not None
+    # Wait for voltage requests to be sent, one request per phase.
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+
+    for count in range(10):
+        volt_delta = 1 if count % 2 == 0 else -1
+        volt_phases: list[float | None] = [
+            220.0 * volt_delta,
+            219.8 * volt_delta,
+            220.2 * volt_delta,
+        ]
+
+        await mockgrid.mock_resampler.send_meter_voltage([volt_phases, volt_phases])
+
+        val = await voltage_recv.receive()
+        assert val is not None
+        assert val.value_p1 and val.value_p2 and val.value_p3
+        assert val.value_p1.as_volts() == volt_phases[0]
+        assert val.value_p2.as_volts() == volt_phases[1]
+        assert val.value_p3.as_volts() == volt_phases[2]
+
+    await mockgrid.cleanup()
+
+
+async def test_voltage_2(mocker: MockerFixture) -> None:
+    """Test the phase-to-neutral voltage without a grid side meter."""
+    mockgrid = MockMicrogrid(grid_meter=False)
+    mockgrid.add_batteries(1, no_meter=False)
+    mockgrid.add_batteries(1, no_meter=True)
+    await mockgrid.start(mocker)
+
+    voltage = microgrid.voltage()
+    voltage_recv = voltage.new_receiver()
+
+    assert voltage._task is not None
+    # Wait for voltage requests to be sent, one request per phase.
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+
+    for count in range(10):
+        volt_delta = 1 if count % 2 == 0 else -1
+        volt_phases: list[float | None] = [
+            220.0 * volt_delta,
+            219.8 * volt_delta,
+            220.2 * volt_delta,
+        ]
+
+        await mockgrid.mock_resampler.send_meter_voltage([volt_phases])
+
+        val = await voltage_recv.receive()
+        assert val is not None
+        assert val.value_p1 and val.value_p2 and val.value_p3
+        assert val.value_p1.as_volts() == volt_phases[0]
+        assert val.value_p2.as_volts() == volt_phases[1]
+        assert val.value_p3.as_volts() == volt_phases[2]
+
+    await mockgrid.cleanup()
+
+
+async def test_voltage_3(mocker: MockerFixture) -> None:
+    """Test the phase-to-neutral voltage with None values."""
+    mockgrid = MockMicrogrid(grid_meter=True)
+    mockgrid.add_batteries(2, no_meter=False)
+    await mockgrid.start(mocker)
+
+    voltage = microgrid.voltage()
+    voltage_recv = voltage.new_receiver()
+
+    assert voltage._task is not None
+    # Wait for voltage requests to be sent, one request per phase.
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+
+    for count in range(10):
+        volt_delta = 1 if count % 2 == 0 else -1
+        volt_phases: list[float | None] = [
+            220.0 * volt_delta,
+            219.8 * volt_delta,
+            220.2 * volt_delta,
+        ]
+
+        await mockgrid.mock_resampler.send_meter_voltage(
+            [volt_phases, [None, None, None], [None, 219.8, 220.2]]
+        )
+
+        val = await voltage_recv.receive()
+        assert val is not None
+        assert val.value_p1 and val.value_p2 and val.value_p3
+        assert val.value_p1.as_volts() == volt_phases[0]
+        assert val.value_p2.as_volts() == volt_phases[1]
+        assert val.value_p3.as_volts() == volt_phases[2]
+
+    await mockgrid.cleanup()