""" tp-link tapo p110 zentraler controller für myp platform sammelt alle operativen tapo-steckdosen-funktionalitäten an einem ort. """ import time import socket import signal import ipaddress from datetime import datetime from typing import Dict, Tuple, Optional, List, Any from concurrent.futures import ThreadPoolExecutor, as_completed from models import get_db_session, Printer, PlugStatusLog from utils.logging_config import get_logger from utils.settings import TAPO_USERNAME, TAPO_PASSWORD, DEFAULT_TAPO_IPS, TAPO_TIMEOUT, TAPO_RETRY_COUNT # tp-link tapo p110 unterstützung prüfen try: from PyP100 import PyP100 TAPO_AVAILABLE = True except ImportError: TAPO_AVAILABLE = False # logger initialisieren logger = get_logger("tapo_controller") class TapoController: """ zentraler controller für alle tp-link tapo p110 operationen. """ def __init__(self): """initialisiere den tapo controller.""" self.username = TAPO_USERNAME self.password = TAPO_PASSWORD self.timeout = TAPO_TIMEOUT self.retry_count = TAPO_RETRY_COUNT self.auto_discovered = False if not TAPO_AVAILABLE: logger.error("❌ PyP100-modul nicht installiert - tapo-funktionalität eingeschränkt") else: logger.info("✅ tapo controller initialisiert") def toggle_plug(self, ip: str, state: bool, username: str = None, password: str = None) -> bool: """ schaltet eine tp-link tapo p100/p110-steckdose ein oder aus. args: ip: ip-adresse der steckdose state: true = ein, false = aus username: benutzername (optional, nutzt standard wenn nicht angegeben) password: passwort (optional, nutzt standard wenn nicht angegeben) returns: bool: true wenn erfolgreich geschaltet """ if not TAPO_AVAILABLE: logger.error("❌ PyP100-modul nicht installiert - steckdose kann nicht geschaltet werden") return False # immer globale anmeldedaten verwenden username = self.username password = self.password logger.debug(f"🔧 verwende globale tapo-anmeldedaten für {ip}") for attempt in range(self.retry_count): try: # p100-verbindung herstellen p100 = PyP100.P100(ip, username, password) p100.handshake() p100.login() # steckdose schalten if state: p100.turnOn() logger.info(f"✅ tapo-steckdose {ip} erfolgreich eingeschaltet") else: p100.turnOff() logger.info(f"✅ tapo-steckdose {ip} erfolgreich ausgeschaltet") return True except Exception as e: action = "ein" if state else "aus" logger.warning(f"⚠️ versuch {attempt+1}/{self.retry_count} fehlgeschlagen beim {action}schalten von {ip}: {str(e)}") if attempt < self.retry_count - 1: time.sleep(1) # kurze pause vor erneutem versuch else: logger.error(f"❌ fehler beim {action}schalten der tapo-steckdose {ip}: {str(e)}") return False def turn_off(self, ip: str, username: str = None, password: str = None, printer_id: int = None) -> bool: """ schaltet eine tp-link tapo p110-steckdose aus. args: ip: ip-adresse der steckdose username: benutzername (optional) password: passwort (optional) printer_id: id des zugehörigen druckers für logging (optional) returns: bool: true wenn erfolgreich ausgeschaltet """ if not TAPO_AVAILABLE: logger.error("⚠️ PyP100-modul nicht verfügbar - kann tapo-steckdose nicht schalten") self._log_plug_status(printer_id, "disconnected", ip, error_message="PyP100-modul nicht verfügbar") return False # immer globale anmeldedaten verwenden username = self.username password = self.password start_time = time.time() try: # tp-link tapo p100 verbindung herstellen p100 = PyP100.P100(ip, username, password) p100.handshake() p100.login() # steckdose ausschalten p100.turnOff() response_time = int((time.time() - start_time) * 1000) # in millisekunden logger.debug(f"✅ tapo-steckdose {ip} erfolgreich ausgeschaltet") # logging: erfolgreich ausgeschaltet self._log_plug_status(printer_id, "off", ip, response_time_ms=response_time) return True except Exception as e: response_time = int((time.time() - start_time) * 1000) logger.debug(f"⚠️ fehler beim ausschalten der tapo-steckdose {ip}: {str(e)}") # logging: fehlgeschlagener versuch self._log_plug_status(printer_id, "disconnected", ip, response_time_ms=response_time, error_message=str(e)) return False def check_outlet_status(self, ip: str, username: str = None, password: str = None, printer_id: int = None) -> Tuple[bool, str]: """ überprüft den status einer tp-link tapo p110-steckdose. args: ip: ip-adresse der steckdose username: benutzername (optional) password: passwort (optional) printer_id: id des zugehörigen druckers für logging (optional) returns: tuple[bool, str]: (erreichbar, status) - status: "on", "off", "unknown" """ if not TAPO_AVAILABLE: logger.debug("⚠️ PyP100-modul nicht verfügbar - kann tapo-steckdosen-status nicht abfragen") self._log_plug_status(printer_id, "disconnected", ip, error_message="PyP100-modul nicht verfügbar", notes="status-check fehlgeschlagen") return False, "unknown" # immer globale anmeldedaten verwenden username = self.username password = self.password start_time = time.time() try: # tp-link tapo p100 verbindung herstellen p100 = PyP100.P100(ip, username, password) p100.handshake() p100.login() # geräteinformationen abrufen device_info = p100.getDeviceInfo() # status auswerten device_on = device_info.get('device_on', False) status = "on" if device_on else "off" response_time = int((time.time() - start_time) * 1000) logger.debug(f"✅ tapo-steckdose {ip}: status = {status}") # erweiterte informationen sammeln extra_info = self._collect_device_info(p100, device_info) # logging: erfolgreicher status-check self._log_plug_status(printer_id, status, ip, response_time_ms=response_time, power_consumption=extra_info.get('power_consumption'), voltage=extra_info.get('voltage'), current=extra_info.get('current'), firmware_version=extra_info.get('firmware_version'), notes="automatischer status-check") return True, status except Exception as e: response_time = int((time.time() - start_time) * 1000) logger.debug(f"⚠️ fehler bei tapo-steckdosen-status-check {ip}: {str(e)}") # logging: fehlgeschlagener status-check self._log_plug_status(printer_id, "disconnected", ip, response_time_ms=response_time, error_message=str(e), notes="status-check fehlgeschlagen") return False, "unknown" def test_connection(self, ip: str, username: str = None, password: str = None) -> dict: """ testet die verbindung zu einer tp-link tapo p110-steckdose. args: ip: ip-adresse der steckdose username: benutzername (optional) password: passwort (optional) returns: dict: ergebnis mit status und informationen """ result = { "success": False, "message": "", "device_info": None, "error": None } if not TAPO_AVAILABLE: result["message"] = "PyP100-modul nicht verfügbar" result["error"] = "ModuleNotFound" logger.error("PyP100-modul nicht verfügbar - kann tapo-steckdosen nicht testen") return result # verwende globale anmeldedaten falls nicht angegeben if not username or not password: username = self.username password = self.password logger.debug(f"verwende globale tapo-anmeldedaten für {ip}") try: # tp-link tapo p100 verbindung herstellen p100 = PyP100.P100(ip, username, password) p100.handshake() p100.login() # geräteinformationen abrufen device_info = p100.getDeviceInfo() result["success"] = True result["message"] = "verbindung erfolgreich" result["device_info"] = device_info logger.info(f"tapo-verbindung zu {ip} erfolgreich: {device_info.get('nickname', 'unbekannt')}") except Exception as e: result["success"] = False result["message"] = f"verbindungsfehler: {str(e)}" result["error"] = str(e) logger.error(f"fehler bei tapo-test zu {ip}: {str(e)}") return result def ping_address(self, ip: str, timeout: int = 3) -> bool: """ führt einen konnektivitätstest zu einer ip-adresse durch. verwendet tcp-verbindung statt ping für bessere kompatibilität. args: ip: zu testende ip-adresse timeout: timeout in sekunden returns: bool: true wenn verbindung erfolgreich """ try: # ip-adresse validieren ipaddress.ip_address(ip.strip()) # standard-ports für tapo-steckdosen testen test_ports = [9999, 80, 443] # tapo-standard, http, https for port in test_ports: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) result = sock.connect_ex((ip.strip(), port)) sock.close() if result == 0: logger.debug(f"✅ verbindung zu {ip}:{port} erfolgreich") return True logger.debug(f"❌ keine verbindung zu {ip} auf standard-ports möglich") return False except Exception as e: logger.debug(f"❌ fehler beim verbindungstest zu {ip}: {str(e)}") return False def auto_discover_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: logger.info("🔍 tapo-steckdosen wurden bereits erkannt") return {} logger.info("🔍 starte automatische tapo-steckdosenerkennung...") results = {} start_time = time.time() # standard-ips aus der konfiguration testen logger.info(f"🔄 teste {len(DEFAULT_TAPO_IPS)} standard-ips aus der konfiguration") for i, ip in enumerate(DEFAULT_TAPO_IPS): try: logger.info(f"🔍 teste ip {i+1}/{len(DEFAULT_TAPO_IPS)}: {ip}") # schneller ping-test if self.ping_address(ip, timeout=2): logger.info(f"✅ steckdose mit ip {ip} ist erreichbar") # tapo-verbindung testen test_result = self.test_connection(ip) if test_result["success"]: device_info = test_result["device_info"] nickname = device_info.get('nickname', f"tapo p110 ({ip})") state = "on" if device_info.get('device_on', False) else "off" logger.info(f"✅ tapo-steckdose '{nickname}' ({ip}) gefunden - status: {state}") results[ip] = True # steckdose in datenbank speichern/aktualisieren try: self._ensure_outlet_in_database(ip, nickname) except Exception as db_error: logger.warning(f"⚠️ fehler beim speichern in db für {ip}: {str(db_error)}") else: logger.debug(f"❌ ip {ip} ist erreichbar, aber keine tapo-steckdose") results[ip] = False else: logger.debug(f"❌ ip {ip} nicht erreichbar") results[ip] = False except Exception as e: logger.warning(f"❌ fehler bei steckdosen-erkennung für ip {ip}: {str(e)}") results[ip] = False continue # erfolgsstatistik success_count = sum(1 for success in results.values() if success) elapsed_time = time.time() - start_time logger.info(f"✅ steckdosen-erkennung abgeschlossen: {success_count}/{len(results)} steckdosen gefunden in {elapsed_time:.1f}s") self.auto_discovered = True return results def initialize_all_outlets(self) -> Dict[str, bool]: """ schaltet alle gespeicherten steckdosen aus (einheitlicher startzustand). returns: dict[str, bool]: ergebnis der initialisierung pro drucker """ logger.info("🚀 starte steckdosen-initialisierung...") results = {} try: db_session = get_db_session() printers = db_session.query(Printer).filter(Printer.active == True).all() if not printers: logger.warning("⚠️ keine aktiven drucker zur initialisierung gefunden") db_session.close() return results # alle steckdosen ausschalten for printer in printers: try: if printer.plug_ip: success = self.turn_off( printer.plug_ip, printer_id=printer.id ) results[printer.name] = success if success: logger.info(f"✅ {printer.name}: steckdose ausgeschaltet") printer.status = "offline" printer.last_checked = datetime.now() else: logger.warning(f"❌ {printer.name}: steckdose konnte nicht ausgeschaltet werden") else: logger.warning(f"⚠️ {printer.name}: keine steckdosen-ip konfiguriert") results[printer.name] = False except Exception as e: logger.error(f"❌ fehler bei initialisierung von {printer.name}: {str(e)}") results[printer.name] = False # änderungen speichern db_session.commit() db_session.close() success_count = sum(1 for success in results.values() if success) total_count = len(results) logger.info(f"🎯 steckdosen-initialisierung abgeschlossen: {success_count}/{total_count} erfolgreich") except Exception as e: logger.error(f"❌ kritischer fehler bei steckdosen-initialisierung: {str(e)}") return results def get_all_outlet_status(self) -> Dict[str, Dict[str, Any]]: """ holt den status aller konfigurierten tapo-steckdosen. returns: dict[str, dict]: status aller steckdosen mit ip als schlüssel """ status_dict = {} try: db_session = get_db_session() printers = db_session.query(Printer).filter( Printer.active == True, Printer.plug_ip.isnot(None) ).all() if not printers: logger.info("ℹ️ keine drucker mit tapo-steckdosen konfiguriert") db_session.close() return status_dict logger.info(f"🔍 prüfe status von {len(printers)} tapo-steckdosen...") # parallel-status-prüfung with ThreadPoolExecutor(max_workers=min(len(printers), 8)) as executor: future_to_printer = { executor.submit( self.check_outlet_status, printer.plug_ip, printer_id=printer.id ): printer for printer in printers } for future in as_completed(future_to_printer, timeout=15): printer = future_to_printer[future] try: reachable, status = future.result() status_dict[printer.plug_ip] = { "printer_name": printer.name, "printer_id": printer.id, "reachable": reachable, "status": status, "ip": printer.plug_ip, "last_checked": datetime.now().isoformat() } except Exception as e: logger.error(f"❌ fehler bei status-check für {printer.name}: {str(e)}") status_dict[printer.plug_ip] = { "printer_name": printer.name, "printer_id": printer.id, "reachable": False, "status": "error", "ip": printer.plug_ip, "error": str(e), "last_checked": datetime.now().isoformat() } db_session.close() logger.info(f"✅ status-update abgeschlossen für {len(status_dict)} steckdosen") except Exception as e: logger.error(f"❌ kritischer fehler beim abrufen des steckdosen-status: {str(e)}") return status_dict def _collect_device_info(self, p100: PyP100.P100, device_info: dict) -> dict: """ sammelt erweiterte geräteinformationen von der tapo-steckdose. args: p100: pyp100-instanz device_info: basis-geräteinformationen returns: dict: erweiterte informationen """ extra_info = {} try: # firmware-version extra_info['firmware_version'] = device_info.get('fw_ver', None) # versuche energiedaten zu holen (nur p110) try: energy_usage = p100.getEnergyUsage() if energy_usage: extra_info['power_consumption'] = energy_usage.get('current_power', None) extra_info['voltage'] = energy_usage.get('voltage', None) extra_info['current'] = energy_usage.get('current', None) except: pass # p100 unterstützt keine energiedaten except Exception as e: logger.debug(f"fehler beim sammeln erweiterter geräteinformationen: {str(e)}") return extra_info def _log_plug_status(self, printer_id: int, status: str, ip_address: str, **kwargs): """ protokolliert steckdosen-status in der datenbank. args: printer_id: id des druckers status: status der steckdose ip_address: ip-adresse der steckdose **kwargs: zusätzliche parameter für das logging """ if not printer_id: return try: PlugStatusLog.log_status_change( printer_id=printer_id, status=status, source="system", ip_address=ip_address, **kwargs ) except Exception as e: logger.warning(f"fehler beim loggen des steckdosen-status: {e}") def _ensure_outlet_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 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 if not existing_printer.plug_username or not existing_printer.plug_password: existing_printer.plug_username = self.username existing_printer.plug_password = self.password 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 logger.info(f"✅ drucker {old_name} umbenannt zu {nickname}") # drucker als aktiv markieren if not existing_printer.active: existing_printer.active = True logger.info(f"✅ drucker {existing_printer.name} als aktiv markiert") existing_printer.last_checked = datetime.now() db_session.commit() db_session.close() return True else: # neuen drucker erstellen printer_name = nickname or f"tapo p110 ({ip_address})" mac_address = f"tapo:{ip_address.replace('.', '-')}" new_printer = Printer( name=printer_name, model="TP-Link Tapo P110", location="automatisch erkannt", ip_address=ip_address, mac_address=mac_address, plug_ip=ip_address, plug_username=self.username, plug_password=self.password, status="offline", active=True, last_checked=datetime.now() ) db_session.add(new_printer) db_session.commit() logger.info(f"✅ neuer drucker '{printer_name}' mit tapo-steckdose {ip_address} erstellt") db_session.close() return True except Exception as e: 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 für einfachen zugriff tapo_controller = TapoController() # convenience-funktionen für rückwärtskompatibilität def toggle_plug(ip: str, state: bool, username: str = None, password: str = None) -> bool: """schaltet eine tapo-steckdose ein/aus.""" return tapo_controller.toggle_plug(ip, state, username, password) def test_tapo_connection(ip: str, username: str = None, password: str = None) -> dict: """testet die verbindung zu einer tapo-steckdose.""" return tapo_controller.test_connection(ip, username, password) def check_outlet_status(ip: str, username: str = None, password: str = None, printer_id: int = None) -> Tuple[bool, str]: """prüft den status einer tapo-steckdose.""" return tapo_controller.check_outlet_status(ip, username, password, printer_id) def auto_discover_tapo_outlets() -> Dict[str, bool]: """führt automatische erkennung von tapo-steckdosen durch.""" return tapo_controller.auto_discover_outlets() def initialize_all_outlets() -> Dict[str, bool]: """initialisiert alle tapo-steckdosen (schaltet sie aus).""" return tapo_controller.initialize_all_outlets()