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

Ersetze tapo durch PyP100

Browse Source 
- Ersetzt tapo==0.8.1 mit PyP100==0.0.19 in requirements.txt
- Ändert Import von ApiClient zu PyP100
- Implementiert TapoControl Klasse neu mit PyP100 API
- Entfernt async/await-Aufrufe, da PyP100 synchrone API benutzt

🤖 Generated with [Claude Code](https://claude.ai/code)
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
root 2025-03-12 10:48:23 +01:00
parent f3cd2ba730
commit ea4b903d63
2 changed files with 29 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
backend/app.py
View File

@ -14,7 +14,7 @@ from logging.handlers import RotatingFileHandler
from datetime import timedelta
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass
from tapo import ApiClient
from PyP100 import PyP100
from dotenv import load_dotenv
# Lade Umgebungsvariablen
@ -433,46 +433,46 @@ class PrintJob:
'remainingMinutes': self.remaining_time()
}
# Tapo Steckdosen-Steuerung
# TP-Link Steckdosen-Steuerung mit PyP100
class TapoControl:
def __init__(self, username, password):
self.username = username
self.password = password
self.clients = {}
self.devices = {}
async def get_client(self, ip_address):
if ip_address not in self.clients:
def get_device(self, ip_address):
if ip_address not in self.devices:
try:
client = ApiClient(self.username, self.password)
await client.login()
self.clients[ip_address] = client
device = PyP100.P100(ip_address, self.username, self.password)
device.handshake() # Erstellt die erforderlichen Cookies
device.login() # Sendet Anmeldedaten und erstellt AES-Schlüssel
self.devices[ip_address] = device
app.logger.info(f"PyP100 Verbindung zu {ip_address} hergestellt")
except Exception as e:
app.logger.error(f"Fehler bei der Anmeldung an Tapo-Gerät {ip_address}: {e}")
app.logger.error(f"Fehler bei der Anmeldung an P100-Gerät {ip_address}: {e}")
return None
return self.clients[ip_address]
return self.devices[ip_address]
async def turn_on(self, ip_address):
client = await self.get_client(ip_address)
if client:
def turn_on(self, ip_address):
device = self.get_device(ip_address)
if device:
try:
device = await client.p115(ip_address)
await device.on()
app.logger.info(f"Tapo-Steckdose {ip_address} eingeschaltet")
device.turnOn()
app.logger.info(f"P100-Steckdose {ip_address} eingeschaltet")
return True
except Exception as e:
app.logger.error(f"Fehler beim Einschalten der Tapo-Steckdose {ip_address}: {e}")
app.logger.error(f"Fehler beim Einschalten der P100-Steckdose {ip_address}: {e}")
return False
async def turn_off(self, ip_address):
client = await self.get_client(ip_address)
if client:
def turn_off(self, ip_address):
device = self.get_device(ip_address)
if device:
try:
device = await client.p115(ip_address)
await device.off()
app.logger.info(f"Tapo-Steckdose {ip_address} ausgeschaltet")
device.turnOff()
app.logger.info(f"P100-Steckdose {ip_address} ausgeschaltet")
return True
except Exception as e:
app.logger.error(f"Fehler beim Ausschalten der Tapo-Steckdose {ip_address}: {e}")
app.logger.error(f"Fehler beim Ausschalten der P100-Steckdose {ip_address}: {e}")
return False
tapo_control = TapoControl(TAPO_USERNAME, TAPO_PASSWORD)
@ -723,7 +723,7 @@ def create_job():
# Steckdose einschalten, falls IP-Adresse hinterlegt ist
if printer.ip_address:
try:
asyncio.run(tapo_control.turn_on(printer.ip_address))
tapo_control.turn_on(printer.ip_address)
except Exception as e:
app.logger.error(f"Fehler beim Einschalten der Steckdose: {e}")
@ -768,7 +768,7 @@ def abort_job(job_id):
# Steckdose ausschalten, falls IP-Adresse hinterlegt ist
if printer and printer.ip_address:
try:
asyncio.run(tapo_control.turn_off(printer.ip_address))
tapo_control.turn_off(printer.ip_address)
except Exception as e:
app.logger.error(f"Fehler beim Ausschalten der Steckdose: {e}")
@ -800,7 +800,7 @@ def finish_job(job_id):
# Steckdose ausschalten, falls IP-Adresse hinterlegt ist
if printer and printer.ip_address:
try:
asyncio.run(tapo_control.turn_off(printer.ip_address))
tapo_control.turn_off(printer.ip_address)
except Exception as e:
app.logger.error(f"Fehler beim Ausschalten der Steckdose: {e}")
@ -976,7 +976,7 @@ def check_jobs():
# Steckdose ausschalten, falls IP-Adresse hinterlegt ist
if printer and printer.ip_address:
try:
asyncio.run(tapo_control.turn_off(printer.ip_address))
tapo_control.turn_off(printer.ip_address)
app.logger.info(f"Steckdose {printer.ip_address} für abgelaufenen Job {job.id} ausgeschaltet.")
except Exception as e:
app.logger.error(f"Fehler beim Ausschalten der Steckdose {printer.ip_address}: {e}")

2
backend/requirements.txt
View File

@ -4,4 +4,4 @@ pyjwt==2.8.0
python-dotenv==1.0.0
werkzeug==2.3.7
gunicorn==21.2.0
tapo==0.8.1
PyP100==0.0.19