frequenz-sdk-python/tests/timeseries/test_logical_meter.py at 187621c954d1301074f429ed79c65b183bdaa94a · frequenz-floss/frequenz-sdk-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# License: MIT
# Copyright © 2022 Frequenz Energy-as-a-Service GmbH

"""Tests for the logical meter."""


from contextlib import AsyncExitStack

from frequenz.quantities import Power
from pytest_mock import MockerFixture

from frequenz.sdk import microgrid

from .mock_microgrid import MockMicrogrid

# pylint: disable=too-many-locals
# pylint: disable=protected-access


class TestLogicalMeter:  # pylint: disable=too-many-public-methods
    """Tests for the logical meter."""

    async def test_chp_power(self, mocker: MockerFixture) -> None:
        """Test the chp power formula."""
        mockgrid = MockMicrogrid(grid_meter=False, mocker=mocker)
        mockgrid.add_chps(1)
        mockgrid.add_batteries(2)

        async with mockgrid, AsyncExitStack() as stack:
            logical_meter = microgrid.logical_meter()
            stack.push_async_callback(logical_meter.stop)

            chp_power_receiver = logical_meter.chp_power.new_receiver()

            await mockgrid.mock_resampler.send_meter_power([2.0, 3.0, 4.0])
            await mockgrid.mock_resampler.send_chp_power([2.0])
            assert (await chp_power_receiver.receive()).value == Power.from_watts(2.0)

            await mockgrid.mock_resampler.send_meter_power([-12.0, None, 10.2])
            await mockgrid.mock_resampler.send_chp_power([-12.0])
            assert (await chp_power_receiver.receive()).value == Power.from_watts(-12.0)

    async def test_pv_power(self, mocker: MockerFixture) -> None:
        """Test the pv power formula."""
        mockgrid = MockMicrogrid(grid_meter=False, mocker=mocker)
        mockgrid.add_solar_inverters(2)

        async with mockgrid, AsyncExitStack() as stack:
            pv_pool = microgrid.new_pv_pool(priority=5)
            stack.push_async_callback(pv_pool.stop)
            pv_power_receiver = pv_pool.power.new_receiver()

            await mockgrid.mock_resampler.send_meter_power([-10.0, -20.0])
            assert (await pv_power_receiver.receive()).value == Power.from_watts(-30.0)