from __future__ import annotations
import html
from datetime import datetime, timezone
from typing import Iterable, List
from .boinc import BoincSnapshot, Task
def progress_bar(percent: float, width: int = 16) -> str:
clamped = max(0, min(percent, 100))
filled = int(round((clamped / 100.0) * width))
return "[" + "#" * filled + "." * (width - filled) + "]"
def humanize_seconds(seconds: float) -> str:
if seconds <= 0:
return "0s"
minutes, sec = divmod(int(seconds), 60)
hours, minutes = divmod(minutes, 60)
days, hours = divmod(hours, 24)
parts = []
if days:
parts.append(f"{days}d")
if hours:
parts.append(f"{hours}h")
if minutes:
parts.append(f"{minutes}m")
if sec and len(parts) < 2:
parts.append(f"{sec}s")
return " ".join(parts) if parts else f"{sec}s"
def format_timestamp(dt: datetime) -> str:
if not dt:
return "-"
if not dt.tzinfo:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
def format_task(task: Task) -> str:
safe_name = html.escape(task.name)
safe_project = html.escape(task.project)
eta = humanize_seconds(task.remaining_seconds or 0)
deadline = format_timestamp(task.deadline) if task.deadline else "-"
bar = progress_bar(task.progress_percent)
return (
f"{safe_name} ({safe_project})\n"
f"{bar} {task.progress_percent}% | ETA {eta} | Deadline {deadline}"
)
def format_tasks_section(title: str, tasks: Iterable[Task]) -> str:
items = list(tasks)
if not items:
return f"{html.escape(title)}: ничего нет"
lines: List[str] = [f"{html.escape(title)}:"]
for task in items:
lines.append(format_task(task))
return "\n".join(lines)
def format_dashboard(snapshot: BoincSnapshot) -> str:
lines = [
"BOINC / grcpool монитор",
f"Обновлено: {format_timestamp(snapshot.fetched_at)}",
"",
f"Активные: {len(snapshot.active)} | В очереди: {len(snapshot.queued)} | Завершены: {len(snapshot.completed)}",
f"Средний прогресс: {snapshot.average_progress:.1f}% | Оставшееся время: {humanize_seconds(snapshot.total_remaining_seconds)}",
"",
format_tasks_section("Активные задачи", snapshot.active[:5]),
]
if snapshot.queued:
lines.extend(["", format_tasks_section("Ожидают запуска", snapshot.queued[:5])])
if snapshot.completed:
lines.extend(["", format_tasks_section("Готовы к отправке", snapshot.completed[:5])])
return "\n".join(lines)