"""EDA-Report einlesen, Mitgliederdaten aktualisieren, KPI-Kurven berechnen.""" import logging import os from datetime import datetime from openpyxl import load_workbook from core.models import Mitglied, DEFAULT_DATE logger = logging.getLogger(__name__) def read_eda_report(system, report_path: str): """Liest den EDA-Report aus Blatt 2 und aktualisiert Mitglieder-Daten sowie Gemeinschafts-KPIs. system: Abrechnungssystem-Instanz. Gibt rows2 zurück, damit _validate_eda_sheet_consistency das Workbook nicht nochmal öffnet. """ try: wb = load_workbook(report_path, data_only=True, read_only=True) except Exception as e: logger.error(f"EDA-Report konnte nicht geöffnet werden: {e} – ABBRUCH") exit(1) sheet_names = wb.sheetnames if len(sheet_names) < 2: logger.error("EDA-Report hat weniger als 2 Tabellenblätter – ABBRUCH") exit(1) rows2 = list(wb[sheet_names[1]].iter_rows(values_only=True)) wb.close() if len(rows2) < 16: logger.error("Blatt 2 (Energiedaten) enthält zu wenige Zeilen – ABBRUCH") exit(1) row_zp = rows2[1] # Z2: MeteringPointId row_dir = rows2[3] # Z4: Energydirection row_ps = rows2[6] # Z7: Data Period Start row_pe = rows2[7] # Z8: Data Period End row_code = rows2[13] # Z14: MeterCode row_sum = rows2[15] # Z16: Spaltensumme dir_c = system.config.get("EDA_direction_C", "CONSUMPTION") dir_g = system.config.get("EDA_direction_G", "GENERATION") code_eigendeckung = system.config["EDA_Eigendeckung"] code_erzeugung = system.config["EDA_Erzeugung"] code_overflow = system.config["EDA_Overflow"] code_gesamterzeugung = system.config["EDA_Gesamterzeugung"] code_gesamtverbrauch = system.config["EDA_Gesamtverbrauch"] def _parse_date(v): if v is None: return None s = str(v).strip() for fmt in ("%d.%m.%Y %H:%M:%S", "%d.%m.%Y %H:%M", "%Y-%m-%d %H:%M:%S"): try: return datetime.strptime(s, fmt) except ValueError: pass return None # zp_data[zp_str][direction] = {'ps': dt, 'pe': dt, 'vals': {code: float}} zp_data: dict = {} n_cols = len(row_zp) for ci in range(1, n_cols): zp_val = row_zp[ci] dir_val = row_dir[ci] code = row_code[ci] val = row_sum[ci] if (zp_val is None or zp_val in ("MM", "TOTAL") or dir_val not in (dir_c, dir_g) or not isinstance(val, (int, float))): continue zp_str = str(zp_val) if zp_str not in zp_data: zp_data[zp_str] = {} if dir_val not in zp_data[zp_str]: zp_data[zp_str][dir_val] = {"ps": None, "pe": None, "vals": {}} entry = zp_data[zp_str][dir_val] entry["vals"][code] = val ps = _parse_date(row_ps[ci]) pe = _parse_date(row_pe[ci]) if ps: entry["ps"] = ps if entry["ps"] is None else min(entry["ps"], ps) if pe: entry["pe"] = pe if entry["pe"] is None else max(entry["pe"], pe) if not zp_data: logger.error("Blatt 2 enthält keine auswertbaren ZP-Spalten – ABBRUCH") exit(1) # ── Mitglieder aktualisieren ─────────────────────────────────────────────── for zp_str, directions in zp_data.items(): mitglied_match = None for m in system.mitglieder: if zp_str in (str(m.ZP_Consumption), str(m.ZP_Generation)): mitglied_match = m break if mitglied_match is None: logger.error(f"Zählpunkt {zp_str} nicht in der Mitgliederliste – ABBRUCH") exit(1) for dir_val, entry in directions.items(): vals = entry["vals"] ps = entry["ps"] pe = entry["pe"] if dir_val == dir_c: eigendeckung = vals.get(code_eigendeckung, 0.0) or 0.0 mitglied_match.Consumption += eigendeckung system.total_eigendeckung += eigendeckung system.total_gesamtverbrauch += vals.get(code_gesamtverbrauch, 0.0) or 0.0 if ps: mitglied_match.Start_C = ps if mitglied_match.Start_C == DEFAULT_DATE \ else min(mitglied_match.Start_C, ps) if pe: mitglied_match.Ende_C = pe if mitglied_match.Ende_C == DEFAULT_DATE \ else max(mitglied_match.Ende_C, pe) elif dir_val == dir_g: erzeugung = vals.get(code_erzeugung, 0.0) or 0.0 overflow = vals.get(code_overflow, 0.0) or 0.0 mitglied_match.Generation += (erzeugung - overflow) system.total_gesamterzeugung += vals.get(code_gesamterzeugung, 0.0) or 0.0 system.total_erzeugung_tf += erzeugung system.total_restüberschuss += overflow if ps: mitglied_match.Start_G = ps if mitglied_match.Start_G == DEFAULT_DATE \ else min(mitglied_match.Start_G, ps) if pe: mitglied_match.Ende_G = pe if mitglied_match.Ende_G == DEFAULT_DATE \ else max(mitglied_match.Ende_G, pe) mitglied_match.DataCheck = True for mitglied in system.mitglieder: if not mitglied.DataCheck: logger.info(f"Mitglied {mitglied.Name} nicht im EDA-File") logger.info( f"EDA-Report eingelesen (Blatt 2): {len(zp_data)} Zählpunkte, " f"Datei: {os.path.basename(report_path)}" ) return rows2 def compute_hourly_kpis(system, rows2: list) -> None: """Berechnet stündliche EVQ- und Autarkiegrad-Kurven aus der 15-Min-Zeitreihe. Schreibt die Ergebnisse direkt in system.hourly_eigenverbrauch / hourly_autarkiegrad / hourly_avg_consumption / hourly_avg_generation. Wird von _validate_eda_sheet_consistency() aufgerufen. """ from collections import defaultdict IDX_DATA_START = 16 IDX_ZP = 1 IDX_DIR = 3 IDX_CODE = 13 IDX_SUM_PRE = 15 if len(rows2) <= IDX_DATA_START: return row_zp = rows2[IDX_ZP] row_dir = rows2[IDX_DIR] row_code = rows2[IDX_CODE] row_sum_pre = rows2[IDX_SUM_PRE] dir_g_val = system.config.get("EDA_direction_G", "GENERATION") dir_c_val = system.config.get("EDA_direction_C", "CONSUMPTION") code_erz = system.config.get("EDA_Erzeugung", "") code_eig = system.config.get("EDA_Eigendeckung", "") code_ges_verb = system.config.get("EDA_Gesamtverbrauch", "") data_cols = [] for ci in range(1, len(row_zp)): zp = row_zp[ci] dirn = row_dir[ci] code = row_code[ci] pre = row_sum_pre[ci] if (zp and zp not in ("MM", "TOTAL") and dirn in (dir_c_val, dir_g_val) and code and isinstance(pre, (int, float))): data_cols.append({"idx": ci, "dir": dirn, "code": code}) erz_cols = [c["idx"] for c in data_cols if c["code"] == code_erz and c["dir"] == dir_g_val] eig_cols = [c["idx"] for c in data_cols if c["code"] == code_eig and c["dir"] == dir_c_val] ges_verb_cols = [c["idx"] for c in data_cols if c["code"] == code_ges_verb and c["dir"] == dir_c_val] def _parse_ts(v): if v is None: return None s = str(v).strip() for fmt in ("%d.%m.%Y %H:%M", "%d.%m.%Y %H:%M:%S", "%Y-%m-%d %H:%M:%S"): try: return datetime.strptime(s, fmt) except ValueError: pass return None evq_buckets: dict = defaultdict(list) aut_buckets: dict = defaultdict(list) cons_buckets: dict = defaultdict(list) gen_buckets: dict = defaultdict(list) for row in rows2[IDX_DATA_START:]: ts = _parse_ts(row[0]) if ts is None: continue erz_tf = sum(row[ci] for ci in erz_cols if isinstance(row[ci], (int, float))) eig = sum(row[ci] for ci in eig_cols if isinstance(row[ci], (int, float))) ges_verb = sum(row[ci] for ci in ges_verb_cols if isinstance(row[ci], (int, float))) if erz_tf > 0: evq_buckets[ts.hour].append(max(0.0, min(100.0, (eig / erz_tf) * 100.0))) if ges_verb > 0: aut_buckets[ts.hour].append(max(0.0, min(100.0, (eig / ges_verb) * 100.0))) cons_buckets[ts.hour].append(ges_verb * 4.0) gen_buckets[ts.hour].append(erz_tf * 4.0) system.hourly_eigenverbrauch = [ round(sum(evq_buckets[h]) / len(evq_buckets[h]), 1) if evq_buckets[h] else None for h in range(24) ] system.hourly_autarkiegrad = [ round(sum(aut_buckets[h]) / len(aut_buckets[h]), 1) if aut_buckets[h] else None for h in range(24) ] system.hourly_avg_consumption = [ round(sum(cons_buckets[h]) / len(cons_buckets[h]), 3) if cons_buckets[h] else None for h in range(24) ] system.hourly_avg_generation = [ round(sum(gen_buckets[h]) / len(gen_buckets[h]), 3) if gen_buckets[h] else None for h in range(24) ] vals_evq = [v for v in system.hourly_eigenverbrauch if v is not None] vals_aut = [v for v in system.hourly_autarkiegrad if v is not None] logger.info( f"Stündliche EVQ berechnet – " f"Tagesmittel EVQ: {round(sum(vals_evq)/max(1,len(vals_evq)),1)} % | " f"Autarkie: {round(sum(vals_aut)/max(1,len(vals_aut)),1)} %" ) return ges_verb_cols # für nighttime-Berechnung weiterreichen