Dynamische Netzentgelte

Author: LKuemmel

packages/control/ev/charge_template.py

@@ -292,8 +292,8 @@ def eco_charging(self,
                 current = 0
                 sub_mode = "stop"
                 message = self.AMOUNT_REACHED
-            elif data.data.optional_data.et_provider_available():
-                if data.data.optional_data.et_is_charging_allowed_price_threshold(eco_charging.max_price):
+            elif data.data.optional_data.data.electricity_pricing.configured:
+                if data.data.optional_data.ep_is_charging_allowed_price_threshold(eco_charging.max_price):
                     sub_mode = "instant_charging"
                     message = self.CHARGING_PRICE_LOW
                     phases = max_phases_hw
@@ -608,7 +608,7 @@ def end_of_today_timestamp() -> int:
                             hour=23, minute=59, second=59, microsecond=999000).timestamp()
 
                     def is_loading_hour(hour: int) -> bool:
-                        return data.data.optional_data.et_is_charging_allowed_hours_list(hour)
+                        return data.data.optional_data.ep_is_charging_allowed_hours_list(hour)
 
                     def convert_loading_hours_to_string(hour_list: List[int]) -> str:
                         if 1 < len(hour_list):
@@ -639,11 +639,11 @@ def convert_loading_hours_to_string(hour_list: List[int]) -> str:
                                         else '')
                     return loading_message + '.'
 
-                hour_list = data.data.optional_data.et_get_loading_hours(
+                hour_list = data.data.optional_data.ep_get_loading_hours(
                     selected_plan.duration, selected_plan.duration + selected_plan.remaining_time)
 
                 log.debug(f"Günstige Ladezeiten: {hour_list}")
-                if data.data.optional_data.et_is_charging_allowed_hours_list(hour_list):
+                if data.data.optional_data.ep_is_charging_allowed_hours_list(hour_list):
                     message = self.SCHEDULED_CHARGING_CHEAP_HOUR.format(get_hours_message())
                     current = plan_current
                     submode = "instant_charging"

packages/control/ev/charge_template_test.py

@@ -88,7 +88,6 @@ def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amou
 def test_instant_charging(selected: str, current_soc: float, used_amount: float,
                           expected: Tuple[int, str, Optional[str]]):
     # setup
-    data.data.optional_data.data.et.active = False
     ct = ChargeTemplate()
     ct.data.chargemode.instant_charging.limit.selected = selected
     ct.data.chargemode.instant_charging.limit.amount = 1000
@@ -377,10 +376,10 @@ def test_scheduled_charging_calc_current_electricity_tariff(
     plan.limit.selected = "soc"
     ct.data.chargemode.scheduled_charging.plans = [plan]
     # für Github-Test keinen Zeitstempel verwenden
-    mock_et_get_loading_hours = Mock(return_value=loading_hours)
-    monkeypatch.setattr(data.data.optional_data, "et_get_loading_hours", mock_et_get_loading_hours)
+    mock_ep_get_loading_hours = Mock(return_value=loading_hours)
+    monkeypatch.setattr(data.data.optional_data, "ep_get_loading_hours", mock_ep_get_loading_hours)
     mock_is_list_valid = Mock(return_value=is_loading_hour)
-    monkeypatch.setattr(data.data.optional_data, "et_is_charging_allowed_hours_list", mock_is_list_valid)
+    monkeypatch.setattr(data.data.optional_data, "ep_is_charging_allowed_hours_list", mock_is_list_valid)
 
     # execution
     ret = ct.scheduled_charging_calc_current(

packages/control/optional.py

@@ -7,24 +7,54 @@
 
 
 @dataclass
-class EtGet:
+class PricingGet:
     fault_state: int = 0
     fault_str: str = NO_ERROR
+    prices: Dict = field(default_factory=empty_dict_factory)
+
+
+def get_factory() -> PricingGet:
+    return PricingGet()
+
+
+@dataclass
+class FlexibleTariff:
+    get: PricingGet = field(default_factory=get_factory)
+
+
+def get_flexible_tariff_factory() -> FlexibleTariff:
+    return FlexibleTariff()
+
+
+@dataclass
+class GridFee:
+    get: PricingGet = field(default_factory=get_factory)
+
+
+def get_grid_fee_factory() -> GridFee:
+    return GridFee()
+
+
+@dataclass
+class ElectricityPricingGet:
     next_query_time: Optional[float] = None
     prices: Dict = field(default_factory=empty_dict_factory)
 
 
-def get_factory() -> EtGet:
-    return EtGet()
+def electricity_pricing_get_factory() -> ElectricityPricingGet:
+    return ElectricityPricingGet()
 
 
 @dataclass
-class Et:
-    get: EtGet = field(default_factory=get_factory)
+class ElectricityPricing:
+    configured: bool = False
+    flexible_tariff: FlexibleTariff = field(default_factory=get_flexible_tariff_factory)
+    grid_fee: GridFee = field(default_factory=get_grid_fee_factory)
+    get: ElectricityPricingGet = field(default_factory=electricity_pricing_get_factory)
 
 
-def et_factory() -> Et:
-    return Et()
+def ep_factory() -> ElectricityPricing:
+    return ElectricityPricing()
 
 
 @dataclass
@@ -84,7 +114,7 @@ def ocpp_factory() -> Ocpp:
 
 @dataclass
 class OptionalData:
-    et: Et = field(default_factory=et_factory)
+    electricity_pricing: ElectricityPricing = field(default_factory=ep_factory)
     int_display: InternalDisplay = field(default_factory=int_display_factory)
     led: Led = field(default_factory=led_factory)
     rfid: Rfid = field(default_factory=rfid_factory)

packages/control/optional_data.py

@@ -1,7 +1,8 @@
 from unittest.mock import Mock
-from control.optional import Optional
-from helpermodules import timecheck
 import pytest
+from helpermodules import timecheck
+from control.optional import Optional
+
 
 ONE_HOUR_SECONDS = 3600
 IGNORED = 0.0001
@@ -220,7 +221,7 @@
         ),
     ],
 )
-def test_et_get_loading_hours(granularity,
+def test_ep_get_loading_hours(granularity,
                               now_ts,
                               duration,
                               remaining_time,
@@ -229,17 +230,16 @@ def test_et_get_loading_hours(granularity,
                               monkeypatch):
     # setup
     opt = Optional()
-    opt.data.et.get.prices = price_list
-    mock_et_provider_available = Mock(return_value=True)
-    monkeypatch.setattr(opt, "et_provider_available", mock_et_provider_available)
+    opt.data.electricity_pricing.get.prices = price_list
+    opt.data.electricity_pricing.configured = True
     monkeypatch.setattr(
         timecheck,
         "create_timestamp",
         Mock(return_value=now_ts)
     )
 
     # execution
-    loading_hours = opt.et_get_loading_hours(duration=duration, remaining_time=remaining_time)
+    loading_hours = opt.ep_get_loading_hours(duration=duration, remaining_time=remaining_time)
 
     # evaluation
     assert loading_hours == expected_loading_hours
@@ -256,18 +256,18 @@ def test_et_get_loading_hours(granularity,
 )
 def test_et_charging_allowed(monkeypatch, provider_available, current_price, max_price, expected):
     opt = Optional()
-    monkeypatch.setattr(opt, "et_provider_available", Mock(return_value=provider_available))
+    opt.data.electricity_pricing.configured = provider_available
     if provider_available:
-        monkeypatch.setattr(opt, "et_get_current_price", Mock(return_value=current_price))
-    result = opt.et_is_charging_allowed_price_threshold(max_price)
+        monkeypatch.setattr(opt, "ep_get_current_price", Mock(return_value=current_price))
+    result = opt.ep_is_charging_allowed_price_threshold(max_price)
     assert result == expected
 
 
 def test_et_charging_allowed_exception(monkeypatch):
     opt = Optional()
-    monkeypatch.setattr(opt, "et_provider_available", Mock(return_value=True))
-    monkeypatch.setattr(opt, "et_get_current_price", Mock(side_effect=Exception))
-    result = opt.et_is_charging_allowed_price_threshold(0.15)
+    opt.data.electricity_pricing.configured = True
+    monkeypatch.setattr(opt, "ep_get_current_price", Mock(side_effect=Exception))
+    result = opt.ep_is_charging_allowed_price_threshold(0.15)
     assert result is False
 
 
@@ -425,17 +425,18 @@ def test_et_charging_available(now_ts, provider_available, price_list, selected_
         Mock(return_value=now_ts)
     )
     opt = Optional()
-    opt.data.et.get.prices = price_list
-    monkeypatch.setattr(opt, "et_provider_available", Mock(return_value=provider_available))
-    result = opt.et_is_charging_allowed_hours_list(selected_hours)
+    opt.data.electricity_pricing.get.prices = price_list
+    opt.data.electricity_pricing.configured = provider_available
+    result = opt.ep_is_charging_allowed_hours_list(selected_hours)
     assert result == expected
 
 
 def test_et_charging_available_exception(monkeypatch):
     opt = Optional()
-    monkeypatch.setattr(opt, "et_provider_available", Mock(return_value=True))
-    opt.data.et.get.prices = {}  # empty prices list raises exception
-    result = opt.et_is_charging_allowed_hours_list([])
+    opt.data.electricity_pricing.configured = True
+
+    opt.data.electricity_pricing.get.prices = {}  # empty prices list raises exception
+    result = opt.ep_is_charging_allowed_hours_list([])
     assert result is False
 
 
@@ -446,10 +447,6 @@ def test_et_charging_available_exception(monkeypatch):
             {}, None, 1698224400, True,
             id="update_required_when_no_prices"
         ),
-        pytest.param(
-            {"1698224400": 0.1, "1698228000": 0.2}, None, 1698224400, False,
-            id="no_update_required_when_prices_available_and_recent"
-        ),
         pytest.param(
             {"1698224400": 0.1, "1698228000": 0.2}, 1698310800, 1698224400, False,
             id="no_update_required_when_next_query_time_not_reached"
@@ -467,10 +464,11 @@ def test_et_charging_available_exception(monkeypatch):
 def test_et_price_update_required(monkeypatch, prices, next_query_time, current_timestamp, expected):
     # setup
     opt = Optional()
-    opt.data.et.get.prices = prices
-    opt.data.et.get.next_query_time = next_query_time
+    opt.data.electricity_pricing.get.prices = prices
+    opt.data.electricity_pricing.get.next_query_time = next_query_time
 
     monkeypatch.setattr(timecheck, "create_timestamp", Mock(return_value=current_timestamp))
+    opt.data.electricity_pricing.configured = True
 
     # execution
     result = opt.et_price_update_required()

packages/control/optional_test.py

@@ -504,7 +504,7 @@ def __on_connect_broker_essentials(self, client, userdata, flags, rc):
         client.subscribe("openWB/counter/#", 2)
         client.subscribe("openWB/pv/#", 2)
         client.subscribe("openWB/bat/#", 2)
-        client.subscribe("openWB/optional/et/provider", 2)
+        client.subscribe("openWB/optional/ep/flexible_tariff/provider", 2)
 
     def __on_connect_bridges(self, client, userdata, flags, rc):
         client.subscribe("openWB/system/mqtt/#", 2)

packages/helpermodules/create_debug.py

@@ -199,8 +199,9 @@ def create_entry(log_type: LogType, sh_log_data: LegacySmartHomeLogData, previou
     try:
         prices = data.data.general_data.data.prices
         try:
-            grid_price = data.data.optional_data.et_get_current_price()
+            grid_price = data.data.optional_data.ep_get_current_price()
         except Exception:
+            log.exception("Fehler im Werte-Logging-Modul für aktuellen Netzpreis, nutze hinterlegten Netzpreis")
             grid_price = prices.grid
         prices_dict = {"grid": grid_price,
                        "pv": prices.pv,

packages/helpermodules/measurement_logging/write_log.py

@@ -847,16 +847,27 @@ def process_optional_topic(self, msg: mqtt.MQTTMessage):
             enthält Topic und Payload
         """
         try:
-            if "openWB/set/optional/et/get/prices" in msg.topic:
+            pricing_regex = "openWB/set/optional/ep/(flexible_tariff|grid_fee)/"
+            if re.search(pricing_regex, msg.topic) is not None:
+                if re.search(f"{pricing_regex}provider$", msg.topic) is not None:
+                    self._validate_value(msg, "json")
+                elif re.search(f"{pricing_regex}get/prices$", msg.topic) is not None:
+                    self._validate_value(msg, "json")
+                elif re.search(f"{pricing_regex}get/price$", msg.topic) is not None:
+                    self._validate_value(msg, float)
+                elif re.search(f"{pricing_regex}get/fault_state$", msg.topic) is not None:
+                    self._validate_value(msg, int, [(0, 2)])
+                elif re.search(f"{pricing_regex}get/fault_str$", msg.topic) is not None:
+                    self._validate_value(msg, str)
+            elif "openWB/set/optional/ep/get/prices" in msg.topic:
                 self._validate_value(msg, "json")
-            elif "openWB/set/optional/et/get/next_query_time" in msg.topic:
+            elif "openWB/set/optional/ep/get/next_query_time" in msg.topic:
                 self._validate_value(msg, float)
-            elif "openWB/set/optional/et/get/fault_state" in msg.topic:
-                self._validate_value(msg, int, [(0, 2)])
-            elif "openWB/set/optional/et/get/fault_str" in msg.topic:
-                self._validate_value(msg, str)
-            elif ("openWB/set/optional/et/provider" in msg.topic or
-                  "openWB/set/optional/ocpp/config" in msg.topic):
+            elif "openWB/set/optional/ep/configured" in msg.topic:
+                self._validate_value(msg, bool)
+            elif "module_update_completed" in msg.topic:
+                self._validate_value(msg, bool)
+            elif "openWB/set/optional/ocpp/config" in msg.topic:
                 self._validate_value(msg, "json")
             elif "openWB/set/optional/monitoring" in msg.topic:
                 self._validate_value(msg, "json")

packages/helpermodules/setdata.py

@@ -30,7 +30,7 @@
 from dataclass_utils import dataclass_from_dict
 from modules.common.abstract_vehicle import CalculatedSocState, GeneralVehicleConfig
 from modules.common.configurable_backup_cloud import ConfigurableBackupCloud
-from modules.common.configurable_tariff import ConfigurableElectricityTariff
+from modules.common.configurable_tariff import ConfigurableFlexibleTariff, ConfigurableGridFee
 from modules.common.simcount.simcounter_state import SimCounterState
 from modules.internal_chargepoint_handler.internal_chargepoint_handler_config import (
     GlobalHandlerData, InternalChargepoint, RfidData)
@@ -712,23 +712,42 @@ def process_optional_topic(self, var: optional.Optional, msg: mqtt.MQTTMessage):
                         run_command([
                             str(Path(__file__).resolve().parents[2] / "runs" / "update_local_display.sh")
                         ], process_exception=True)
-                elif re.search("/optional/et/", msg.topic) is not None:
-                    if re.search("/optional/et/get/prices", msg.topic) is not None:
-                        var.data.et.get.prices = decode_payload(msg.payload)
-                    elif re.search("/optional/et/get/", msg.topic) is not None:
-                        self.set_json_payload_class(var.data.et.get, msg)
-                    elif re.search("/optional/et/provider$", msg.topic) is not None:
+                elif re.search("/optional/ep/(flexible_tariff|grid_fee)/", msg.topic) is not None:
+                    if re.search("/optional/ep/flexible_tariff/provider$", msg.topic) is not None:
                         config_dict = decode_payload(msg.payload)
                         if config_dict["type"] is None:
-                            var.et_module = None
+                            var.flexible_tariff_module = None
                         else:
                             mod = importlib.import_module(
-                                f".electricity_tariffs.{config_dict['type']}.tariff", "modules")
+                                f".electricity_pricing.flexible_tariffs.{config_dict['type']}.tariff", "modules")
                             config = dataclass_from_dict(mod.device_descriptor.configuration_factory, config_dict)
-                            var.et_module = ConfigurableElectricityTariff(config, mod.create_electricity_tariff)
-                            var.et_get_prices()
-                    else:
-                        self.set_json_payload_class(var.data.et, msg)
+                            var.flexible_tariff_module = ConfigurableFlexibleTariff(
+                                config, mod.create_electricity_tariff)
+                    elif re.search("/optional/ep/flexible_tariff/get/prices", msg.topic) is not None:
+                        var.data.electricity_pricing.flexible_tariff.get.prices = decode_payload(msg.payload)
+                    elif re.search("/optional/ep/flexible_tariff/get/", msg.topic) is not None:
+                        self.set_json_payload_class(var.data.electricity_pricing.flexible_tariff.get, msg)
+                    elif re.search("/optional/ep/grid_fee/provider$", msg.topic) is not None:
+                        config_dict = decode_payload(msg.payload)
+                        if config_dict["type"] is None:
+                            var.grid_fee_module = None
+                        else:
+                            mod = importlib.import_module(
+                                f".electricity_pricing.grid_fees.{config_dict['type']}.tariff", "modules")
+                            config = dataclass_from_dict(mod.device_descriptor.configuration_factory, config_dict)
+                            var.grid_fee_module = ConfigurableGridFee(config, mod.create_electricity_tariff)
+                    elif re.search("/optional/ep/grid_fee/get/prices", msg.topic) is not None:
+                        var.data.electricity_pricing.grid_fee.get.prices = decode_payload(msg.payload)
+                    elif re.search("/optional/ep/grid_fee/get/", msg.topic) is not None:
+                        self.set_json_payload_class(var.data.electricity_pricing.grid_fee.get, msg)
+                elif re.search("/optional/ep/get/prices", msg.topic) is not None:
+                    var.data.electricity_pricing.get.prices = decode_payload(msg.payload)
+                elif re.search("/optional/ep/get/", msg.topic) is not None:
+                    self.set_json_payload_class(var.data.electricity_pricing.get, msg)
+                elif re.search("/optional/ep/", msg.topic) is not None:
+                    self.set_json_payload_class(var.data.electricity_pricing, msg)
+                elif "module_update_completed" in msg.topic:
+                    self.event_module_update_completed.set()
                 elif re.search("/optional/ocpp/", msg.topic) is not None:
                     config_dict = decode_payload(msg.payload)
                     var.data.ocpp = dataclass_from_dict(Ocpp, config_dict)