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

🛠️ "Implementiere TP-Link Tapo P110 Unterstützung für Druckerüberwachung und -steuerung"

Browse Source 
- Aktualisiere die `check_printer_status()` Funktion zur Verwendung des PyP100-Moduls für die Tapo-Steckdosen.
- Füge neue API-Endpunkte hinzu: `test-tapo` für die Verbindungstests einzelner Drucker und `test-all-tapo` für Massentests.
- Verbessere die Fehlerbehandlung und Logging für Tapo-Verbindungen.
- Aktualisiere die Benutzeroberfläche, um den Datei-Upload als optional zu kennzeichnen.
- Implementiere umfassende Tests für die Tapo-Verbindungen in `debug_drucker_erkennung.py` und verbessere die Validierung der Konfiguration in `job_scheduler.py`.
This commit is contained in:
Till Tomczak
2025-05-29 21:19:30 +02:00
parent da1d531c16
commit de1b87f833
8 changed files with 585 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
backend/app/utils/debug_drucker_erkennung.py
View File

@@ -177,6 +177,108 @@ def test_network_connectivity():
except Exception as e:
log_message(f"❌ Fehler beim Testen der Netzwerkverbindung: {str(e)}", "ERROR")
def test_tapo_connections():
"""Teste TP-Link Tapo P110-Steckdosen-Verbindungen"""
log_message("Teste TP-Link Tapo P110-Steckdosen-Verbindungen...")
try:
# PyP100 importieren
from PyP100 import PyP110
log_message("✅ PyP100-Modul erfolgreich importiert")
except ImportError:
log_message("❌ PyP100-Modul nicht verfügbar", "ERROR")
log_message(" Installiere mit: pip install PyP100", "INFO")
return
# Lade Drucker aus Datenbank
try:
db_files = ['database.db', 'app.db', 'myp.db']
printers = []
for db_file in db_files:
if os.path.exists(db_file):
conn = sqlite3.connect(db_file)
cursor = conn.cursor()
cursor.execute("SELECT id, name, plug_ip, plug_username, plug_password FROM printer WHERE plug_ip IS NOT NULL;")
printers = cursor.fetchall()
conn.close()
break
if not printers:
log_message("❌ Keine Drucker mit Tapo-Konfiguration gefunden")
return
successful_connections = 0
total_printers = len(printers)
for printer_id, name, plug_ip, plug_username, plug_password in printers:
log_message(f"Teste Tapo-Verbindung zu {name} ({plug_ip})...")
# Konfiguration validieren
if not all([plug_ip, plug_username, plug_password]):
log_message(f" ❌ Unvollständige Konfiguration")
missing = []
if not plug_ip: missing.append("IP-Adresse")
if not plug_username: missing.append("Benutzername")
if not plug_password: missing.append("Passwort")
log_message(f" Fehlend: {', '.join(missing)}")
continue
try:
# Tapo-Verbindung herstellen
p110 = PyP110.P110(plug_ip, plug_username, plug_password)
p110.handshake() # Authentifizierung
p110.login() # Login
# Geräteinformationen abrufen
device_info = p110.getDeviceInfo()
log_message(f" ✅ Tapo-Verbindung erfolgreich")
log_message(f" 📛 Gerätename: {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")
if 'power_usage' in device_info:
power_usage = device_info.get('power_usage', {})
current_power = power_usage.get('power_mw', 0) / 1000 # mW zu W
log_message(f" 🔋 Aktueller Verbrauch: {current_power:.1f}W")
successful_connections += 1
except Exception as e:
log_message(f" ❌ Tapo-Verbindung fehlgeschlagen: {str(e)}")
# Detaillierte Fehleranalyse
if "login" in str(e).lower():
log_message(f" 🔐 Mögliche Ursache: Falsche Anmeldedaten")
elif "timeout" in str(e).lower():
log_message(f" ⏱️ Mögliche Ursache: Netzwerk-Timeout")
elif "connect" in str(e).lower():
log_message(f" 🌐 Mögliche Ursache: Steckdose nicht erreichbar")
elif "handshake" in str(e).lower():
log_message(f" 🤝 Mögliche Ursache: Protokoll-Handshake fehlgeschlagen")
# Zusammenfassung
success_rate = (successful_connections / total_printers * 100) if total_printers > 0 else 0
log_message(f"📊 Tapo-Verbindungs-Zusammenfassung:")
log_message(f" Getestete Drucker: {total_printers}")
log_message(f" Erfolgreiche Verbindungen: {successful_connections}")
log_message(f" Erfolgsrate: {success_rate:.1f}%")
if successful_connections == total_printers:
log_message("🎉 Alle Tapo-Verbindungen erfolgreich!")
elif successful_connections > 0:
log_message("⚠️ Einige Tapo-Verbindungen fehlgeschlagen")
else:
log_message("❌ Keine Tapo-Verbindungen erfolgreich", "ERROR")
except Exception as e:
log_message(f"❌ Fehler beim Testen der Tapo-Verbindungen: {str(e)}", "ERROR")
def test_flask_app_status():
"""Teste den Status der Flask-Anwendung"""
log_message("Teste Flask-Anwendung...")
@@ -292,6 +394,10 @@ def run_comprehensive_test():
test_network_connectivity()
print()
# Tapo-Verbindungen testen
test_tapo_connections()
print()
log_message("=== Diagnose abgeschlossen ===")
print()

54
backend/app/utils/job_scheduler.py
View File

@@ -313,7 +313,14 @@ def toggle_plug(printer_id: int, state: bool) -> bool:
db_session.close()
return False
# Konfiguration validieren
if not printer.plug_ip or not printer.plug_username or not printer.plug_password:
logger.error(f"Unvollständige Tapo-Konfiguration für Drucker {printer.name} (ID: {printer_id})")
db_session.close()
return False
# TP-Link Tapo P110 Verbindung herstellen
logger.debug(f"Verbinde zu Tapo-Steckdose {printer.plug_ip} für Drucker {printer.name}")
p110 = PyP110.P110(printer.plug_ip, printer.plug_username, printer.plug_password)
p110.handshake() # Authentifizierung
p110.login() # Login
@@ -321,19 +328,60 @@ def toggle_plug(printer_id: int, state: bool) -> bool:
# Steckdose ein-/ausschalten
if state:
p110.turnOn()
logger.info(f"Steckdose für Drucker {printer.name} (ID: {printer_id}) eingeschaltet")
logger.info(f"✅ Steckdose für Drucker {printer.name} (ID: {printer_id}) eingeschaltet")
else:
p110.turnOff()
logger.info(f"Steckdose für Drucker {printer.name} (ID: {printer_id}) ausgeschaltet")
logger.info(f"✅ Steckdose für Drucker {printer.name} (ID: {printer_id}) ausgeschaltet")
db_session.close()
return True
except Exception as e:
logger.error(f"Fehler beim Schalten der Steckdose für Drucker {printer_id}: {str(e)}")
logger.error(f"❌ Fehler beim Schalten der Steckdose für Drucker {printer_id}: {str(e)}")
db_session.close()
return False
def test_tapo_connection(ip_address: str, username: str, password: str) -> dict:
"""
Testet die Verbindung zu einer Tapo-Steckdose und gibt detaillierte Informationen zurück.
Args:
ip_address: IP-Adresse der Steckdose
username: Benutzername
password: Passwort
Returns:
dict: Testergebnis mit Status und Informationen
"""
logger = get_logger("printers")
result = {
"success": False,
"error": None,
"device_info": None,
"status": "unknown"
}
try:
logger.debug(f"Teste Tapo-Verbindung zu {ip_address}")
p110 = PyP110.P110(ip_address, username, password)
p110.handshake() # Authentifizierung
p110.login() # Login
# Geräteinformationen abrufen
device_info = p110.getDeviceInfo()
result["device_info"] = device_info
result["status"] = "on" if device_info.get('device_on', False) else "off"
result["success"] = True
logger.debug(f"✅ Tapo-Verbindung zu {ip_address} erfolgreich")
except Exception as e:
result["error"] = str(e)
logger.warning(f"❌ Tapo-Verbindung zu {ip_address} fehlgeschlagen: {str(e)}")
return result
def check_jobs():
"""
Überprüft alle geplanten und laufenden Jobs und schaltet Steckdosen entsprechend.

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

@@ -13,11 +13,19 @@ from typing import Dict, Tuple, List, Optional
from flask import session
from sqlalchemy import func
from sqlalchemy.orm import Session
import os
from models import get_db_session, Printer
from utils.logging_config import get_logger
from config.settings import PRINTERS
# TP-Link Tapo P110 Unterstützung hinzufügen
try:
from PyP100 import PyP110
TAPO_AVAILABLE = True
except ImportError:
TAPO_AVAILABLE = False
# Logger initialisieren
monitor_logger = get_logger("printer_monitor")
@@ -110,57 +118,35 @@ class PrinterMonitor:
def _turn_outlet_off(self, ip_address: str, username: str, password: str, timeout: int = 5) -> bool:
"""
Schaltet eine Smart-Steckdose aus.
Schaltet eine TP-Link Tapo P110-Steckdose aus.
Args:
ip_address: IP-Adresse der Steckdose
username: Benutzername für die Steckdose
password: Passwort für die Steckdose
timeout: Timeout in Sekunden
timeout: Timeout in Sekunden (wird ignoriert, da PyP100 eigenes Timeout hat)
Returns:
bool: True wenn erfolgreich ausgeschaltet
"""
if not TAPO_AVAILABLE:
monitor_logger.error("⚠️ PyP100-Modul nicht verfügbar - kann Tapo-Steckdose nicht schalten")
return False
try:
# Für TP-Link Tapo und ähnliche Smart Plugs
auth = (username, password)
# TP-Link Tapo P110 Verbindung herstellen
p110 = PyP110.P110(ip_address, username, password)
p110.handshake() # Authentifizierung
p110.login() # Login
# Verschiedene API-Endpunkte versuchen
endpoints = [
f"http://{ip_address}/relay/off",
f"http://{ip_address}/api/relay/off",
f"http://{ip_address}/set?relay=off",
f"http://{ip_address}/control?action=off"
]
# Steckdose ausschalten
p110.turnOff()
monitor_logger.debug(f"✅ Tapo-Steckdose {ip_address} erfolgreich ausgeschaltet")
return True
for endpoint in endpoints:
try:
response = requests.post(endpoint, auth=auth, timeout=timeout)
if response.status_code in [200, 201, 204]:
monitor_logger.debug(f"✅ Steckdose {ip_address} ausgeschaltet via {endpoint}")
return True
except requests.RequestException:
continue
# Wenn spezifische Endpunkte fehlschlagen, versuche generische JSON-API
try:
payload = {"system": {"set_relay_state": {"state": 0}}}
response = requests.post(
f"http://{ip_address}/api",
json=payload,
auth=auth,
timeout=timeout
)
if response.status_code in [200, 201]:
monitor_logger.debug(f"✅ Steckdose {ip_address} ausgeschaltet via JSON-API")
return True
except requests.RequestException:
pass
except Exception as e:
monitor_logger.debug(f"⚠️ Fehler beim Ausschalten der Steckdose {ip_address}: {str(e)}")
return False
monitor_logger.debug(f"⚠️ Fehler beim Ausschalten der Tapo-Steckdose {ip_address}: {str(e)}")
return False
def get_live_printer_status(self, use_session_cache: bool = True) -> Dict[int, Dict]:
"""
@@ -364,7 +350,6 @@ class PrinterMonitor:
ipaddress.ip_address(ip_address.strip())
# Platform-spezifische Ping-Befehle
import os
if os.name == 'nt': # Windows
cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address.strip()]
else: # Unix/Linux/macOS
@@ -386,79 +371,40 @@ class PrinterMonitor:
def _check_outlet_status(self, ip_address: str, username: str, password: str, timeout: int = 5) -> Tuple[bool, str]:
"""
Überprüft den Status einer Smart-Steckdose.
Überprüft den Status einer TP-Link Tapo P110-Steckdose.
Args:
ip_address: IP-Adresse der Steckdose
username: Benutzername für die Steckdose
password: Passwort für die Steckdose
timeout: Timeout in Sekunden
timeout: Timeout in Sekunden (wird ignoriert, da PyP100 eigenes Timeout hat)
Returns:
Tuple[bool, str]: (Erreichbar, Status) - Status: "on", "off", "unknown"
"""
if not TAPO_AVAILABLE:
monitor_logger.debug("⚠️ PyP100-Modul nicht verfügbar - kann Tapo-Steckdosen-Status nicht abfragen")
return False, "unknown"
try:
auth = (username, password)
# TP-Link Tapo P110 Verbindung herstellen
p110 = PyP110.P110(ip_address, username, password)
p110.handshake() # Authentifizierung
p110.login() # Login
# Verschiedene API-Endpunkte für Status-Abfrage versuchen
status_endpoints = [
f"http://{ip_address}/status",
f"http://{ip_address}/api/status",
f"http://{ip_address}/relay/status",
f"http://{ip_address}/state"
]
# Geräteinformationen abrufen
device_info = p110.getDeviceInfo()
for endpoint in status_endpoints:
try:
response = requests.get(endpoint, auth=auth, timeout=timeout)
if response.status_code == 200:
try:
data = response.json()
# Verschiedene JSON-Strukturen handhaben
if isinstance(data, dict):
# TP-Link Format
if 'system' in data and 'get_sysinfo' in data['system']:
relay_state = data['system']['get_sysinfo'].get('relay_state')
return True, "on" if relay_state == 1 else "off"
# Direktes Format
if 'relay_state' in data:
return True, "on" if data['relay_state'] in [1, True, "on"] else "off"
if 'state' in data:
return True, "on" if data['state'] in [1, True, "on"] else "off"
if 'power' in data:
return True, "on" if data['power'] in [1, True, "on"] else "off"
# Wenn JSON-Parsing funktioniert, aber Format unbekannt
return True, "unknown"
except ValueError:
# Nicht-JSON Antwort - versuche Text-Parsing
text = response.text.lower()
if "on" in text or "1" in text or "true" in text:
return True, "on"
elif "off" in text or "0" in text or "false" in text:
return True, "off"
return True, "unknown"
except requests.RequestException:
continue
# Status auswerten
device_on = device_info.get('device_on', False)
status = "on" if device_on else "off"
monitor_logger.debug(f"✅ Tapo-Steckdose {ip_address}: Status = {status}")
return True, status
# Wenn spezifische Endpunkte fehlschlagen, einfache Konnektivitätsprüfung
try:
response = requests.get(f"http://{ip_address}", auth=auth, timeout=timeout)
if response.status_code in [200, 401, 403]: # Erreichbar, aber möglicherweise Authentifizierungsproblem
return True, "unknown"
except requests.RequestException:
pass
except Exception as e:
monitor_logger.debug(f"⚠️ Fehler bei Steckdosen-Status-Check {ip_address}: {str(e)}")
return False, "unknown"
monitor_logger.debug(f"⚠️ Fehler bei Tapo-Steckdosen-Status-Check {ip_address}: {str(e)}")
return False, "unknown"
def clear_all_caches(self):
"""Löscht alle Caches (Session und DB)."""