Imrayya
0b4f739d12
- FastAPI server with audio mixer (Windows + Voicemeeter), media tracking, notifications (progress + alerts), telemetry, and system tray .exe build - Test suite covering notifications, audio mixer, media, telemetry, and API - E-Ink web client (vanilla JS, DOM API, block meters, swipe gestures) - Android WebView APK for BOOX Go 7 Color Gen II (Android 13, Kotlin) - Design decision records in .pi/docs/design/ - PyInstaller build script for server .exe
132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""Notification system supporting progress bars and rich text alerts.
|
|
|
|
External scripts push notifications via POST /api/v1/notify.
|
|
Two types are supported:
|
|
- "progress": long-running tasks with a measurable progress bar
|
|
- "alert": one-off informational/warning/error messages
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from typing import Literal
|
|
|
|
logger = logging.getLogger("paperdash.notifications")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data Models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ProgressState:
|
|
"""Progress tracking for long-running tasks."""
|
|
|
|
current: float = 0.0
|
|
min: float = 0.0
|
|
max: float = 100.0
|
|
unit: str = ""
|
|
|
|
@property
|
|
def percentage(self) -> float:
|
|
if self.max <= self.min:
|
|
return 0.0
|
|
return max(
|
|
0.0, min(100.0, (self.current - self.min) / (self.max - self.min) * 100)
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Notification:
|
|
"""A notification to display on the dashboard."""
|
|
|
|
id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
|
type: Literal["progress", "alert"] = "alert"
|
|
title: str = ""
|
|
message: str = ""
|
|
progress: ProgressState | None = None
|
|
eta: str | None = None
|
|
priority: Literal["info", "warning", "error"] = "info"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Notification Manager
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class NotificationManager:
|
|
"""Manages active notifications and their lifecycle."""
|
|
|
|
def __init__(self) -> None:
|
|
self._notifications: dict[str, Notification] = {}
|
|
self._max_active: int = 10 # Keep at most 10 active at a time
|
|
|
|
async def create(self, notification: Notification) -> Notification:
|
|
"""Create a new notification or update an existing one by ID."""
|
|
# If a notification with this ID already exists, update it
|
|
if notification.id in self._notifications:
|
|
existing = self._notifications[notification.id]
|
|
# Preserve type if not explicitly set on update
|
|
if not notification.type:
|
|
notification.type = existing.type
|
|
self._notifications[notification.id] = notification
|
|
logger.debug(f"Notification updated: {notification.id}")
|
|
else:
|
|
self._notifications[notification.id] = notification
|
|
logger.info(
|
|
f"Notification created: {notification.id} ({notification.type})"
|
|
)
|
|
|
|
# Prune old notifications if we exceed the limit
|
|
self._prune_old()
|
|
|
|
return notification
|
|
|
|
def _prune_old(self) -> None:
|
|
"""Remove oldest notifications if we exceed the limit."""
|
|
if len(self._notifications) <= self._max_active:
|
|
return
|
|
|
|
# Remove oldest first (by insertion order)
|
|
ids_to_remove = list(self._notifications.keys())[
|
|
: len(self._notifications) - self._max_active
|
|
]
|
|
for nid in ids_to_remove:
|
|
del self._notifications[nid]
|
|
logger.debug(f"Notification pruned: {nid}")
|
|
|
|
def get_active(self) -> list[dict]:
|
|
"""Return all active notifications as serializable dicts."""
|
|
result = []
|
|
for n in self._notifications.values():
|
|
entry: dict = {
|
|
"id": n.id,
|
|
"type": n.type,
|
|
"title": n.title,
|
|
"message": n.message,
|
|
"priority": n.priority,
|
|
}
|
|
if n.type == "progress" and n.progress:
|
|
entry["progress"] = {
|
|
"current": n.progress.current,
|
|
"min": n.progress.min,
|
|
"max": n.progress.max,
|
|
"unit": n.progress.unit,
|
|
"percentage": n.progress.percentage,
|
|
}
|
|
if n.eta:
|
|
entry["eta"] = n.eta
|
|
result.append(entry)
|
|
return result
|
|
|
|
async def dismiss(self, notification_id: str) -> bool:
|
|
"""Dismiss (remove) a notification by ID."""
|
|
if notification_id in self._notifications:
|
|
del self._notifications[notification_id]
|
|
logger.debug(f"Notification dismissed: {notification_id}")
|
|
return True
|
|
return False
|