64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
import time
|
|
import asyncio
|
|
import inspect
|
|
|
|
class AlertManager:
|
|
def __init__(self):
|
|
self.alerts = []
|
|
self.on_alert_callback = None
|
|
|
|
def set_on_alert_callback(self, callback):
|
|
self.on_alert_callback = callback
|
|
|
|
def call_on_alert_callback(self, alert):
|
|
if self.on_alert_callback:
|
|
if inspect.iscoroutinefunction(self.on_alert_callback):
|
|
# Schedule coroutine on the running loop without blocking
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
loop.create_task(self.on_alert_callback(alert))
|
|
except RuntimeError:
|
|
# Fallback if called outside an active event loop
|
|
asyncio.run(self.on_alert_callback(alert))
|
|
else:
|
|
self.on_alert_callback(alert)
|
|
|
|
def add_alert(self, alert):
|
|
self.alerts.append(alert)
|
|
self.call_on_alert_callback(alert)
|
|
|
|
def get_alerts(self):
|
|
return self.alerts
|
|
|
|
def clear_alerts(self):
|
|
self.alerts.clear()
|
|
|
|
class Alert:
|
|
def __init__(self, alertType: "AlertType", message: str, redistributed: bool = False, small_message: str = None):
|
|
self.alertType = alertType
|
|
self.message = message
|
|
self.small_message = small_message
|
|
self.timestamp = time.time()
|
|
self.redistributed = redistributed
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"type": self.alertType,
|
|
"message": self.message,
|
|
"timestamp": self.timestamp
|
|
}
|
|
|
|
def to_json(self):
|
|
import json
|
|
return json.dumps(self.to_dict())
|
|
|
|
def from_json(json_string: str) -> "Alert":
|
|
import json
|
|
data = json.loads(json_string)
|
|
alert = Alert(data["type"], data["message"])
|
|
alert.timestamp = data["timestamp"]
|
|
return alert
|
|
|
|
class AlertType:
|
|
COOKING_SAFETY = "COOKING_SAFETY"
|
|
TEMPERATURE_ALERT = "TEMPERATURE_ALERT" |