Improve Monthly/Yearly Log

.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 

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",

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,

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

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()

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()