Ladeprotokoll: Energie und Kosten je Tarifzeitraum (Feste Tarifstunden)

packages/control/chargelog/chargelog.py

@@ -56,7 +56,9 @@
 #                 "grid": 0.25,
 #                 "pv": 0.25,
 #                 "bat": 0.5,
-#                 "cp": 0}
+#                 "cp": 0},
+#         "tariff_split": {  # optional, nur bei Festen Tarifstunden: geladene Energie und Kosten je Tarifzeitraum
+#                 "NT": {"energy": 1000, "costs": 0.25}}
 #     }
 # }
 log = logging.getLogger("chargelog")
@@ -300,6 +302,9 @@ def _create_entry(chargepoint, charging_ev):
             "power_source": energy_source
         }
     }
+    tariff_split = get_value_or_default(lambda: _get_tariff_split(log_data), {})
+    if tariff_split:
+        new_entry["data"]["tariff_split"] = tariff_split
     return new_entry
 
 
@@ -343,10 +348,31 @@ def calc_energy_costs(cp, create_log_entry: bool = False):
             else:
                 costs = _calc_costs(charged_energy_by_source, reference_entries[-1]["prices"])
                 cp.data.set.log.costs += costs
+                _add_tariff_split(cp.data.set.log, charged_energy_by_source, costs,
+                                  reference_entries[-1]["prices"].get("tariff_name"))
     except Exception:
         log.exception(f"Fehler beim Berechnen der Ladekosten für Ladepunkt {cp.num}")
 
 
+def _add_tariff_split(log_data, charged_energy_by_source: Dict[str, float], costs: float,
+                      tariff_name: Optional[str]) -> None:
+    """ Geladene Energie des Intervalls (über alle Quellen) und seine Kosten, wie sie in costs eingehen, dem
+    Tarifzeitraum (Feste Tarifstunden) zuordnen. Die Kosten je Tarifzeitraum ergeben so in Summe costs.
+    Ohne Tarifname (dynamische Tarife, kein Preismodul) bleibt es bei der Summe in costs.
+    """
+    energy = sum(charged_energy_by_source.values())
+    if tariff_name is None or energy <= 0:
+        return
+    tariff = log_data.tariff_split.setdefault(tariff_name, {"energy": 0, "costs": 0})
+    tariff["energy"] += energy
+    tariff["costs"] += costs
+
+
+def _get_tariff_split(log_data) -> Dict[str, Dict[str, float]]:
+    return {name: {"energy": round(tariff["energy"], 2), "costs": round(tariff["costs"], 2)}
+            for name, tariff in log_data.tariff_split.items()}
+
+
 def calculate_charged_energy_by_source(cp, processed_entries, reference_entries, create_log_entry: bool = False):
     try:
         reference = _get_reference_position(cp, create_log_entry)

packages/control/chargelog/chargelog_test.py

@@ -8,6 +8,7 @@
 from control.chargelog import chargelog
 from control.chargelog.chargelog import calc_energy_costs
 from control.chargepoint.chargepoint import Chargepoint
+from control.chargepoint.chargepoint_data import Log
 from control.general import Prices
 from helpermodules.measurement_logging import process_log_entry_builder
 
@@ -277,6 +278,114 @@ def test_calc_charge_cost_reference_middle_day_change(mock_data, monkeypatch):
     assert round(cp.data.set.log.costs, 5) == 0.5
 
 
+def test_calc_charge_cost_tariff_split(mock_data, monkeypatch):
+    cp = Chargepoint(4, None)
+    cp.data.set.log.imported_since_plugged = cp.data.set.log.imported_since_mode_switch = 3950
+    cp.data.set.log.timestamp_mode_switch = 1652682600  # 8:30
+    cp.data.get.imported = 4050
+    cp.data.set.log.charged_energy_by_source = {'bat': 100, 'cp': 0, 'grid': 100, 'pv': 100}
+    cp.data.set.log.tariff_split = {"NT": {"energy": 1000, "costs": 0.25},
+                                    "Standardpreis": {"energy": 100, "costs": 0.03}}
+    daily_log = mock_daily_log(monkeypatch)
+    daily_log["entries"][-1]["prices"]["tariff_name"] = "NT"
+
+    with patch("builtins.open", mock_open(read_data=json.dumps(daily_log))):
+        calc_energy_costs(cp)
+
+    # die gesamte geladene Energie dieses Intervalls (4000 - 2000 Wh, über alle Quellen) und seine Kosten
+    assert round(cp.data.set.log.tariff_split["NT"]["energy"], 2) == 3000
+    assert round(cp.data.set.log.tariff_split["NT"]["costs"], 5) == 0.75
+    assert cp.data.set.log.tariff_split["Standardpreis"] == {"energy": 100, "costs": 0.03}
+    assert round(cp.data.set.log.costs, 5) == 0.5
+
+
+def test_calc_charge_cost_tariff_split_end_uses_current_entry(mock_data, monkeypatch):
+    cp = Chargepoint(4, None)
+    cp.data.set.log.imported_since_plugged = cp.data.set.log.imported_since_mode_switch = 3950
+    cp.data.set.log.timestamp_mode_switch = 1652682600  # 8:30
+    cp.data.get.imported = 4100
+    cp.data.set.log.charged_energy_by_source = {'grid': 1243, 'pv': 386, 'bat': 671, 'cp': 0.0}
+    daily_log = mock_daily_log(monkeypatch)
+    daily_log["entries"][-1]["prices"]["tariff_name"] = "NT"
+    current_entry = json.loads(json.dumps(daily_log["entries"][-1]))
+    current_entry["bat"]["all"]["exported"] = 4000
+    current_entry["bat"]["bat2"]["exported"] = 4000
+    current_entry["counter"]["counter0"]["imported"] = 4500
+    current_entry["cp"]["cp4"]["imported"] = 4100
+    current_entry["pv"]["all"]["exported"] = 3000
+    current_entry["pv"]["pv1"]["exported"] = 3000
+    current_entry["prices"]["tariff_name"] = "HT"
+    monkeypatch.setattr(chargelog, "create_entry", Mock(return_value=current_entry))
+
+    with patch("builtins.open", mock_open(read_data=json.dumps(daily_log))):
+        calc_energy_costs(cp, True)
+
+    # das angebrochene letzte Intervall gehört zum Tarifzeitraum der aktuellen Messung
+    assert {name: {key: round(value, 5) for key, value in tariff.items()}
+            for name, tariff in cp.data.set.log.tariff_split.items()} == {"HT": {"energy": 100.0, "costs": 0.025}}
+
+
+def test_add_tariff_split_counts_all_sources():
+    log_data = Log()
+    prices = {'grid': 0.0003, 'pv': 0.0001, 'bat': 0.0002, 'cp': 0, 'tariff_name': "NT"}
+    charged_energy_by_source = {'grid': 600, 'pv': 400, 'bat': 0, 'cp': 0}
+    costs = chargelog._calc_costs(charged_energy_by_source, prices)
+
+    chargelog._add_tariff_split(log_data, charged_energy_by_source, costs, prices["tariff_name"])
+    chargelog._add_tariff_split(log_data, {'grid': 0, 'pv': 0, 'bat': 0, 'cp': 0}, 0, "HT")
+
+    # Energie = Zähler des Ladepunkts (PV + Netz), Kosten = Ergebnis von _calc_costs; leere Intervalle zählen nicht
+    assert log_data.tariff_split == {"NT": {"energy": 1000, "costs": costs}}
+    assert costs == round(600 * 0.0003 + 400 * 0.0001, 4)
+
+
+@pytest.mark.parametrize("tariff_name", [pytest.param(None, id="dynamischer Tarif"),
+                                         pytest.param("missing", id="alter Tageslog-Eintrag")])
+def test_calc_charge_cost_without_tariff_name(tariff_name, mock_data, monkeypatch):
+    cp = Chargepoint(4, None)
+    cp.data.set.log.imported_since_plugged = cp.data.set.log.imported_since_mode_switch = 3950
+    cp.data.set.log.timestamp_mode_switch = 1652682600  # 8:30
+    cp.data.get.imported = 4050
+    cp.data.set.log.charged_energy_by_source = {'bat': 100, 'cp': 0, 'grid': 100, 'pv': 100}
+    daily_log = mock_daily_log(monkeypatch)
+    if tariff_name != "missing":
+        daily_log["entries"][-1]["prices"]["tariff_name"] = tariff_name
+
+    with patch("builtins.open", mock_open(read_data=json.dumps(daily_log))):
+        calc_energy_costs(cp)
+
+    assert cp.data.set.log.tariff_split == {}
+    assert round(cp.data.set.log.costs, 5) == 0.5
+
+
+@pytest.mark.parametrize("tariff_split, expected",
+                         [pytest.param({}, None, id="ohne Aufteilung kein Schlüssel"),
+                          pytest.param({"NT": {"energy": 18400.123, "costs": 4.73456}},
+                                       {"NT": {"energy": 18400.12, "costs": 4.73}}, id="mit Aufteilung")])
+def test_create_entry_tariff_split(tariff_split, expected, mock_data, monkeypatch):
+    monkeypatch.setattr(chargelog, "calc_energy_costs", Mock())
+    cp = Chargepoint(4, None)
+    cp.data.get.imported = 4050
+    cp.data.set.log.imported_at_mode_switch = 100
+    cp.data.set.log.timestamp_mode_switch = 1652682600
+    cp.data.set.log.costs = 4.9
+    cp.data.set.log.tariff_split = tariff_split
+
+    entry = chargelog._create_entry(cp, Mock())
+
+    assert entry["data"]["costs"] == 4.9
+    assert entry["data"].get("tariff_split") == expected
+
+
+def test_reset_log_data_chargemode_switch_resets_tariff_split(mock_data):
+    cp = Chargepoint(4, None)
+    cp.data.set.log.tariff_split = {"NT": {"energy": 1000, "costs": 0.25}}
+
+    cp.reset_log_data_chargemode_switch()
+
+    assert cp.data.set.log.tariff_split == {}
+
+
 def test_small_positive_charge_is_assigned_to_energy_sources(mock_data, monkeypatch):
     cp = Chargepoint(4, None)
     charged_energy = 80

packages/control/chargelog/process_chargelog.py

@@ -1,7 +1,7 @@
 import json
 import logging
 import pathlib
-from typing import Dict
+from typing import Dict, List
 
 from helpermodules import timecheck
 
@@ -112,13 +112,33 @@ def get_sum(entry_name: str) -> float:
         power_sum = get_sum("power")
         costs_sum = get_sum("costs")
         power_sum = power_sum / len(log_data["entries"])
-        return {
+        totals = {
             "time_charged": duration_sum,
             "range_charged": range_charged_sum,
             "imported_since_mode_switch": mode_sum,
             "power": power_sum,
             "costs": costs_sum,
         }
+        tariff_split_sum = get_tariff_split_sum(log_data["entries"])
+        if tariff_split_sum:
+            totals["tariff_split"] = tariff_split_sum
+        return totals
+
+
+def get_tariff_split_sum(entries: List[Dict]) -> Dict[str, Dict[str, float]]:
+    """ Geladene Energie und Kosten je Tarifzeitraum summieren. Einträge ohne Aufteilung (ältere Einträge,
+    dynamische Tarife) werden übersprungen, data.costs bleibt die maßgebliche Summe.
+    """
+    tariff_split_sum = {}
+    for entry in entries:
+        try:
+            for name, tariff in (entry["data"].get("tariff_split") or {}).items():
+                tariff_sum = tariff_split_sum.setdefault(name, {"energy": 0, "costs": 0})
+                tariff_sum["energy"] = round(tariff_sum["energy"] + tariff.get("energy", 0), 2)
+                tariff_sum["costs"] = round(tariff_sum["costs"] + tariff.get("costs", 0), 2)
+        except Exception:
+            log.exception("Fehler beim Summieren der Tarifaufteilung")
+    return tariff_split_sum
 
 
 def _get_parent_file() -> pathlib.Path:

packages/control/chargelog/process_chargelog_test.py

@@ -0,0 +1,30 @@
+from control.chargelog.process_chargelog import get_totals_of_filtered_log_data
+
+
+def _entry(costs: float, tariff_split=None) -> dict:
+    data = {"range_charged": 10, "imported_since_mode_switch": 1000, "power": 1000, "costs": costs}
+    if tariff_split is not None:
+        data["tariff_split"] = tariff_split
+    return {"time": {"time_charged": "0:30"}, "data": data}
+
+
+def test_totals_tariff_split():
+    log_data = {"entries": [
+        _entry(1.0),  # Eintrag ohne Aufteilung (ältere Version, dynamischer Tarif)
+        _entry(2.0, {"NT": {"energy": 1000.1, "costs": 0.25}, "Standardpreis": {"energy": 500, "costs": 0.15}}),
+        _entry(3.0, {"NT": {"energy": 2000.2, "costs": 0.5}}),
+        _entry(4.0, {"NT": {"energy": None}}),  # unbrauchbarer Eintrag wird übersprungen
+    ]}
+
+    totals = get_totals_of_filtered_log_data(log_data)
+
+    assert totals["costs"] == 10.0
+    assert totals["tariff_split"] == {"NT": {"energy": 3000.3, "costs": 0.75},
+                                      "Standardpreis": {"energy": 500, "costs": 0.15}}
+
+
+def test_totals_without_tariff_split():
+    totals = get_totals_of_filtered_log_data({"entries": [_entry(1.0), _entry(2.0)]})
+
+    assert totals["costs"] == 3.0
+    assert "tariff_split" not in totals

packages/control/chargepoint/chargepoint_data.py

@@ -99,6 +99,8 @@ class Log:
     range_at_start: Optional[float] = None
     range_at_end: Optional[float] = None
     odometer: Optional[float] = None
+    # geladene Energie und Kosten je Tarifzeitraum (Feste Tarifstunden): {Name: {"energy": Wh, "costs": €}}
+    tariff_split: Dict[str, Dict[str, float]] = field(default_factory=empty_dict_factory)
 
 
 def connected_vehicle_factory() -> ConnectedVehicle:

packages/control/optional.py

@@ -18,10 +18,12 @@
 from modules.common.configurable_forecast import ConfigurableForecast
 from modules.common.configurable_tariff import ConfigurableFlexibleTariff, ConfigurableGridFee
 from modules.common.configurable_monitoring import ConfigurableMonitoring
+from modules.electricity_pricing.flexible_tariffs.fixed_hours.tariff import get_tariff_name_for_time_slot
 
 log = logging.getLogger(__name__)
 AS_EURO_PER_KWH = 1000.0  # Umrechnung von €/Wh in €/kWh
 MQTT_PREFIX = "openWB/set/optional/ep"
+DEFAULT_TARIFF_NAME = "Standardpreis"
 
 
 class Optional(OcppMixin):
@@ -220,6 +222,24 @@ def ep_get_current_price(self) -> float:
         else:
             raise Exception("Kein Anbieter für strompreisbasiertes Laden konfiguriert.")
 
+    def ep_get_current_tariff_name(self) -> TypingOptional[str]:
+        """ Name des Tarifzeitraums (Feste Tarifstunden) für denselben Zeitslot, dessen Preis ep_get_current_price()
+        liefert. None, wenn ein konfiguriertes Preismodul keine festen Tarifstunden hat (z.B. dynamische Tarife).
+        """
+        try:
+            modules = [module for module in (self._flexible_tariff_module, self._grid_fee_module)
+                       if module is not None]
+            if (self.data.electricity_pricing.configured is False or len(modules) == 0 or
+                    any(module.config.type != "fixed_hours" for module in modules)):
+                return None
+            time_slot = datetime.fromtimestamp(self.__get_current_timeslot_start())
+            return " / ".join(get_tariff_name_for_time_slot(module.config.configuration, time_slot,
+                                                            DEFAULT_TARIFF_NAME)
+                              for module in modules)
+        except Exception as e:
+            log.exception("Fehler beim Ermitteln des Tarifzeitraums: %s", e)
+            return None
+
     def __calculate_price_timeslot_length(self, prices: dict) -> int:
         first_timestamps = sorted(list(prices.keys()))[:2]
         return float(first_timestamps[1]) - float(first_timestamps[0])

packages/control/optional_test.py

@@ -1,8 +1,11 @@
+import datetime
 from dataclasses import dataclass
 from unittest.mock import Mock
 import pytest
 from helpermodules import timecheck
 from control.optional import Optional
+from modules.electricity_pricing.flexible_tariffs.fixed_hours.config import (FixedHoursTariff,
+                                                                             FixedHoursTariffConfiguration)
 
 
 @dataclass
@@ -499,6 +502,68 @@ def test_et_price_update_required(monkeypatch, flexible_tariff, grid_fee, curren
     assert result == expected
 
 
+def _fixed_hours_module(name: str, times: list) -> Mock:
+    return Mock(config=FixedHoursTariff(configuration=FixedHoursTariffConfiguration(
+        default_price=0.0003,
+        tariffs=[{"name": name, "price": 0.0002,
+                  "active_times": {"dates": [("01-01", "31-12")], "times": times, "weekdays": [0, 1, 2, 3, 4]}}])))
+
+
+NT = _fixed_hours_module("NT", [("00:00", "07:00"), ("23:00", "24:00")])
+MODULE_3 = _fixed_hours_module("Modul 3 NT", [("22:00", "24:00")])
+DYNAMIC = Mock(config=Mock(type="tibber"))
+# Montag 16.05.2022 23:15 und 23:30: der erste Preis-Eintrag bestimmt den Zeitslot, nicht now()
+SLOT_NT = int(datetime.datetime(2022, 5, 16, 23, 15).timestamp())
+SLOT_DAY = int(datetime.datetime(2022, 5, 16, 12, 0).timestamp())
+
+
+@pytest.mark.parametrize(
+    "flexible_tariff, grid_fee, first_slot, expected",
+    [
+        pytest.param(NT, None, SLOT_NT, "NT", id="feste Tarifstunden, Tarifzeitraum"),
+        pytest.param(NT, None, SLOT_DAY, "Standardpreis", id="feste Tarifstunden, Standardpreis"),
+        pytest.param(None, MODULE_3, SLOT_NT, "Modul 3 NT", id="nur Netzentgelt mit festen Tarifstunden"),
+        pytest.param(NT, MODULE_3, SLOT_NT, "NT / Modul 3 NT", id="Tarif und Netzentgelt mit festen Tarifstunden"),
+        pytest.param(DYNAMIC, None, SLOT_NT, None, id="dynamischer Tarif"),
+        pytest.param(DYNAMIC, MODULE_3, SLOT_NT, None, id="dynamischer Tarif mit festem Netzentgelt"),
+    ]
+)
+def test_ep_get_current_tariff_name(flexible_tariff, grid_fee, first_slot, expected):
+    # setup
+    opt = Optional()
+    opt._flexible_tariff_module = flexible_tariff
+    opt._grid_fee_module = grid_fee
+    opt.data.electricity_pricing.configured = True
+    opt.data.electricity_pricing.get.prices = {str(first_slot): 0.0002, str(first_slot + 900): 0.0003}
+
+    # execution
+    result = opt.ep_get_current_tariff_name()
+
+    # evaluation
+    assert result == expected
+
+
+@pytest.mark.parametrize(
+    "configured, prices",
+    [
+        pytest.param(False, {str(SLOT_NT): 0.0002}, id="nicht konfiguriert"),
+        pytest.param(True, {}, id="keine Preise"),
+    ]
+)
+def test_ep_get_current_tariff_name_none(configured, prices):
+    # setup
+    opt = Optional()
+    opt._flexible_tariff_module = NT
+    opt.data.electricity_pricing.configured = configured
+    opt.data.electricity_pricing.get.prices = prices
+
+    # execution
+    result = opt.ep_get_current_tariff_name()
+
+    # evaluation
+    assert result is None
+
+
 class TestForecastModule:
     def _configured_opt(self) -> Optional:
         opt = Optional()

packages/helpermodules/measurement_logging/process_log.py

@@ -30,6 +30,7 @@ def get_default_charge_log_columns() -> Dict:
         "time_time_charged": True,
         "data_costs": True,
         "data_power_source": True,
+        "data_tariff_split": False,
         "vehicle_name": True,
         "vehicle_chargemode": True,
         "vehicle_prio": True,