diff --git a/backend/app/test_p110.py b/backend/app/test_p110.py new file mode 100644 index 00000000..0519ecba --- /dev/null +++ b/backend/app/test_p110.py @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/backend/app/test_tapo_direkt.py b/backend/app/test_tapo_direkt.py new file mode 100644 index 00000000..c54cda69 --- /dev/null +++ b/backend/app/test_tapo_direkt.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +DIREKT-TEST für TP-Link Tapo P110-Steckdosen +Umgeht Ping-Befehle und testet direkte TCP-Verbindung +""" + +import sys +import os +import socket +import time +from datetime import datetime + +# Anmeldedaten für Tapo-Steckdosen +TAPO_USERNAME = "till.tomczak@mercedes-benz.com" +TAPO_PASSWORD = "744563017196" + +# Standard-IPs für Tapo-Steckdosen +# (falls nicht verfügbar, passen Sie diese an die tatsächlichen IPs in Ihrem Netzwerk an) +TAPO_IPS = [ + # Typische IP-Bereiche + "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", + + # Mercedes-Benz Netzwerk spezifisch + "10.0.0.100", + "10.0.0.101", + "10.0.0.102", + "10.0.0.103", + "10.0.0.104", + "10.0.0.105", + + # Zusätzliche mögliche IPs + "192.168.178.100", + "192.168.178.101", + "192.168.178.102", + "192.168.178.103", + "192.168.178.104", + "192.168.178.105", +] + +def log_message(message, level="INFO"): + """Logge eine Nachricht mit Zeitstempel""" + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"[{timestamp}] [{level}] {message}") + +def check_tcp_connection(host, port=80, timeout=1): + """ + Prüft ob eine TCP-Verbindung zu einem Host und Port möglich ist. + Vermeidet Ping und charmap-Probleme. + + Args: + host: Hostname oder IP-Adresse + port: TCP-Port (Standard: 80) + timeout: Timeout in Sekunden + + Returns: + bool: True wenn Verbindung erfolgreich + """ + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + result = sock.connect_ex((host, port)) + sock.close() + return result == 0 + except: + return False + +def test_tapo_connection(): + """ + Testet die Verbindung zu TP-Link Tapo P110-Steckdosen. + """ + log_message("🔄 Überprüfe ob PyP100-Modul installiert ist...") + + try: + from PyP100 import PyP110 + log_message("✅ PyP100-Modul erfolgreich importiert") + except ImportError: + log_message("❌ PyP100-Modul nicht installiert", "ERROR") + log_message(" Installiere jetzt mit: pip install PyP100==0.1.2") + + try: + import subprocess + 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 + 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}") + + # TCP-Verbindungstest für Grundkonnektivität (Port 80 für HTTP) + conn_success = check_tcp_connection(ip, port=80) + + if not conn_success: + # Alternativ Port 443 testen für HTTPS + conn_success = check_tcp_connection(ip, port=443) + + if not conn_success: + log_message(f" ❌ IP {ip} nicht erreichbar (Verbindung fehlgeschlagen)") + continue + + log_message(f" ✅ IP {ip} ist erreichbar (TCP-Verbindung 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() + + # Status abrufen + is_on = device_info.get('device_on', False) + nickname = device_info.get('nickname', "Unbekannt") + + 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) + + # Steckdose testen: EIN/AUS + 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) + + # 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" ❌ 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(" Bitte überprüfen Sie die IP-Adressen und Anmeldedaten") + + # Fehlerbehebungs-Tipps + log_message("\n🔧 Fehlerbehebungs-Tipps:") + log_message("1. Stellen Sie sicher, dass die Steckdosen mit dem WLAN verbunden sind") + log_message("2. Prüfen Sie die IP-Adressen in der Tapo-App oder im Router") + log_message("3. Stellen Sie sicher, dass die Anmeldedaten korrekt sind") + log_message("4. Prüfen Sie ob die Steckdosen über die Tapo-App erreichbar sind") + log_message("5. Führen Sie einen Neustart der Steckdosen durch (aus- und wieder einstecken)") + +if __name__ == "__main__": + print("\n====== TAPO P110 DIREKT-TEST (OHNE PING) ======\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 6a34b860..677287e6 100644 --- a/backend/app/utils/printer_monitor.py +++ b/backend/app/utils/printer_monitor.py @@ -341,37 +341,42 @@ class PrinterMonitor: def _ping_address(self, ip_address: str, timeout: int = 3) -> bool: """ - Führt einen Ping-Test zu einer IP-Adresse durch. + Führt einen Konnektivitätstest zu einer IP-Adresse durch. + Verwendet TCP-Verbindung statt Ping, um Encoding-Probleme zu vermeiden. Args: ip_address: Zu testende IP-Adresse timeout: Timeout in Sekunden Returns: - bool: True wenn Ping erfolgreich + bool: True wenn Verbindung erfolgreich """ try: # IP-Adresse validieren ipaddress.ip_address(ip_address.strip()) - # Platform-spezifische Ping-Befehle - if os.name == 'nt': # Windows - cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address.strip()] - else: # Unix/Linux/macOS - cmd = ['ping', '-c', '1', '-W', str(timeout), ip_address.strip()] + # TCP-Verbindung zu Port 80 (HTTP) oder 443 (HTTPS) versuchen + import socket - result = subprocess.run( - cmd, - capture_output=True, - text=True, - encoding='utf-8', - errors='ignore', - timeout=timeout + 1 - ) + # Erst Port 80 versuchen (HTTP) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + result = sock.connect_ex((ip_address.strip(), 80)) + sock.close() - return result.returncode == 0 + if result == 0: + return True + + # Falls Port 80 nicht erfolgreich, Port 443 versuchen (HTTPS) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + result = sock.connect_ex((ip_address.strip(), 443)) + sock.close() - except Exception: + return result == 0 + + except Exception as e: + monitor_logger.debug(f"❌ Fehler beim Verbindungstest zu {ip_address}: {str(e)}") return False def _check_outlet_status(self, ip_address: str, username: str, password: str, timeout: int = 5) -> Tuple[bool, str]: