core/Projektarbeit-MYP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

"Add test files for P110 and Tapo Direct integration in backend"

Browse Source
This commit is contained in:
Till Tomczak 2025-05-29 22:17:10 +02:00
parent c7c27452e7
commit b3a1c66f8b
3 changed files with 235 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
backend/app/test_p110.py Normal file
View File

@ -0,0 +1 @@

212
backend/app/test_tapo_direkt.py Normal file
View File

@ -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")

39
backend/app/utils/printer_monitor.py
View File

@ -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]: