removing meter-only consumers from load management priority

packages/control/counter_all/counter_all_data.py

@@ -2,6 +2,7 @@
 from typing import Callable, Dict, Generator, List, Protocol, Tuple, Union
 
 from control.chargepoint.chargepoint import Chargepoint
+from control.consumer.usage import ConsumerUsage
 from dataclass_utils.factories import empty_list_factory
 from modules.common.component_type import ComponentType
 
@@ -98,6 +99,10 @@ class LoadmanagementPrioProtocol(Protocol):
     @property
     def data(self) -> CounterAllData: ...
     def add_loadmanagement_prio_item(self, type: ComponentType, id: int) -> None: ...
+    def update_consumer_loadmanagement_prio(self, consumer_id: int, usage_type: ConsumerUsage) -> bool: ...
+    def _has_loadmanagement_prio_item(
+        self, type: ComponentType, id: int, entries: List[Dict]) -> bool: ...
+
     def remove_loadmanagement_prio_item(self, type: ComponentType, id: int) -> None: ...
     def _remove_loadmanagement_prio_item(self, type: ComponentType, id: int, entry: List[Dict]) -> bool: ...
     def sort_cps_by_loadmanagement_prios_nested(self, filtered_cps: List[Chargepoint]) -> List[List[Chargepoint]]: ...

packages/control/counter_all/loadmanagement_prio.py

@@ -2,6 +2,7 @@
 from typing import Dict, Generator, List, Tuple
 
 from control.chargepoint.chargepoint import Chargepoint
+from control.consumer.usage import ConsumerUsage, NOT_CONTROLLED
 from control.counter_all.counter_all_data import LoadmanagementPrioProtocol
 from modules.common.component_type import ComponentType
 
@@ -13,6 +14,30 @@ class LoadmanagementPrioMixin:
     def add_loadmanagement_prio_item(self: LoadmanagementPrioProtocol, type: ComponentType, id: int) -> None:
         self.data.get.loadmanagement_prios.append({"type": type.value, "id": id})
 
+    def update_consumer_loadmanagement_prio(
+            self: LoadmanagementPrioProtocol, consumer_id: int, usage_type: ConsumerUsage) -> bool:
+        consumer_in_prios = self._has_loadmanagement_prio_item(
+            ComponentType.CONSUMER, consumer_id, self.data.get.loadmanagement_prios)
+        if usage_type in NOT_CONTROLLED:
+            if consumer_in_prios:
+                self.remove_loadmanagement_prio_item(ComponentType.CONSUMER, consumer_id)
+                return True
+        elif consumer_in_prios is False:
+            self.add_loadmanagement_prio_item(ComponentType.CONSUMER, consumer_id)
+            return True
+        return False
+
+    def _has_loadmanagement_prio_item(self: LoadmanagementPrioProtocol,
+                                      type: ComponentType,
+                                      id: int,
+                                      entries: List[Dict]) -> bool:
+        for entry in entries:
+            if entry["type"] == type.value and entry["id"] == id:
+                return True
+            if entry["type"] == "group" and self._has_loadmanagement_prio_item(type, id, entry["children"]):
+                return True
+        return False
+
     def remove_loadmanagement_prio_item(self: LoadmanagementPrioProtocol, type: ComponentType, id: int) -> None:
         if self._remove_loadmanagement_prio_item(type, id, self.data.get.loadmanagement_prios) is False:
             # Kein Grund, die Löschung des restlichen Elements (Topics, Hierarchie) abzubrechen, wenn der

packages/control/counter_all/loadmanagement_prio_test.py

@@ -5,6 +5,7 @@
 import pytest
 
 from control.chargepoint.chargepoint import Chargepoint
+from control.consumer.usage import ConsumerUsage
 from control.counter_all.counter_all import CounterAll
 from modules.common.component_type import ComponentType
 
@@ -76,6 +77,36 @@ def test_add_item(loadmanagement_prios: List[Dict],
     assert c.data.get.loadmanagement_prios == expected_loadmanagement_prios
 
 
+@pytest.mark.parametrize(
+    "loadmanagement_prios, usage_type, expected_loadmanagement_prios, expected_changed",
+    [
+        pytest.param([], ConsumerUsage.CONTINUOUS, [{"type": "consumer", "id": 2}], True, id="add consumer"),
+        pytest.param([{"type": "consumer", "id": 2}], ConsumerUsage.CONTINUOUS,
+                     [{"type": "consumer", "id": 2}], False, id="do not add duplicate"),
+        pytest.param([{
+            "type": "group",
+            "label": "Verbraucher",
+            "children": [{"type": "consumer", "id": 2}],
+        }], ConsumerUsage.METER_ONLY, [], True, id="remove consumer from group"),
+        pytest.param([], ConsumerUsage.METER_ONLY, [], False, id="ignore absent meter"),
+        pytest.param([{"type": "consumer", "id": 2}], ConsumerUsage.SELF_CONTROLLED, [], True,
+                     id="remove self-controlled consumer"),
+        pytest.param([], ConsumerUsage.SELF_CONTROLLED, [], False, id="ignore absent self-controlled consumer"),
+    ],
+)
+def test_update_consumer_loadmanagement_prio(loadmanagement_prios: List[Dict],
+                                             usage_type: ConsumerUsage,
+                                             expected_loadmanagement_prios: List[Dict],
+                                             expected_changed: bool):
+    c = CounterAll()
+    c.data.get.loadmanagement_prios = loadmanagement_prios
+
+    changed = c.update_consumer_loadmanagement_prio(2, usage_type)
+
+    assert c.data.get.loadmanagement_prios == expected_loadmanagement_prios
+    assert changed is expected_changed
+
+
 @pytest.mark.parametrize(
     "loadmanagement_prios, type, id, expected_loadmanagement_prios",
     [

packages/helpermodules/subdata.py

@@ -1279,7 +1279,12 @@ def process_consumer_topic(self, client: mqtt.Client, var: Dict[str, Consumer],
                     elif re.search("openWB/consumer/[0-9]+/extra_meter", msg.topic) is not None:
                         self.set_json_payload_class(var[f"consumer{index}"].data, msg)
                     elif re.search("openWB/consumer/[0-9]+/usage$", msg.topic) is not None:
-                        var[f"consumer{index}"].data.usage = dataclass_from_dict(Usage, decode_payload(msg.payload))
+                        usage = dataclass_from_dict(Usage, decode_payload(msg.payload))
+                        var[f"consumer{index}"].data.usage = usage
+                        if (self.event_subdata_initialized.is_set() and
+                                self.counter_all_data.update_consumer_loadmanagement_prio(int(index), usage.type)):
+                            Pub().pub("openWB/set/counter/get/loadmanagement_prios",
+                                      self.counter_all_data.data.get.loadmanagement_prios)
                     elif re.search("openWB/consumer/[0-9]+/control_parameter/", msg.topic) is not None:
                         if re.search("openWB/consumer/[0-9]+/control_parameter/limit", msg.topic) is not None:
                             payload = decode_payload(msg.payload)

packages/helpermodules/subdata_test.py

@@ -0,0 +1,100 @@
+import json
+from typing import Dict, List
+from unittest.mock import Mock
+
+import pytest
+
+from control.consumer.consumer import Consumer
+from control.consumer.usage import ConsumerUsage
+from control.counter_all.counter_all import CounterAll
+from helpermodules.subdata import SubData
+
+
+@pytest.mark.parametrize(
+    "usage_type, loadmanagement_prios, expected_loadmanagement_prios, expected_changed",
+    [
+        pytest.param(
+            ConsumerUsage.SUSPENDABLE_TUNABLE,
+            [],
+            [{"type": "consumer", "id": 2}],
+            True,
+            id="add controllable consumer",
+        ),
+        pytest.param(
+            ConsumerUsage.CONTINUOUS,
+            [{"type": "group", "label": "Gruppe", "children": [{"type": "consumer", "id": 2}]}],
+            [{"type": "group", "label": "Gruppe", "children": [{"type": "consumer", "id": 2}]}],
+            False,
+            id="do not duplicate grouped consumer",
+        ),
+        pytest.param(
+            ConsumerUsage.METER_ONLY,
+            [{"type": "group", "label": "Gruppe", "children": [{"type": "consumer", "id": 2}]}],
+            [],
+            True,
+            id="remove meter-only consumer from group",
+        ),
+        pytest.param(
+            ConsumerUsage.METER_ONLY,
+            [],
+            [],
+            False,
+            id="ignore absent meter-only consumer",
+        ),
+        pytest.param(
+            ConsumerUsage.SELF_CONTROLLED,
+            [{"type": "consumer", "id": 2}],
+            [],
+            True,
+            id="remove self-controlled consumer",
+        ),
+    ],
+)
+def test_process_consumer_usage_updates_loadmanagement_prios(usage_type: ConsumerUsage,
+                                                             loadmanagement_prios: List[Dict],
+                                                             expected_loadmanagement_prios: List[Dict],
+                                                             expected_changed: bool,
+                                                             monkeypatch: pytest.MonkeyPatch,
+                                                             mock_pub: Mock):
+    subdata = SubData.__new__(SubData)
+    subdata.event_subdata_initialized = Mock()
+    subdata.event_subdata_initialized.is_set.return_value = True
+    counter_all_data = CounterAll()
+    counter_all_data.data.get.loadmanagement_prios = loadmanagement_prios
+    monkeypatch.setattr(SubData, "counter_all_data", counter_all_data)
+    consumers = {"consumer2": Consumer(2)}
+    msg = Mock(
+        topic="openWB/consumer/2/usage",
+        payload=json.dumps({"type": usage_type.value}).encode("utf-8"),
+    )
+
+    subdata.process_consumer_topic(Mock(), consumers, msg)
+
+    assert consumers["consumer2"].data.usage.type == usage_type
+    assert counter_all_data.data.get.loadmanagement_prios == expected_loadmanagement_prios
+    if expected_changed:
+        mock_pub.pub.assert_called_once_with(
+            "openWB/set/counter/get/loadmanagement_prios", expected_loadmanagement_prios)
+    else:
+        mock_pub.pub.assert_not_called()
+
+
+def test_process_consumer_usage_does_not_update_loadmanagement_prios_on_startup(monkeypatch: pytest.MonkeyPatch,
+                                                                                mock_pub: Mock):
+    subdata = SubData.__new__(SubData)
+    subdata.event_subdata_initialized = Mock()
+    subdata.event_subdata_initialized.is_set.return_value = False
+    counter_all_data = CounterAll()
+    counter_all_data.data.get.loadmanagement_prios = [{"type": "consumer", "id": 2}]
+    monkeypatch.setattr(SubData, "counter_all_data", counter_all_data)
+    consumers = {"consumer2": Consumer(2)}
+    msg = Mock(
+        topic="openWB/consumer/2/usage",
+        payload=json.dumps({"type": ConsumerUsage.METER_ONLY.value}).encode("utf-8"),
+    )
+
+    subdata.process_consumer_topic(Mock(), consumers, msg)
+
+    assert consumers["consumer2"].data.usage.type == ConsumerUsage.METER_ONLY
+    assert counter_all_data.data.get.loadmanagement_prios == [{"type": "consumer", "id": 2}]
+    mock_pub.pub.assert_not_called()

packages/helpermodules/update_config.py

@@ -12,6 +12,7 @@
 from paho.mqtt.client import Client as MqttClient, MQTTMessage
 
 from control.chargemode import Chargemode
+from control.consumer.usage import NOT_CONTROLLED
 from control.limiting_value import LoadmanagementLimit
 
 from control.chargepoint.chargepoint_template import get_chargepoint_template_default
@@ -58,7 +59,7 @@
 
 class UpdateConfig:
 
-    DATASTORE_VERSION = 148
+    DATASTORE_VERSION = 149
 
     valid_topic = [
         "^openWB/bat/config/bat_control_activated$",
@@ -3864,3 +3865,33 @@ def upgrade(topic: str, payload) -> Optional[dict]:
                     return {topic: NO_MODULE}
         self._loop_all_received_topics(upgrade)
         self._append_datastore_version(148)
+
+    def upgrade_datastore_149(self) -> None:
+        """Verbraucher ohne Lastmanagement aus der Prioritätensteuerung entfernen."""
+        not_controlled_usage_types = {usage_type.value for usage_type in NOT_CONTROLLED}
+        not_controlled_consumers = {
+            int(get_index(topic))
+            for topic, payload in self.all_received_topics.items()
+            if re.search(r"^openWB/consumer/[0-9]+/usage$", topic) is not None
+            and decode_payload(payload).get("type") in not_controlled_usage_types
+        }
+
+        def remove_not_controlled_consumers(entries: list) -> None:
+            for entry in entries.copy():
+                if entry.get("type") == "consumer" and entry.get("id") in not_controlled_consumers:
+                    entries.remove(entry)
+                elif entry.get("type") == "group":
+                    children = entry.get("children", [])
+                    had_children = bool(children)
+                    remove_not_controlled_consumers(children)
+                    if had_children and not children:
+                        entries.remove(entry)
+
+        topic = "openWB/counter/get/loadmanagement_prios"
+        if topic in self.all_received_topics:
+            loadmanagement_prios = decode_payload(self.all_received_topics[topic])
+            migrated_loadmanagement_prios = copy.deepcopy(loadmanagement_prios)
+            remove_not_controlled_consumers(migrated_loadmanagement_prios)
+            if migrated_loadmanagement_prios != loadmanagement_prios:
+                self.__update_topic(topic, migrated_loadmanagement_prios)
+        self._append_datastore_version(149)

packages/helpermodules/update_config_test.py

@@ -363,3 +363,40 @@ def test_upgrade_datastore_142_ev_chargemode_conversion(ev0_prio: bool,
     assert uc.all_received_topics["openWB/counter/get/loadmanagement_prios"] == expected
     assert uc.all_received_topics["openWB/system/datastore_version"] == [131, 132, 142]
     assert mock_pub.pub.call_count == 2  # einmal publishen für Upgrade der Datastore-Version
+
+
+def test_upgrade_datastore_149_removes_not_controlled_consumers(mock_pub: Mock):
+    uc = UpdateConfig()
+    uc.all_received_topics = {
+        "openWB/consumer/1/usage": {"type": "continuous"},
+        "openWB/consumer/2/usage": {"type": "meter_only"},
+        "openWB/consumer/4/usage": {"type": "self_controlled"},
+        "openWB/counter/get/loadmanagement_prios": [
+            {"type": "vehicle", "id": 0},
+            {
+                "type": "group",
+                "label": "Verbraucher",
+                "children": [
+                    {"type": "consumer", "id": 1},
+                    {"type": "consumer", "id": 2},
+                    {"type": "consumer", "id": 4},
+                ],
+            },
+            {"type": "consumer", "id": 3},
+        ],
+        "openWB/system/datastore_version": list(range(149)),
+    }
+
+    uc.upgrade_datastore_149()
+
+    assert uc.all_received_topics["openWB/counter/get/loadmanagement_prios"] == [
+        {"type": "vehicle", "id": 0},
+        {
+            "type": "group",
+            "label": "Verbraucher",
+            "children": [{"type": "consumer", "id": 1}],
+        },
+        {"type": "consumer", "id": 3},
+    ]
+    assert uc.all_received_topics["openWB/system/datastore_version"] == list(range(150))
+    assert mock_pub.pub.call_count == 2