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

🐛 Refactor: Update printer status handling and session management. Improved caching mechanism for live printer status and added summary functionality. Removed obsolete database files for better organization.

Browse Source
This commit is contained in:
Till Tomczak
2025-06-12 19:48:45 +02:00
parent 61607ae082
commit 10423eb1e3
29 changed files with 399 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
backend/utils/hardware_integration.py
View File

@ -667,28 +667,154 @@ class PrinterMonitor:
def __init__(self):
self.cache = {}
self._cache_timeout = 300 # 5 Minuten Cache
hardware_logger.info("✅ Printer Monitor initialisiert")
def get_live_printer_status(self) -> Dict[int, Dict]:
"""Holt Live-Druckerstatus"""
def get_live_printer_status(self, use_session_cache: bool = True) -> Dict[int, Dict]:
"""
Holt Live-Druckerstatus mit Cache-Unterstützung.
Args:
use_session_cache: Ob Cache verwendet werden soll
Returns:
Dict: Druckerstatus mit Drucker-ID als Schlüssel
"""
try:
# Cache prüfen wenn aktiviert
if use_session_cache and 'live_status' in self.cache:
cache_entry = self.cache['live_status']
if (datetime.now() - cache_entry['timestamp']).total_seconds() < self._cache_timeout:
hardware_logger.debug("Live-Status aus Cache abgerufen")
return cache_entry['data']
db_session = get_db_session()
printers = db_session.query(Printer).filter(Printer.active == True).all()
status_dict = {}
for printer in printers:
status_dict[printer.id] = {
# Basis-Status
printer_status = {
"id": printer.id,
"name": printer.name,
"model": printer.model,
"location": printer.location,
"status": printer.status,
"last_checked": datetime.now().isoformat()
"ip_address": printer.ip_address,
"plug_ip": printer.plug_ip,
"has_plug": bool(printer.plug_ip),
"active": printer.active,
"last_checked": printer.last_checked.isoformat() if printer.last_checked else None,
"created_at": printer.created_at.isoformat() if printer.created_at else None
}
# Tapo-Status wenn verfügbar
if printer.plug_ip and TAPO_AVAILABLE:
try:
tapo_controller = get_tapo_controller()
reachable, plug_status = tapo_controller.check_outlet_status(
printer.plug_ip, printer_id=printer.id
)
printer_status.update({
"plug_reachable": reachable,
"plug_status": plug_status,
"can_control": reachable
})
except Exception as e:
hardware_logger.error(f"Tapo-Status-Fehler für {printer.name}: {e}")
printer_status.update({
"plug_reachable": False,
"plug_status": "error",
"can_control": False,
"error": str(e)
})
else:
printer_status.update({
"plug_reachable": False,
"plug_status": "no_plug",
"can_control": False
})
status_dict[printer.id] = printer_status
db_session.close()
# Cache aktualisieren
if use_session_cache:
self.cache['live_status'] = {
'data': status_dict,
'timestamp': datetime.now()
}
hardware_logger.info(f"Live-Status für {len(status_dict)} Drucker abgerufen")
return status_dict
except Exception as e:
hardware_logger.error(f"Status-Fehler: {e}")
return {}
def get_printer_summary(self) -> Dict[str, Any]:
"""
Erstellt eine Zusammenfassung des Druckerstatus.
Returns:
Dict: Zusammenfassung mit Zählern und Statistiken
"""
try:
status_data = self.get_live_printer_status(use_session_cache=True)
summary = {
'total': len(status_data),
'online': 0,
'offline': 0,
'standby': 0,
'unreachable': 0,
'with_plug': 0,
'plug_online': 0,
'plug_offline': 0
}
for printer_id, printer_data in status_data.items():
status = printer_data.get('status', 'offline')
# Status-Zähler
if status == 'online':
summary['online'] += 1
elif status == 'standby':
summary['standby'] += 1
elif status == 'unreachable':
summary['unreachable'] += 1
else:
summary['offline'] += 1
# Plug-Zähler
if printer_data.get('has_plug'):
summary['with_plug'] += 1
plug_status = printer_data.get('plug_status', 'unknown')
if plug_status == 'on':
summary['plug_online'] += 1
elif plug_status == 'off':
summary['plug_offline'] += 1
return summary
except Exception as e:
hardware_logger.error(f"Summary-Fehler: {e}")
return {
'total': 0,
'online': 0,
'offline': 0,
'standby': 0,
'unreachable': 0,
'with_plug': 0,
'plug_online': 0,
'plug_offline': 0
}
def clear_all_caches(self):
"""Leert alle Caches des Printer Monitors."""
self.cache.clear()
hardware_logger.debug("Printer Monitor Cache geleert")
# ===== GLOBALE INSTANZEN =====