""" AutoRecover - Drucker Fehler Handler Erkennt das "Error"-Fenster (Java AWT) und klickt automatisch "Recover" (max. 3x). Zusätzlich werden die erkannten Fehlermeldungen (best-effort) an Directus gemeldet (Collection agent_logs, source="recover"). Die Directus-Zugangsdaten kommen aus einer config.json neben diesem Skript ODER aus der des Agenten (../skrift-agent/config.json) ODER aus Umgebungsvariablen (SKRIFT_DIRECTUS_URL / SKRIFT_DIRECTUS_TOKEN / SKRIFT_MACHINE_ID). """ import win32gui import win32con import ctypes import time import os import json try: import requests except Exception: # requests optional – ohne läuft AutoRecover trotzdem (nur ohne Directus-Log) requests = None MAX_VERSUCHE = 3 PRUEF_INTERVAL = 2 # Sekunden zwischen den Checks _HIER = os.path.dirname(os.path.abspath(__file__)) def _lade_directus_cfg(): """Directus-Zugang aus AutoRecover- oder Agenten-config.json bzw. Env-Variablen.""" for pfad in (os.path.join(_HIER, "config.json"), os.path.join(_HIER, "..", "skrift-agent", "config.json")): try: with open(pfad, "r", encoding="utf-8") as f: c = json.load(f) if c.get("directus_url") and c.get("directus_token"): return c.get("directus_url"), c.get("directus_token"), c.get("machine_id") except Exception: pass url = os.environ.get("SKRIFT_DIRECTUS_URL") token = os.environ.get("SKRIFT_DIRECTUS_TOKEN") mid = os.environ.get("SKRIFT_MACHINE_ID") return url, token, (int(mid) if mid and str(mid).isdigit() else None) _DX_URL, _DX_TOKEN, _DX_MID = _lade_directus_cfg() def _lade_esp_cfg(): """ESP-Zugang (MOSFET-Trigger) aus config.json bzw. Env-Variablen. Erwartete Schlüssel in der config.json (neben diesem Skript oder beim Agenten): "esp_url": "http://192.168.2.50" Basis-URL des ESP (Pflicht, sonst aus) "esp_path": "/trigger" Endpunkt (Default /trigger) "esp_key": "geheim" optionaler Schutz-Schlüssel (?key=) "esp_timeout": 5 Sekunden (Default 5) """ url = path = key = None timeout = 5 for pfad in (os.path.join(_HIER, "config.json"), os.path.join(_HIER, "..", "skrift-agent", "config.json")): try: with open(pfad, "r", encoding="utf-8") as f: c = json.load(f) if c.get("esp_url"): url = c.get("esp_url") path = c.get("esp_path") key = c.get("esp_key") timeout = c.get("esp_timeout", 5) break except Exception: pass url = url or os.environ.get("SKRIFT_ESP_URL") path = path or os.environ.get("SKRIFT_ESP_PATH") or "/trigger" key = key or os.environ.get("SKRIFT_ESP_KEY") or "" try: timeout = float(timeout) except Exception: timeout = 5.0 return (url.rstrip("/") if url else None), path, key, timeout _ESP_URL, _ESP_PATH, _ESP_KEY, _ESP_TIMEOUT = _lade_esp_cfg() def esp_trigger(): """Sagt dem ESP „Stromzufuhr aktivieren" (schaltet den MOSFET). Nie fatal. Wird bei jedem erkannten Recover-Fenster aufgerufen. Ohne konfigurierte esp_url passiert nichts (AutoRecover läuft normal weiter). """ if not (requests and _ESP_URL): return try: params = {"key": _ESP_KEY} if _ESP_KEY else None r = requests.get(_ESP_URL + _ESP_PATH, params=params, timeout=_ESP_TIMEOUT) print(f" → ESP getriggert ({_ESP_URL}{_ESP_PATH}): HTTP {r.status_code}") dx_log("info", f"ESP-Stromzufuhr aktiviert (HTTP {r.status_code}).") except Exception as e: print(f" → ESP-Trigger fehlgeschlagen: {e}") dx_log("warn", f"ESP-Trigger fehlgeschlagen: {e}") def dx_log(level, message): """Meldet ein Ereignis an Directus (agent_logs). Fehler nie fatal.""" if not (requests and _DX_URL and _DX_TOKEN): return try: requests.post( _DX_URL.rstrip("/") + "/items/agent_logs", headers={"Authorization": f"Bearer {_DX_TOKEN}"}, json={"machine": _DX_MID, "level": level, "source": "recover", "message": str(message)[:2000]}, timeout=15, ) except Exception: pass def fenster_text(hwnd): """Liest die Beschriftungen der Kind-Fenster (die eigentliche Fehlermeldung).""" teile = [] def _cb(h, _): try: t = win32gui.GetWindowText(h) if t and t.strip(): teile.append(t.strip()) except Exception: pass return True try: win32gui.EnumChildWindows(hwnd, _cb, None) except Exception: pass return " / ".join(teile)[:500] def click_recover_button(hwnd): """ Holt die Fensterposition und klickt auf den Recover-Button. Der Button ist zentriert am unteren Rand des Fensters. """ # Fenster in den Vordergrund holen win32gui.SetForegroundWindow(hwnd) time.sleep(0.3) # Fensterposition und -größe ermitteln left, top, right, bottom = win32gui.GetWindowRect(hwnd) width = right - left height = bottom - top # Recover-Button liegt ca. bei: horizontal zentriert, ~85% der Höhe click_x = left + width // 2 click_y = top + int(height * 0.85) # Mausklick simulieren ctypes.windll.user32.SetCursorPos(click_x, click_y) time.sleep(0.1) ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0) # linke Taste runter ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0) # linke Taste hoch print(f" → Klick auf Position ({click_x}, {click_y})") def main(): versuche = 0 print("AutoRecover läuft... (Strg+C zum Beenden)") print(f"Maximal {MAX_VERSUCHE} Versuche.\n") while versuche < MAX_VERSUCHE: hwnd = win32gui.FindWindow(None, "Error") if hwnd and win32gui.IsWindowVisible(hwnd): versuche += 1 meldung = fenster_text(hwnd) or "Error" print(f"[Versuch {versuche}/{MAX_VERSUCHE}] Fehlerfenster gefunden!") dx_log("error", f"Plotter-Fehlerfenster erkannt (Versuch {versuche}/{MAX_VERSUCHE}): {meldung}") # ESP zuerst: MOSFET/„Stromzufuhr aktivieren", dann Recover klicken. esp_trigger() time.sleep(0.5) click_recover_button(hwnd) if versuche >= MAX_VERSUCHE: print(f"\nMaximale Versuche ({MAX_VERSUCHE}) erreicht. Programm beendet.") dx_log("error", f"Maximale Recover-Versuche ({MAX_VERSUCHE}) erreicht – manueller Eingriff nötig.") break time.sleep(3) else: time.sleep(PRUEF_INTERVAL) if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\nProgramm manuell beendet.")