diff --git a/backend/app/blueprints/printers.py b/backend/app/blueprints/printers.py index f273acc5..15f4cb1e 100644 --- a/backend/app/blueprints/printers.py +++ b/backend/app/blueprints/printers.py @@ -74,7 +74,7 @@ def get_live_printer_status(): @printers_blueprint.route("/control//power", methods=["POST"]) @login_required -@require_permission(Permission.MANAGE_PRINTERS) +@require_permission(Permission.CONTROL_PRINTER) # Verwende die bereits vorhandene Berechtigung @measure_execution_time(logger=printers_logger, task_name="API-Drucker-Stromversorgung-Steuerung") def control_printer_power(printer_id): """ diff --git a/backend/app/test_tapo_sofort.py b/backend/app/test_tapo_sofort.py index 92dc7ebc..5a3ae7bf 100644 --- a/backend/app/test_tapo_sofort.py +++ b/backend/app/test_tapo_sofort.py @@ -2,206 +2,183 @@ # -*- coding: utf-8 -*- """ -SOFORT-TEST für TP-Link Tapo P110-Steckdosen -Testet die Verbindung mit echten Anmeldedaten und findet alle verfügbaren Steckdosen. +Sofort-Test für TP-Link Tapo P110-Steckdosen +Testet direkt die Verbindung zu allen konfigurierten Steckdosen """ import sys -import time +import os import subprocess +import time from datetime import datetime -# Tapo-Anmeldedaten (HARDKODIERT) +# Anmeldedaten für Tapo-Steckdosen TAPO_USERNAME = "till.tomczak@mercedes-benz.com" TAPO_PASSWORD = "744563017196" -# IP-Bereiche zum Testen -IP_RANGES_TO_SCAN = [ - "192.168.1.{}", - "192.168.0.{}", - "10.0.0.{}", - "172.16.0.{}" +# Standard-IPs für Tapo-Steckdosen +# (falls nicht verfügbar, passen Sie diese an die tatsächlichen IPs in Ihrem Netzwerk an) +TAPO_IPS = [ + "192.168.1.100", + "192.168.1.101", + "192.168.1.102", + "192.168.1.103", + "192.168.1.104", + "192.168.1.105", + "192.168.0.100", + "192.168.0.101", + "192.168.0.102", + "192.168.0.103", + "192.168.0.104", + "192.168.0.105", + "192.168.1.200", + "192.168.1.201", + "192.168.0.200", + "192.168.0.201", ] def log_message(message, level="INFO"): """Logge eine Nachricht mit Zeitstempel""" - timestamp = datetime.now().strftime("%H:%M:%S") + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] [{level}] {message}") -def test_ping(ip_address): - """Teste ob eine IP erreichbar ist""" +def ping_address(ip_address, timeout=2): + """Führt einen Ping-Test durch""" try: + # Platform-spezifische Ping-Befehle + import platform + if platform.system().lower() == "windows": # Windows + cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address] + else: # Unix/Linux/macOS + cmd = ['ping', '-c', '1', '-W', str(timeout), ip_address] + result = subprocess.run( - ['ping', '-n', '1', '-w', '1000', ip_address], - capture_output=True, - text=True, - timeout=2 + cmd, + capture_output=True, + text=True, + timeout=timeout + 1 ) + return result.returncode == 0 - except: + except Exception: return False -def test_tapo_connection(ip_address): - """Teste Tapo-Verbindung zu einer IP""" - try: - from PyP100 import PyP110 - - log_message(f"Teste Tapo-Verbindung zu {ip_address}...") - p110 = PyP110.P110(ip_address, TAPO_USERNAME, TAPO_PASSWORD) - p110.handshake() # Authentifizierung - p110.login() # Login - - # Geräteinformationen abrufen - device_info = p110.getDeviceInfo() - - log_message(f"✅ ERFOLG: {ip_address} ist eine Tapo-Steckdose!") - log_message(f" 📛 Name: {device_info.get('nickname', 'Unbekannt')}") - log_message(f" ⚡ Status: {'EIN' if device_info.get('device_on', False) else 'AUS'}") - - if 'on_time' in device_info: - on_time = device_info.get('on_time', 0) - hours, minutes = divmod(on_time // 60, 60) - log_message(f" ⏱️ Betriebszeit: {hours}h {minutes}m") - - return True, device_info - - except Exception as e: - error_msg = str(e).lower() - if "login" in error_msg or "credentials" in error_msg: - log_message(f"❌ {ip_address}: Falsche Anmeldedaten") - elif "timeout" in error_msg: - log_message(f"❌ {ip_address}: Timeout") - elif "connect" in error_msg or "unreachable" in error_msg: - log_message(f"❌ {ip_address}: Nicht erreichbar") - else: - log_message(f"❌ {ip_address}: {str(e)}") - return False, None - -def main(): - """Hauptfunktion: Scanne Netzwerk nach Tapo-Steckdosen""" - log_message("🚀 SOFORT-TEST: TP-Link Tapo P110-Steckdosen") - log_message(f"👤 Benutzername: {TAPO_USERNAME}") - log_message(f"🔐 Passwort: {'*' * len(TAPO_PASSWORD)}") - print() +def test_tapo_connection(): + """ + Testet die Verbindung zu TP-Link Tapo P110-Steckdosen. + """ + log_message("🔄 Überprüfe ob PyP100-Modul installiert ist...") - # PyP100 Import testen try: from PyP100 import PyP110 log_message("✅ PyP100-Modul erfolgreich importiert") except ImportError: - log_message("❌ FEHLER: PyP100-Modul nicht gefunden!", "ERROR") - log_message(" Installiere mit: pip install PyP100", "ERROR") - return - - print() - log_message("🔍 Scanne Netzwerk nach erreichbaren Geräten...") - - # Erst bekannte IP-Adressen testen - known_ips = [ - "192.168.1.100", "192.168.1.101", "192.168.1.102", - "192.168.1.103", "192.168.1.104", "192.168.1.105", - "192.168.0.100", "192.168.0.101", "192.168.0.102" - ] - - found_steckdosen = [] - - for ip in known_ips: - if test_ping(ip): - log_message(f"🌐 {ip} ist erreichbar - teste Tapo...") - success, device_info = test_tapo_connection(ip) - if success: - found_steckdosen.append((ip, device_info)) - else: - log_message(f"⚫ {ip} nicht erreichbar") - - # Wenn keine gefunden, erweiterte Suche - if not found_steckdosen: - log_message("🔍 Keine Steckdosen in Standard-IPs gefunden - erweitere Suche...") - - for ip_range in IP_RANGES_TO_SCAN: - log_message(f"Scanne Bereich {ip_range.format('1-20')}...") - - for i in range(1, 21): # Erste 20 IPs pro Bereich - ip = ip_range.format(i) - if test_ping(ip): - log_message(f"🌐 {ip} erreichbar - teste Tapo...") - success, device_info = test_tapo_connection(ip) - if success: - found_steckdosen.append((ip, device_info)) - - # Ergebnisse - print() - log_message("📊 ERGEBNISSE:") - - if found_steckdosen: - log_message(f"🎉 {len(found_steckdosen)} Tapo-Steckdose(n) gefunden!") - print() - - for i, (ip, device_info) in enumerate(found_steckdosen, 1): - log_message(f"Steckdose #{i}:") - log_message(f" IP: {ip}") - log_message(f" Name: {device_info.get('nickname', 'Unbekannt')}") - log_message(f" Status: {'EIN' if device_info.get('device_on', False) else 'AUS'}") - log_message(f" Modell: {device_info.get('model', 'Unbekannt')}") - print() - - # Steckdosen-Test durchführen - log_message("🧪 Teste Ein/Aus-Schaltung der ersten Steckdose...") - test_ip, _ = found_steckdosen[0] + log_message("❌ PyP100-Modul nicht installiert", "ERROR") + log_message(" Installiere jetzt mit: pip install PyP100==0.1.2") try: + subprocess.run([sys.executable, "-m", "pip", "install", "PyP100==0.1.2"], check=True) + log_message("✅ PyP100-Modul erfolgreich installiert") + + # Erneut importieren from PyP100 import PyP110 - p110 = PyP110.P110(test_ip, TAPO_USERNAME, TAPO_PASSWORD) - p110.handshake() - p110.login() + except Exception as e: + log_message(f"❌ Fehler bei Installation von PyP100: {str(e)}", "ERROR") + log_message(" Bitte installieren Sie manuell: pip install PyP100==0.1.2") + return + + log_message("🔍 Starte Test für Tapo-Steckdosen...") + log_message(f"🔐 Anmeldedaten: {TAPO_USERNAME} / {TAPO_PASSWORD}") + + successful_connections = 0 + found_ips = [] + + for ip in TAPO_IPS: + log_message(f"🔄 Teste IP-Adresse: {ip}") + + # Ping-Test für Grundkonnektivität + ping_success = ping_address(ip) + if not ping_success: + log_message(f" ❌ IP {ip} nicht erreichbar (Ping fehlgeschlagen)") + continue - # Aktuelle Status abrufen + log_message(f" ✅ IP {ip} ist erreichbar (Ping erfolgreich)") + + # Tapo-Verbindung testen + try: + log_message(f" 🔄 Verbinde zu Tapo-Steckdose {ip}...") + p110 = PyP110.P110(ip, TAPO_USERNAME, TAPO_PASSWORD) + p110.handshake() # Authentifizierung + p110.login() # Login + + # Geräteinformationen abrufen device_info = p110.getDeviceInfo() - current_state = device_info.get('device_on', False) - log_message(f"Aktueller Status: {'EIN' if current_state else 'AUS'}") + # Status abrufen + is_on = device_info.get('device_on', False) + nickname = device_info.get('nickname', "Unbekannt") - # Gegenteil schalten - if current_state: - log_message("Schalte AUS...") - p110.turnOff() - time.sleep(2) - log_message("✅ Erfolgreich ausgeschaltet!") + log_message(f" ✅ Verbindung zu Tapo-Steckdose '{nickname}' ({ip}) erfolgreich") + log_message(f" 📱 Gerätename: {nickname}") + log_message(f" ⚡ Status: {'Eingeschaltet' if is_on else 'Ausgeschaltet'}") + + if 'on_time' in device_info: + on_time = device_info.get('on_time', 0) + hours, minutes = divmod(on_time // 60, 60) + log_message(f" ⏱️ Betriebszeit: {hours}h {minutes}m") + + successful_connections += 1 + found_ips.append(ip) + + # Optionales Ein-/Ausschalten zum Testen + if len(sys.argv) > 1 and sys.argv[1] == '--toggle': + if is_on: + log_message(f" 🔄 Schalte Steckdose {nickname} AUS...") + p110.turnOff() + log_message(f" ✅ Steckdose ausgeschaltet") + else: + log_message(f" 🔄 Schalte Steckdose {nickname} EIN...") + p110.turnOn() + log_message(f" ✅ Steckdose eingeschaltet") + + # Kurze Pause + time.sleep(1) - log_message("Schalte wieder EIN...") - p110.turnOn() - time.sleep(2) - log_message("✅ Erfolgreich eingeschaltet!") - else: - log_message("Schalte EIN...") - p110.turnOn() - time.sleep(2) - log_message("✅ Erfolgreich eingeschaltet!") - - log_message("Schalte wieder AUS...") - p110.turnOff() - time.sleep(2) - log_message("✅ Erfolgreich ausgeschaltet!") - - log_message("🎉 STECKDOSEN-STEUERUNG FUNKTIONIERT PERFEKT!") + # Status erneut abrufen + device_info = p110.getDeviceInfo() + is_on = device_info.get('device_on', False) + log_message(f" ⚡ Neuer Status: {'Eingeschaltet' if is_on else 'Ausgeschaltet'}") except Exception as e: - log_message(f"❌ Fehler beim Testen der Schaltung: {str(e)}", "ERROR") + log_message(f" ❌ Verbindung zu Tapo-Steckdose {ip} fehlgeschlagen: {str(e)}", "ERROR") + # Zusammenfassung + log_message("\n📊 Zusammenfassung:") + log_message(f" Getestete IPs: {len(TAPO_IPS)}") + log_message(f" Gefundene Tapo-Steckdosen: {successful_connections}") + + if successful_connections > 0: + log_message("✅ Tapo-Steckdosen erfolgreich gefunden und getestet!") + log_message(f"📝 Gefundene IPs: {found_ips}") + + # Ausgabe für Konfiguration + log_message("\n🔧 Konfiguration für settings.py:") + log_message(f""" +# TP-Link Tapo Standard-Anmeldedaten +TAPO_USERNAME = "{TAPO_USERNAME}" +TAPO_PASSWORD = "{TAPO_PASSWORD}" + +# Automatische Steckdosen-Erkennung aktivieren +TAPO_AUTO_DISCOVERY = True + +# Standard-Steckdosen-IPs +DEFAULT_TAPO_IPS = {found_ips} +""") else: - log_message("❌ KEINE Tapo-Steckdosen gefunden!", "ERROR") - log_message("Mögliche Ursachen:", "ERROR") - log_message("1. Steckdosen sind in einem anderen IP-Bereich", "ERROR") - log_message("2. Anmeldedaten sind falsch", "ERROR") - log_message("3. Netzwerkprobleme", "ERROR") - log_message("4. Steckdosen sind nicht im Netzwerk", "ERROR") + log_message("❌ Keine Tapo-Steckdosen gefunden!", "ERROR") + log_message(" Bitte überprüfen Sie die IP-Adressen und Anmeldedaten") if __name__ == "__main__": - try: - main() - except KeyboardInterrupt: - log_message("Test abgebrochen durch Benutzer", "INFO") - except Exception as e: - log_message(f"Unerwarteter Fehler: {str(e)}", "ERROR") - import traceback - traceback.print_exc() \ No newline at end of file + print("\n====== TAPO P110 STECKDOSEN-TEST ======\n") + test_tapo_connection() + print("\n====== TEST ABGESCHLOSSEN ======\n") \ No newline at end of file diff --git a/backend/app/utils/printer_monitor.py b/backend/app/utils/printer_monitor.py index bd636a02..6a34b860 100644 --- a/backend/app/utils/printer_monitor.py +++ b/backend/app/utils/printer_monitor.py @@ -17,7 +17,7 @@ import os from models import get_db_session, Printer from utils.logging_config import get_logger -from config.settings import PRINTERS, TAPO_USERNAME, TAPO_PASSWORD +from config.settings import PRINTERS, TAPO_USERNAME, TAPO_PASSWORD, DEFAULT_TAPO_IPS, TAPO_AUTO_DISCOVERY # TP-Link Tapo P110 Unterstützung hinzufügen try: @@ -42,12 +42,17 @@ class PrinterMonitor: self.monitoring_active = False self.monitor_thread = None self.startup_initialized = False + self.auto_discovered_tapo = False # Cache-Konfiguration self.session_cache_ttl = 30 # 30 Sekunden für Session-Cache self.db_cache_ttl = 300 # 5 Minuten für DB-Cache monitor_logger.info("🖨️ Drucker-Monitor initialisiert") + + # Automatische Steckdosenerkennung starten, falls aktiviert + if TAPO_AUTO_DISCOVERY: + self.auto_discover_tapo_outlets() def initialize_all_outlets_on_startup(self) -> Dict[str, bool]: """ @@ -450,6 +455,140 @@ class PrinterMonitor: summary["offline"] += 1 return summary + + def auto_discover_tapo_outlets(self) -> Dict[str, bool]: + """ + Automatische Erkennung und Konfiguration von TP-Link Tapo P110-Steckdosen im Netzwerk. + + Returns: + Dict[str, bool]: Ergebnis der Steckdosenerkennung mit IP als Schlüssel + """ + if self.auto_discovered_tapo: + monitor_logger.info("🔍 Tapo-Steckdosen wurden bereits erkannt") + return {} + + monitor_logger.info("🔍 Starte automatische Tapo-Steckdosenerkennung...") + results = {} + + # 1. Zuerst die Standard-IPs aus der Konfiguration testen + monitor_logger.info(f"🔄 Teste {len(DEFAULT_TAPO_IPS)} Standard-IPs aus der Konfiguration") + + for ip in DEFAULT_TAPO_IPS: + try: + # Ping-Test für Grundkonnektivität + ping_success = self._ping_address(ip, timeout=2) + + if ping_success: + monitor_logger.info(f"✅ Steckdose mit IP {ip} ist pingbar") + + # Tapo-Verbindung testen + if TAPO_AVAILABLE: + try: + p110 = PyP110.P110(ip, TAPO_USERNAME, TAPO_PASSWORD) + p110.handshake() + p110.login() + device_info = p110.getDeviceInfo() + + # Steckdose gefunden und verbunden + nickname = device_info.get('nickname', f"Tapo P110 ({ip})") + state = "on" if device_info.get('device_on', False) else "off" + + monitor_logger.info(f"✅ Tapo-Steckdose '{nickname}' ({ip}) gefunden - Status: {state}") + results[ip] = True + + # Steckdose in Datenbank speichern/aktualisieren + self._ensure_tapo_in_database(ip, nickname) + + except Exception as e: + monitor_logger.debug(f"❌ IP {ip} ist pingbar, aber keine Tapo-Steckdose: {str(e)}") + results[ip] = False + else: + monitor_logger.warning("⚠️ PyP100-Modul nicht verfügbar - kann Tapo-Verbindung nicht testen") + results[ip] = False + else: + monitor_logger.debug(f"❌ IP {ip} nicht erreichbar") + results[ip] = False + + except Exception as e: + monitor_logger.error(f"❌ Fehler bei Steckdosen-Erkennung für IP {ip}: {str(e)}") + results[ip] = False + + # Erfolgsstatistik berechnen + success_count = sum(1 for success in results.values() if success) + monitor_logger.info(f"✅ Steckdosen-Erkennung abgeschlossen: {success_count}/{len(results)} Steckdosen gefunden") + + # Markieren, dass automatische Erkennung durchgeführt wurde + self.auto_discovered_tapo = True + + return results + + def _ensure_tapo_in_database(self, ip_address: str, nickname: str = None) -> bool: + """ + Stellt sicher, dass eine erkannte Tapo-Steckdose in der Datenbank existiert. + + Args: + ip_address: IP-Adresse der Steckdose + nickname: Name der Steckdose (optional) + + Returns: + bool: True wenn erfolgreich in Datenbank gespeichert/aktualisiert + """ + try: + db_session = get_db_session() + + # Prüfen, ob Drucker mit dieser IP bereits existiert + existing_printer = db_session.query(Printer).filter(Printer.plug_ip == ip_address).first() + + if existing_printer: + # Drucker aktualisieren, falls nötig + if not existing_printer.plug_username or not existing_printer.plug_password: + existing_printer.plug_username = TAPO_USERNAME + existing_printer.plug_password = TAPO_PASSWORD + monitor_logger.info(f"✅ Drucker {existing_printer.name} mit Tapo-Anmeldedaten aktualisiert") + + if nickname and existing_printer.name != nickname and "Tapo P110" not in existing_printer.name: + old_name = existing_printer.name + existing_printer.name = nickname + monitor_logger.info(f"✅ Drucker {old_name} umbenannt zu {nickname}") + + # Status aktualisieren + existing_printer.last_checked = datetime.now() + db_session.commit() + db_session.close() + return True + else: + # Neuen Drucker erstellen, falls keiner existiert + printer_name = nickname or f"Tapo P110 ({ip_address})" + mac_address = f"tapo:{ip_address.replace('.', '-')}" # Pseudo-MAC-Adresse + + new_printer = Printer( + name=printer_name, + model="TP-Link Tapo P110", + location="Automatisch erkannt", + ip_address=ip_address, # Drucker-IP setzen wir gleich Steckdosen-IP + mac_address=mac_address, + plug_ip=ip_address, + plug_username=TAPO_USERNAME, + plug_password=TAPO_PASSWORD, + status="offline", + active=True, + last_checked=datetime.now() + ) + + db_session.add(new_printer) + db_session.commit() + monitor_logger.info(f"✅ Neuer Drucker '{printer_name}' mit Tapo-Steckdose {ip_address} erstellt") + db_session.close() + return True + + except Exception as e: + monitor_logger.error(f"❌ Fehler beim Speichern der Tapo-Steckdose {ip_address}: {str(e)}") + try: + db_session.rollback() + db_session.close() + except: + pass + return False # Globale Instanz printer_monitor = PrinterMonitor() \ No newline at end of file