#!/usr/bin/env python3.11 """ Hardware Integration - VOLLSTÄNDIGE Backend-Steuerung für Drucker/Steckdosen ============================================================================ NEUE PHILOSOPHIE - BACKEND DIKTIERT FRONTEND: - Drucker werden AUSSCHLIESSLICH über ihre Tapo-Steckdosen gesteuert - KEIN JavaScript für Hardware-Steuerung - nur Flask/Jinja - Backend sammelt ALLE Daten und übergibt sie komplett an Templates - Frontend ist PASSIV und zeigt nur an, was Backend vorgibt Autor: Till Tomczak - Mercedes-Benz TBA Marienfelde Datum: 2025-06-19 (Komplett-Neuschreibung für Backend-Kontrolle) """ import time import socket import threading from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Tuple from concurrent.futures import ThreadPoolExecutor, as_completed # Hardware-Bibliotheken try: from PyP100.PyP100 import P100 as PyP100 from PyP100.PyP110 import P110 as PyP110 TAPO_AVAILABLE = True except ImportError: PyP100 = None PyP110 = None TAPO_AVAILABLE = False # MYP Models & Utils from models import get_db_session, Printer, PlugStatusLog from utils.logging_config import get_logger import os # Logger hardware_logger = get_logger("hardware_integration") # ===== DRUCKER-STEUERUNGS-KLASSE ===== class DruckerSteuerung: """ VOLLSTÄNDIGE Backend-Steuerung aller Drucker über Tapo-Steckdosen. Diese Klasse übernimmt die KOMPLETTE Kontrolle: - Status-Sammlung für alle Drucker - Ein/Aus-Schaltung über Steckdosen - Energiemonitoring - Template-Daten-Vorbereitung Das Frontend erhält fertige Daten und muss NICHTS selbst berechnen! """ def __init__(self): """Initialisiere die Drucker-Steuerung""" # Tapo-Zugangsdaten für Mercedes-Benz self.tapo_username = "till.tomczak@mercedes-benz.com" self.tapo_password = "744563017196A" self.timeout = 10 self.retry_count = 3 # Backend-State-Management self.drucker_stati = {} # Aktueller Status aller Drucker self.energie_daten = {} # Energie-Monitoring-Daten self.letztes_update = {} # Letzte Aktualisierung pro Drucker self.verbindung_cache = {} # Connection-Pool für Performance hardware_logger.info("🎯 DruckerSteuerung initialisiert - BACKEND ÜBERNIMMT KONTROLLE") if not TAPO_AVAILABLE: hardware_logger.warning("⚠️ PyP100 nicht verfügbar - Simulation-Modus aktiv") # ===== KERN-STEUERUNGS-FUNKTIONEN ===== def drucker_einschalten(self, drucker_id: int, grund: str = "Manuell") -> Dict[str, Any]: """ Schaltet einen Drucker über seine Tapo-Steckdose EIN. Args: drucker_id: ID des Druckers grund: Grund für das Einschalten (für Logging) Returns: Dict mit Ergebnis und neuen Template-Daten """ hardware_logger.info(f"🟢 Drucker {drucker_id} wird eingeschaltet - Grund: {grund}") try: with get_db_session() as session: drucker = session.query(Printer).filter(Printer.id == drucker_id).first() if not drucker: return { 'success': False, 'error': f'Drucker {drucker_id} nicht gefunden', 'template_data': {} } if not drucker.plug_ip: return { 'success': False, 'error': f'Keine Steckdosen-IP für {drucker.name} konfiguriert', 'template_data': {} } # Steckdose einschalten erfolg = self._steckdose_schalten(drucker.plug_ip, True) if erfolg: # Drucker-Status in DB aktualisieren drucker.status = 'online' drucker.last_checked = datetime.now() session.commit() # Status-Log erstellen self._status_log_erstellen(drucker_id, 'turned_on', grund) # Backend-State aktualisieren self._drucker_state_aktualisieren(drucker) hardware_logger.info(f"✅ Drucker {drucker.name} erfolgreich eingeschaltet") return { 'success': True, 'message': f'Drucker {drucker.name} eingeschaltet', 'template_data': self._template_daten_sammeln() } else: hardware_logger.error(f"❌ Drucker {drucker.name} konnte nicht eingeschaltet werden") return { 'success': False, 'error': f'Steckdose {drucker.plug_ip} nicht erreichbar', 'template_data': self._template_daten_sammeln() } except Exception as e: hardware_logger.error(f"❌ Fehler beim Einschalten von Drucker {drucker_id}: {e}") return { 'success': False, 'error': f'Technischer Fehler: {str(e)}', 'template_data': {} } def drucker_ausschalten(self, drucker_id: int, grund: str = "Manuell") -> Dict[str, Any]: """ Schaltet einen Drucker über seine Tapo-Steckdose AUS. Args: drucker_id: ID des Druckers grund: Grund für das Ausschalten (für Logging) Returns: Dict mit Ergebnis und neuen Template-Daten """ hardware_logger.info(f"🔴 Drucker {drucker_id} wird ausgeschaltet - Grund: {grund}") try: with get_db_session() as session: drucker = session.query(Printer).filter(Printer.id == drucker_id).first() if not drucker: return { 'success': False, 'error': f'Drucker {drucker_id} nicht gefunden', 'template_data': {} } if not drucker.plug_ip: return { 'success': False, 'error': f'Keine Steckdosen-IP für {drucker.name} konfiguriert', 'template_data': {} } # Steckdose ausschalten erfolg = self._steckdose_schalten(drucker.plug_ip, False) if erfolg: # Drucker-Status in DB aktualisieren drucker.status = 'offline' drucker.last_checked = datetime.now() session.commit() # Status-Log erstellen self._status_log_erstellen(drucker_id, 'turned_off', grund) # Backend-State aktualisieren self._drucker_state_aktualisieren(drucker) hardware_logger.info(f"✅ Drucker {drucker.name} erfolgreich ausgeschaltet") return { 'success': True, 'message': f'Drucker {drucker.name} ausgeschaltet', 'template_data': self._template_daten_sammeln() } else: hardware_logger.error(f"❌ Drucker {drucker.name} konnte nicht ausgeschaltet werden") return { 'success': False, 'error': f'Steckdose {drucker.plug_ip} nicht erreichbar', 'template_data': self._template_daten_sammeln() } except Exception as e: hardware_logger.error(f"❌ Fehler beim Ausschalten von Drucker {drucker_id}: {e}") return { 'success': False, 'error': f'Technischer Fehler: {str(e)}', 'template_data': {} } def drucker_toggle(self, drucker_id: int, grund: str = "Toggle") -> Dict[str, Any]: """ Wechselt den Status eines Druckers (Ein <-> Aus). Args: drucker_id: ID des Druckers grund: Grund für den Wechsel Returns: Dict mit Ergebnis und neuen Template-Daten """ try: with get_db_session() as session: drucker = session.query(Printer).filter(Printer.id == drucker_id).first() if not drucker: return { 'success': False, 'error': f'Drucker {drucker_id} nicht gefunden', 'template_data': {} } # Aktueller Status bestimmen if drucker.status == 'online': return self.drucker_ausschalten(drucker_id, f"{grund} (war online)") else: return self.drucker_einschalten(drucker_id, f"{grund} (war offline)") except Exception as e: hardware_logger.error(f"❌ Fehler beim Toggle von Drucker {drucker_id}: {e}") return { 'success': False, 'error': f'Technischer Fehler: {str(e)}', 'template_data': {} } # ===== DATEN-SAMMLUNG FÜR TEMPLATES ===== def template_daten_sammeln(self) -> Dict[str, Any]: """ Sammelt ALLE Daten für die Frontend-Templates. Das Frontend muss NICHTS selbst berechnen! Returns: Dict mit allen Template-Daten für Jinja """ hardware_logger.debug("📊 Sammle Template-Daten für Frontend") try: with get_db_session() as session: drucker_liste = session.query(Printer).order_by(Printer.name).all() # Drucker-Daten mit Status sammeln drucker_daten = [] gesamt_online = 0 gesamt_offline = 0 for drucker in drucker_liste: # Status aktualisieren falls nötig aktueller_status = self._drucker_status_pruefen(drucker) drucker_info = { 'id': drucker.id, 'name': drucker.name, 'model': drucker.model or 'Unbekannt', 'location': drucker.location or 'TBA Marienfelde', 'status': aktueller_status, 'plug_ip': drucker.plug_ip, 'ip_address': drucker.ip_address, 'active': drucker.active, 'last_checked': drucker.last_checked, 'created_at': drucker.created_at, # UI-Hilfsdaten 'status_class': 'success' if aktueller_status == 'online' else 'danger', 'status_text': 'Online' if aktueller_status == 'online' else 'Offline', 'status_icon': '🟢' if aktueller_status == 'online' else '🔴', 'kann_gesteuert_werden': bool(drucker.plug_ip), 'toggle_text': 'Ausschalten' if aktueller_status == 'online' else 'Einschalten', 'toggle_action': 'off' if aktueller_status == 'online' else 'on', # Energie-Daten (Mock für Demo) 'current_power': 125.5 if aktueller_status == 'online' else 0.0, 'daily_consumption': 2.4 if aktueller_status == 'online' else 0.0, 'monthly_consumption': 45.8, } drucker_daten.append(drucker_info) if aktueller_status == 'online': gesamt_online += 1 else: gesamt_offline += 1 # System-Statistiken statistiken = { 'gesamt_drucker': len(drucker_liste), 'online_drucker': gesamt_online, 'offline_drucker': gesamt_offline, 'verfügbarkeits_rate': round((gesamt_online / len(drucker_liste) * 100) if drucker_liste else 0, 1), 'letztes_update': datetime.now(), # Energie-Gesamtdaten (Mock) 'gesamt_verbrauch': round(sum(d['daily_consumption'] for d in drucker_daten), 2), 'aktuelle_leistung': round(sum(d['current_power'] for d in drucker_daten), 1), 'geschätzte_kosten': round(sum(d['daily_consumption'] for d in drucker_daten) * 0.30, 2) } return { 'drucker': drucker_daten, 'stats': statistiken, 'system_status': 'healthy' if gesamt_online > 0 else 'warning', 'timestamp': datetime.now().isoformat(), 'tapo_verfügbar': TAPO_AVAILABLE } except Exception as e: hardware_logger.error(f"❌ Fehler beim Sammeln der Template-Daten: {e}") return { 'drucker': [], 'stats': { 'gesamt_drucker': 0, 'online_drucker': 0, 'offline_drucker': 0, 'verfügbarkeits_rate': 0.0 }, 'system_status': 'error', 'error': str(e), 'timestamp': datetime.now().isoformat(), 'tapo_verfügbar': TAPO_AVAILABLE } # ===== PRIVATE HILFSFUNKTIONEN ===== def _steckdose_schalten(self, ip: str, einschalten: bool) -> bool: """Schaltet eine Tapo-Steckdose ein oder aus""" if not TAPO_AVAILABLE: hardware_logger.warning(f"⚠️ Simulation: Steckdose {ip} würde {'eingeschaltet' if einschalten else 'ausgeschaltet'}") return True # Simulation immer erfolgreich # Zuerst Netzwerk-Erreichbarkeit prüfen if not self._erweiterte_netzwerk_prüfung(ip): hardware_logger.error(f"❌ Steckdose {ip} ist im Netzwerk nicht erreichbar") return False retry_count = 0 max_retries = 3 while retry_count < max_retries: try: action = "einschalten" if einschalten else "ausschalten" hardware_logger.debug(f"🔌 Versuche Steckdose {ip} zu {action} (Versuch {retry_count + 1}/{max_retries})") # P100-Verbindung herstellen mit Timeout p100 = PyP100(ip, self.tapo_username, self.tapo_password) p100.handshake() p100.login() # Schalten if einschalten: p100.turnOn() else: p100.turnOff() hardware_logger.info(f"✅ Steckdose {ip} erfolgreich {action}") return True except Exception as e: retry_count += 1 error_msg = str(e) # Spezifische Fehlerbehandlung if "Connection refused" in error_msg: hardware_logger.error(f"❌ Verbindung zu {ip} verweigert - Steckdose antwortet nicht auf Port 80") elif "timeout" in error_msg.lower(): hardware_logger.error(f"❌ Zeitüberschreitung bei Verbindung zu {ip}") elif "handshake" in error_msg.lower(): hardware_logger.error(f"❌ Tapo-Handshake fehlgeschlagen für {ip} - Möglicherweise falsche Credentials") elif "login" in error_msg.lower(): hardware_logger.error(f"❌ Tapo-Login fehlgeschlagen für {ip} - Benutzername/Passwort prüfen") else: hardware_logger.error(f"❌ Fehler beim Schalten der Steckdose {ip}: {e}") if retry_count < max_retries: hardware_logger.info(f"🔄 Warte 2 Sekunden vor erneutem Versuch...") time.sleep(2) hardware_logger.error(f"❌ Alle {max_retries} Versuche für Steckdose {ip} fehlgeschlagen") return False def _drucker_status_pruefen(self, drucker: Printer) -> str: """Prüft den aktuellen Status eines Druckers""" if not drucker.plug_ip: return 'unknown' # Ping-Test zur Steckdose if self._ping_test(drucker.plug_ip): return 'online' else: return 'offline' def _ping_test(self, ip: str, timeout: int = 3) -> bool: """Einfacher Ping-Test zu einer IP""" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) result = sock.connect_ex((ip, 80)) sock.close() return result == 0 except: return False def _erweiterte_netzwerk_prüfung(self, ip: str) -> bool: """ Erweiterte Netzwerk-Prüfung mit mehreren Tests. Args: ip: IP-Adresse zum Prüfen Returns: bool: True wenn erreichbar """ hardware_logger.debug(f"🔍 Erweiterte Netzwerk-Prüfung für {ip}") # Test 1: Port 80 (HTTP) if self._ping_test(ip, timeout=2): hardware_logger.debug(f"✅ {ip} auf Port 80 erreichbar") return True # Test 2: Port 9999 (Tapo-spezifisch für manche Modelle) try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2) result = sock.connect_ex((ip, 9999)) sock.close() if result == 0: hardware_logger.debug(f"✅ {ip} auf Port 9999 erreichbar") return True except: pass # Test 3: ICMP Ping (falls verfügbar) try: from utils.core_system import safe_subprocess_run # Windows und Linux kompatibel param = '-n' if os.name == 'nt' else '-c' command = ['ping', param, '1', '-w', '2000', ip] result = safe_subprocess_run(command, capture_output=True, timeout=3) if result and result.returncode == 0: hardware_logger.debug(f"✅ {ip} via ICMP Ping erreichbar") return True except Exception as e: hardware_logger.debug(f"ICMP Ping Test fehlgeschlagen: {e}") pass hardware_logger.warning(f"⚠️ {ip} ist über keine Methode erreichbar") return False def _status_log_erstellen(self, drucker_id: int, action: str, grund: str): """Erstellt einen Eintrag im Status-Log""" try: with get_db_session() as session: log_entry = PlugStatusLog( printer_id=drucker_id, timestamp=datetime.now(), action=action, reason=grund, success=True ) session.add(log_entry) session.commit() except Exception as e: hardware_logger.warning(f"⚠️ Status-Log konnte nicht erstellt werden: {e}") def _drucker_state_aktualisieren(self, drucker: Printer): """Aktualisiert den internen Backend-State""" self.drucker_stati[drucker.id] = { 'name': drucker.name, 'status': drucker.status, 'last_update': datetime.now(), 'plug_ip': drucker.plug_ip } def _template_daten_sammeln(self) -> Dict[str, Any]: """Wrapper für template_daten_sammeln (Backward-Compatibility)""" return self.template_daten_sammeln() def check_outlet_status(self, ip: str, printer_id: Optional[int] = None) -> tuple: """ Prüft den Status einer Tapo-Steckdose. Args: ip: IP-Adresse der Steckdose printer_id: Optional - ID des Druckers (für Legacy-Kompatibilität) Returns: tuple: (reachable: bool, status: str) - Legacy-Format für Kompatibilität """ hardware_logger.debug(f"🔍 Prüfe Steckdosen-Status: {ip}" + (f" (Drucker ID: {printer_id})" if printer_id else "")) if not TAPO_AVAILABLE: # Legacy-Format: (reachable, status) return (True, 'online') # Zuerst Netzwerk-Erreichbarkeit prüfen if not self._erweiterte_netzwerk_prüfung(ip): hardware_logger.warning(f"⚠️ Steckdose {ip} ist im Netzwerk nicht erreichbar") return (False, 'unreachable') retry_count = 0 max_retries = 2 while retry_count < max_retries: try: # Tapo P100/P110 Verbindung p100 = PyP100(ip, self.tapo_username, self.tapo_password) p100.handshake() p100.login() # Device Info abrufen device_info = p100.getDeviceInfo() if device_info and 'error_code' in device_info: if device_info['error_code'] == 0: device_on = device_info.get('result', {}).get('device_on', False) hardware_logger.debug(f"✅ Steckdose {ip}: {'EIN' if device_on else 'AUS'}") # Legacy-Format: (reachable, status) return (True, 'on' if device_on else 'off') else: hardware_logger.warning(f"⚠️ Steckdose {ip} Error Code: {device_info['error_code']}") return (False, 'error') else: hardware_logger.error(f"❌ Steckdose {ip}: Keine gültige Antwort") return (False, 'unreachable') except Exception as e: retry_count += 1 error_msg = str(e) if "Connection refused" in error_msg: hardware_logger.error(f"❌ Verbindung zu {ip} verweigert") elif "timeout" in error_msg.lower(): hardware_logger.error(f"❌ Zeitüberschreitung bei {ip}") elif "handshake" in error_msg.lower(): hardware_logger.error(f"❌ Handshake-Fehler bei {ip}") else: hardware_logger.error(f"❌ Fehler beim Prüfen von Steckdose {ip}: {e}") if retry_count < max_retries: time.sleep(1) return (False, 'unreachable') def ping_address(self, ip: str, timeout: int = 5) -> bool: """ Prüft die Netzwerk-Erreichbarkeit einer IP-Adresse. Args: ip: IP-Adresse zum Testen timeout: Timeout in Sekunden Returns: bool: True wenn erreichbar, False sonst """ hardware_logger.debug(f"📡 Teste Netzwerk-Erreichbarkeit: {ip}") try: # Socket-basierter Ping-Test auf Port 80 (HTTP) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) result = sock.connect_ex((ip, 80)) sock.close() is_reachable = (result == 0) hardware_logger.debug(f"📡 {ip}: {'✅ erreichbar' if is_reachable else '❌ nicht erreichbar'}") return is_reachable except Exception as e: hardware_logger.debug(f"❌ Ping-Test für {ip} fehlgeschlagen: {e}") return False def get_energy_statistics(self) -> Dict[str, Any]: """ Sammelt Energiestatistiken für alle Drucker für das Energiedashboard. Returns: Dict: Umfassende Energiestatistiken mit allen erforderlichen Daten """ hardware_logger.debug("📊 Sammle Energiestatistiken für alle Drucker") try: with get_db_session() as session: drucker_liste = session.query(Printer).all() device_data = [] total_current_power = 0.0 total_today_energy = 0.0 total_month_energy = 0.0 online_count = 0 for drucker in drucker_liste: # Aktuellen Status prüfen is_online = False current_power = 0.0 if drucker.plug_ip: # Status über Steckdose prüfen reachable, status = self.check_outlet_status(drucker.plug_ip, drucker.id) is_online = reachable and status == 'on' if is_online: # Mock-Energiedaten für online Drucker current_power = 125.5 # Simulierter Verbrauch online_count += 1 total_current_power += current_power total_today_energy += 2.4 total_month_energy += 45.8 device_info = { 'id': drucker.id, 'name': drucker.name, 'model': drucker.model or 'Unbekannt', 'location': drucker.location or 'TBA Marienfelde', 'online': is_online, 'current_power': current_power, 'today_energy': 2.4 if is_online else 0.0, 'month_energy': 45.8, 'past24h': 2.4 if is_online else 0.0, 'past30d': 45.8, 'past1y': 547.6, 'voltage': 230.0 if is_online else 0.0, 'current': current_power / 230.0 if current_power > 0 else 0.0, 'ip': drucker.plug_ip } device_data.append(device_info) # Berechne Durchschnittswerte avg_current_power = total_current_power / len(drucker_liste) if drucker_liste else 0.0 avg_today_energy = total_today_energy / len(drucker_liste) if drucker_liste else 0.0 avg_month_energy = total_month_energy / len(drucker_liste) if drucker_liste else 0.0 # Mock-Zeitreihen-Daten für Charts hourly_consumption = [round(125.5 + (i % 3) * 25.2, 1) for i in range(24)] daily_consumption = [round(2.4 + (i % 5) * 0.5, 1) for i in range(30)] monthly_consumption = [round(45.8 + (i % 3) * 15.2, 1) for i in range(12)] statistiken = { 'total_devices': len(drucker_liste), 'online_devices': online_count, 'offline_devices': len(drucker_liste) - online_count, 'total_current_power': round(total_current_power, 2), 'avg_current_power': round(avg_current_power, 2), 'total_today_energy': round(total_today_energy, 1), 'total_month_energy': round(total_month_energy, 1), 'avg_today_energy': round(avg_today_energy, 1), 'avg_month_energy': round(avg_month_energy, 1), 'hourly_consumption': hourly_consumption, 'daily_consumption': daily_consumption, 'monthly_consumption': monthly_consumption, 'devices': device_data, 'timestamp': datetime.now().isoformat() } hardware_logger.info(f"✅ Energiestatistiken erstellt: {online_count}/{len(drucker_liste)} Drucker online") return statistiken except Exception as e: hardware_logger.error(f"❌ Fehler beim Sammeln der Energiestatistiken: {e}") return { 'total_devices': 0, 'online_devices': 0, 'offline_devices': 0, 'total_current_power': 0.0, 'avg_current_power': 0.0, 'total_today_energy': 0.0, 'total_month_energy': 0.0, 'avg_today_energy': 0.0, 'avg_month_energy': 0.0, 'hourly_consumption': [0.0] * 24, 'daily_consumption': [0.0] * 30, 'monthly_consumption': [0.0] * 12, 'devices': [], 'timestamp': datetime.now().isoformat(), 'error': str(e) } def turn_off(self, ip: str, username: str = None, password: str = None, printer_id: int = None) -> bool: """ Schaltet eine Tapo-Steckdose aus. Args: ip: IP-Adresse der Steckdose username: Benutzername (wird ignoriert, verwendet interne Credentials) password: Passwort (wird ignoriert, verwendet interne Credentials) printer_id: Optional - ID des Druckers für Logging Returns: bool: True wenn erfolgreich ausgeschaltet """ hardware_logger.debug(f"🔴 Schalte Steckdose aus: {ip}" + (f" (Drucker ID: {printer_id})" if printer_id else "")) if not TAPO_AVAILABLE: hardware_logger.info(f"🔄 SIMULATION: Steckdose {ip} ausgeschaltet") return True try: # P100-Verbindung mit internen Credentials p100 = PyP100(ip, self.tapo_username, self.tapo_password) p100.handshake() p100.login() # Steckdose ausschalten p100.turnOff() hardware_logger.info(f"✅ Steckdose {ip} erfolgreich ausgeschaltet") # Status-Log erstellen falls Drucker-ID verfügbar if printer_id: try: self._status_log_erstellen(printer_id, 'turned_off', 'Startup-Initialisierung') except: pass # Fehler beim Logging nicht kritisch return True except Exception as e: hardware_logger.error(f"❌ Fehler beim Ausschalten der Steckdose {ip}: {e}") return False def turn_on(self, ip: str, username: str = None, password: str = None, printer_id: int = None) -> bool: """ Schaltet eine Tapo-Steckdose ein. Args: ip: IP-Adresse der Steckdose username: Benutzername (wird ignoriert, verwendet interne Credentials) password: Passwort (wird ignoriert, verwendet interne Credentials) printer_id: Optional - ID des Druckers für Logging Returns: bool: True wenn erfolgreich eingeschaltet """ hardware_logger.debug(f"🟢 Schalte Steckdose ein: {ip}" + (f" (Drucker ID: {printer_id})" if printer_id else "")) if not TAPO_AVAILABLE: hardware_logger.info(f"🔄 SIMULATION: Steckdose {ip} eingeschaltet") return True try: # P100-Verbindung mit internen Credentials p100 = PyP100(ip, self.tapo_username, self.tapo_password) p100.handshake() p100.login() # Steckdose einschalten p100.turnOn() hardware_logger.info(f"✅ Steckdose {ip} erfolgreich eingeschaltet") # Status-Log erstellen falls Drucker-ID verfügbar if printer_id: try: self._status_log_erstellen(printer_id, 'turned_on', 'Manuell') except: pass # Fehler beim Logging nicht kritisch return True except Exception as e: hardware_logger.error(f"❌ Fehler beim Einschalten der Steckdose {ip}: {e}") return False # ===== GLOBALE INSTANZ ===== # Singleton-Pattern für globale Drucker-Steuerung _drucker_steuerung_instanz = None def get_drucker_steuerung() -> DruckerSteuerung: """ Gibt die globale DruckerSteuerung-Instanz zurück (Singleton). Returns: DruckerSteuerung: Die globale Instanz """ global _drucker_steuerung_instanz if _drucker_steuerung_instanz is None: _drucker_steuerung_instanz = DruckerSteuerung() return _drucker_steuerung_instanz # ===== LEGACY-KOMPATIBILITÄT ===== # Backward-Compatibility für bestehenden Code def get_tapo_controller(): """Legacy-Funktion für Rückwärtskompatibilität""" return get_drucker_steuerung() def toggle_plug(ip: str, state: bool) -> bool: """Legacy-Funktion für direktes Steckdosen-Schalten""" controller = get_drucker_steuerung() return controller._steckdose_schalten(ip, state) def get_printer_monitor(): """Legacy-Funktion für Drucker-Monitoring""" return get_drucker_steuerung() # Export für andere Module __all__ = [ 'DruckerSteuerung', 'get_drucker_steuerung', 'get_tapo_controller', # Legacy 'toggle_plug', # Legacy 'get_printer_monitor' # Legacy ] hardware_logger.info("🚀 Hardware Integration (Backend-Kontrolle) erfolgreich geladen")