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"