#!/usr/bin/env python3 """ WashOS — Seko Dosierpumpen-Poller v1 ======================================== Pollt die Seko Web API für Waschmittel-Verbrauchsdaten. Architektur: - Einfache HTTP-Session (requests) mit Cookie-Auth - Login über POST /login/checkUser - Chemikalien-Verbrauch über POST /application/ctrl_get_chemical_usage - Gerätestatus über GET /devices/ctrl_get_list - Ausgabe: seko.json (wird vom Dashboard gelesen) Starten: Wird automatisch von washos-server.py gestartet. """ import json, os, time, threading, requests from datetime import datetime, timezone, timedelta # ── KONFIGURATION ───────────────────────────────────────────────── # Defaults — werden bei Bedarf aus data/settings.json überschrieben. SEKO_BASE = "https://gbl2146seko.sekoweb.com" SEKO_USER = "" SEKO_PASS = "" SEKO_OWNER = "GBL2146SEKO" SEKO_APP_ID = "75d34340-26fd-4a8b-ab2e-84d408d89b61" POLL_INTERVAL = 300 # Alle 5 Minuten (Verbrauch ändert sich nicht sekündlich) def load_credentials_from_settings(output_dir): """Lädt Seko-Credentials aus settings.json. Gibt True zurück wenn Credentials verfügbar sind, sonst False.""" global SEKO_USER, SEKO_PASS settings_file = os.path.join(output_dir, "settings.json") if not os.path.exists(settings_file): return bool(SEKO_USER and SEKO_PASS) try: with open(settings_file, encoding="utf-8") as f: data = json.load(f) creds = (data.get("seko") or {}) if creds.get("email"): SEKO_USER = creds["email"] if creds.get("password"): SEKO_PASS = creds["password"] except Exception: pass return bool(SEKO_USER and SEKO_PASS) # Waschmaschinen-Zuordnung: Pumpe → Kanal → Maschine WASHER_MAP = [ "012500003B9D_Ga6c3579c61134d31b8b1b82ea8e31136#1#A Small (1)", "012500003B9D_Ga6c3579c61134d31b8b1b82ea8e31136#2#B Small (2)", "012500003B9D_Ga6c3579c61134d31b8b1b82ea8e31136#3#C Medium (3)", "012500003B90_G2d00cb1d22fd42dba6554472c83eaaf3#1#D Large (1)", "012500003B90_G2d00cb1d22fd42dba6554472c83eaaf3#2#E Large (2)", "012500003B90_G2d00cb1d22fd42dba6554472c83eaaf3#3#F Large (3)", ] DEVICE_IDS = ["012500003B9D", "012500003B90"] # ── NEU: GID → Pumpe → Maschinen-Zuordnung # Jede GID ist eine Dosier-Pumpe mit 3 Kanälen (= 3 Maschinen). # Aus dem WASHER_MAP abgeleitet: GID#Kanal → Maschinen-Buchstabe GIDS = [ "012500003B9D_Ga6c3579c61134d31b8b1b82ea8e31136", # Pumpe 1 (Small-Block A/B/C) "012500003B90_G2d00cb1d22fd42dba6554472c83eaaf3", # Pumpe 2 (Large-Block D/E/F) ] # GID → Pumpen-Name (menschenlesbar, für Logs/Dashboard) PUMP_NAMES = { "012500003B9D_Ga6c3579c61134d31b8b1b82ea8e31136": "Pumpe 1 (Small-Block)", "012500003B90_G2d00cb1d22fd42dba6554472c83eaaf3": "Pumpe 2 (Large-Block)", } # Info zu Maschinen-Zuordnung (nur Doku): # Die Seko-API liefert pro Event im GID-Bucket das Feld WN ("Washer Name", # z.B. "A Small"). Daraus extrahieren wir die Maschine direkt. Das Feld DC # im flachen ALL-Bucket ist KEIN Maschinen-Channel (war ursprüngliche Annahme, # live verifiziert: DC ist immer 1). Maßgeblich ist der WAS-Dict-Key im # GID-Bucket. MACHINE_NAMES = { 0: "A Small", 1: "B Small", 2: "C Medium", 3: "D Large", 4: "E Large", 5: "F Large", } # Einkaufspreise Bloomest Store (Stand April 2026) # Monatlich prüfen unter: store.bloomest.de → Waschmittel CHEMICAL_PRICES = { "ENERGY": 5.25, # 105,00 € / 20l Kanister "MEGASOFT": 3.85, # 77,00 € / 20l Kanister "SANYPLUS": 2.95, # 59,00 € / 20l Kanister "TOTAL": 5.20, # 104,00 € / 20l Kanister "SG CLEAN": 7.70, # 77,00 € / 10l Kanister "NSG-CLEAN": 7.70, # Alias "PROOFING": 12.90, # 129,00 € / 10l Kanister } class SekoPoller: def __init__(self, output_dir="."): self.output_dir = output_dir self.output_file = os.path.join(output_dir, "seko.json") # NEU: Persistentes Event-Archiv (wächst kontinuierlich) self.archive_file = os.path.join(output_dir, "seko_events_archive.json") self.session = requests.Session() self.session.headers.update({ "User-Agent": "WashOS/1.0", "Accept": "application/json, text/html, */*", }) self.logged_in = False def login(self) -> bool: """Login bei Seko Web.""" try: r = self.session.post( f"{SEKO_BASE}/login/checkUser", data={"username": SEKO_USER, "password": SEKO_PASS}, timeout=45, allow_redirects=True, ) if r.status_code == 200 and "login" not in r.url: self.logged_in = True print(" [Seko] Login OK") return True # Manchmal kommt ein Redirect zum Dashboard if r.status_code == 200: # Prüfe ob wir eingeloggt sind test = self.session.get( f"{SEKO_BASE}/devices/ctrl_get_list/{SEKO_OWNER}", timeout=10, ) if test.status_code == 200 and "retCode" in test.text: self.logged_in = True print(" [Seko] Login OK (via redirect)") return True print(f" [Seko] Login fehlgeschlagen: HTTP {r.status_code}") return False except Exception as e: print(f" [Seko] Login-Fehler: {e}") return False def _get_devices(self) -> list: """Geräteliste abrufen.""" try: r = self.session.get( f"{SEKO_BASE}/devices/ctrl_get_list/{SEKO_OWNER}", timeout=45, ) if r.status_code == 200: data = r.json() if data.get("retCode"): return data.get("table", []) if r.status_code == 401 or r.status_code == 302: self.logged_in = False return [] except Exception as e: print(f" [Seko] Geräte-Fehler: {e}") return [] def _get_chemical_usage(self, range_type="month", start_date=None, end_date=None) -> dict: """Chemikalien-Verbrauch abrufen.""" now = datetime.now() if start_date and end_date: start = start_date end = end_date elif range_type == "today": start = now.replace(hour=0, minute=0, second=0) end = now elif range_type == "week": start = now - timedelta(days=7) end = now elif range_type == "month": start = now - timedelta(days=30) end = now elif range_type == "year": start = now - timedelta(days=365) end = now else: start = now - timedelta(days=30) end = now params = { "applicationID": SEKO_APP_ID, "type": "table", "rangeType": "custom" if start_date else range_type, "startDate": int(start.timestamp()), "endDate": int(end.timestamp()), "washer": WASHER_MAP, "owner": [], "devicesIDS": DEVICE_IDS, "customers": {}, } try: r = self.session.post( f"{SEKO_BASE}/application/ctrl_get_chemical_usage/{SEKO_OWNER}", json=params, timeout=30, ) if r.status_code == 200: return r.json() if r.status_code == 401 or r.status_code == 302: self.logged_in = False return {} except Exception as e: print(f" [Seko] Chemical-Usage-Fehler: {e}") return {} def _get_flow_records(self, gid: str, hours_back: int = 24) -> dict: """NEU: Dosier-Events pro Pumpe (GID) abrufen. Liefert für jede Chemikalie eine Event-Liste mit: - T (timestamp ms), DQ (dosed quantity ml), DT (duration s), FR (flow rate ml/min), DC (device channel 1-3), CV, RV Der Endpoint verlangt pro GID einen separaten Call (die Seko-Web-UI macht das auch so — ein Call pro Pumpe). hours_back: Standard 24h (für Zyklus-Korrelation reicht das; längere Zeiträume bringen kilobyteweise Daten pro Poll).""" now_s = int(datetime.now().timestamp()) start_s = now_s - hours_back * 3600 params = { "startTime": start_s, "endTime": now_s, "washer": WASHER_MAP, "ownerID": SEKO_OWNER, "applicationID": SEKO_APP_ID, "GID": gid, "installationSiteName": "Waschsalon Nord GmbH", } try: r = self.session.post( f"{SEKO_BASE}/application/ctrl_getflowrecords", json=params, timeout=30, ) if r.status_code == 200: return r.json() if r.status_code == 401 or r.status_code == 302: self.logged_in = False return {} except Exception as e: print(f" [Seko] Flow-Records-Fehler ({gid[:16]}…): {e}") return {} def poll_once(self) -> dict: """Einmal alle Seko-Daten abfragen.""" if not self.logged_in: return {"error": "not_logged_in"} # Geräte-Status devices_raw = self._get_devices() devices = [] for d in devices_raw: # HTML aus device ID extrahieren did = d.get("deviceID", "") # Entferne HTML-Tags import re did_clean = re.sub(r'<[^>]+>', '', did).strip() devices.append({ "device_id": did_clean, "model": d.get("systemModel", ""), "name": d.get("name", ""), "status": "online" if "device-online" in d.get("deviceID", "") else "offline", "firmware": d.get("firmware", ""), "build": d.get("build", ""), "last_update": d.get("LASTUPDATE", ""), }) # Chemikalien-Verbrauch: heute, Monat, Jahr, Referenz ab 1.1.2026 usage_today = self._get_chemical_usage("today") usage_month = self._get_chemical_usage("month") usage_year = self._get_chemical_usage("year") # Referenzzeitraum ab 1.1.2026 für saubere Kosten-pro-Waschgang-Berechnung ref_start = datetime(2026, 1, 1) ref_end = datetime.now() ref_days = (ref_end - ref_start).days or 1 usage_ref = self._get_chemical_usage( "custom", start_date=ref_start, end_date=ref_end) def parse_usage(data): """Parse die Seko chemical_usage Antwort.""" values = data.get("values", {}) chemicals = [] for key, v in values.items(): total_list = v.get("totalList", []) per_machine = {} for i, litres in enumerate(total_list): machine = MACHINE_NAMES.get(i, f"Maschine {i+1}") per_machine[machine] = round(litres, 3) total_l = round(v.get("total", 0), 3) # Preis aus Bloomest Store (Fallback: Seko-Plattform) price_per_l = CHEMICAL_PRICES.get( key.upper(), CHEMICAL_PRICES.get( v.get("name", "").upper(), 0)) seko_cost = round(v.get("cost", 0), 2) actual_cost = round(total_l * price_per_l, 2) if price_per_l else seko_cost chemicals.append({ "key": key, "name": v.get("name", key), "runtime_seconds": v.get("runningTime", 0), "runtime_formatted": format_duration( v.get("runningTime", 0)), "total_litres": total_l, "cost_eur": actual_cost, "price_per_litre": price_per_l, "cost_unit": "EUR", "per_machine": per_machine, }) # Sortiere nach Verbrauch absteigend chemicals.sort(key=lambda x: x["total_litres"], reverse=True) return chemicals chemicals_today = parse_usage(usage_today) chemicals_month = parse_usage(usage_month) chemicals_year = parse_usage(usage_year) chemicals_ref = parse_usage(usage_ref) # Zusammenfassungs-KPIs total_litres_month = sum( c["total_litres"] for c in chemicals_month) total_litres_year = sum( c["total_litres"] for c in chemicals_year) total_litres_ref = sum( c["total_litres"] for c in chemicals_ref) total_cost_month = sum( c["cost_eur"] for c in chemicals_month) total_cost_year = sum( c["cost_eur"] for c in chemicals_year) total_cost_ref = sum( c["cost_eur"] for c in chemicals_ref) # Kosten pro Liter (gewichtet) cost_per_litre = ( round(total_cost_ref / total_litres_ref, 2) if total_litres_ref > 0 else 0 ) # Verbrauch pro Maschine (alle Chemikalien summiert) für Referenzzeitraum machine_totals = {} for c in chemicals_ref: for m, litres in c.get("per_machine", {}).items(): machine_totals[m] = round( machine_totals.get(m, 0) + litres, 3) # ── NEU: Dosier-Events der letzten 24h pro Pumpe holen # Das ist die Basis für Zyklus-Korrelation (Miele-Zyklus-Zeitfenster × # Seko-Events im selben Fenster = Verbrauch pro Zyklus). # # Struktur-Erkenntnis nach Live-Verifikation: # Der Response-Pfad CHEMICAL_FLOW.ALL.CHEM.DDQ[] ist ein SUMMEN-View # (alle Events der Pumpe; DC im Event ist KEIN Maschinen-Channel). # Die maschinen-aufgelöste Version steht im GID-Bucket: # CHEMICAL_FLOW[GID].CHEM.WAS["{DID}_{N}"].CHS["1"] = [events] # Dabei ist: # - WAS = Washer-Dict, Keys sind "{DID}_{1|2|3}" (DID = Pumpen-Device-ID) # - Jeder WAS-Entry hat "WN" (Washer Name, z.B. "A Small") + "CHS" (events) # Wir lesen die Maschine aus dem WN-Feld ab. dosing_events = [] # Flach, bereits mit pump + machine angereichert events_by_pump = {} # Statistik für Log for gid in GIDS: pump_name = PUMP_NAMES.get(gid, gid[:16]) events_by_pump[pump_name] = { "event_count": 0, "total_litres": 0.0, "chemicals": set()} flow_data = self._get_flow_records(gid, hours_back=24) if not flow_data: continue chem_flow = (flow_data.get("devicedata") or {}).get("CHEMICAL_FLOW") or {} # Der GID-Bucket hat die maschinen-aufgelöste Struktur gid_bucket = chem_flow.get(gid) or {} for chem_key, chem_data in gid_bucket.items(): if not isinstance(chem_data, dict): continue chem_name = chem_data.get("CN", chem_key) price_per_l = CHEMICAL_PRICES.get( chem_key.upper(), CHEMICAL_PRICES.get(chem_name.upper(), 0)) was_dict = chem_data.get("WAS") or {} for was_key, was_data in was_dict.items(): if not isinstance(was_data, dict): continue # Maschinen-Name aus WN extrahieren ("A Small" → "A") wn = was_data.get("WN", "") # Die 6 Maschinen heißen alle "X Small/Medium/Large" — erstes Zeichen ist die Maschine machine = wn.split(" ")[0] if wn else "?" # CHS ist ein Dict: { "1": [events], "2": [events], ... } # In der Praxis aktuell immer nur "1", aber wir iterieren robust chs = was_data.get("CHS") or {} for chs_key, events in chs.items(): if not isinstance(events, list): continue for ev in events: if not isinstance(ev, dict): continue ts_ms = ev.get("T") or ev.get("timestamp") or 0 if ts_ms is None or not ts_ms: continue # Ohne Timestamp unbrauchbar litres = _safe_float(ev.get("TQ")) ml = _safe_float(ev.get("DQ")) duration_s = _safe_int(ev.get("DT")) dosing_events.append({ "ts_ms": ts_ms, "ts_iso": datetime.fromtimestamp( ts_ms / 1000, tz=timezone.utc).isoformat(), "pump_gid": gid, "pump_name": pump_name, "washer_key": was_key, # z.B. "012500003B9D_1" "washer_name": wn, # z.B. "A Small" "channel_set": chs_key, # idR "1" "machine": machine, # z.B. "A" "chemical_key": chem_key, "chemical_name": chem_name, "ml": round(ml, 3), "litres": round(litres, 6), "duration_s": duration_s, "flow_rate": round(_safe_float(ev.get("FR")), 1), "cost_eur": round(ml / 1000.0 * price_per_l, 4) if price_per_l else 0, }) events_by_pump[pump_name]["event_count"] += 1 events_by_pump[pump_name]["total_litres"] += litres events_by_pump[pump_name]["chemicals"].add(chem_name) # Sortiere nach Timestamp aufsteigend (macht Korrelation leichter) dosing_events.sort(key=lambda e: e["ts_ms"]) # Chemicals-Set in Liste umwandeln (JSON-serialisierbar) for pname in events_by_pump: events_by_pump[pname]["chemicals"] = sorted( events_by_pump[pname]["chemicals"]) events_by_pump[pname]["total_litres"] = round( events_by_pump[pname]["total_litres"], 3) # NEU: Events ins persistente Archiv mergen archive_total, archive_added = self._merge_to_archive(dosing_events) return { "meta": { "last_update": datetime.now(timezone.utc).isoformat(), "source": "seko_web", "owner_id": SEKO_OWNER, "device_count": len(devices), "ref_start": "2026-01-01", "ref_days": ref_days, "archive_total_events": archive_total, "archive_new_events_this_poll": archive_added, }, "devices": devices, "summary": { "total_litres_month": round(total_litres_month, 1), "total_litres_year": round(total_litres_year, 1), "total_litres_ref": round(total_litres_ref, 1), "total_cost_month": round(total_cost_month, 2), "total_cost_year": round(total_cost_year, 2), "total_cost_ref": round(total_cost_ref, 2), "cost_per_litre": cost_per_litre, "chemicals_count": len(chemicals_year), "pumps_online": sum( 1 for d in devices if d["status"] == "online"), "pumps_total": len(devices), "ref_days": ref_days, }, "chemicals_today": chemicals_today, "chemicals_month": chemicals_month, "chemicals_year": chemicals_year, "chemicals_ref": chemicals_ref, "machine_totals_ref": machine_totals, # ── NEU: Einzel-Dosier-Events (letzte 24h) "dosing_events_24h": dosing_events, "dosing_events_meta": { "window_hours": 24, "total_events": len(dosing_events), "by_pump": events_by_pump, }, # ── NEU: Zyklus-Korrelation (Miele × Seko) # Wird in write_output nachträglich befüllt, sobald das Dict steht } def _load_archive(self) -> list: """Lade das persistente Event-Archiv. Liefert eine Liste von Events. Schema-Check: Wenn die Archiv-Version älter ist als die aktuelle Schema-Version (= Dedup-Logik hat sich geändert), wird das Archiv verworfen und beim Backfill neu aufgebaut.""" if not os.path.exists(self.archive_file): return [] try: with open(self.archive_file, encoding="utf-8") as f: data = json.load(f) meta = data.get("meta") or {} stored_version = meta.get("schema_version", 1) if stored_version < self.ARCHIVE_SCHEMA_VERSION: print( f" [Seko] Archiv-Schema veraltet " f"(v{stored_version} < v{self.ARCHIVE_SCHEMA_VERSION}) — " f"Archiv wird verworfen und neu aufgebaut.") return [] return data.get("events", []) except Exception as e: print(f" [Seko] Archiv-Lade-Fehler: {e}") return [] def _save_archive(self, events: list, meta: dict = None): """Schreibe das Event-Archiv atomar.""" try: payload = { "meta": meta or {}, "events": events, } tmp = self.archive_file + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=1) try: os.replace(tmp, self.archive_file) except OSError: import shutil shutil.move(tmp, self.archive_file) except Exception as e: print(f" [Seko] Archiv-Schreib-Fehler: {e}") # Schema-Version des Archivs. Wird beim Laden geprüft — bei Mismatch # wird das Archiv verworfen und neu befüllt. Bumpen wenn sich die # Event-Struktur oder Dedup-Logik ändert. ARCHIVE_SCHEMA_VERSION = 2 def _merge_to_archive(self, new_events: list) -> tuple: """Merge neue Events ins Archiv mit Payload-Dedup. WICHTIG: Die Seko-API (ctrl_getflowrecords) liefert bei jedem Poll den *zuletzt gemessenen Dosiervorgang* pro Pumpe/Maschine/Chemikalie. Wenn seit dem letzten Poll nichts Neues dosiert wurde, kommt der gleiche Event mit neuem Timestamp. Naives Dedup auf `ts_ms` würde jeden Poll als neuen Event zählen → Mehrfachzählung. Korrekte Logik: Dedup auf (machine, chemical, ml, duration_s, flow_rate). Wenn der gleiche Dosierwert bei identischer Dauer und Flow-Rate erneut kommt, ist's derselbe Vorgang wiederholt. Edge-Case: Zwei echt aufeinanderfolgende Dosierungen mit exakt identischen Werten werden als eine gezählt — bei ml auf 3 Nachkomma- stellen + Flow-Rate (0.1 ml/min Präzision) ist das praktisch ausgeschlossen; Flow-Rate variiert durch Druckschwankungen. Liefert (total_count, added_count) zurück.""" existing = self._load_archive() # Pro (machine, chemical_key): letzter bekannter Event last_by_mc = {} for ev in existing: key = (ev.get("machine"), ev.get("chemical_key")) last = last_by_mc.get(key) if not last or ev.get("ts_ms", 0) > last.get("ts_ms", 0): last_by_mc[key] = ev # Neue Events chronologisch verarbeiten new_events_sorted = sorted(new_events, key=lambda e: e.get("ts_ms", 0)) added = 0 for ev in new_events_sorted: mc_key = (ev.get("machine"), ev.get("chemical_key")) last = last_by_mc.get(mc_key) if last: same_payload = ( round(ev.get("ml", 0), 3) == round(last.get("ml", 0), 3) and ev.get("duration_s") == last.get("duration_s") and round(ev.get("flow_rate", 0), 1) == round(last.get("flow_rate", 0), 1) ) if same_payload: continue # Wiederholung des letzten Polls existing.append(ev) last_by_mc[mc_key] = ev added += 1 existing.sort(key=lambda e: e.get("ts_ms", 0)) meta = { "schema_version": self.ARCHIVE_SCHEMA_VERSION, "last_update": datetime.now(timezone.utc).isoformat(), "total_events": len(existing), "earliest_ts_ms": existing[0]["ts_ms"] if existing else None, "latest_ts_ms": existing[-1]["ts_ms"] if existing else None, "earliest_iso": existing[0].get("ts_iso") if existing else None, "latest_iso": existing[-1].get("ts_iso") if existing else None, } self._save_archive(existing, meta) return len(existing), added def correlate_with_miele(self, seko_data: dict) -> dict: """Korreliert Seko-Dosier-Events mit Miele-Zyklen. Ansatz: 1. water_log.json laden (enthält alle Miele-Zyklen mit machine/program/started_at/ended_at/max_temp) 2. Für jeden Zyklus alle Seko-Events der gleichen Maschine im Zeitfenster [started_at - 2 min, ended_at + 5 min] sammeln 3. Aggregat pro Zyklus → Liste Chemikalien mit ml + € 4. Aggregat pro (Programm + Temperatur) → Ø + Spannbreite Das Zeitfenster ist asymmetrisch: -2 min (Dosierpumpen starten manchmal kurz vor Miele-Zyklus-Erkennung), +5 min (Miele meldet Ende oft bevor die letzte Spül-Dosierung vorbei ist). """ events = seko_data.get("dosing_events_24h") or [] if not events: return {"status": "no_events", "cycles": [], "by_program": {}} # water_log laden water_log_file = os.path.join(self.output_dir, "water_log.json") if not os.path.exists(water_log_file): return {"status": "no_miele_data", "cycles": [], "by_program": {}} try: with open(water_log_file, encoding="utf-8") as f: water_log = json.load(f) except Exception as e: return {"status": f"water_log_error: {e}", "cycles": [], "by_program": {}} all_cycles = water_log.get("cycles", []) # Nur Zyklen aus dem Event-Fenster (letzte 24h + Puffer) cutoff_ms = (datetime.now().timestamp() - 26 * 3600) * 1000 relevant_cycles = [] for c in all_cycles: try: ended = datetime.fromisoformat(c["ended_at"].replace("Z", "+00:00")) if ended.timestamp() * 1000 >= cutoff_ms: relevant_cycles.append(c) except Exception: continue if not relevant_cycles: return {"status": "no_recent_cycles", "cycles": [], "by_program": {}, "miele_cycles_total": len(all_cycles)} # Events indizieren nach Maschine für schnelle Suche events_by_machine = {} for ev in events: m = ev.get("machine", "?") events_by_machine.setdefault(m, []).append(ev) # Zeitfenster-Puffer: vor Start und nach Ende PRE_BUFFER_MS = 2 * 60 * 1000 # 2 Minuten vor Zyklus-Start POST_BUFFER_MS = 5 * 60 * 1000 # 5 Minuten nach Zyklus-Ende correlated_cycles = [] for cyc in relevant_cycles: try: started = datetime.fromisoformat( cyc["started_at"].replace("Z", "+00:00")).timestamp() * 1000 ended = datetime.fromisoformat( cyc["ended_at"].replace("Z", "+00:00")).timestamp() * 1000 except Exception: continue machine = cyc.get("machine", "?") machine_events = events_by_machine.get(machine, []) matching_events = [] for ev in machine_events: ts = ev.get("ts_ms", 0) if (started - PRE_BUFFER_MS) <= ts <= (ended + POST_BUFFER_MS): matching_events.append(ev) # Pro Chemikalie aggregieren by_chem = {} total_ml = 0.0 total_cost = 0.0 for ev in matching_events: chem = ev.get("chemical_name") or ev.get("chemical_key", "?") if chem not in by_chem: by_chem[chem] = {"ml": 0.0, "events": 0, "cost_eur": 0.0} by_chem[chem]["ml"] += ev.get("ml", 0) by_chem[chem]["events"] += 1 by_chem[chem]["cost_eur"] += ev.get("cost_eur", 0) total_ml += ev.get("ml", 0) total_cost += ev.get("cost_eur", 0) # Chemikalien-Liste sortiert nach ml chems_list = [] for chem, data in sorted( by_chem.items(), key=lambda x: -x[1]["ml"]): chems_list.append({ "chemical": chem, "ml": round(data["ml"], 1), "events": data["events"], "cost_eur": round(data["cost_eur"], 4), }) correlated_cycles.append({ "machine": machine, "program": cyc.get("program", "?"), "started_at": cyc.get("started_at"), "ended_at": cyc.get("ended_at"), "max_temp": cyc.get("max_temp"), "water_liters": cyc.get("water_liters"), "event_count": len(matching_events), "total_ml": round(total_ml, 1), "total_cost_eur": round(total_cost, 4), "chemicals": chems_list, }) # Aggregation pro Programm (+ Temperatur) # Key: "{program} @ {temp}°C" (oder "{program}" wenn temp fehlt) by_program = {} for cc in correlated_cycles: # Nur Zyklen mit Events in Aggregate (ohne Dosierung = wurden # korreliert aber die Zeitfenster passten nicht oder Trockner) if cc["event_count"] == 0: continue prog = cc["program"] or "?" temp = cc.get("max_temp") key = f"{prog} @ {temp}°C" if temp else prog if key not in by_program: by_program[key] = { "program": prog, "max_temp": temp, "cycle_count": 0, "total_ml": 0.0, "total_cost_eur": 0.0, "by_chemical": {}, "machines_used": set(), } agg = by_program[key] agg["cycle_count"] += 1 agg["total_ml"] += cc["total_ml"] agg["total_cost_eur"] += cc["total_cost_eur"] agg["machines_used"].add(cc["machine"]) for ch in cc["chemicals"]: chem = ch["chemical"] if chem not in agg["by_chemical"]: agg["by_chemical"][chem] = {"total_ml": 0.0, "cycle_count": 0} agg["by_chemical"][chem]["total_ml"] += ch["ml"] agg["by_chemical"][chem]["cycle_count"] += 1 # Ø pro Zyklus berechnen + Sets in Listen for key, agg in by_program.items(): n = agg["cycle_count"] or 1 agg["avg_ml_per_cycle"] = round(agg["total_ml"] / n, 1) agg["avg_cost_per_cycle"] = round(agg["total_cost_eur"] / n, 4) agg["total_ml"] = round(agg["total_ml"], 1) agg["total_cost_eur"] = round(agg["total_cost_eur"], 4) agg["machines_used"] = sorted(agg["machines_used"]) # Pro-Chemikalie Ø berechnen chems_out = [] for chem, cd in sorted( agg["by_chemical"].items(), key=lambda x: -x[1]["total_ml"]): chems_out.append({ "chemical": chem, "total_ml": round(cd["total_ml"], 1), "avg_ml_per_cycle": round(cd["total_ml"] / n, 1), "cycles_with_this": cd["cycle_count"], }) agg["chemicals"] = chems_out del agg["by_chemical"] return { "status": "ok", "window_hours": 24, "miele_cycles_in_window": len(relevant_cycles), "miele_cycles_correlated": sum( 1 for c in correlated_cycles if c["event_count"] > 0), "cycles": sorted( correlated_cycles, key=lambda c: c["ended_at"] or "", reverse=True), "by_program": by_program, } def compute_archive_aggregations(self, settings: dict = None, aggregated_data: dict = None) -> dict: """Große Aggregation über das gesamte Event-Archiv. Nutzt das persistente seko_events_archive.json + water_log.json und produziert: - Programm × Maschinengröße (Small/Medium/Large) Aggregate - Verbrauch im Zeitverlauf (täglich) - Hochrechnung Monat/Jahr - Wirtschaftlichkeit (Umsatz − Chemie pro Programm) - Anomalien (Zyklen die >X% vom Programm-Schnitt abweichen) WICHTIG: Die Event-basierten Aggregationen sind UNVOLLSTÄNDIG — der Seko-Event-Endpoint (ctrl_getflowrecords) liefert historisch nur ~30% der tatsächlichen Dosierungen. Die Haupt-KPIs (Monatskosten, Hochrechnung) kommen deshalb aus aggregated_data (= chemicals_today/month/year aus ctrl_get_chemical_usage). Die Event-Auswertung ist für relative Programm-Vergleiche und Anomalien weiter nützlich. aggregated_data: Das komplette seko.json-Data-Dict mit chemicals_today, chemicals_month, chemicals_year, summary. settings: Optional dict mit: - 'pump_start_date': ISO-Datum, ab wann ausgewertet wird - 'anomaly_thresholds': dict {chemical_or_program: percent} - 'machine_size_map': {machine: 'small'|'medium'|'large'} - 'machine_prices': {size: euro_per_wash} """ # Defaults settings = settings or {} pump_start_iso = settings.get("pump_start_date", "2026-03-21") # Default-Maschinengrößen aus WASHER_MAP ableitbar (Small/Medium/Large) size_map = settings.get("machine_size_map") or { "A": "small", "B": "small", "C": "medium", "D": "large", "E": "large", "F": "large", } # Default-Preise (Brutto): kann der Betreiber überschreiben prices_per_wash = settings.get("machine_prices") or { "small": 4.50, "medium": 6.50, "large": 8.00, } # Anomalie-Schwellen (% Abweichung vom Programm-Schnitt) anomaly_thresholds = settings.get("anomaly_thresholds") or {} default_anomaly_pct = settings.get("default_anomaly_pct", 50) try: pump_start_ms = datetime.fromisoformat( pump_start_iso.replace("Z", "+00:00") ).timestamp() * 1000 except Exception: # Fallback: 4 Wochen zurück pump_start_ms = (datetime.now().timestamp() - 28 * 86400) * 1000 pump_start_iso = datetime.fromtimestamp( pump_start_ms / 1000).isoformat() # Archiv laden archive = self._load_archive() # Nur Events ab Pumpen-Start archive = [e for e in archive if e.get("ts_ms", 0) >= pump_start_ms] if not archive: return { "status": "no_archive_data", "pump_start": pump_start_iso, } # water_log laden — alle Zyklen ab Pumpen-Start all_cycles = [] water_log_file = os.path.join(self.output_dir, "water_log.json") if os.path.exists(water_log_file): try: with open(water_log_file, encoding="utf-8") as f: wl = json.load(f) for c in wl.get("cycles", []): try: ended = datetime.fromisoformat( c["ended_at"].replace("Z", "+00:00")) if ended.timestamp() * 1000 >= pump_start_ms: all_cycles.append(c) except Exception: continue except Exception: pass # ── 1. Korrelation aller Zyklen mit allen Events ── events_by_machine = {} for ev in archive: m = ev.get("machine", "?") events_by_machine.setdefault(m, []).append(ev) PRE_BUFFER_MS = 2 * 60 * 1000 POST_BUFFER_MS = 5 * 60 * 1000 correlated_cycles = [] for cyc in all_cycles: try: started = datetime.fromisoformat( cyc["started_at"].replace("Z", "+00:00")).timestamp() * 1000 ended = datetime.fromisoformat( cyc["ended_at"].replace("Z", "+00:00")).timestamp() * 1000 except Exception: continue machine = cyc.get("machine", "?") machine_events = events_by_machine.get(machine, []) matching = [ev for ev in machine_events if (started - PRE_BUFFER_MS) <= ev.get("ts_ms", 0) <= (ended + POST_BUFFER_MS)] if not matching: continue # Nur Zyklen mit Dosierung by_chem = {} total_ml = 0.0 total_cost = 0.0 for ev in matching: chem = ev.get("chemical_name") or ev.get("chemical_key", "?") if chem not in by_chem: by_chem[chem] = {"ml": 0.0, "cost_eur": 0.0} by_chem[chem]["ml"] += ev.get("ml", 0) by_chem[chem]["cost_eur"] += ev.get("cost_eur", 0) total_ml += ev.get("ml", 0) total_cost += ev.get("cost_eur", 0) correlated_cycles.append({ "machine": machine, "machine_size": size_map.get(machine, "unknown"), "program": cyc.get("program", "?"), "started_at": cyc.get("started_at"), "ended_at": cyc.get("ended_at"), "max_temp": cyc.get("max_temp"), "water_liters": cyc.get("water_liters"), "ts_ms_end": ended, "total_ml": round(total_ml, 1), "total_cost_eur": round(total_cost, 4), "by_chem": {k: {"ml": round(v["ml"], 1), "cost_eur": round(v["cost_eur"], 4)} for k, v in by_chem.items()}, }) # ── 2. Aggregate pro Programm × Maschinengröße ── # Key: (program, size) prog_size_agg = {} for cc in correlated_cycles: prog = cc["program"] size = cc["machine_size"] key = f"{prog}__{size}" if key not in prog_size_agg: prog_size_agg[key] = { "program": prog, "machine_size": size, "cycle_count": 0, "total_ml": 0.0, "total_cost_eur": 0.0, "total_water_l": 0.0, "by_chem_total_ml": {}, "by_chem_total_cost": {}, "machines": set(), "ml_per_cycle_list": [], "cost_per_cycle_list": [], } agg = prog_size_agg[key] agg["cycle_count"] += 1 agg["total_ml"] += cc["total_ml"] agg["total_cost_eur"] += cc["total_cost_eur"] agg["total_water_l"] += cc.get("water_liters") or 0 agg["machines"].add(cc["machine"]) agg["ml_per_cycle_list"].append(cc["total_ml"]) agg["cost_per_cycle_list"].append(cc["total_cost_eur"]) for chem, cd in cc["by_chem"].items(): agg["by_chem_total_ml"][chem] = ( agg["by_chem_total_ml"].get(chem, 0) + cd["ml"]) agg["by_chem_total_cost"][chem] = ( agg["by_chem_total_cost"].get(chem, 0) + cd["cost_eur"]) # Aggregate finalisieren (Ø + Standardabweichung) for key, agg in prog_size_agg.items(): n = agg["cycle_count"] or 1 agg["avg_ml_per_cycle"] = round(agg["total_ml"] / n, 1) agg["avg_cost_per_cycle"] = round(agg["total_cost_eur"] / n, 4) agg["avg_water_l_per_cycle"] = round(agg["total_water_l"] / n, 1) # Standardabweichung if n > 1: mean = agg["avg_ml_per_cycle"] variance = sum((x - mean) ** 2 for x in agg["ml_per_cycle_list"]) / n agg["std_ml_per_cycle"] = round(variance ** 0.5, 1) else: agg["std_ml_per_cycle"] = 0 # Wirtschaftlichkeit price = prices_per_wash.get(agg["machine_size"], 0) agg["price_per_wash"] = price agg["margin_per_cycle"] = round(price - agg["avg_cost_per_cycle"], 4) agg["margin_pct"] = round( (agg["margin_per_cycle"] / price * 100), 1) if price > 0 else 0 agg["total_ml"] = round(agg["total_ml"], 1) agg["total_cost_eur"] = round(agg["total_cost_eur"], 2) agg["total_water_l"] = round(agg["total_water_l"], 1) agg["machines"] = sorted(agg["machines"]) # Pro-Chemikalie Ø chems_out = [] for chem, total_ml in sorted( agg["by_chem_total_ml"].items(), key=lambda x: -x[1]): chems_out.append({ "chemical": chem, "total_ml": round(total_ml, 1), "total_cost_eur": round( agg["by_chem_total_cost"].get(chem, 0), 4), "avg_ml_per_cycle": round(total_ml / n, 1), }) agg["chemicals"] = chems_out # Sets/Listen die nicht gebraucht werden rauswerfen del agg["by_chem_total_ml"] del agg["by_chem_total_cost"] del agg["ml_per_cycle_list"] del agg["cost_per_cycle_list"] # ── 3. Aggregate pro Programm (alle Größen zusammen) ── prog_only_agg = {} for cc in correlated_cycles: prog = cc["program"] if prog not in prog_only_agg: prog_only_agg[prog] = { "program": prog, "cycle_count": 0, "total_ml": 0.0, "total_cost_eur": 0.0, "machines": set(), "sizes": set(), } a = prog_only_agg[prog] a["cycle_count"] += 1 a["total_ml"] += cc["total_ml"] a["total_cost_eur"] += cc["total_cost_eur"] a["machines"].add(cc["machine"]) a["sizes"].add(cc["machine_size"]) for prog, a in prog_only_agg.items(): n = a["cycle_count"] or 1 a["avg_ml_per_cycle"] = round(a["total_ml"] / n, 1) a["avg_cost_per_cycle"] = round(a["total_cost_eur"] / n, 4) a["total_ml"] = round(a["total_ml"], 1) a["total_cost_eur"] = round(a["total_cost_eur"], 2) a["machines"] = sorted(a["machines"]) a["sizes"] = sorted(a["sizes"]) # ── 4. Tagesverlauf (für Linienchart) ── # Aggregate pro Tag: total_ml, total_cost from collections import defaultdict daily = defaultdict(lambda: {"events": 0, "total_ml": 0.0, "total_cost_eur": 0.0, "by_chem_ml": {}}) for ev in archive: ts = ev.get("ts_ms", 0) if ts <= 0: continue day = datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d") daily[day]["events"] += 1 daily[day]["total_ml"] += ev.get("ml", 0) daily[day]["total_cost_eur"] += ev.get("cost_eur", 0) chem = ev.get("chemical_name") or ev.get("chemical_key", "?") daily[day]["by_chem_ml"][chem] = ( daily[day]["by_chem_ml"].get(chem, 0) + ev.get("ml", 0)) daily_series = [] for day in sorted(daily.keys()): d = daily[day] daily_series.append({ "date": day, "events": d["events"], "total_ml": round(d["total_ml"], 1), "total_cost_eur": round(d["total_cost_eur"], 4), "by_chem_ml": {k: round(v, 1) for k, v in d["by_chem_ml"].items()}, }) # ── 5. Hochrechnung Monat/Jahr ── # Basis: Ø-Tagesverbrauch der letzten N Tage (mind. 7) × 30 bzw. 365 days_with_data = len(daily_series) recent_days = daily_series[-min(28, days_with_data):] if recent_days: avg_daily_cost = sum(d["total_cost_eur"] for d in recent_days) / len(recent_days) avg_daily_ml = sum(d["total_ml"] for d in recent_days) / len(recent_days) else: avg_daily_cost = 0 avg_daily_ml = 0 projection = { "based_on_days": len(recent_days), "avg_daily_cost_eur": round(avg_daily_cost, 2), "avg_daily_ml": round(avg_daily_ml, 1), "monthly_cost_eur": round(avg_daily_cost * 30, 2), "monthly_litres": round(avg_daily_ml * 30 / 1000, 1), "yearly_cost_eur": round(avg_daily_cost * 365, 2), "yearly_litres": round(avg_daily_ml * 365 / 1000, 1), } # ── 6. Anomalie-Erkennung ── # Pro Programm × Größe: Zyklen die mehr als X% vom Schnitt abweichen anomalies = [] for cc in correlated_cycles: key = f"{cc['program']}__{cc['machine_size']}" agg = prog_size_agg.get(key) if not agg or agg["cycle_count"] < 3: continue # Nicht genug Datenbasis avg = agg["avg_ml_per_cycle"] if avg <= 0: continue actual = cc["total_ml"] deviation_pct = ((actual - avg) / avg) * 100 # Schwelle: programm-spezifisch oder Default threshold = anomaly_thresholds.get( cc["program"], default_anomaly_pct) if abs(deviation_pct) > threshold: anomalies.append({ "machine": cc["machine"], "program": cc["program"], "ended_at": cc["ended_at"], "actual_ml": actual, "avg_ml": avg, "deviation_pct": round(deviation_pct, 1), "actual_cost": cc["total_cost_eur"], "threshold_pct": threshold, }) # Neueste zuerst anomalies.sort(key=lambda a: a["ended_at"] or "", reverse=True) # ── 7. Gesamt-KPIs ── total_archive_ml = sum(e.get("ml", 0) for e in archive) total_archive_cost = sum(e.get("cost_eur", 0) for e in archive) total_archive_events = len(archive) first_day = daily_series[0]["date"] if daily_series else None last_day = daily_series[-1]["date"] if daily_series else None days_active = days_with_data # ── 7.5 Aggregierte Sicht (OFFIZIELLE Seko-Zahlen) ── # Diese Werte kommen vom ctrl_get_chemical_usage Endpoint und sind # mit der Realität abgeglichen (BWA-Übereinstimmung). Sie werden als # Haupt-KPIs angezeigt, während die Event-basierten Zahlen oben nur # für relative Vergleiche (Programm × Größe, Anomalien) taugen. aggregated_view = None if aggregated_data: chem_month = aggregated_data.get("chemicals_month") or [] chem_year = aggregated_data.get("chemicals_year") or [] chem_today = aggregated_data.get("chemicals_today") or [] chem_ref = aggregated_data.get("chemicals_ref") or [] summary = aggregated_data.get("summary") or {} # Pro-Maschine-Aggregation (Summe über alle Chemikalien) per_machine_month = {} for chem in chem_month: pm = chem.get("per_machine", {}) or {} price = chem.get("price_per_litre", 0) or 0 for machine_label, litres in pm.items(): if not isinstance(litres, (int, float)) or litres <= 0: continue # machine_label ist z.B. "A Small" → wir wollen "A" mkey = machine_label.split(" ")[0] if mkey not in per_machine_month: per_machine_month[mkey] = { "machine": mkey, "size_label": " ".join(machine_label.split(" ")[1:]), "total_litres": 0.0, "total_cost_eur": 0.0, "by_chem": {}, } per_machine_month[mkey]["total_litres"] += litres per_machine_month[mkey]["total_cost_eur"] += litres * price per_machine_month[mkey]["by_chem"][chem.get("name", chem.get("key"))] = { "litres": round(litres, 3), "cost_eur": round(litres * price, 2), } for mkey, d in per_machine_month.items(): d["total_litres"] = round(d["total_litres"], 3) d["total_cost_eur"] = round(d["total_cost_eur"], 2) # Pro-Chemikalie kompakt (nur die nicht-leeren) chem_month_clean = [ { "key": c.get("key"), "name": c.get("name"), "litres": round(c.get("total_litres", 0), 3), "cost_eur": round(c.get("cost_eur", 0), 2), "price_per_litre": c.get("price_per_litre", 0), "runtime_seconds": c.get("runtime_seconds", 0), } for c in chem_month if (c.get("total_litres") or 0) > 0 ] # Hochrechnung auf 30d basierend auf 28-Tage-Fenster # (chemicals_month ist laut API rollendes 30d-Fenster, wir # übernehmen die Werte direkt) monthly_cost = summary.get("total_cost_month", 0) yearly_cost = summary.get("total_cost_year", 0) monthly_litres = summary.get("total_litres_month", 0) yearly_litres = summary.get("total_litres_year", 0) aggregated_view = { "source": "seko_ctrl_get_chemical_usage", "note": ("Diese Werte kommen vom offiziellen Seko-Aggregat-" "Endpoint und stimmen mit deiner BWA überein."), "summary": { "total_cost_month": round(monthly_cost, 2), "total_cost_year": round(yearly_cost, 2), "total_litres_month": round(monthly_litres, 2), "total_litres_year": round(yearly_litres, 2), "avg_cost_per_litre": summary.get("cost_per_litre", 0), }, "chemicals_today": chem_today, "chemicals_month": chem_month_clean, "per_machine_month": list(per_machine_month.values()), "ref_days": summary.get("ref_days"), "total_cost_ref": summary.get("total_cost_ref", 0), "total_litres_ref": summary.get("total_litres_ref", 0), } return { "status": "ok", "pump_start": pump_start_iso, "first_event_day": first_day, "last_event_day": last_day, "days_with_data": days_active, "aggregated_view": aggregated_view, "totals": { "total_events": total_archive_events, "total_ml": round(total_archive_ml, 1), "total_litres": round(total_archive_ml / 1000, 2), "total_cost_eur": round(total_archive_cost, 2), "total_cycles_correlated": len(correlated_cycles), "note": ("Event-basierte Zahlen: unvollständig (~30% Abdeckung). " "Für absolute Mengen/Kosten: aggregated_view nutzen."), }, "projection": projection, "by_program_size": prog_size_agg, "by_program": prog_only_agg, "daily_series": daily_series, "anomalies": anomalies, "settings_used": { "pump_start_date": pump_start_iso, "machine_size_map": size_map, "machine_prices": prices_per_wash, "default_anomaly_pct": default_anomaly_pct, "anomaly_thresholds": anomaly_thresholds, }, } def write_output(self, data: dict): """Schreibe seko.json (atomar).""" # ── Korrelation hinzufügen bevor geschrieben wird data["cycle_correlation"] = self.correlate_with_miele(data) # ── NEU: Archiv-Aggregationen (Settings aus settings.json laden) analysis_settings = {} settings_file = os.path.join(self.output_dir, "settings.json") if os.path.exists(settings_file): try: with open(settings_file, encoding="utf-8") as f: s = json.load(f) analysis_settings = s.get("waschmittel_analyse") or {} except Exception: pass data["chemical_analysis"] = self.compute_archive_aggregations( analysis_settings, aggregated_data=data) tmp = self.output_file + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) try: os.replace(tmp, self.output_file) except OSError: import shutil shutil.move(tmp, self.output_file) ts = datetime.now().strftime("%H:%M:%S") s = data.get("summary", {}) ev_meta = data.get("dosing_events_meta", {}) ev_total = ev_meta.get("total_events", 0) corr = data.get("cycle_correlation", {}) corr_info = "" if corr.get("status") == "ok": corr_info = (f" | Zyklen korreliert: " f"{corr.get('miele_cycles_correlated', 0)}" f"/{corr.get('miele_cycles_in_window', 0)}") # Archiv-Info archive_total = data.get("meta", {}).get("archive_total_events", 0) archive_added = data.get("meta", {}).get("archive_new_events_this_poll", 0) archive_info = f" | Archiv: {archive_total} (+{archive_added})" print( f"[{ts}] SEKO | " f"{s.get('pumps_online', 0)}/{s.get('pumps_total', 0)} Pumpen | " f"Monat: {s.get('total_litres_month', 0):.1f} l " f"({s.get('total_cost_month', 0):.2f} €) | " f"Jahr: {s.get('total_litres_year', 0):.1f} l | " f"Events 24h: {ev_total}" f"{corr_info}" f"{archive_info}" ) def _backfill_archive_if_empty(self, days_back: int = 28): """Wenn das Archiv leer ist (oder sehr klein), hol einmal die letzten N Tage und befülle es. Wird beim Poller-Start aufgerufen. Geht in Tages-Chunks vor (Seko-API liefert sonst zu viele Daten auf einmal). """ existing = self._load_archive() if len(existing) >= 100: # Wir haben schon eine substanzielle Menge Daten — kein Backfill return print(f" [Seko] Backfill: hole letzte {days_back} Tage in Tages-Chunks...") all_new_events = [] for day_offset in range(days_back, 0, -1): end_s = int(datetime.now().timestamp() - (day_offset - 1) * 86400) start_s = end_s - 86400 # 1 Tag day_str = datetime.fromtimestamp(start_s).strftime("%Y-%m-%d") events_this_day = 0 for gid in GIDS: pump_name = PUMP_NAMES.get(gid, gid[:16]) params = { "startTime": start_s, "endTime": end_s, "washer": WASHER_MAP, "ownerID": SEKO_OWNER, "applicationID": SEKO_APP_ID, "GID": gid, "installationSiteName": "Waschsalon Nord GmbH", } try: r = self.session.post( f"{SEKO_BASE}/application/ctrl_getflowrecords", json=params, timeout=60, ) if r.status_code != 200: continue flow_data = r.json() except Exception as e: print(f" [Seko] Backfill-Fehler Tag {day_str} {gid[:8]}: {e}") continue # Parsing in try-Block pro GID — damit ein einzelnes schlechtes # Event oder eine defekte Struktur nicht den ganzen Backfill reißt try: chem_flow = (flow_data.get("devicedata") or {}).get( "CHEMICAL_FLOW") or {} gid_bucket = chem_flow.get(gid) or {} for chem_key, chem_data in gid_bucket.items(): if not isinstance(chem_data, dict): continue chem_name = chem_data.get("CN", chem_key) price_per_l = CHEMICAL_PRICES.get( chem_key.upper(), CHEMICAL_PRICES.get(chem_name.upper(), 0)) was_dict = chem_data.get("WAS") or {} for was_key, was_data in was_dict.items(): if not isinstance(was_data, dict): continue wn = was_data.get("WN", "") machine = wn.split(" ")[0] if wn else "?" chs = was_data.get("CHS") or {} for chs_key, events in chs.items(): if not isinstance(events, list): continue for ev in events: if not isinstance(ev, dict): continue ts_ms = ev.get("T") or ev.get("timestamp") or 0 if ts_ms is None or not ts_ms: continue litres = _safe_float(ev.get("TQ")) ml = _safe_float(ev.get("DQ")) duration_s = _safe_int(ev.get("DT")) all_new_events.append({ "ts_ms": ts_ms, "ts_iso": datetime.fromtimestamp( ts_ms / 1000, tz=timezone.utc).isoformat(), "pump_gid": gid, "pump_name": pump_name, "washer_key": was_key, "washer_name": wn, "channel_set": chs_key, "machine": machine, "chemical_key": chem_key, "chemical_name": chem_name, "ml": round(ml, 3), "litres": round(litres, 6), "duration_s": duration_s, "flow_rate": round(_safe_float(ev.get("FR")), 1), "cost_eur": round(ml / 1000.0 * price_per_l, 4) if price_per_l else 0, }) events_this_day += 1 except Exception as e: print(f" [Seko] Backfill Parse-Fehler Tag {day_str} {gid[:8]}: {e}") continue # Progress pro Tag (alle 7 Tage eine Zeile damit's nicht zu viel wird) if day_offset % 7 == 0 or day_offset == 1: print(f" [Seko] Backfill: bei Tag {day_str} · bisher {len(all_new_events)} Events gesammelt") # Etwas Schonzeit pro Tag — Server nicht überfordern time.sleep(0.3) if all_new_events: total, added = self._merge_to_archive(all_new_events) print(f" [Seko] Backfill fertig: {added} neue Events ins Archiv " f"(jetzt {total} insgesamt)") else: print(" [Seko] Backfill ohne Ergebnis (keine Events).") def run_poller(self): """Endlos-Polling.""" print(" [Seko] Starte Poller...") if not self.login(): print(" [Seko] Login fehlgeschlagen, retry in 60s...") time.sleep(60) return self.run_poller() # NEU: Beim ersten Start einmal die letzten 28 Tage holen try: self._backfill_archive_if_empty(days_back=28) except Exception as e: print(f" [Seko] Backfill-Fehler: {e}") print(f" [Seko] Polling gestartet (alle {POLL_INTERVAL}s)") fail_count = 0 while True: try: data = self.poll_once() if "error" not in data: self.write_output(data) fail_count = 0 else: fail_count += 1 if fail_count >= 3: print(" [Seko] 3x fehlgeschlagen, Re-Login...") self.login() fail_count = 0 except Exception as e: print(f" [Seko] Fehler: {e}") fail_count += 1 time.sleep(POLL_INTERVAL) def format_duration(seconds): """Sekunden in HH:MM:SS formatieren.""" h = int(seconds // 3600) m = int((seconds % 3600) // 60) s = int(seconds % 60) return f"{h:02d}:{m:02d}:{s:02d}" def _safe_float(val, default=0.0): """Robuste float-Konvertierung: None, leere Strings, Nicht-Zahlen werden zu default statt TypeError. Seko liefert gelegentlich None für einzelne Event-Felder (DQ/TQ/FR/DT), das darf nicht crashen.""" if val is None: return default try: return float(val) except (TypeError, ValueError): return default def _safe_int(val, default=0): """Robuste int-Konvertierung (siehe _safe_float).""" if val is None: return default try: return int(val) except (TypeError, ValueError): return default def start_seko_thread(output_dir="."): """Startet den Seko-Poller als Daemon-Thread. Wenn keine Credentials vorhanden: gibt None zurück.""" has_creds = load_credentials_from_settings(output_dir) if not has_creds: print(" [Seko] Keine Credentials in settings.json — Poller pausiert.") print(" [Seko] Bitte in den Einstellungen eintragen.") return None poller = SekoPoller(output_dir=output_dir) t = threading.Thread( target=poller.run_poller, daemon=True, name="SekoPoller", ) t.start() return poller def try_login_with_settings(output_dir="."): """Einmaliger Login-Test für /api/seko/test. Gibt (ok: bool, message: str, pumps: int|None) zurück.""" load_credentials_from_settings(output_dir) poller = SekoPoller(output_dir=output_dir) try: ok = poller.login() if not ok: return False, "Login fehlgeschlagen — E-Mail/Passwort prüfen", None # Optional: ein Poll für Pumpenzahl try: data = poller.poll_once() pumps = len((data or {}).get("pumps", [])) or len((data or {}).get("chemicals", [])) except Exception: pumps = None return True, "Verbunden", pumps except Exception as e: return False, str(e), None if __name__ == "__main__": print("Seko Web Poller — Standalone-Test") poller = SekoPoller() if poller.login(): data = poller.poll_once() poller.write_output(data) print(json.dumps(data, indent=2, ensure_ascii=False)[:3000])