diff --git a/.gitignore b/.gitignore index b44b30afe8..c2d4c15890 100644 --- a/.gitignore +++ b/.gitignore @@ -16,11 +16,14 @@ data/config/eebus/certs/* data/log/* data/charge_log/* data/daily_log/* +data/daily_totals/* data/monthly_log/* +data/monthly_totals/* data/backup/*.tar.gz data/backup/*.openwb-backup* data/restore/*.tar.gz data/data_migration/*.tar.gz +data/generate_totals.lock ramdisk/* web/lastcommit !.gitignore diff --git a/data/config/mosquitto/public/default-dynamic-security.json b/data/config/mosquitto/public/default-dynamic-security.json index 5e850ad871..8e0cbba91a 100644 --- a/data/config/mosquitto/public/default-dynamic-security.json +++ b/data/config/mosquitto/public/default-dynamic-security.json @@ -194,6 +194,12 @@ "priority": 0, "allow": true }, + { + "acltype": "publishClientReceive", + "topic": "openWB/system/log_data_ready", + "priority": 0, + "allow": true + }, { "acltype": "publishClientReceive", "topic": "openWB/system/usage_terms_acknowledged", diff --git a/packages/control/chargelog/chargelog.py b/packages/control/chargelog/chargelog.py index 621dbb06d4..6468f55c8c 100644 --- a/packages/control/chargelog/chargelog.py +++ b/packages/control/chargelog/chargelog.py @@ -11,7 +11,7 @@ from typing import Any, Dict, List, Optional, Tuple from control import data -from helpermodules.measurement_logging.process_log import ( +from helpermodules.measurement_logging.process_log_calculation import ( FILE_ERRORS, CalculationType, _analyse_energy_source, _process_entries, get_totals) # alte Daten: Startzeitpunkt der Ladung, Endzeitpunkt, Geladene Reichweite, Energie, Leistung, Ladedauer, LP-Nummer, diff --git a/packages/helpermodules/logger.py b/packages/helpermodules/logger.py index d3c62246de..73fdf53719 100644 --- a/packages/helpermodules/logger.py +++ b/packages/helpermodules/logger.py @@ -229,6 +229,33 @@ def combine_logs(base_path: str, name: str): f.write(logs) +# The totals subprocess needs its own log file, including messages from process_log. +# In the main process, process_log must keep propagating to main log. +def setup_generate_totals_logging(include_process_log: bool = False) -> None: + generate_totals_log = logging.getLogger("generate_totals") + generate_totals_log.setLevel(logging.DEBUG) + generate_totals_log.propagate = False + RAMDISK_PATH.mkdir(parents=True, exist_ok=True) + handler = next( + (handler for handler in generate_totals_log.handlers + if isinstance(handler, RotatingFileHandler) + and handler.baseFilename == str(RAMDISK_PATH / 'generate_totals.log')), + None, + ) + if handler is None: + handler = RotatingFileHandler(RAMDISK_PATH / 'generate_totals.log', maxBytes=5 * 1000000, backupCount=1) + handler.setFormatter(logging.Formatter(FORMAT_STR_SHORT)) + handler.addFilter(RedactingFilter()) + generate_totals_log.addHandler(handler) + + if include_process_log: + process_log = logging.getLogger("helpermodules.measurement_logging.process_log") + process_log.setLevel(logging.DEBUG) + process_log.propagate = False + if handler not in process_log.handlers: + process_log.addHandler(handler) + + def setup_logging() -> None: def mb_to_bytes(megabytes: int) -> int: return megabytes * 1000000 @@ -274,6 +301,9 @@ def mb_to_bytes(megabytes: int) -> int: mqtt_file_handler.addFilter(RedactingFilter()) mqtt_log.addHandler(mqtt_file_handler) + # Totals generation logger + setup_generate_totals_logging() + # Steuve control command logger steuve_control_command_log = logging.getLogger("steuve_control_command") steuve_control_command_log.propagate = False diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py new file mode 100644 index 0000000000..7f722ae55a --- /dev/null +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess.py @@ -0,0 +1,93 @@ +import fcntl +import json +from pathlib import Path +from datetime import date + +from typing import List +from helpermodules import logger, pub +from helpermodules.measurement_logging.process_log import (save_daily_source_totals, + save_monthly_source_totals) + +import logging + + +BASE_PATH = Path(__file__).resolve().parents[3] +logger.setup_generate_totals_logging(include_process_log=True) +log = logging.getLogger("generate_totals") + +LOCK_FILE = BASE_PATH / "data" / "generate_totals.lock" + + +def get_all_days_to_calc(): + try: + daily_log_dir = BASE_PATH / "data" / "daily_log" + if not daily_log_dir.is_dir(): + return None + + today_stem = date.today().strftime("%Y%m%d") + daily_log_files = sorted( + [p for p in daily_log_dir.glob("*.json") + if p.stem.isdigit() and p.stem < today_stem], + key=lambda p: p.stem, + ) + + log.debug(f"Anzahl zu berechnender Tageslogs: {len(daily_log_files)}") + return daily_log_files + except Exception: + log.debug("Fehler beim Abrufen der Tageslogs. Es werden keine Tageslogs berechnet.") + return None + + +def get_all_months_to_calc(days: List[str] = None): + if days is None: + return None + current_month = date.today().strftime("%Y%m") + months = sorted({p.stem[:6] for p in days if len(p.stem) == 8 and p.stem[:6] < current_month}) + log.debug(f"Anzahl zu berechnender Monatslogs: {len(months)}") + return months + + +def _generate_totals(): + + errors = 0 + days_to_calc = get_all_days_to_calc() or [] + months_to_calc = get_all_months_to_calc(days_to_calc) or [] + + for day in days_to_calc: + try: + save_daily_source_totals(day.stem, saving=True) + except Exception: + log.exception(f"Fehler beim Generieren der Tageserträge für {day.stem}, Tag wird übersprungen.") + errors += 1 + continue + + for month in months_to_calc: + try: + save_monthly_source_totals(month, saving=True) + except Exception: + log.exception(f"Fehler beim Generieren der Monatswerte für {month}, Monat wird übersprungen.") + errors += 1 + continue + + log.info(f"Totals-Migration abgeschlossen. Fehlerhafte Logs: {errors}.") + publisher = pub.Pub() + publish_info = publisher.publisher.client.publish( + "openWB/set/system/log_data_ready", json.dumps(True), qos=1, retain=True) + publish_info.wait_for_publish() + + +def generate_totals(): + LOCK_FILE.parent.mkdir(parents=True, exist_ok=True) + + with LOCK_FILE.open("w") as lock_file: + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except Exception: + log.exception("generate_totals_subprocess läuft bereits. " + "Der zweite Prozess wird beendet.") + return + _generate_totals() + + +if __name__ == "__main__": + generate_totals() diff --git a/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py new file mode 100644 index 0000000000..81d03cb521 --- /dev/null +++ b/packages/helpermodules/measurement_logging/generate_totals_subprocess_test.py @@ -0,0 +1,43 @@ +from unittest.mock import Mock + +from helpermodules.measurement_logging import generate_totals_subprocess + + +def test_generate_totals_calls_generator_when_lock_is_acquired(monkeypatch, tmp_path): + # setup + lock_file = tmp_path / "locks" / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) + + flock_mock = Mock() + generate_mock = Mock() + + monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", flock_mock) + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) + + # execution + generate_totals_subprocess.generate_totals() + + # evaluation + flock_mock.assert_called_once() + generate_mock.assert_called_once() + assert lock_file.parent.is_dir() + + +def test_generate_totals_skips_generator_when_lock_is_not_acquired(monkeypatch, tmp_path): + # setup + lock_file = tmp_path / "locks" / "generate_totals.lock" + monkeypatch.setattr(generate_totals_subprocess, "LOCK_FILE", lock_file) + + flock_mock = Mock(side_effect=RuntimeError("already locked")) + generate_mock = Mock() + + monkeypatch.setattr(generate_totals_subprocess.fcntl, "flock", flock_mock) + monkeypatch.setattr(generate_totals_subprocess, "_generate_totals", generate_mock) + + # execution + generate_totals_subprocess.generate_totals() + + # evaluation + flock_mock.assert_called_once() + generate_mock.assert_not_called() + assert lock_file.parent.is_dir() diff --git a/packages/helpermodules/measurement_logging/process_log.py b/packages/helpermodules/measurement_logging/process_log.py index b7415e7480..8df895de7f 100644 --- a/packages/helpermodules/measurement_logging/process_log.py +++ b/packages/helpermodules/measurement_logging/process_log.py @@ -1,27 +1,28 @@ -from enum import Enum + +from copy import deepcopy import json import logging from pathlib import Path -from typing import Dict, List, Optional, Tuple, Union - +from typing import Dict, List, Optional +from datetime import datetime from helpermodules import timecheck -from helpermodules.measurement_logging.write_log import (LegacySmartHomeLogData, LogType, create_entry, - get_previous_entry) -from helpermodules.messaging import MessageType, pub_system_message -from helpermodules.utils.precision_math import decimal_add, decimal_divide, decimal_multiply, decimal_subtract +from helpermodules.utils.precision_math import decimal_divide +from helpermodules.measurement_logging.process_log_calculation import (CalculationType, + FILE_ERRORS, + analyse_percentage_totals, + analyse_percentage, + get_totals, + _analyse_energy_source, + _process_entries) +from helpermodules.measurement_logging.write_log import ( + LegacySmartHomeLogData, + create_entry, + get_previous_entry +) log = logging.getLogger(__name__) -class CalculationType(Enum): - ALL = 0 - POWER = 1 - ENERGY = 2 - - -FILE_ERRORS = (FileNotFoundError, json.decoder.JSONDecodeError) - - def get_default_charge_log_columns() -> Dict: return { "time_begin": True, @@ -184,52 +185,6 @@ def convert_legacy_units(data: dict) -> dict: return data -def safe_get_nested(data: Dict, *keys, default: Union[int, float] = 0) -> Union[int, float]: - current = data - for key in keys: - if isinstance(current, dict) and key in current: - current = current[key] - else: - return default - return current if isinstance(current, (int, float)) else default - - -def get_totals(entries: List, process_entries: bool = True) -> Dict: - """ Berechnet aus der übergebenen Liste "entries" die Summen (totals). - "process_entries" besagt, ob die Differenzen der einzelnen Einträge noch - berechnet werden müssen. - """ - if process_entries: - entries = _process_entries(entries, CalculationType.ENERGY) - totals = {"consumer": {}, "cp": {}, "counter": {}, "pv": {}, "bat": {}, "sh": {}, "hc": {}} - for totals_group in totals.keys(): - for entry in entries: - if totals_group in entry: - for entry_module in entry[totals_group]: - try: - if entry_module not in totals[totals_group]: - if totals_group == "hc": - totals[totals_group][entry_module] = {"energy_imported": 0.0} - elif totals_group == "pv": - totals[totals_group][entry_module] = {"energy_exported": 0.0} - else: - totals[totals_group][entry_module] = {"energy_imported": 0.0, "energy_exported": 0.0} - if totals_group == "counter" and "grid" in entry[totals_group][entry_module]: - totals[totals_group][entry_module]["grid"] = entry[ - totals_group][entry_module]["grid"] - for entry_module_key, entry_module_value in entry[totals_group][entry_module].items(): - if "grid" != entry_module_key and entry_module_key in totals[totals_group][entry_module]: - # avoid floating point issues with using Decimal - current_total = totals[totals_group][entry_module][entry_module_key] - totals[totals_group][entry_module][entry_module_key] = decimal_add( - current_total, entry_module_value) # totals in Wh! - - except Exception: - log.exception(f"Fehler beim Berechnen der Summe von {entry_module}; " - f"group:{totals_group}, module:{entry_module}, key:{entry_module_key}") - return totals - - def get_daily_log(date: str): data = _collect_daily_log_data(date) data["entries"] = _process_entries(data["entries"], CalculationType.ALL) @@ -245,8 +200,8 @@ def _collect_daily_log_data(date: str): log_data = json.load(json_file) if date == timecheck.create_timestamp_YYYYMMDD(): # beim aktuellen Tag den aktuellen Datensatz ergänzen - log_data["entries"].append(create_entry( - LogType.DAILY, LegacySmartHomeLogData(), get_previous_entry(parent_file, log_data))) + log_data["entries"].append(create_entry(LegacySmartHomeLogData(), + get_previous_entry(parent_file, log_data))) else: # bei älteren als letzten Datensatz den des nächsten Tags try: @@ -263,481 +218,305 @@ def _collect_daily_log_data(date: str): def get_monthly_log(date: str): - data = _collect_monthly_log_data(date) - data["entries"] = _process_entries(data["entries"], CalculationType.ENERGY) - data["totals"] = get_totals(data["entries"], False) - data = _analyse_energy_source(data) - return data + + if not (len(date) == 6 and date.isdigit()): + log.debug(f"Ungültiges Datum für Monats-Summen: {date}") + return {"entries": [], "names": {}, "colors": {}, "totals": {}} + # Nur Logs ab dem ältesten Tageslog auswerten + # Sonst werden unnötige Totals-Werte gespeichert + oldest_log_day = _oldest_log_day() + if (oldest_log_day is None + or date < oldest_log_day[:6]): # Jahr und Monat + return {"entries": [], "names": {}, "colors": {}, "totals": {}} + + monthly_entries = [] + monthly_names = {} + monthly_colors = {} + + today = timecheck.create_timestamp_YYYYMMDD() + day = f"{date}01" + + while day.startswith(date): + # Zukunftstage/-monate nicht verarbeiten, um keine leeren daily_totals zu erzeugen. + if day > today: + break + if day < oldest_log_day: + day = timecheck.get_relative_date_string(day, day_offset=1) + continue + + content = load_daily_source_totals_content(day) + if content is None: + # Dürfte eigentlich nie passieren... + content = save_daily_source_totals(day, saving=(day != today)) + + if isinstance(content, dict): + daily_totals = content.get("totals") + daily_entry = content.get("entry") + + if isinstance(daily_totals, dict) and isinstance(daily_entry, dict) and len(daily_entry) > 0: + daily_entry = deepcopy(daily_entry) + daily_entry["date"] = day + _apply_source_totals(daily_entry, daily_totals) + + monthly_entries.append(daily_entry) + if isinstance(content.get("names"), dict): + monthly_names.update(content["names"]) + if isinstance(content.get("colors"), dict): + monthly_colors.update(content["colors"]) + + day = timecheck.get_relative_date_string(day, day_offset=1) + + if len(monthly_entries) > 0: + data = {"entries": monthly_entries, "names": monthly_names, "colors": monthly_colors} + data["totals"] = get_totals(data["entries"], False) + data["totals"] = analyse_percentage_totals(data["entries"], data["totals"]) + return data + + # Fallback, wenn keine Daten vorhanden sind + return {"entries": [], "names": {}, "colors": {}, "totals": {}} + + +def get_yearly_log(year: str): + # Nur Logs ab dem ältesten Tageslog auswerten + # Sonst werden unnötige totals Werte gespeichert + oldest_log_day = _oldest_log_day() + if oldest_log_day is None or year < oldest_log_day[:4]: + return {"entries": [], "names": {}, "colors": {}, "totals": {}} + + monthly_entries = [] + monthly_names = {} + monthly_colors = {} + this_month = timecheck.create_timestamp_YYYYMM() + month = f"{year}01" + + while month.startswith(year): + if month > this_month: + break + + if month < oldest_log_day[:6]: + month = timecheck.get_relative_date_string(month, month_offset=1) + continue + + content = load_monthly_source_totals_content(month) + if content is None: + # Dürfte eigentlich nie passieren... + content = save_monthly_source_totals(month, None, saving=(month != this_month)) + + if isinstance(content, dict): + monthly_totals = content.get("totals") + monthly_entry = content.get("entry") + + if isinstance(monthly_totals, dict) and isinstance(monthly_entry, dict) and len(monthly_entry) > 0: + monthly_entry = deepcopy(monthly_entry) + monthly_entry["date"] = month + _apply_source_totals(monthly_entry, monthly_totals) + + monthly_entries.append(monthly_entry) + if isinstance(content.get("names"), dict): + monthly_names.update(content["names"]) + if isinstance(content.get("colors"), dict): + monthly_colors.update(content["colors"]) + month = timecheck.get_relative_date_string(month, month_offset=1) + + if len(monthly_entries) > 0: + data = {"entries": monthly_entries, "names": monthly_names, "colors": monthly_colors} + data["totals"] = get_totals(data["entries"], False) + data["totals"] = analyse_percentage_totals(data["entries"], data["totals"]) + return data + + # Fallback, wenn keine Daten vorhanden sind + return {"entries": [], "names": {}, "colors": {}, "totals": {}} + + +def _get_data_folder_path() -> str: + return str(Path(__file__).resolve().parents[3] / "data") -def _collect_monthly_log_data(date: str): +def save_daily_source_totals(date: str, saving: bool = True): try: - with open(f"{_get_data_folder_path()}/monthly_log/{date}.json", "r") as jsonFile: - log_data = json.load(jsonFile) - this_month = timecheck.create_timestamp_YYYYMM() - if date == this_month: - # add last entry of current day, if current month is requested - try: - today = timecheck.create_timestamp_YYYYMMDD() - with open(f"{_get_data_folder_path()}/daily_log/{today}.json", - "r") as todayJsonFile: - today_log_data = json.load(todayJsonFile) - if len(today_log_data["entries"]) > 0: - log_data["entries"].append(today_log_data["entries"][-1]) - except FILE_ERRORS: - pass - else: - # add first entry of next month - try: - next_date = timecheck.get_relative_date_string(date, month_offset=1) - with open(f"{_get_data_folder_path()}/monthly_log/{next_date}.json", - "r") as nextJsonFile: - next_log_data = json.load(nextJsonFile) - log_data["entries"].append(next_log_data["entries"][0]) - except FILE_ERRORS: - pass + if not (len(date) == 8 and date.isdigit()): + log.debug(f"Ungültiges Datum für Tages-Summen: {date}") + return None + data = _collect_daily_log_data(date) + source_entries = data.get("entries", []) + + # Keep one source snapshot for the daily output entry and process entries in place to avoid full-copy peaks. + source_daily_entry = _get_last_entry_for_period(source_entries, date, "%Y%m%d") + processed_entries = _process_entries(source_entries, calculation=CalculationType.ENERGY) + + totals = get_totals(processed_entries, process_entries=False) + analysed_data = _analyse_energy_source({ + "entries": processed_entries, + "totals": totals, + "names": data.get("names", {}) + }) + totals = analysed_data["totals"] + + daily_entry = {} + if source_daily_entry is not None: + daily_entry = deepcopy(source_daily_entry) + daily_entry["date"] = date + _apply_source_totals(daily_entry, totals) + + # Erzeugt Ordner daily_totals, falls nicht vorhanden + totals_dir = Path(_get_data_folder_path()) / "daily_totals" + filepath = totals_dir / f"{date}_totals.json" + + content = { + "date": date, + "totals": totals, + "entry": daily_entry, + "names": data.get("names", {}), + "colors": data.get("colors", {}) + } + + if saving: + totals_dir.mkdir(parents=True, exist_ok=True) + with open(str(filepath), "w") as jsonFile: + json.dump(content, jsonFile, ensure_ascii=False, indent=2) + + log.debug(f"Tages-Summen für {date} gespeichert in {filepath}") + + return content + except FILE_ERRORS: - log_data = {"entries": [], "names": {}} - return log_data + log.exception(f"Fehler beim Speichern der Tages-Summen für {date}") -def get_yearly_log(year: str): - data = _collect_yearly_log_data(year) - data["entries"] = _process_entries(data["entries"], CalculationType.ENERGY) - data["totals"] = get_totals(data["entries"], False) - data = _analyse_energy_source(data) - return data +def load_daily_source_totals_content(date: str): + try: + if not (len(date) == 8 and date.isdigit()): + log.debug(f"Ungültiges Datum für Tages-Summen: {date}") + return None + filepath = f"{_get_data_folder_path()}/daily_totals/{date}_totals.json" + if not Path(filepath).is_file(): + log.debug(f"Keine Tages-Summen-Datei gefunden: {filepath}") + return None -def _collect_yearly_log_data(year: str): - def add_monthly_log(month: str, check_next_month: bool = False) -> None: - monthly_log_path = Path(__file__).resolve().parents[3]/"data"/"monthly_log" - try: - with open(monthly_log_path / f"{month}.json", "r") as jsonFile: - content = json.load(jsonFile) - entries.append(content["entries"][0]) - # add last entry of current file if next file is missing - if check_next_month: - next_month = timecheck.get_relative_date_string(month, month_offset=1) - if not (monthly_log_path / (next_month+".json")).is_file(): - entries.append(content["entries"][-1]) - log.debug(f"Keine Logdatei für Monat {next_month} gefunden, " - f"füge letzten Datensatz von {month} ein: {entries[-1]['date']}") - names.update(content["names"]) - colors.update(content["colors"]) - except FILE_ERRORS: - log.debug(f"Kein Log für Monat {month} gefunden.") - - def add_daily_log(day: str) -> None: - try: - with open(f"{_get_data_folder_path()}/daily_log/{day}.json", "r") as dayJsonFile: - day_log_data = json.load(dayJsonFile) - if len(day_log_data["entries"]) > 0: - entries.append(day_log_data["entries"][-1]) - except FILE_ERRORS: - pass - - entries = [] - names = {} - colors = {} - dates = [] - - # we have to find a valid data range - this_year = timecheck.create_timestamp_YYYY() - this_month = timecheck.create_timestamp_YYYYMM() - if year < this_year: - # if the requested year is in the past, just add all possible months - for month in range(1, 13): - dates.append(f"{year}{month:02}") - else: - # add all months until current month - for month in range(1, int(this_month[-2:])+1): - dates.append(f"{year}{month:02}") - # add data for month range - for date in dates: - try: - log.debug(f"add regular month: {date}") - add_monthly_log(date, date != this_month) - except Exception: - log.exception(f"Fehler beim Zusammenstellen der Jahresdaten für Monat {date}") - - # now we have to find a valid "next" entry for proper calculation - if year == this_year: # current year - # add todays last entry - this_day = timecheck.create_timestamp_YYYYMMDD() - try: - log.debug(f"add today: {this_day}") - add_daily_log(this_day) - except Exception: - log.exception(f"Fehler beim Zusammenstellen der Jahresdaten für den aktuellen Tag {this_day}") - else: - # no special handling here, just add first entry of next month - next_date = f"{int(year)+1}01" - try: - log.debug(f"add next month: {next_date}") - add_monthly_log(next_date) - except Exception: - log.exception(f"Fehler beim Zusammenstellen der Jahresdaten für Monat {next_date}") - - # return our data - return {"entries": entries, "names": names, "colors": colors} - - -def _analyse_energy_source(data, calc_cp: Optional[str] = None) -> Dict: - if data and len(data["entries"]) > 0: - try: - if data.get("message") is None: - data["message"] = "" - for i in range(0, len(data["entries"])): - data["entries"][i], message_analyse = analyse_percentage(data["entries"][i]) - data["entries"][i], message_calc = calc_energy_imported_by_source( - data["entries"][i], data["names"], message_key_filter=calc_cp) - data["message"] += message_analyse + message_calc - data["totals"] = analyse_percentage_totals(data["entries"], data["totals"]) - except Exception: - log.exception("Fehler beim Analysieren der Energiequellen") - pub_system_message({}, "Fehler beim Berechnen des Strom-Mix", MessageType.ERROR) - data["message"] = "Fehler beim Berechnen des Strom-Mix." - return data + with open(str(filepath), "r") as jsonFile: + content = json.load(jsonFile) + log.debug(f"Tages-Summen für {date} geladen aus {filepath}") + return content -def analyse_percentage(entry) -> Tuple[Dict, str]: - EOOR_STATE_MSG = ("Der Strom-Mix um " + entry["date"] + - " konnte nicht berechnet werden, da sich {} im Fehlerzustand befindet. Alle Verbräuche werden" + - " dem Netz zugerechnet.\n") + except FILE_ERRORS: + log.exception(f"Fehler beim Laden der Tages-Summen für {date}") - def format(value): - return round(value, 4) - def get_grid_counter(entry) -> Dict: - # es gibt nur einen Zähler am EVU-Punkt - for counter in entry["counter"].values(): - if counter.get("grid") is True: - return counter - else: - raise KeyError(f"Kein Zähler für das Netz gefunden in Eintrag '{entry['timestamp']}'.") +def save_monthly_source_totals(date: str, data: Optional[Dict] = None, saving: bool = True): try: - message = "" - grid_counter = get_grid_counter(entry) - # Wenn neben dem "all" Eintrag kein weiterer Eintrag existiert, dann gibt es keine Komponenten. - if ((safe_get_nested(entry, "bat", "all", "fault_state") == 2 and len(entry.get("bat", {})) > 1) or - (safe_get_nested(entry, "cp", "all", "fault_state") == 2 and len(entry.get("cp", {})) > 1) or - (safe_get_nested(entry, "pv", "all", "fault_state") == 2 and len(entry.get("pv", {})) > 1) or - grid_counter.get("fault_state", None) == 2): - - entry["energy_source"] = {"grid": 1, "pv": 0, "bat": 0, "cp": 0} - if safe_get_nested(entry, "bat", "all", "fault_state") == 2 and len(entry.get("bat", {})) > 1: - message += EOOR_STATE_MSG.format("mind. einer der Speicher") - if safe_get_nested(entry, "cp", "all", "fault_state") == 2 and len(entry.get("cp", {})) > 1: - message += EOOR_STATE_MSG.format("mind. einer der Ladepunkte") - if safe_get_nested(entry, "pv", "all", "fault_state") == 2 and len(entry.get("pv", {})) > 1: - message += EOOR_STATE_MSG.format("mind. einer der Wechselrichter") - if grid_counter.get("fault_state", None) == 2: - message += EOOR_STATE_MSG.format("der Zähler für das Netz") - - else: - bat_imported = safe_get_nested(entry, "bat", "all", "energy_imported") - bat_exported = safe_get_nested(entry, "bat", "all", "energy_exported") - cp_exported = safe_get_nested(entry, "cp", "all", "energy_exported") - pv_exported = safe_get_nested(entry, "pv", "all", "energy_exported") - grid_imported = grid_counter.get("energy_imported", 0) - grid_exported = grid_counter.get("energy_exported", 0) - consumption = grid_imported - grid_exported + pv_exported + bat_exported - bat_imported + cp_exported - if consumption < 0: - consumption = 0 - - try: - # Berechnung der Energiequellenanteile: - # Da die genaue Aufteilung der Energiequellen nicht bekannt ist, - # wird die Einspeisung (grid_exported) entsprechend der folgenden Priorität aufgeteilt: - # 1. PV - # 2. Batterie - # 3. CP - # Sollte die Einspeisung nicht komplett von PV gedeckt werden, - # wird der Rest von der Batterie übernommen, und falls nötig, vom CP. - # - # Entsprechend ähnlich wird der Batterieimport nach folgender Priorität aufgeteilt: - # 1. PV - # 2. Grid - # 3. CP - # - # Anschließend wird der Verbrauch (ohne Einspeisung und Batterieimport) auf energy_source aufgeteilt. - if consumption <= 0: - entry["energy_source"] = {"grid": 0, "pv": 0, "bat": 0, "cp": 0} - else: - - direct = {"grid": grid_imported, "pv": pv_exported, "bat": bat_exported, "cp": cp_exported} - - # Einspeißung aufteilen - unassigned_export = grid_exported - for source in ("pv", "bat", "cp"): - if direct[source] > unassigned_export: - direct[source] -= unassigned_export - unassigned_export = 0 - break - else: - unassigned_export -= direct[source] - direct[source] = 0 - - if unassigned_export > 0: - # Fehler / inkonsistente Energiebilanz - log.warning( - f"grid_exported konnte nicht vollständig verteilt werden. " - f"Unverteilter Anteil: {unassigned_export}" - ) - - # Batterieimport aufteilen - unassigned_bat_import = bat_imported - for source in ("pv", "grid", "cp"): - if direct[source] > unassigned_bat_import: - direct[source] -= unassigned_bat_import - unassigned_bat_import = 0 - break - else: - unassigned_bat_import -= direct[source] - direct[source] = 0 - - if unassigned_bat_import > 0: - # Fehler / inkonsistente Energiebilanz - log.warning( - f"bat_imported konnte nicht vollständig verteilt werden. " - f"Unverteilter Anteil: {unassigned_bat_import}" - ) - - # Anschließend Verbrauch aufteilen, wenn vorhanden - direct_total = sum(direct.values()) - - if direct_total <= 0: - entry["energy_source"] = {"grid": 0, "pv": 0, "bat": 0, "cp": 0} - else: - entry["energy_source"] = { - "grid": format(direct["grid"] / direct_total), - "pv": format(direct["pv"] / direct_total), - "bat": format(direct["bat"] / direct_total), - "cp": format(direct["cp"] / direct_total)} - except ZeroDivisionError: - entry["energy_source"] = {"grid": 0, "pv": 0, "bat": 0, "cp": 0} - except Exception: - log.exception(f"Fehler beim Berechnen des Strom-Mix von {entry['timestamp']}") - message += f"Fehler beim Berechnen des Strom-Mix von {entry['timestamp']}.\n" - finally: - return entry, message + # Hauptsächlich für Midnight-Handler + # Wenn keine Daten übergeben werden, dann die Monatswerte berechnen + if data is None: + data = get_monthly_log(date) + + totals = data["totals"] + source_entries = data.get("entries", []) + monthly_entry = {} + source_monthly_entry = _get_last_entry_for_period(source_entries, date, "%Y%m") + if source_monthly_entry is not None: + # Nur den letzten Eintrag des Monats nehmen + monthly_entry = deepcopy(source_monthly_entry) + monthly_entry["date"] = date + + # Erzeugt Ordner monthly_totals, falls nicht vorhanden + totals_dir = Path(_get_data_folder_path()) / "monthly_totals" + filepath = totals_dir / f"{date}_totals.json" + content = { + "date": date, + "totals": totals, + "entry": monthly_entry, + "names": data.get("names", {}), + "colors": data.get("colors", {}) + } + + if saving: + totals_dir.mkdir(parents=True, exist_ok=True) + with open(str(filepath), "w") as jsonFile: + json.dump(content, jsonFile, ensure_ascii=False, indent=2) + + log.debug(f"Monats-Summen für {date} gespeichert in {filepath}") + return content - -ERROR_STATE_MESSAGE = ("Die Anteile der Energiequellen für {} konnten nicht berechnet werden, da er sich im " + - "Fehlerzustand befindet. Die Verbräuche werden mit 0 kWh angesetzt.\n") + except FILE_ERRORS: + log.exception(f"Fehler beim Speichern der Monats-Summen für {date}") -def calc_energy_imported_by_source(entry, names, message_key_filter: Optional[str] = None) -> Tuple[Dict, str]: +def load_monthly_source_totals_content(date: str): try: - message = "" - - if "energy_source" in entry.keys(): - energy_source = entry["energy_source"] - hc_section = entry.get("hc") - if isinstance(hc_section, dict) and "all" in hc_section: - hc_all = hc_section["all"] - if isinstance(hc_all, dict): - if hc_all.get("fault_state", 0) != 2 and "energy_imported" in hc_all: - for source in ("grid", "pv", "bat", "cp"): - hc_all[f"energy_imported_{source}"] = decimal_multiply( - hc_all["energy_imported"], energy_source[source]) - else: - for source in ("grid", "pv", "bat", "cp"): - hc_all[f"energy_imported_{source}"] = 0 - if message_key_filter is None or message_key_filter == "hc": - message += ERROR_STATE_MESSAGE.format("den Hausverbrauch") - - cp_section = entry.get("cp") - if isinstance(cp_section, dict): - for cp_key, cp_data in cp_section.items(): - if isinstance(cp_data, dict): - if cp_data.get("fault_state", 0) != 2 and "energy_imported" in cp_data: - for source in ("grid", "pv", "bat", "cp"): - cp_data[f"energy_imported_{source}"] = decimal_multiply( - cp_data["energy_imported"], energy_source[source]) - else: - for source in ("grid", "pv", "bat", "cp"): - cp_data[f"energy_imported_{source}"] = 0 - if message_key_filter is None or message_key_filter == cp_key: - message += ERROR_STATE_MESSAGE.format(f"Ladepunkt {names.get(cp_key, cp_key)}") - - consumer_section = entry.get("consumer") - if isinstance(consumer_section, dict): - for consumer_key, consumer_data in consumer_section.items(): - if isinstance(consumer_data, dict): - if consumer_data.get("fault_state", 0) != 2 and "energy_imported" in consumer_data: - for source in ("grid", "pv", "bat", "cp"): - consumer_data[f"energy_imported_{source}"] = decimal_multiply( - consumer_data["energy_imported"], energy_source[source]) - else: - for source in ("grid", "pv", "bat", "cp"): - consumer_data[f"energy_imported_{source}"] = 0 - if message_key_filter is None or message_key_filter == consumer_key: - message += ERROR_STATE_MESSAGE.format( - f"Verbraucher {names.get(consumer_key, consumer_key)}") - - counter_section = entry.get("counter") - if isinstance(counter_section, dict): - for counter_key, counter_data in counter_section.items(): - if isinstance(counter_data, dict) and counter_data.get("grid") is False: - if counter_data.get("fault_state", 0) != 2 and "energy_imported" in counter_data: - for source in ("grid", "pv", "bat", "cp"): - counter_data[f"energy_imported_{source}"] = decimal_multiply( - counter_data["energy_imported"], energy_source[source]) - else: - for source in ("grid", "pv", "bat", "cp"): - counter_data[f"energy_imported_{source}"] = 0 - if message_key_filter is None or message_key_filter == counter_key: - message += ERROR_STATE_MESSAGE.format(f"Zähler {names.get(counter_key, counter_key)}") - except Exception: - log.exception(f"Fehler beim Berechnen der Energie-Anteile aus dem Strom-Mix von {entry['timestamp']}") - message += f"Fehler beim Berechnen des Strom-Mix von {entry['timestamp']}.\n" - finally: - return entry, message - - -def analyse_percentage_totals(entries, totals): - for section in ("consumer", "cp", "hc"): - if "all" not in totals[section].keys(): - totals[section]["all"] = {} - for source in ("grid", "pv", "bat", "cp"): - totals["hc"]["all"].update({f"energy_imported_{source}": 0}) - for entry in entries: - if "hc" in entry.keys() and "all" in entry["hc"].keys(): - current_value = totals["hc"]["all"][f"energy_imported_{source}"] - add_value = entry["hc"]["all"].get(f"energy_imported_{source}", 0) - totals["hc"]["all"][f"energy_imported_{source}"] = decimal_add( - current_value, add_value) - for key in entry["cp"].keys(): - if f"energy_imported_{source}" in entry["cp"][key].keys(): - if totals["cp"][key].get(f"energy_imported_{source}") is None: - totals["cp"][key].update({f"energy_imported_{source}": 0}) - current_value = totals["cp"][key][f"energy_imported_{source}"] - add_value = entry["cp"][key][f"energy_imported_{source}"] - totals["cp"][key][f"energy_imported_{source}"] = decimal_add( - current_value, add_value) - for key in entry["consumer"].keys(): - if f"energy_imported_{source}" in entry["consumer"][key].keys(): - if totals["consumer"][key].get(f"energy_imported_{source}") is None: - totals["consumer"][key].update({f"energy_imported_{source}": 0}) - current_value = totals["consumer"][key][f"energy_imported_{source}"] - add_value = entry["consumer"][key][f"energy_imported_{source}"] - totals["consumer"][key][f"energy_imported_{source}"] = decimal_add( - current_value, add_value) - for key, counter in entry["counter"].items(): - if counter["grid"] is False and f"energy_imported_{source}" in counter: - if totals["counter"][key].get(f"energy_imported_{source}") is None: - totals["counter"][key].update({f"energy_imported_{source}": 0}) - current_value = totals["counter"][key][f"energy_imported_{source}"] - add_value = counter[f"energy_imported_{source}"] - totals["counter"][key][f"energy_imported_{source}"] = decimal_add( - current_value, add_value) - return totals - - -def _process_entries(entries: List, calculation: CalculationType): - if entries: - if len(entries) == 1: - # Wenn es nur einen Eintrag gibt, kann keine Differenz berechnet werden und die Werte sind 0. - entry = entries[0] - for type in ("bat", "consumer", "counter", "cp", "pv", "sh", "hc"): - if type in entry: - for module in entry[type].keys(): - if calculation in [CalculationType.POWER, CalculationType.ALL]: - entry[type][module].update({ - "power_average": 0, - "power_imported": 0, - "power_exported": 0 - }) - if calculation in [CalculationType.ENERGY, CalculationType.ALL]: - entry[type][module].update({ - "energy_imported": 0, - "energy_exported": 0 - }) - elif len(entries) > 1: - for i in range(0, len(entries)-1): - entry = entries[i] - next_entry = entries[i+1] - entries[i] = process_entry(entry, next_entry, calculation) - entries.pop() - return entries - - -def process_entry(entry: dict, next_entry: dict, calculation: CalculationType): - time_diff = next_entry["timestamp"] - entry["timestamp"] - for type in ("bat", "consumer", "counter", "cp", "pv", "sh", "hc"): - if type in entry: - for module in entry[type].keys(): - try: - new_data = {} - if "imported" in entry[type][module].keys() or "exported" in entry[type][module].keys(): - def get_current_and_next(value_key: str) -> Tuple[float, float]: - def get_single_value(source: dict, default: int = 0) -> float: - try: - return source[type][module][value_key] - except KeyError: - return default - current_value = get_single_value(entry) - return current_value, get_single_value(next_entry, current_value) - value_imported, next_value_imported = get_current_and_next("imported") - value_exported, next_value_exported = get_current_and_next("exported") - if calculation in [CalculationType.POWER, CalculationType.ALL]: - if next_value_imported < value_imported or next_value_exported < value_exported: - # do not calculate as we have a backwards jump in our meter value! - average_power = 0 - else: - average_power = _calculate_average_power( - time_diff, value_imported, next_value_imported, - value_exported, next_value_exported) - new_data.update({ - "power_average": average_power, - "power_imported": average_power if average_power >= 0 else 0, - "power_exported": average_power * -1 if average_power < 0 else 0 - }) - if calculation in [CalculationType.ENERGY, CalculationType.ALL]: - if next_value_imported < value_imported: - # do not calculate as we have a backwards jump in our meter value! - energy_imported = 0 - else: - energy_imported = decimal_subtract(next_value_imported, - value_imported) - if next_value_exported < value_exported: - # do not calculate as we have a backwards jump in our meter value! - energy_exported = 0 - else: - energy_exported = decimal_subtract(next_value_exported, - value_exported) - new_data.update({ - "energy_imported": energy_imported, - "energy_exported": energy_exported - }) - entry[type][module].update(new_data) - except Exception: - log.exception("Fehler beim Berechnen der Leistung") - # next_entry may contain new modules, we add them here - try: - for module, module_data in next_entry[type].items(): - if module not in entry[type].keys(): - log.debug(f"adding module {module} from next entry") - if calculation in [CalculationType.POWER, CalculationType.ALL]: - module_data.update({"power_average": 0, "power_imported": 0, "power_exported": 0}) - if calculation in [CalculationType.ENERGY, CalculationType.ALL]: - module_data.update({"energy_imported": 0, "energy_exported": 0}) - entry[type].update({module: module_data}) - except KeyError: - # catch missing "type" - pass + filepath = f"{_get_data_folder_path()}/monthly_totals/{date}_totals.json" + if not Path(filepath).is_file(): + log.debug(f"Keine Monats-Summen-Datei gefunden: {filepath}") + return None + + with open(str(filepath), "r") as jsonFile: + content = json.load(jsonFile) + + log.debug(f"Monats-Summen für {date} geladen aus {filepath}") + return content + + except FILE_ERRORS: + log.exception(f"Fehler beim Laden der Monats-Summen für {date}") + + +def _get_last_entry_for_period(entries: List, period: str, period_format: str) -> Optional[Dict]: + # Suche den letzten Eintrag in der Liste, der dem angegebenen Zeitraum entspricht. + for entry in reversed(entries): + if isinstance(entry, dict) and isinstance(entry.get("timestamp"), (int, float)): + entry_period = datetime.fromtimestamp(entry["timestamp"]).strftime(period_format) + if entry_period == period: + return entry + + # Fallback: Falls kein passender Zeitstempel gefunden wird, letzten gueltigen Eintrag verwenden. + for entry in reversed(entries): + if isinstance(entry, dict): + return entry + return None + + +def _apply_source_totals(entry: Dict, daily_totals: Dict): + for section, section_totals in daily_totals.items(): + section_data = entry.get(section) + if not isinstance(section_totals, dict): + continue + + if not isinstance(section_data, dict): + section_data = {} + entry[section] = section_data + + for module, module_totals in section_totals.items(): + module_data = section_data.get(module) + if not isinstance(module_totals, dict): + continue + + if not isinstance(module_data, dict): + module_data = {} + section_data[module] = module_data + + # Alle vorhandenen Summenfelder des Moduls mit den aggregierten Summen ueberschreiben. + module_data.update(module_totals) + # Der kopierte Eintrag enthält ggf. noch den energy_source des letzten + # Intervalls/Tages. Nach dem Einspielen der aggregierten Energiewerte + # muss der Strom-Mix deshlab ebenfalls neu berechnet werden. + entry.pop("energy_source", None) + entry, _ = analyse_percentage(entry) + return entry -def _calculate_average_power(time_diff: float, current_imported: float = 0, next_imported: float = 0, - current_exported: float = 0, next_exported: float = 0) -> float: - imported_diff = decimal_subtract(next_imported, current_imported) - exported_diff = decimal_subtract(next_exported, current_exported) - energy_diff = decimal_subtract(imported_diff, exported_diff) - return decimal_multiply(energy_diff, 3600 / time_diff) # Ws -> W +def _oldest_log_day() -> Optional[str]: + try: + daily_log_dir = Path(_get_data_folder_path()) / "daily_log" + if not daily_log_dir.is_dir(): + return None + daily_log_files = [p for p in daily_log_dir.glob("*.json") if p.stem.isdigit()] + if not daily_log_files: + return None -def _get_data_folder_path() -> str: - return str(Path(__file__).resolve().parents[3] / "data") + oldest_file = min(daily_log_files, key=lambda f: f.stem) + return oldest_file.stem + except Exception: + log.exception("Fehler beim Ermitteln des ältesten Tageslogs") + return None diff --git a/packages/helpermodules/measurement_logging/process_log_calculation.py b/packages/helpermodules/measurement_logging/process_log_calculation.py new file mode 100644 index 0000000000..10b18af36a --- /dev/null +++ b/packages/helpermodules/measurement_logging/process_log_calculation.py @@ -0,0 +1,460 @@ +from enum import Enum +from typing import Dict, List, Optional, Tuple, Union +import json +import logging +from helpermodules.utils.precision_math import (decimal_add, + decimal_subtract, + decimal_multiply) + +from helpermodules.messaging import MessageType, pub_system_message + +log = logging.getLogger(__name__) + + +class CalculationType(Enum): + ALL = 0 + POWER = 1 + ENERGY = 2 + + +FILE_ERRORS = (FileNotFoundError, json.decoder.JSONDecodeError) + + +def safe_get_nested(data: Dict, *keys, default: Union[int, float] = 0) -> Union[int, float]: + current = data + for key in keys: + if isinstance(current, dict) and key in current: + current = current[key] + else: + return default + return current if isinstance(current, (int, float)) else default + + +def get_totals(entries: List, process_entries: bool = True) -> Dict: + """ Berechnet aus der übergebenen Liste "entries" die Summen (totals). + "process_entries" besagt, ob die Differenzen der einzelnen Einträge noch + berechnet werden müssen. + """ + if process_entries: + entries = _process_entries(entries, CalculationType.ENERGY) + totals = {"consumer": {}, "cp": {}, "counter": {}, "pv": {}, "bat": {}, "sh": {}, "hc": {}} + for totals_group in totals.keys(): + for entry in entries: + if totals_group in entry: + for entry_module in entry[totals_group]: + try: + if entry_module not in totals[totals_group]: + if totals_group == "hc": + totals[totals_group][entry_module] = {"energy_imported": 0.0} + elif totals_group == "pv": + totals[totals_group][entry_module] = {"energy_exported": 0.0} + else: + totals[totals_group][entry_module] = {"energy_imported": 0.0, "energy_exported": 0.0} + if totals_group == "counter" and "grid" in entry[totals_group][entry_module]: + totals[totals_group][entry_module]["grid"] = entry[ + totals_group][entry_module]["grid"] + for entry_module_key, entry_module_value in entry[totals_group][entry_module].items(): + if "grid" != entry_module_key and entry_module_key in totals[totals_group][entry_module]: + # avoid floating point issues with using Decimal + current_total = totals[totals_group][entry_module][entry_module_key] + totals[totals_group][entry_module][entry_module_key] = decimal_add( + current_total, entry_module_value) # totals in Wh! + + except Exception: + log.exception(f"Fehler beim Berechnen der Summe von {entry_module}; " + f"group:{totals_group}, module:{entry_module}, key:{entry_module_key}") + return totals + + +def _process_entries(entries: List, calculation: CalculationType): + if entries: + if len(entries) == 1: + # Wenn es nur einen Eintrag gibt, kann keine Differenz berechnet werden und die Werte sind 0. + entry = entries[0] + for type in ("bat", "consumer", "counter", "cp", "pv", "sh", "hc"): + if type in entry: + for module in entry[type].keys(): + if calculation in [CalculationType.POWER, CalculationType.ALL]: + entry[type][module].update({ + "power_average": 0, + "power_imported": 0, + "power_exported": 0 + }) + if calculation in [CalculationType.ENERGY, CalculationType.ALL]: + entry[type][module].update({ + "energy_imported": 0, + "energy_exported": 0 + }) + elif len(entries) > 1: + for i in range(0, len(entries)-1): + entry = entries[i] + next_entry = entries[i+1] + entries[i] = process_entry(entry, next_entry, calculation) + entries.pop() + return entries + + +def process_entry(entry: dict, next_entry: dict, calculation: CalculationType): + time_diff = next_entry["timestamp"] - entry["timestamp"] + for type in ("bat", "consumer", "counter", "cp", "pv", "sh", "hc"): + if type in entry: + for module in entry[type].keys(): + try: + new_data = {} + if "imported" in entry[type][module].keys() or "exported" in entry[type][module].keys(): + def get_current_and_next(value_key: str) -> Tuple[float, float]: + def get_single_value(source: dict) -> Optional[float]: + try: + value = source[type][module][value_key] + if isinstance(value, (int, float)): + return float(value) + except KeyError: + pass + return None + + current_value = get_single_value(entry) + next_value = get_single_value(next_entry) + + # Keep meter deltas neutral if one side is invalid/missing. + if current_value is None and next_value is None: + return 0.0, 0.0 + if current_value is None: + return next_value, next_value + if next_value is None: + return current_value, current_value + return current_value, next_value + value_imported, next_value_imported = get_current_and_next("imported") + value_exported, next_value_exported = get_current_and_next("exported") + if calculation in [CalculationType.POWER, CalculationType.ALL]: + if next_value_imported < value_imported or next_value_exported < value_exported: + # do not calculate as we have a backwards jump in our meter value! + average_power = 0 + else: + average_power = _calculate_average_power( + time_diff, value_imported, next_value_imported, + value_exported, next_value_exported) + new_data.update({ + "power_average": average_power, + "power_imported": average_power if average_power >= 0 else 0, + "power_exported": average_power * -1 if average_power < 0 else 0 + }) + if calculation in [CalculationType.ENERGY, CalculationType.ALL]: + if next_value_imported < value_imported: + # do not calculate as we have a backwards jump in our meter value! + energy_imported = 0 + else: + energy_imported = decimal_subtract(next_value_imported, + value_imported) + if next_value_exported < value_exported: + # do not calculate as we have a backwards jump in our meter value! + energy_exported = 0 + else: + energy_exported = decimal_subtract(next_value_exported, + value_exported) + new_data.update({ + "energy_imported": energy_imported, + "energy_exported": energy_exported + }) + entry[type][module].update(new_data) + except Exception: + log.exception("Fehler beim Berechnen der Leistung") + # next_entry may contain new modules, we add them here + try: + for module, module_data in next_entry[type].items(): + if module not in entry[type].keys(): + log.debug(f"adding module {module} from next entry") + if calculation in [CalculationType.POWER, CalculationType.ALL]: + module_data.update({"power_average": 0, "power_imported": 0, "power_exported": 0}) + if calculation in [CalculationType.ENERGY, CalculationType.ALL]: + module_data.update({"energy_imported": 0, "energy_exported": 0}) + entry[type].update({module: module_data}) + except KeyError: + # catch missing "type" + pass + return entry + + +def _calculate_average_power(time_diff: float, current_imported: float = 0, next_imported: float = 0, + current_exported: float = 0, next_exported: float = 0) -> float: + imported_diff = decimal_subtract(next_imported, current_imported) + exported_diff = decimal_subtract(next_exported, current_exported) + energy_diff = decimal_subtract(imported_diff, exported_diff) + return decimal_multiply(energy_diff, 3600 / time_diff) # Ws -> W + + +def analyse_percentage(entry) -> Tuple[Dict, str]: + EOOR_STATE_MSG = ("Der Strom-Mix um " + entry["date"] + + " konnte nicht berechnet werden, da sich {} im Fehlerzustand befindet. Alle Verbräuche werden" + + " dem Netz zugerechnet.\n") + + def format(value): + return round(value, 4) + + def get_grid_counter(entry) -> Dict: + # es gibt nur einen Zähler am EVU-Punkt + for counter in entry["counter"].values(): + if counter.get("grid") is True: + return counter + else: + raise KeyError(f"Kein Zähler für das Netz gefunden in Eintrag '{entry['timestamp']}'.") + try: + message = "" + grid_counter = get_grid_counter(entry) + # Wenn neben dem "all" Eintrag kein weiterer Eintrag existiert, dann gibt es keine Komponenten. + if ((safe_get_nested(entry, "bat", "all", "fault_state") == 2 and len(entry.get("bat", {})) > 1) or + (safe_get_nested(entry, "cp", "all", "fault_state") == 2 and len(entry.get("cp", {})) > 1) or + (safe_get_nested(entry, "pv", "all", "fault_state") == 2 and len(entry.get("pv", {})) > 1) or + grid_counter.get("fault_state", None) == 2): + + entry["energy_source"] = {"grid": 1, "pv": 0, "bat": 0, "cp": 0} + if safe_get_nested(entry, "bat", "all", "fault_state") == 2 and len(entry.get("bat", {})) > 1: + message += EOOR_STATE_MSG.format("mind. einer der Speicher") + if safe_get_nested(entry, "cp", "all", "fault_state") == 2 and len(entry.get("cp", {})) > 1: + message += EOOR_STATE_MSG.format("mind. einer der Ladepunkte") + if safe_get_nested(entry, "pv", "all", "fault_state") == 2 and len(entry.get("pv", {})) > 1: + message += EOOR_STATE_MSG.format("mind. einer der Wechselrichter") + if grid_counter.get("fault_state", None) == 2: + message += EOOR_STATE_MSG.format("der Zähler für das Netz") + + else: + bat_imported = safe_get_nested(entry, "bat", "all", "energy_imported") + bat_exported = safe_get_nested(entry, "bat", "all", "energy_exported") + cp_exported = safe_get_nested(entry, "cp", "all", "energy_exported") + pv_exported = safe_get_nested(entry, "pv", "all", "energy_exported") + grid_imported = grid_counter.get("energy_imported", 0) + grid_exported = grid_counter.get("energy_exported", 0) + consumption = grid_imported - grid_exported + pv_exported + bat_exported - bat_imported + cp_exported + if consumption < 0: + consumption = 0 + + try: + # Berechnung der Energiequellenanteile: + # Da die genaue Aufteilung der Energiequellen nicht bekannt ist, + # wird die Einspeisung (grid_exported) entsprechend der folgenden Priorität aufgeteilt: + # 1. PV + # 2. Batterie + # 3. CP + # Sollte die Einspeisung nicht komplett von PV gedeckt werden, + # wird der Rest von der Batterie übernommen, und falls nötig, vom CP. + # + # Entsprechend ähnlich wird der Batterieimport nach folgender Priorität aufgeteilt: + # 1. PV + # 2. Grid + # 3. CP + # + # Anschließend wird der Verbrauch (ohne Einspeisung und Batterieimport) auf energy_source aufgeteilt. + if consumption <= 0: + entry["energy_source"] = {"grid": 0, "pv": 0, "bat": 0, "cp": 0} + else: + + direct = {"grid": grid_imported, "pv": pv_exported, "bat": bat_exported, "cp": cp_exported} + + # Einspeißung aufteilen + unassigned_export = grid_exported + for source in ("pv", "bat", "cp"): + if direct[source] > unassigned_export: + direct[source] -= unassigned_export + unassigned_export = 0 + break + else: + unassigned_export -= direct[source] + direct[source] = 0 + + if unassigned_export > 0: + # Fehler / inkonsistente Energiebilanz + log.warning( + f"grid_exported konnte nicht vollständig verteilt werden. " + f"Unverteilter Anteil: {unassigned_export}" + ) + + # Batterieimport aufteilen + unassigned_bat_import = bat_imported + for source in ("pv", "grid", "cp"): + if direct[source] > unassigned_bat_import: + direct[source] -= unassigned_bat_import + unassigned_bat_import = 0 + break + else: + unassigned_bat_import -= direct[source] + direct[source] = 0 + + if unassigned_bat_import > 0: + # Fehler / inkonsistente Energiebilanz + log.warning( + f"bat_imported konnte nicht vollständig verteilt werden. " + f"Unverteilter Anteil: {unassigned_bat_import}" + ) + + # Anschließend Verbrauch aufteilen, wenn vorhanden + direct_total = sum(direct.values()) + + if direct_total <= 0: + entry["energy_source"] = {"grid": 0, "pv": 0, "bat": 0, "cp": 0} + else: + entry["energy_source"] = { + "grid": format(direct["grid"] / direct_total), + "pv": format(direct["pv"] / direct_total), + "bat": format(direct["bat"] / direct_total), + "cp": format(direct["cp"] / direct_total)} + except ZeroDivisionError: + entry["energy_source"] = {"grid": 0, "pv": 0, "bat": 0, "cp": 0} + except Exception: + log.exception(f"Fehler beim Berechnen des Strom-Mix von {entry['timestamp']}") + message += f"Fehler beim Berechnen des Strom-Mix von {entry['timestamp']}.\n" + finally: + return entry, message + + +ERROR_STATE_MESSAGE = ("Die Anteile der Energiequellen für {} konnten nicht berechnet werden, da er sich im " + + "Fehlerzustand befindet. Die Verbräuche werden mit 0 kWh angesetzt.\n") + + +def calc_energy_imported_by_source(entry, names, message_key_filter: Optional[str] = None) -> Tuple[Dict, str]: + try: + message = "" + + if "energy_source" in entry.keys(): + energy_source = entry["energy_source"] + hc_section = entry.get("hc") + if isinstance(hc_section, dict) and "all" in hc_section: + hc_all = hc_section["all"] + if isinstance(hc_all, dict): + if hc_all.get("fault_state", 0) != 2 and "energy_imported" in hc_all: + for source in ("grid", "pv", "bat", "cp"): + hc_all[f"energy_imported_{source}"] = decimal_multiply( + hc_all["energy_imported"], energy_source[source]) + else: + for source in ("grid", "pv", "bat", "cp"): + hc_all[f"energy_imported_{source}"] = 0 + if message_key_filter is None or message_key_filter == "hc": + message += ERROR_STATE_MESSAGE.format("den Hausverbrauch") + + cp_section = entry.get("cp") + if isinstance(cp_section, dict): + for cp_key, cp_data in cp_section.items(): + if isinstance(cp_data, dict): + if cp_data.get("fault_state", 0) != 2 and "energy_imported" in cp_data: + for source in ("grid", "pv", "bat", "cp"): + cp_data[f"energy_imported_{source}"] = decimal_multiply( + cp_data["energy_imported"], energy_source[source]) + else: + for source in ("grid", "pv", "bat", "cp"): + cp_data[f"energy_imported_{source}"] = 0 + if message_key_filter is None or message_key_filter == cp_key: + message += ERROR_STATE_MESSAGE.format(f"Ladepunkt {names.get(cp_key, cp_key)}") + + consumer_section = entry.get("consumer") + if isinstance(consumer_section, dict): + for consumer_key, consumer_data in consumer_section.items(): + if isinstance(consumer_data, dict): + if consumer_data.get("fault_state", 0) != 2 and "energy_imported" in consumer_data: + for source in ("grid", "pv", "bat", "cp"): + consumer_data[f"energy_imported_{source}"] = decimal_multiply( + consumer_data["energy_imported"], energy_source[source]) + else: + for source in ("grid", "pv", "bat", "cp"): + consumer_data[f"energy_imported_{source}"] = 0 + if message_key_filter is None or message_key_filter == consumer_key: + message += ERROR_STATE_MESSAGE.format( + f"Verbraucher {names.get(consumer_key, consumer_key)}") + + counter_section = entry.get("counter") + if isinstance(counter_section, dict): + for counter_key, counter_data in counter_section.items(): + if isinstance(counter_data, dict) and counter_data.get("grid") is False: + if counter_data.get("fault_state", 0) != 2 and "energy_imported" in counter_data: + for source in ("grid", "pv", "bat", "cp"): + counter_data[f"energy_imported_{source}"] = decimal_multiply( + counter_data["energy_imported"], energy_source[source]) + else: + for source in ("grid", "pv", "bat", "cp"): + counter_data[f"energy_imported_{source}"] = 0 + if message_key_filter is None or message_key_filter == counter_key: + message += ERROR_STATE_MESSAGE.format(f"Zähler {names.get(counter_key, counter_key)}") + except Exception: + log.exception(f"Fehler beim Berechnen der Energie-Anteile aus dem Strom-Mix von {entry['timestamp']}") + message += f"Fehler beim Berechnen des Strom-Mix von {entry['timestamp']}.\n" + finally: + return entry, message + + +def analyse_percentage_totals(entries, totals): + sources = ("grid", "pv", "bat", "cp") + + def ensure_zero_source_keys(module_totals: Dict): + if isinstance(module_totals, dict): + for source in sources: + module_totals[f"energy_imported_{source}"] = 0 + try: + for section in ("consumer", "cp", "hc"): + if "all" not in totals[section].keys(): + totals[section]["all"] = {} + for source in sources: + totals["hc"]["all"].update({f"energy_imported_{source}": 0}) + for entry in entries: + if "hc" in entry.keys() and "all" in entry["hc"].keys(): + current_value = totals["hc"]["all"][f"energy_imported_{source}"] + add_value = entry["hc"]["all"].get(f"energy_imported_{source}", 0) + totals["hc"]["all"][f"energy_imported_{source}"] = decimal_add( + current_value, add_value) + for key in entry["cp"].keys(): + if f"energy_imported_{source}" in entry["cp"][key].keys(): + if totals["cp"][key].get(f"energy_imported_{source}") is None: + totals["cp"][key].update({f"energy_imported_{source}": 0}) + current_value = totals["cp"][key][f"energy_imported_{source}"] + add_value = entry["cp"][key][f"energy_imported_{source}"] + totals["cp"][key][f"energy_imported_{source}"] = decimal_add( + current_value, add_value) + for key in entry["consumer"].keys(): + if f"energy_imported_{source}" in entry["consumer"][key].keys(): + if totals["consumer"][key].get(f"energy_imported_{source}") is None: + totals["consumer"][key].update({f"energy_imported_{source}": 0}) + current_value = totals["consumer"][key][f"energy_imported_{source}"] + add_value = entry["consumer"][key][f"energy_imported_{source}"] + totals["consumer"][key][f"energy_imported_{source}"] = decimal_add( + current_value, add_value) + for key, counter in entry["counter"].items(): + if counter["grid"] is False and f"energy_imported_{source}" in counter: + if totals["counter"][key].get(f"energy_imported_{source}") is None: + totals["counter"][key].update({f"energy_imported_{source}": 0}) + current_value = totals["counter"][key][f"energy_imported_{source}"] + add_value = counter[f"energy_imported_{source}"] + totals["counter"][key][f"energy_imported_{source}"] = decimal_add( + current_value, add_value) + except Exception: + log.exception("Fehler beim Berechnen der Summen der Energiequellen") + # Im Fehlerfall werden die Totals auf 0 gesetzt + # -> dann wird nur die Ladung/Entladung-Leistung angezeigt. + # -> sprich keine Aufteilung der Energiequellen, sondern nur die Gesamtwerte. + totals.setdefault("hc", {}) + totals["hc"].setdefault("all", {}) + ensure_zero_source_keys(totals["hc"]["all"]) + + totals.setdefault("cp", {}) + for key, module_totals in totals["cp"].items(): + ensure_zero_source_keys(module_totals) + + totals.setdefault("counter", {}) + for key, module_totals in totals["counter"].items(): + ensure_zero_source_keys(module_totals) + + return totals + return totals + + +def _analyse_energy_source(data, calc_cp: Optional[str] = None) -> Dict: + if data and len(data["entries"]) > 0: + try: + if data.get("message") is None: + data["message"] = "" + for i in range(0, len(data["entries"])): + data["entries"][i], message_analyse = analyse_percentage(data["entries"][i]) + data["entries"][i], message_calc = calc_energy_imported_by_source( + data["entries"][i], data["names"], message_key_filter=calc_cp) + data["message"] += message_analyse + message_calc + data["totals"] = analyse_percentage_totals(data["entries"], data["totals"]) + except Exception: + log.exception("Fehler beim Analysieren der Energiequellen") + pub_system_message({}, "Fehler beim Berechnen des Strom-Mix", MessageType.ERROR) + data["message"] = "Fehler beim Berechnen des Strom-Mix." + return data diff --git a/packages/helpermodules/measurement_logging/process_log_calculation_test.py b/packages/helpermodules/measurement_logging/process_log_calculation_test.py new file mode 100644 index 0000000000..f46b705f9a --- /dev/null +++ b/packages/helpermodules/measurement_logging/process_log_calculation_test.py @@ -0,0 +1,392 @@ +from copy import deepcopy +from typing import Dict +import pytest + +from helpermodules.measurement_logging.process_log_calculation import ( + analyse_percentage, + _calculate_average_power, + process_entry, + calc_energy_imported_by_source, + CalculationType) + + +@pytest.mark.parametrize("name", + ["regular", + "negative consumption", + "cp export"]) +def test_analyse_percentage(name: str, + daily_log_entry_percentage: Dict, + daily_log_entry_percentage_negative_consumption: Dict, + daily_log_entry_percentage_cp_discharge: Dict): + # setup + if name == "regular": + data = daily_log_entry_percentage + expected = deepcopy(data) + expected.update({"energy_source": {'bat': 0.2398, 'cp': 0.0, 'grid': 0.6504, 'pv': 0.1099}}) + elif name == "negative consumption": + data = daily_log_entry_percentage_negative_consumption + expected = deepcopy(data) + expected.update({"energy_source": {'bat': 0.0, 'cp': 0.0, 'grid': 0.0, 'pv': 0.0}}) + elif name == "cp export": + data = daily_log_entry_percentage_cp_discharge + expected = deepcopy(data) + expected.update({"energy_source": {'bat': 0.2006, 'cp': 0.1459, 'grid': 0.5441, 'pv': 0.1094}}) + + # execution + entry, message = analyse_percentage(data) + + # evaluation + assert entry == expected + assert message == "" + + +@pytest.mark.parametrize("test_case, entry_data, expected_energy_source, should_be_unchanged", [ + ( + "zero_consumption", + { + "timestamp": 1234567890, + "date": "00:31", + "bat": {"all": {"energy_imported": 5.0, "energy_exported": 5.0, "fault_state": 0}}, + "cp": {"all": {"energy_exported": 0.0, "fault_state": 0}}, + "pv": {"all": {"energy_exported": 0.0, "fault_state": 0}}, + "counter": {"counter0": {"grid": True, "energy_imported": 5.0, "energy_exported": 5.0, "fault_state": 0}} + }, + {"grid": 0, "pv": 0, "bat": 0, "cp": 0}, + False + ), + ( + "missing_sections", + { + "timestamp": 1234567890, + "date": "00:31", + "counter": {"counter0": {"grid": True, "energy_imported": 10.0, "energy_exported": 2.0, "fault_state": 0}} + }, + {"grid": 1.0, "pv": 0.0, "bat": 0.0, "cp": 0.0}, + False + ), + ( + "no_grid_counter", + { + "timestamp": 1234567890, + "date": "00:31", + "bat": {"all": {"energy_imported": 0.0, "energy_exported": 5.0, "fault_state": 0}}, + "counter": {"counter0": {"grid": False, "energy_imported": 10.0, "energy_exported": 2.0, "fault_state": 0}} + }, + None, + True + ) +]) +def test_analyse_percentage_edge_cases(test_case, entry_data, expected_energy_source, should_be_unchanged): + # execution + result, message = analyse_percentage(entry_data) + + # evaluation + if should_be_unchanged: + # Entry should be unchanged, no energy_source added due to error + assert result["timestamp"] == entry_data["timestamp"] + assert "energy_source" not in result or result.get("energy_source") is None + else: + # Energy source should be calculated correctly + assert result["energy_source"] == expected_energy_source + assert message == "" + + +def test_calculate_average_power(): + # setup and execution + power = _calculate_average_power(100, 250, 300) + + # evaluation + assert power == 1800 + + +def test_calc_energy_imported_by_source(): + # setup + entry = { + "timestamp": 1234567890, + "energy_source": {"grid": 0.6523, "pv": 0.2487, "bat": 0.0789, "cp": 0.0201}, + "hc": {"all": {"energy_imported": 2345.6, "fault_state": 0}}, + "cp": { + "cp1": {"energy_imported": 15723.4, "fault_state": 0}, + "cp2": {"energy_imported": 22108.7, "fault_state": 0} + }, + "counter": { + "counter0": {"grid": True, "energy_imported": 45892.3, "fault_state": 0}, + "counter1": {"grid": False, "energy_imported": 8956.7, "fault_state": 0} + } + } + + # execution + result, message = calc_energy_imported_by_source(entry, {}) + + # evaluation - realistic Wh values with decimal precision + assert result["hc"]["all"]["energy_imported_grid"] == 1530.035 + assert result["hc"]["all"]["energy_imported_pv"] == 583.351 + assert result["hc"]["all"]["energy_imported_bat"] == 185.068 + assert result["hc"]["all"]["energy_imported_cp"] == 47.147 + + assert result["cp"]["cp1"]["energy_imported_grid"] == 10256.374 + assert result["cp"]["cp1"]["energy_imported_pv"] == 3910.41 + assert result["cp"]["cp1"]["energy_imported_bat"] == 1240.576 + assert result["cp"]["cp1"]["energy_imported_cp"] == 316.04 + + assert result["cp"]["cp2"]["energy_imported_grid"] == 14421.505 + assert result["cp"]["cp2"]["energy_imported_pv"] == 5498.434 + assert result["cp"]["cp2"]["energy_imported_bat"] == 1744.376 + assert result["cp"]["cp2"]["energy_imported_cp"] == 444.385 + + assert result["counter"]["counter1"]["energy_imported_grid"] == 5842.455 + assert result["counter"]["counter1"]["energy_imported_pv"] == 2227.531 + assert result["counter"]["counter1"]["energy_imported_bat"] == 706.684 + assert result["counter"]["counter1"]["energy_imported_cp"] == 180.03 + # counter0 should not have these fields as it's a grid counter + assert "energy_imported_grid" not in result["counter"]["counter0"] + assert "energy_imported_pv" not in result["counter"]["counter0"] + assert "energy_imported_bat" not in result["counter"]["counter0"] + assert "energy_imported_cp" not in result["counter"]["counter0"] + assert message == "" + + +def test_calc_energy_imported_by_source_message_filtering(): + """Test message filtering when component is in fault state and name is missing.""" + # setup + entry = { + "timestamp": 1234567890, + "energy_source": {"grid": 0.6523, "pv": 0.2487, "bat": 0.0789, "cp": 0.0201}, + "cp": { + "cp1": {"energy_imported": 15723.4, "fault_state": 0}, + "cp2": {"energy_imported": 22108.7, "fault_state": 2} # fault state + }, + "counter": { + "counter0": {"grid": True, "energy_imported": 45892.3, "fault_state": 0}, + "counter1": {"grid": False, "energy_imported": 8956.7, "fault_state": 2} # fault state + } + } + + # Names dict is missing keys for cp2 and counter1 + names = { + "cp1": "Ladepunkt 1", + "counter0": "EVU-Zähler" + # cp2 and counter1 intentionally missing + } + + # execution - filter messages only for cp2 + result, message = calc_energy_imported_by_source(entry, names, message_key_filter="cp2") + + # evaluation + # Should only get message for cp2, not counter1 (due to filtering) + expected_message = ("Die Anteile der Energiequellen für Ladepunkt cp2 konnten nicht berechnet werden, da er sich " + "im Fehlerzustand befindet. Die Verbräuche werden mit 0 kWh angesetzt.\n") + assert message == expected_message + + # cp2 should have zero values for all energy sources due to fault state + assert result["cp"]["cp2"]["energy_imported_grid"] == 0 + assert result["cp"]["cp2"]["energy_imported_pv"] == 0 + assert result["cp"]["cp2"]["energy_imported_bat"] == 0 + assert result["cp"]["cp2"]["energy_imported_cp"] == 0 + + # cp1 should have normal calculated values (not in fault state) + assert result["cp"]["cp1"]["energy_imported_grid"] == 10256.374 + + # counter1 should have zero values but no message (filtered out) + assert result["counter"]["counter1"]["energy_imported_grid"] == 0 + assert result["counter"]["counter1"]["energy_imported_pv"] == 0 + assert result["counter"]["counter1"]["energy_imported_bat"] == 0 + assert result["counter"]["counter1"]["energy_imported_cp"] == 0 + + +def test_convert(daily_log_entry_processed, daily_log_sample): + # setup and execution + entry = process_entry(daily_log_sample[0], daily_log_sample[1], CalculationType.ALL) + + # evaluation + assert entry == daily_log_entry_processed + + +def test_pv_export_and_bat_export_and_bat_import(): + entry = { + "timestamp": 1234567890, + "date": "00:00", + "bat": { + "all": { + "energy_imported": 1.0, + "energy_exported": 5.0, + "fault_state": 0, + } + }, + "cp": { + "all": { + "energy_imported": 0.0, + "energy_exported": 0.0, + "fault_state": 0, + } + }, + "pv": { + "all": { + "energy_exported": 10.0, + "fault_state": 0, + } + }, + "counter": { + "counter0": { + "grid": True, + "energy_imported": 2.0, + "energy_exported": 3.0, + "fault_state": 0, + } + } + } + + # Einspeisung - Priorität + # 1 Pv + # 2 Bat + # 3 CP + # -> erst komplette Einspeisung von PV berücksichtigen + # -> wenn dann noch weitere Einspeisung übrig ist -> Bat und dann CP + + # Speicherimport - Priorität + # 1 Pv + # 2 Grid + # 3 CP + + # Pv 10 Exported + # Grid 2 Imported, 3 Exported + # Bat 1 Imported, 5 Exported + + # Realer Verbrauch + # 2 - 3 + 10 + 5 - 1 + 0 = 13 + + # Export aufteilen + # Pv - Export = 10 - 3 = 7 + + # Bat import aufteilen + # Pv - Bat_imported = 7 - 1 = 6 + + # Tatsächlicher Verbrauch: + # Pv = 6 + # Grid = 2 + # Bat = 5 + # --------------------- + # Summe = 13 + + # Anteil der Energiequellen: + # Grid: 2/13 ≈ 0.1538 + # PV = 6/13 ≈ 0.4615 + # Bat = 5/13 ≈ 0.3846 + # CP = 0/13 ≈ 0.0 + result, message = analyse_percentage(entry) + + assert result["energy_source"] == { + "grid": 0.1538, + "pv": 0.4615, + "bat": 0.3846, + "cp": 0.0 + } + assert message == "" + + +@pytest.mark.parametrize( + "name, pv_exported, bat_exported, bat_imported, cp_exported, grid_imported, grid_exported, expected", + [ + ( + "grid import and export", + 10.0, # pv_exported + 0.0, # bat_exported + 0.0, # bat_imported + 0.0, # cp_exported + 2.0, # grid_imported + 3.0, # grid_exported + # 2/9 7/9 + {"grid": 0.2222, "pv": 0.7778, "bat": 0.0, "cp": 0.0} + ), + ( + "grid export proportional pv and bat", + 10.0, # pv_exported + 5.0, # bat_exported + 0.0, # bat_imported + 0.0, # cp_exported + 0.0, # grid_imported + 3.0, # grid_exported + # 0/12 7/12 5/12 + {"grid": 0.0, "pv": 0.5833, "bat": 0.4167, "cp": 0.0} + ), + ( + "grid export and bat import", + 10.0, # pv_exported + 5.0, # bat_exported + 1.0, # bat_imported + 0.0, # cp_exported + 2.0, # grid_imported + 3.0, # grid_exported + # 2/13 6/13 5/13 + {"grid": 0.1538, "pv": 0.4615, "bat": 0.3846, "cp": 0.0} + ), + ( + "grid export proportional pv and cp ", + 6.0, # pv_exported + 0.0, # bat_exported + 0.0, # bat_imported + 3.0, # cp_exported + 0.0, # grid_imported + 3.0, # grid_exported + # 0/6 3/6 0/6 3/6 + {"grid": 0.0, "pv": 0.5, "bat": 0.0, "cp": 0.5} + ), + ( + "bat import proportional pv and grid", + 8.0, # pv_exported + 0.0, # bat_exported + 2.0, # bat_imported + 0.0, # cp_exported + 2.0, # grid_imported + 0.0, # grid_exported + # 2/8 6/8 0/8 8/8 + {"grid": 0.25, "pv": 0.75, "bat": 0.0, "cp": 0.0} + ), + ] +) +def test_analyse_percentage_proportional_distribution(name, + pv_exported, + bat_exported, + bat_imported, + cp_exported, + grid_imported, + grid_exported, expected): + entry = { + "timestamp": 1234567890, + "date": "00:00", + "bat": { + "all": { + "energy_imported": bat_imported, + "energy_exported": bat_exported, + "fault_state": 0, + } + }, + "cp": { + "all": { + "energy_exported": cp_exported, + "fault_state": 0, + } + }, + "pv": { + "all": { + "energy_exported": pv_exported, + "fault_state": 0, + } + }, + "counter": { + "counter0": { + "grid": True, + "energy_imported": grid_imported, + "energy_exported": grid_exported, + "fault_state": 0, + } + } + } + + result, message = analyse_percentage(entry) + + assert result["energy_source"] == expected + assert message == "" + + # Strommix muss zusammen 100% ergeben + assert sum(result["energy_source"].values()) == pytest.approx(1.0, abs=0.0002) diff --git a/packages/helpermodules/measurement_logging/process_log_unit_test.py b/packages/helpermodules/measurement_logging/process_log_unit_test.py index 62a9e0fff2..d9940f0da5 100644 --- a/packages/helpermodules/measurement_logging/process_log_unit_test.py +++ b/packages/helpermodules/measurement_logging/process_log_unit_test.py @@ -1,19 +1,15 @@ from copy import deepcopy import json import os -from typing import Dict from unittest.mock import Mock, mock_open import pytest from helpermodules.measurement_logging.process_log import ( - analyse_percentage, - _calculate_average_power, - process_entry, get_totals, _collect_daily_log_data, - calc_energy_imported_by_source, - analyse_percentage_totals, - CalculationType) + _get_last_entry_for_period, + _apply_source_totals, + analyse_percentage_totals) def test_get_totals(daily_log_sample, daily_log_totals): @@ -25,142 +21,6 @@ def test_get_totals(daily_log_sample, daily_log_totals): assert totals == daily_log_totals -@pytest.mark.parametrize("name", - ["regular", - "negative consumption", - "cp export"]) -def test_analyse_percentage(name: str, - daily_log_entry_percentage: Dict, - daily_log_entry_percentage_negative_consumption: Dict, - daily_log_entry_percentage_cp_discharge: Dict): - # setup - if name == "regular": - data = daily_log_entry_percentage - expected = deepcopy(data) - expected.update({"energy_source": {'bat': 0.2398, 'cp': 0.0, 'grid': 0.6504, 'pv': 0.1099}}) - elif name == "negative consumption": - data = daily_log_entry_percentage_negative_consumption - expected = deepcopy(data) - expected.update({"energy_source": {'bat': 0.0, 'cp': 0.0, 'grid': 0.0, 'pv': 0.0}}) - elif name == "cp export": - data = daily_log_entry_percentage_cp_discharge - expected = deepcopy(data) - expected.update({"energy_source": {'bat': 0.2006, 'cp': 0.1459, 'grid': 0.5441, 'pv': 0.1094}}) - - # execution - entry, message = analyse_percentage(data) - - # evaluation - assert entry == expected - assert message == "" - - -@pytest.mark.parametrize("test_case, entry_data, expected_energy_source, should_be_unchanged", [ - ( - "zero_consumption", - { - "timestamp": 1234567890, - "date": "00:31", - "bat": {"all": {"energy_imported": 5.0, "energy_exported": 5.0, "fault_state": 0}}, - "cp": {"all": {"energy_exported": 0.0, "fault_state": 0}}, - "pv": {"all": {"energy_exported": 0.0, "fault_state": 0}}, - "counter": {"counter0": {"grid": True, "energy_imported": 5.0, "energy_exported": 5.0, "fault_state": 0}} - }, - {"grid": 0, "pv": 0, "bat": 0, "cp": 0}, - False - ), - ( - "missing_sections", - { - "timestamp": 1234567890, - "date": "00:31", - "counter": {"counter0": {"grid": True, "energy_imported": 10.0, "energy_exported": 2.0, "fault_state": 0}} - }, - {"grid": 1.0, "pv": 0.0, "bat": 0.0, "cp": 0.0}, - False - ), - ( - "no_grid_counter", - { - "timestamp": 1234567890, - "date": "00:31", - "bat": {"all": {"energy_imported": 0.0, "energy_exported": 5.0, "fault_state": 0}}, - "counter": {"counter0": {"grid": False, "energy_imported": 10.0, "energy_exported": 2.0, "fault_state": 0}} - }, - None, - True - ) -]) -def test_analyse_percentage_edge_cases(test_case, entry_data, expected_energy_source, should_be_unchanged): - # execution - result, message = analyse_percentage(entry_data) - - # evaluation - if should_be_unchanged: - # Entry should be unchanged, no energy_source added due to error - assert result["timestamp"] == entry_data["timestamp"] - assert "energy_source" not in result or result.get("energy_source") is None - else: - # Energy source should be calculated correctly - assert result["energy_source"] == expected_energy_source - assert message == "" - - -def test_calculate_average_power(): - # setup and execution - power = _calculate_average_power(100, 250, 300) - - # evaluation - assert power == 1800 - - -def test_calc_energy_imported_by_source(): - # setup - entry = { - "timestamp": 1234567890, - "energy_source": {"grid": 0.6523, "pv": 0.2487, "bat": 0.0789, "cp": 0.0201}, - "hc": {"all": {"energy_imported": 2345.6, "fault_state": 0}}, - "cp": { - "cp1": {"energy_imported": 15723.4, "fault_state": 0}, - "cp2": {"energy_imported": 22108.7, "fault_state": 0} - }, - "counter": { - "counter0": {"grid": True, "energy_imported": 45892.3, "fault_state": 0}, - "counter1": {"grid": False, "energy_imported": 8956.7, "fault_state": 0} - } - } - - # execution - result, message = calc_energy_imported_by_source(entry, {}) - - # evaluation - realistic Wh values with decimal precision - assert result["hc"]["all"]["energy_imported_grid"] == 1530.035 - assert result["hc"]["all"]["energy_imported_pv"] == 583.351 - assert result["hc"]["all"]["energy_imported_bat"] == 185.068 - assert result["hc"]["all"]["energy_imported_cp"] == 47.147 - - assert result["cp"]["cp1"]["energy_imported_grid"] == 10256.374 - assert result["cp"]["cp1"]["energy_imported_pv"] == 3910.41 - assert result["cp"]["cp1"]["energy_imported_bat"] == 1240.576 - assert result["cp"]["cp1"]["energy_imported_cp"] == 316.04 - - assert result["cp"]["cp2"]["energy_imported_grid"] == 14421.505 - assert result["cp"]["cp2"]["energy_imported_pv"] == 5498.434 - assert result["cp"]["cp2"]["energy_imported_bat"] == 1744.376 - assert result["cp"]["cp2"]["energy_imported_cp"] == 444.385 - - assert result["counter"]["counter1"]["energy_imported_grid"] == 5842.455 - assert result["counter"]["counter1"]["energy_imported_pv"] == 2227.531 - assert result["counter"]["counter1"]["energy_imported_bat"] == 706.684 - assert result["counter"]["counter1"]["energy_imported_cp"] == 180.03 - # counter0 should not have these fields as it's a grid counter - assert "energy_imported_grid" not in result["counter"]["counter0"] - assert "energy_imported_pv" not in result["counter"]["counter0"] - assert "energy_imported_bat" not in result["counter"]["counter0"] - assert "energy_imported_cp" not in result["counter"]["counter0"] - assert message == "" - - def test_analyse_percentage_totals(): # setup current_dir = os.path.dirname(os.path.abspath(__file__)) @@ -287,62 +147,6 @@ def test_analyse_percentage_totals_entry_missing_grid_counter(): assert result["counter"]["counter0"]["energy_imported_cp"] == 10 -def test_calc_energy_imported_by_source_message_filtering(): - """Test message filtering when component is in fault state and name is missing.""" - # setup - entry = { - "timestamp": 1234567890, - "energy_source": {"grid": 0.6523, "pv": 0.2487, "bat": 0.0789, "cp": 0.0201}, - "cp": { - "cp1": {"energy_imported": 15723.4, "fault_state": 0}, - "cp2": {"energy_imported": 22108.7, "fault_state": 2} # fault state - }, - "counter": { - "counter0": {"grid": True, "energy_imported": 45892.3, "fault_state": 0}, - "counter1": {"grid": False, "energy_imported": 8956.7, "fault_state": 2} # fault state - } - } - - # Names dict is missing keys for cp2 and counter1 - names = { - "cp1": "Ladepunkt 1", - "counter0": "EVU-Zähler" - # cp2 and counter1 intentionally missing - } - - # execution - filter messages only for cp2 - result, message = calc_energy_imported_by_source(entry, names, message_key_filter="cp2") - - # evaluation - # Should only get message for cp2, not counter1 (due to filtering) - expected_message = ("Die Anteile der Energiequellen für Ladepunkt cp2 konnten nicht berechnet werden, da er sich " - "im Fehlerzustand befindet. Die Verbräuche werden mit 0 kWh angesetzt.\n") - assert message == expected_message - - # cp2 should have zero values for all energy sources due to fault state - assert result["cp"]["cp2"]["energy_imported_grid"] == 0 - assert result["cp"]["cp2"]["energy_imported_pv"] == 0 - assert result["cp"]["cp2"]["energy_imported_bat"] == 0 - assert result["cp"]["cp2"]["energy_imported_cp"] == 0 - - # cp1 should have normal calculated values (not in fault state) - assert result["cp"]["cp1"]["energy_imported_grid"] == 10256.374 - - # counter1 should have zero values but no message (filtered out) - assert result["counter"]["counter1"]["energy_imported_grid"] == 0 - assert result["counter"]["counter1"]["energy_imported_pv"] == 0 - assert result["counter"]["counter1"]["energy_imported_bat"] == 0 - assert result["counter"]["counter1"]["energy_imported_cp"] == 0 - - -def test_convert(daily_log_entry_processed, daily_log_sample): - # setup and execution - entry = process_entry(daily_log_sample[0], daily_log_sample[1], CalculationType.ALL) - - # evaluation - assert entry == daily_log_entry_processed - - def test_collect_daily_log_data_current_day(monkeypatch): # setup test_date = "20240422" @@ -459,191 +263,82 @@ def test_collect_daily_log_data_json_decode_error(monkeypatch): assert result == expected_result -def test_pv_export_and_bat_export_and_bat_import(): +def test_apply_source_totals_updates_existing_and_creates_missing_sections(): + # setup entry = { - "timestamp": 1234567890, - "date": "00:00", - "bat": { - "all": { - "energy_imported": 1.0, - "energy_exported": 5.0, - "fault_state": 0, - } + # required by analyse_percentage(), which _apply_source_totals() calls internally + "date": "22.04.2024", + "timestamp": 1713772800, + "counter": { + "counter0": {"grid": True, "energy_imported": 0, "energy_exported": 0} + }, + "cp": { + "all": {"energy_imported": 5, "keep": "x"}, + "cp9": {"keep_cp": True} }, + "meta": {"unchanged": True} + } + daily_totals = { "cp": { - "all": { - "energy_imported": 0.0, - "energy_exported": 0.0, - "fault_state": 0, - } + "all": {"energy_imported": 12, "energy_exported": 3}, + "cp1": {"energy_imported": 7}, + "cp_invalid": 99 }, - "pv": { - "all": { - "energy_exported": 10.0, - "fault_state": 0, - } + "hc": { + "all": {"energy_imported": 2} }, - "counter": { - "counter0": { - "grid": True, - "energy_imported": 2.0, - "energy_exported": 3.0, - "fault_state": 0, - } - } + "invalid_section": "ignore_me" } - # Einspeisung - Priorität - # 1 Pv - # 2 Bat - # 3 CP - # -> erst komplette Einspeisung von PV berücksichtigen - # -> wenn dann noch weitere Einspeisung übrig ist -> Bat und dann CP - - # Speicherimport - Priorität - # 1 Pv - # 2 Grid - # 3 CP - - # Pv 10 Exported - # Grid 2 Imported, 3 Exported - # Bat 1 Imported, 5 Exported - - # Realer Verbrauch - # 2 - 3 + 10 + 5 - 1 + 0 = 13 - - # Export aufteilen - # Pv - Export = 10 - 3 = 7 - - # Bat import aufteilen - # Pv - Bat_imported = 7 - 1 = 6 - - # Tatsächlicher Verbrauch: - # Pv = 6 - # Grid = 2 - # Bat = 5 - # --------------------- - # Summe = 13 - - # Anteil der Energiequellen: - # Grid: 2/13 ≈ 0.1538 - # PV = 6/13 ≈ 0.4615 - # Bat = 5/13 ≈ 0.3846 - # CP = 0/13 ≈ 0.0 - result, message = analyse_percentage(entry) - - assert result["energy_source"] == { - "grid": 0.1538, - "pv": 0.4615, - "bat": 0.3846, - "cp": 0.0 - } - assert message == "" + # execution + result = _apply_source_totals(entry, daily_totals) + + # evaluation + # in-place behavior + assert result is entry + + # existing module gets overwritten/extended, unrelated fields stay + assert result["cp"]["all"]["energy_imported"] == 12 + assert result["cp"]["all"]["energy_exported"] == 3 + assert result["cp"]["all"]["keep"] == "x" + + # missing module and section are created + assert result["cp"]["cp1"]["energy_imported"] == 7 + assert result["hc"]["all"]["energy_imported"] == 2 + + # invalid totals are ignored + assert "cp_invalid" not in result["cp"] + assert "invalid_section" not in result + + # unrelated data stays unchanged + assert result["cp"]["cp9"]["keep_cp"] is True + assert result["meta"]["unchanged"] is True @pytest.mark.parametrize( - "name, pv_exported, bat_exported, bat_imported, cp_exported, grid_imported, grid_exported, expected", + "entries, period, period_format, expected", [ ( - "grid import and export", - 10.0, # pv_exported - 0.0, # bat_exported - 0.0, # bat_imported - 0.0, # cp_exported - 2.0, # grid_imported - 3.0, # grid_exported - # 2/9 7/9 - {"grid": 0.2222, "pv": 0.7778, "bat": 0.0, "cp": 0.0} + [ + {"timestamp": 1711929600, "value": "april_1"}, + {"timestamp": 1712016000, "value": "april_2"}, + {"timestamp": 1714608000, "value": "may_2"}, + ], + "202404", + "%Y%m", + {"timestamp": 1712016000, "value": "april_2"}, ), ( - "grid export proportional pv and bat", - 10.0, # pv_exported - 5.0, # bat_exported - 0.0, # bat_imported - 0.0, # cp_exported - 0.0, # grid_imported - 3.0, # grid_exported - # 0/12 7/12 5/12 - {"grid": 0.0, "pv": 0.5833, "bat": 0.4167, "cp": 0.0} + [], + "202404", + "%Y%m", + None, ), - ( - "grid export and bat import", - 10.0, # pv_exported - 5.0, # bat_exported - 1.0, # bat_imported - 0.0, # cp_exported - 2.0, # grid_imported - 3.0, # grid_exported - # 2/13 6/13 5/13 - {"grid": 0.1538, "pv": 0.4615, "bat": 0.3846, "cp": 0.0} - ), - ( - "grid export proportional pv and cp ", - 6.0, # pv_exported - 0.0, # bat_exported - 0.0, # bat_imported - 3.0, # cp_exported - 0.0, # grid_imported - 3.0, # grid_exported - # 0/6 3/6 0/6 3/6 - {"grid": 0.0, "pv": 0.5, "bat": 0.0, "cp": 0.5} - ), - ( - "bat import proportional pv and grid", - 8.0, # pv_exported - 0.0, # bat_exported - 2.0, # bat_imported - 0.0, # cp_exported - 2.0, # grid_imported - 0.0, # grid_exported - # 2/8 6/8 0/8 8/8 - {"grid": 0.25, "pv": 0.75, "bat": 0.0, "cp": 0.0} - ), - ] + ], ) -def test_analyse_percentage_proportional_distribution(name, - pv_exported, - bat_exported, - bat_imported, - cp_exported, - grid_imported, - grid_exported, expected): - entry = { - "timestamp": 1234567890, - "date": "00:00", - "bat": { - "all": { - "energy_imported": bat_imported, - "energy_exported": bat_exported, - "fault_state": 0, - } - }, - "cp": { - "all": { - "energy_exported": cp_exported, - "fault_state": 0, - } - }, - "pv": { - "all": { - "energy_exported": pv_exported, - "fault_state": 0, - } - }, - "counter": { - "counter0": { - "grid": True, - "energy_imported": grid_imported, - "energy_exported": grid_exported, - "fault_state": 0, - } - } - } - - result, message = analyse_percentage(entry) - - assert result["energy_source"] == expected - assert message == "" +def test_get_last_entry_for_period(entries, period, period_format, expected): + # execution + result = _get_last_entry_for_period(entries, period, period_format) - # Strommix muss zusammen 100% ergeben - assert sum(result["energy_source"].values()) == pytest.approx(1.0, abs=0.0002) + # evaluation + assert result == expected diff --git a/packages/helpermodules/measurement_logging/update_yields.py b/packages/helpermodules/measurement_logging/update_yields.py index 9f8f3a3139..87e845703d 100644 --- a/packages/helpermodules/measurement_logging/update_yields.py +++ b/packages/helpermodules/measurement_logging/update_yields.py @@ -1,11 +1,14 @@ -import json import logging from pathlib import Path from typing import Dict from control import data from helpermodules import timecheck -from helpermodules.measurement_logging.process_log import get_totals +from helpermodules.measurement_logging.process_log import (get_totals, + load_daily_source_totals_content, + load_monthly_source_totals_content, + save_daily_source_totals, + get_monthly_log) log = logging.getLogger(__name__) @@ -17,6 +20,7 @@ def update_daily_yields(entries): totals = get_totals(entries) [update_module_yields(type, totals) for type in ("bat", "consumer", "counter", "cp", "pv")] data.data.counter_all_data.data.set.daily_yield_home_consumption = totals["hc"]["all"]["energy_imported"] + return totals except Exception: log.exception("Fehler beim Veröffentlichen der Tageserträge.") @@ -40,72 +44,115 @@ def update_module_yields(module: str, totals: Dict) -> None: log.exception(f"Fehler beim Aktualisieren der Tageserträge für Modul {m} vom Typ {module}.") -def update_pv_monthly_yearly_yields(): - """ veröffentlicht die monatlichen und jährlichen Erträge für PV +def update_pv_monthly_yearly_yields(daily_totals: Dict) -> None: """ - _update_pv_monthly_yields() - _update_pv_yearly_yields() + veröffentlicht die monatlichen und jährlichen Erträge für PV + """ + # pv_data und pv_all_data werden nur aktualisiert, wenn das Update der Logfiles abgeschlossen ist. + if not data.data.system_data["system"].data.get("log_data_ready", False): + log.debug("Update der Logfiles läuft noch. Monatliche und jährliche PV-Erträge werden nicht aktualisiert.") + return + + monthly_totals = _get_pv_monthly_yields(daily_totals) + yearly_totals = _get_pv_yearly_yields(monthly_totals) + + pv_all_monthly_yield = 0 + pv_all_yearly_yield = 0 + + for pv_key, pv_module in data.data.pv_data.items(): + if pv_key == "all" or not hasattr(pv_module, "num"): + continue + + # Was wurde im Monat/Jahr exportiert + monthly_yield = monthly_totals.get(f"pv{pv_module.num}", {}).get("energy_exported", 0) + yearly_yield = yearly_totals.get(f"pv{pv_module.num}", {}).get("energy_exported", 0) + + data.data.pv_data[f"pv{pv_module.num}"].data.get.monthly_exported = monthly_yield + data.data.pv_data[f"pv{pv_module.num}"].data.get.yearly_exported = yearly_yield + + # Summe über alle Module für pv_all + pv_all_monthly_yield += monthly_yield + pv_all_yearly_yield += yearly_yield + + data.data.pv_all_data.data.get.monthly_exported = pv_all_monthly_yield + data.data.pv_all_data.data.get.yearly_exported = pv_all_yearly_yield -def _update_pv_monthly_yields(): - """ veröffentlicht die monatlichen Erträge für PV - für pv_all nicht die Differenz aus den Logs nehmen, sondern die Summe der Module. Wenn im laufenden Monat ein Modul - gelöscht wurde und keins oder eines mit niedrigerem Zählerstand hinzugefügt wird, wird sonst ein negativer Wert - ermittelt. +def _get_pv_monthly_yields(daily_totals: Dict) -> Dict: """ - try: - pv_all_monthly_yield = 0 - with open(f"data/monthly_log/{timecheck.create_timestamp_YYYYMM()}.json", "r") as f: - monthly_log = json.load(f) - for pv_module in data.data.pv_data.values(): - for entry in monthly_log["entries"]: - if entry["pv"].get(f"pv{pv_module.num}"): - monthly_yield = data.data.pv_data[f"pv{pv_module.num}"].data.get.exported - \ - entry["pv"][f"pv{pv_module.num}"]["exported"] - pv_all_monthly_yield += monthly_yield - data.data.pv_data[f"pv{pv_module.num}"].data.get.monthly_exported = monthly_yield - break - data.data.pv_all_data.data.get.monthly_exported = pv_all_monthly_yield - except FileNotFoundError: - # am Tag der Ersteinrichtung gibt es noch kein Monatslog-File, das wird erst um Mitternacht erstellt. - log.debug("No monthly logfile found for calculation of monthly yield") - except Exception: - log.exception("Fehler beim Veröffentlichen der monatlichen Erträge für PV") + Berechnet den Unterschied zwischen dem Zählerstand vom ersten Tag des aktuellen Monats bis zum aktuellen Tag. + """ + + this_month = timecheck.create_timestamp_YYYYMM() + today = timecheck.create_timestamp_YYYYMMDD() + + pv_totals = {} + + daily_log_path = _get_parent_path()/"data"/"daily_log" + + for logfile in sorted(daily_log_path.glob(f"{this_month}*.json")): + day = logfile.stem + totals = {} + if day == today: + continue # Der aktuelle Tag wird später behandelt + else: + # Lade alle vergangenen Tage des Monats aus dem daily_totals-Logfile, um die Tageserträge zu ermitteln + content = load_daily_source_totals_content(day) + if content is None: + # Fallback/Migration: fehlende Tages-Totals aus den Tageslogs berechnen und speichern. + content = save_daily_source_totals(day, saving=True) + if isinstance(content, dict): + totals = content.get("totals", {}) + + # Totals aufsummieren + _add_pv_totals(pv_totals, totals.get("pv", {})) + # aktueller Tag ergänzen + _add_pv_totals(pv_totals, daily_totals.get("pv", {})) -def _update_pv_yearly_yields(): - """ veröffentlicht die jährlichen Erträge für PV - für pv_all nicht die Differenz aus den Logs nehmen, sondern die Summe der Module. Wenn unterjährig ein Modul - gelöscht wurde und keins oder eines mit niedrigerem Zählerstand hinzugefügt wird, wird sonst ein negativer Wert - ermittelt. + return pv_totals + + +def _get_pv_yearly_yields(current_monthly_totals: Dict) -> Dict: """ - try: - pv_all_yearly_yield = 0 - path_list = list(Path(_get_parent_path()/"data"/"monthly_log").glob(f"{timecheck.create_timestamp_YYYY()}*")) - sorted_path_list = sorted([str(p) for p in path_list]) - for pv_module in data.data.pv_data.values(): - found_pv = False - for path in sorted_path_list: - with open(path, "r") as f: - monthly_log = json.load(f) - for entry in monthly_log["entries"]: - # erster Eintrag mit PV im Jahr, falls WR erst im laufenden Jahr hinzugefügt wurden - if entry["pv"].get(f"pv{pv_module.num}"): - yearly_yield = data.data.pv_data[f"pv{pv_module.num}"].data.get.exported - \ - entry["pv"][f"pv{pv_module.num}"]["exported"] - pv_all_yearly_yield += yearly_yield - data.data.pv_data[f"pv{pv_module.num}"].data.get.yearly_exported = yearly_yield - found_pv = True - break - if found_pv: - break - else: - # am Tag der Ersteinrichtung gibt es noch kein Monatslog-File, das wird erst um Mitternacht erstellt. - log.debug("No monthly logfile found for calculation of yearly yield") - data.data.pv_all_data.data.get.yearly_exported = pv_all_yearly_yield - except Exception: - log.exception("Fehler beim Veröffentlichen der jährlichen Erträge für PV") + Berechnet den Unterschied zwischen dem Zählerstand vom ersten Monat des aktuellen Jahres bis zum aktuellen Monat. + """ + this_year = timecheck.create_timestamp_YYYY() + this_month = timecheck.create_timestamp_YYYYMM() + + pv_totals = {} + + month = f"{this_year}01" + while month < this_month: + totals = {} + content = load_monthly_source_totals_content(month) + if content is None: + # Fallback/Migration: fehlende Monats-Totals berechnen und speichern. + content = get_monthly_log(month) + if isinstance(content, dict): + totals = content.get("totals", {}) + + # Totals aufsummieren + _add_pv_totals(pv_totals, totals.get("pv", {})) + month = timecheck.get_relative_date_string(month, month_offset=1) + + # aktueller Monat ergänzen + _add_pv_totals(pv_totals, current_monthly_totals) + + return pv_totals def _get_parent_path() -> Path: return Path(__file__).resolve().parents[3] + + +def _add_pv_totals(target: Dict, source: Dict) -> None: + for pv_key, values in source.items(): + energy_exported = values.get("energy_exported", 0) + + # Wenn das PV-Modul noch nicht im target ist, initialisiere es mit 0 + if pv_key not in target: + target[pv_key] = {"energy_exported": 0} + + # Addiere die energy_exported Werte für das PV-Modul + target[pv_key]["energy_exported"] += energy_exported diff --git a/packages/helpermodules/measurement_logging/update_yields_test.py b/packages/helpermodules/measurement_logging/update_yields_test.py index 35a3fd7b81..55bbff1a1f 100644 --- a/packages/helpermodules/measurement_logging/update_yields_test.py +++ b/packages/helpermodules/measurement_logging/update_yields_test.py @@ -1,5 +1,11 @@ +from unittest.mock import Mock + from control import data -from helpermodules.measurement_logging.update_yields import update_module_yields +from helpermodules.measurement_logging import update_yields +from helpermodules.measurement_logging.update_yields import ( + update_module_yields, + update_pv_monthly_yearly_yields, +) def test_update_module_yields(daily_log_totals, mock_pub): @@ -21,3 +27,67 @@ def test_update_module_yields(daily_log_totals, mock_pub): data.data.cp_data["cp6"].data.get.daily_exported = 0.0 data.data.pv_all_data.data.get.daily_exported = 251.0 data.data.pv_data["pv1"].data.get.daily_exported = 251.0 + + +def test_update_pv_monthly_yearly_yields_skips_when_generation_not_finished(monkeypatch): + # setup + data.data.system_data = {"system": Mock(data={"log_data_ready": False})} + get_monthly_mock = Mock() + get_yearly_mock = Mock() + + monkeypatch.setattr(update_yields, "_get_pv_monthly_yields", get_monthly_mock) + monkeypatch.setattr(update_yields, "_get_pv_yearly_yields", get_yearly_mock) + + # execution + update_pv_monthly_yearly_yields({"pv": {"pv1": {"energy_exported": 12}}}) + + # evaluation + get_monthly_mock.assert_not_called() + get_yearly_mock.assert_not_called() + + +def test_update_pv_monthly_yearly_yields_with_daily_and_monthly_fallback(monkeypatch, tmp_path): + # setup + data.data.system_data = {"system": Mock(data={"log_data_ready": True})} + + parent_path = tmp_path + daily_log_path = parent_path / "data" / "daily_log" + daily_log_path.mkdir(parents=True) + (daily_log_path / "20260201.json").write_text("{}") + (daily_log_path / "20260214.json").write_text("{}") + (daily_log_path / "20260215.json").write_text("{}") + + monkeypatch.setattr(update_yields, "_get_parent_path", lambda: parent_path) + monkeypatch.setattr(update_yields.timecheck, "create_timestamp_YYYY", lambda: "2026") + monkeypatch.setattr(update_yields.timecheck, "create_timestamp_YYYYMM", lambda: "202602") + monkeypatch.setattr(update_yields.timecheck, "create_timestamp_YYYYMMDD", lambda: "20260215") + monkeypatch.setattr(update_yields.timecheck, "get_relative_date_string", lambda _m, month_offset=1: "202602") + + load_daily_mock = Mock(side_effect=[ + {"totals": {"pv": {"pv1": {"energy_exported": 10}}}}, + None, + ]) + save_daily_mock = Mock(return_value={"totals": {"pv": {"pv1": {"energy_exported": 20}}}}) + load_monthly_mock = Mock(return_value=None) + get_monthly_mock = Mock(return_value={"totals": {"pv": {"pv1": {"energy_exported": 100}}}}) + + monkeypatch.setattr(update_yields, "load_daily_source_totals_content", load_daily_mock) + monkeypatch.setattr(update_yields, "save_daily_source_totals", save_daily_mock) + monkeypatch.setattr(update_yields, "load_monthly_source_totals_content", load_monthly_mock) + monkeypatch.setattr(update_yields, "get_monthly_log", get_monthly_mock) + + daily_totals = {"pv": {"pv1": {"energy_exported": 5}}} + + # execution + update_pv_monthly_yearly_yields(daily_totals) + + # evaluation + assert data.data.pv_data["pv1"].data.get.monthly_exported == 35 + assert data.data.pv_data["pv1"].data.get.yearly_exported == 135 + assert data.data.pv_all_data.data.get.monthly_exported == 35 + assert data.data.pv_all_data.data.get.yearly_exported == 135 + + assert load_daily_mock.call_count == 2 + save_daily_mock.assert_called_once_with("20260214", saving=True) + load_monthly_mock.assert_called_once_with("202601") + get_monthly_mock.assert_called_once_with("202601") diff --git a/packages/helpermodules/measurement_logging/write_log.py b/packages/helpermodules/measurement_logging/write_log.py index d0b83c7b2b..77e35c50a6 100644 --- a/packages/helpermodules/measurement_logging/write_log.py +++ b/packages/helpermodules/measurement_logging/write_log.py @@ -1,4 +1,3 @@ -from enum import Enum import os import json import logging @@ -108,11 +107,6 @@ # } -class LogType(Enum): - DAILY = "daily" - MONTHLY = "monthly" - - class LegacySmartHomeLogData: def __init__(self) -> None: self.all_received_topics: Dict = {} @@ -144,20 +138,11 @@ def on_message(self, client: MqttClient, userdata, msg: MQTTMessage): self.all_received_topics.update({msg.topic: msg.payload}) -def save_log(log_type: LogType): - """ Parameter - --------- - folder: str - gibt an, ob ein Tages-oder Monats-Log-Eintrag erstellt werden soll. - """ +def save_log(): try: - parent_file = Path(__file__).resolve().parents[3] / "data" / \ - ("daily_log" if log_type == LogType.DAILY else "monthly_log") + parent_file = Path(__file__).resolve().parents[3] / "data" / "daily_log" parent_file.mkdir(mode=0o755, parents=True, exist_ok=True) - if log_type == LogType.DAILY: - file_name = timecheck.create_timestamp_YYYYMMDD() - else: - file_name = timecheck.create_timestamp_YYYYMM() + file_name = timecheck.create_timestamp_YYYYMMDD() filepath = str(parent_file / f"{file_name}.json") try: @@ -173,7 +158,7 @@ def save_log(log_type: LogType): previous_entry = get_previous_entry(parent_file, content) sh_log_data = LegacySmartHomeLogData() - new_entry = create_entry(log_type, sh_log_data, previous_entry) + new_entry = create_entry(sh_log_data, previous_entry) # json-Objekt in Datei einfügen @@ -205,11 +190,8 @@ def get_previous_entry(parent_file: Path, content: Dict) -> Optional[Dict]: return previous_entry -def create_entry(log_type: LogType, sh_log_data: LegacySmartHomeLogData, previous_entry: Optional[Dict]) -> Dict: - if log_type == LogType.DAILY: - date = timecheck.create_timestamp_HH_MM() - else: - date = timecheck.create_timestamp_YYYYMMDD() +def create_entry(sh_log_data: LegacySmartHomeLogData, previous_entry: Optional[Dict]) -> Dict: + date = timecheck.create_timestamp_HH_MM() current_timestamp = int(timecheck.create_timestamp()) try: diff --git a/packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py b/packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py index b1c74c086e..6b1b3b3c79 100644 --- a/packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py +++ b/packages/helpermodules/mosquitto_dynsec/missing_role_topics_test.py @@ -101,6 +101,7 @@ def _get_data_path() -> Path: 'openWB/optional/int_display/pin_active', 'openWB/optional/int_display/pin_code', 'openWB/system/datastore_version', + 'openWB/system/file_operation_version', 'openWB/system/device/+/component/+/simulation', 'openWB/system/device/+/component/+/simulation/power_present', 'openWB/system/device/+/component/+/simulation/present_exported', diff --git a/packages/helpermodules/setdata.py b/packages/helpermodules/setdata.py index 7c08918b72..665ff72fb9 100644 --- a/packages/helpermodules/setdata.py +++ b/packages/helpermodules/setdata.py @@ -1069,6 +1069,7 @@ def process_system_topic(self, msg: mqtt.MQTTMessage): "openWB/set/system/perform_update" in msg.topic or "openWB/set/system/wizard_done" in msg.topic or "openWB/set/system/update_in_progress" in msg.topic or + "openWB/set/system/log_data_ready" in msg.topic or "openWB/set/system/backup_cloud/backup_before_update" in msg.topic or "openWB/set/system/installAssistantDone" in msg.topic or "openWB/set/system/dataprotection_acknowledged" in msg.topic or @@ -1083,6 +1084,8 @@ def process_system_topic(self, msg: mqtt.MQTTMessage): self._validate_value(msg, float) elif "openWB/set/system/datastore_version" in msg.topic: self._validate_value(msg, int, [(0, UpdateConfig.DATASTORE_VERSION)], collection=list) + elif "openWB/set/system/file_operation_version" in msg.topic: + self._validate_value(msg, int, [(0, UpdateConfig.FILE_OPERATION_VERSION)], collection=list) elif "openWB/set/system/GetRemoteSupport" in msg.topic: # Server-Topic enthält kein json-Payload. payload = msg.payload.decode("utf-8") diff --git a/packages/helpermodules/update_config.py b/packages/helpermodules/update_config.py index 79824c5eb9..862f29bb68 100644 --- a/packages/helpermodules/update_config.py +++ b/packages/helpermodules/update_config.py @@ -8,6 +8,8 @@ from pathlib import Path import re import time +import sys +import subprocess from typing import List, Optional from paho.mqtt.client import Client as MqttClient, MQTTMessage @@ -61,6 +63,8 @@ class UpdateConfig: DATASTORE_VERSION = 151 + FILE_OPERATION_VERSION = 0 + valid_topic = [ "^openWB/bat/config/bat_control_activated$", "^openWB/bat/config/power_limit_mode$", @@ -570,10 +574,12 @@ class UpdateConfig: "^openWB/system/device/[0-9]+/component/[0-9]+/simulation/timestamp_present$", "^openWB/system/device/[0-9]+/config$", "^openWB/system/device/module_update_completed$", + "^openWB/system/file_operation_version$", "^openWB/system/hostname$", "^openWB/system/io/[0-9]+/config$", "^openWB/system/ip_address$", "^openWB/system/lastlivevaluesJson$", + "^openWB/system/log_data_ready$", "^openWB/system/mac_address$", "^openWB/system/mqtt/bridge/[0-9]+$", "^openWB/system/mqtt/valid_partner_ids$", @@ -778,6 +784,7 @@ def update(self): try: # erst breaking changes auflösen, sonst sind alte Topics schon gelöscht self.__solve_breaking_changes() + self.__solve_breaking_changes_filesystem() self.__remove_outdated_topics() self._remove_invalid_topics() self.__pub_missing_defaults() @@ -878,6 +885,67 @@ def __solve_breaking_changes(self) -> None: pub_system_message( {}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) + def __solve_breaking_changes_filesystem(self) -> None: + """Führt dateisystembezogene Migrationen anhand der file_operation_version aus.""" + file_operation_version = decode_payload(self.all_received_topics.get("openWB/system/file_operation_version")) + if file_operation_version is None: + # Neues Topic (z.B. bei Upgrade von älteren Versionen): alle File-Operation-Upgrades ausführen. + file_operation_version = [] + self.__update_topic("openWB/system/file_operation_version", file_operation_version) + elif isinstance(file_operation_version, int): + # Legacy-Format (int): bereits ausgeführte Upgrades als Liste abbilden. + file_operation_version = list(range(file_operation_version)) + self.__update_topic("openWB/system/file_operation_version", file_operation_version) + + log.debug(f"current file operation version: {file_operation_version}") + log.debug(f"target file operation version: {self.FILE_OPERATION_VERSION}") + for version in list(range(self.FILE_OPERATION_VERSION+1)): + try: + operation_required = version not in file_operation_version + if version == 0: + # Version 0 erneut ausführen, solange die Hintergrund-Generierung der Tages-/Monatssummen + # noch nicht erfolgreich abgeschlossen ist (Flag ist nicht True). + log_totals_generation_finished = decode_payload( + self.all_received_topics.get("openWB/system/log_data_ready")) + operation_required = operation_required or log_totals_generation_finished is not True + + if operation_required: + log.debug(f"upgrading File Operation version '{version}'") + getattr(self, f"upgrade_file_operation_{version}")() + except AttributeError: + log.error(f"missing upgrade function! '{version}'") + except Exception: + log.exception("Fehler bei der Aktualisierung des Brokers.") + pub_system_message( + {}, "Fehler bei der Aktualisierung der Konfiguration des Brokers.", MessageType.ERROR) + + def upgrade_file_operation_0(self) -> None: + """ + Generiere die Totals-Summen für Tage und Monate + """ + + self.__update_topic("openWB/system/log_data_ready", False) + try: + _generate_totals_subprocess = subprocess.Popen( + [ + sys.executable, + "-m", + "helpermodules.measurement_logging.generate_totals_subprocess" + ], + cwd="/var/www/html/openWB/packages", + start_new_session=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=True + ) + + log.debug("generate_totals_subprocess gestartet, PID: %s", _generate_totals_subprocess.pid) + self._append_file_operation_version(0) + except Exception: + _generate_totals_subprocess = None + log.exception("Fehler beim Starten des generate_totals_subprocess.") + def _loop_all_received_topics(self, callback) -> None: modified_topics = {} for topic, payload in self.all_received_topics.items(): @@ -896,6 +964,12 @@ def _append_datastore_version(self, version: int) -> None: datastore_versions.append(version) self.__update_topic("openWB/system/datastore_version", datastore_versions) + def _append_file_operation_version(self, version: int) -> None: + file_operation_versions = decode_payload(self.all_received_topics.get("openWB/system/file_operation_version")) + if version not in file_operation_versions: + file_operation_versions.append(version) + self.__update_topic("openWB/system/file_operation_version", file_operation_versions) + def upgrade_datastore_0(self) -> None: def upgrade(topic: str, payload) -> Optional[dict]: modified_topics = {} diff --git a/packages/helpermodules/update_config_test.py b/packages/helpermodules/update_config_test.py index 5667be977d..1938dcc7ea 100644 --- a/packages/helpermodules/update_config_test.py +++ b/packages/helpermodules/update_config_test.py @@ -497,3 +497,20 @@ def test_upgrade_datastore_151_removes_linked_extra_meter_counters_from_hierarch }] assert uc.all_received_topics["openWB/system/datastore_version"] == list(range(152)) assert mock_pub.pub.call_count == 2 + + +@pytest.mark.parametrize("file_operation_version, finished, expected_calls", [ + ([], False, 1), # erster Start + ([0], True, 0), # bereits fertig -> kein Neustart + ([0], False, 1), # angefangen, aber nicht fertig -> Neustart +]) +def test_file_operation_0_start_behavior(file_operation_version, finished, expected_calls): + update_config = UpdateConfig() + update_config.all_received_topics = { + "openWB/system/file_operation_version": file_operation_version, + "openWB/system/log_data_ready": finished + } + + with patch.object(update_config, "upgrade_file_operation_0") as upgrade_mock: + update_config._UpdateConfig__solve_breaking_changes_filesystem() + assert upgrade_mock.call_count == expected_calls diff --git a/packages/main.py b/packages/main.py index d7d63d337f..a9ad7ae912 100755 --- a/packages/main.py +++ b/packages/main.py @@ -27,7 +27,7 @@ from helpermodules.changed_values_handler import ChangedValuesContext from helpermodules.mosquitto_dynsec.mosquitto_dynsec import check_roles_at_start from helpermodules.measurement_logging.update_yields import update_daily_yields, update_pv_monthly_yearly_yields -from helpermodules.measurement_logging.write_log import LogType, save_log +from helpermodules.measurement_logging.write_log import save_log from helpermodules.modbusserver import start_modbus_server from helpermodules.pub import Pub from modules import configuration, loadvars, update_soc @@ -37,6 +37,7 @@ from modules.utils import wait_for_module_update_completed from smarthome.smarthome import readmq, smarthome_handler +from helpermodules.measurement_logging.process_log import save_daily_source_totals, save_monthly_source_totals class HandlerAlgorithm: def __init__(self): @@ -180,9 +181,11 @@ def handler5MinAlgorithm(self): """ try: with ChangedValuesContext(loadvars_.event_module_update_completed): - totals = save_log(LogType.DAILY) - update_daily_yields(totals) - update_pv_monthly_yearly_yields() + entries = save_log() + daily_totals = update_daily_yields(entries) + if daily_totals is not None: + update_pv_monthly_yearly_yields(daily_totals) + for cp in data.data.cp_data.values(): calc_energy_costs(cp) data.data.general_data.grid_protection() @@ -230,7 +233,15 @@ def handler5Min(self): @__with_handler_lock(error_threshold=60) def handler_midnight(self): try: - save_log(LogType.MONTHLY) + today = timecheck.create_timestamp_YYYYMMDD() + previous_day = timecheck.get_relative_date_string(today, day_offset=-1) + save_daily_source_totals(previous_day) + + prev_month = timecheck.get_relative_date_string(today, month_offset=-1)[:6] + # Neuer Monat hat angefangen, daher Monats Totals speichern + if today[6:8] == "01": + save_monthly_source_totals(prev_month, None, saving=True) + thread_errors_path = Path(Path(__file__).resolve().parents[1]/"ramdisk"/"thread_errors.log") with thread_errors_path.open("w") as f: f.write("")