655 lines
26 KiB
Python
655 lines
26 KiB
Python
"""
|
||
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() |