sirtimbly/luk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dashboard sync app

Browse Source
This commit is contained in:
Tim Bendt
2025-12-16 17:13:26 -05:00
parent 73079f743a
commit d7c82a0da0
25 changed files with 4181 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

680
src/cli/sync_dashboard.py Normal file
View File

@@ -0,0 +1,680 @@
"""TUI dashboard for sync progress with scrollable logs."""
from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical
from textual.widgets import (
Header,
Footer,
Static,
ProgressBar,
Log,
ListView,
ListItem,
Label,
)
from textual.reactive import reactive
from textual.binding import Binding
from rich.text import Text
from datetime import datetime, timedelta
import asyncio
from typing import Dict, Any, Optional, List, Callable
# Default sync interval in seconds (5 minutes)
DEFAULT_SYNC_INTERVAL = 300
# Futuristic spinner frames
# SPINNER_FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
# Alternative spinners you could use:
# SPINNER_FRAMES = ["◢", "◣", "◤", "◥"] # Rotating triangle
SPINNER_FRAMES = ["▰▱▱▱▱", "▰▰▱▱▱", "▰▰▰▱▱", "▰▰▰▰▱", "▰▰▰▰▰", "▱▰▰▰▰", "▱▱▰▰▰", "▱▱▱▰▰", "▱▱▱▱▰"] # Loading bar
# SPINNER_FRAMES = ["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"] # Braille dots
# SPINNER_FRAMES = ["◐", "◓", "◑", "◒"] # Circle quarters
# SPINNER_FRAMES = ["⠁", "⠂", "⠄", "⡀", "⢀", "⠠", "⠐", "⠈"] # Braille orbit
class TaskStatus:
"""Status constants for tasks."""
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
ERROR = "error"
class TaskListItem(ListItem):
"""A list item representing a sync task."""
def __init__(self, task_id: str, task_name: str, *args, **kwargs):
super().__init__(*args, **kwargs)
self.task_id = task_id
self.task_name = task_name
self.status = TaskStatus.PENDING
self.progress = 0
self.total = 100
self.spinner_frame = 0
def compose(self) -> ComposeResult:
"""Compose the task item layout."""
yield Static(self._build_content_text(), id=f"task-content-{self.task_id}")
def _get_status_icon(self) -> str:
"""Get icon based on status."""
if self.status == TaskStatus.RUNNING:
return SPINNER_FRAMES[self.spinner_frame % len(SPINNER_FRAMES)]
icons = {
TaskStatus.PENDING: "○",
TaskStatus.COMPLETED: "✓",
TaskStatus.ERROR: "✗",
}
return icons.get(self.status, "○")
def advance_spinner(self) -> None:
"""Advance the spinner to the next frame."""
self.spinner_frame = (self.spinner_frame + 1) % len(SPINNER_FRAMES)
def _get_status_color(self) -> str:
"""Get color based on status."""
colors = {
TaskStatus.PENDING: "dim",
TaskStatus.RUNNING: "cyan",
TaskStatus.COMPLETED: "bright_white",
TaskStatus.ERROR: "red",
}
return colors.get(self.status, "white")
def _build_content_text(self) -> Text:
"""Build the task content text."""
icon = self._get_status_icon()
color = self._get_status_color()
# Use green checkmark for completed, but white text for readability
if self.status == TaskStatus.RUNNING:
progress_pct = (
int((self.progress / self.total) * 100) if self.total > 0 else 0
)
text = Text()
text.append(f"{icon} ", style="cyan")
text.append(f"{self.task_name} [{progress_pct}%]", style=color)
return text
elif self.status == TaskStatus.COMPLETED:
text = Text()
text.append(f"{icon} ", style="green") # Green checkmark
text.append(f"{self.task_name} [Done]", style=color)
return text
elif self.status == TaskStatus.ERROR:
text = Text()
text.append(f"{icon} ", style="red")
text.append(f"{self.task_name} [Error]", style=color)
return text
else:
return Text(f"{icon} {self.task_name}", style=color)
def update_display(self) -> None:
"""Update the display of this item."""
try:
content = self.query_one(f"#task-content-{self.task_id}", Static)
content.update(self._build_content_text())
except Exception:
pass
class SyncDashboard(App):
"""TUI dashboard for sync operations."""
BINDINGS = [
Binding("q", "quit", "Quit"),
Binding("ctrl+c", "quit", "Quit"),
Binding("s", "sync_now", "Sync Now"),
Binding("r", "refresh", "Refresh"),
Binding("+", "increase_interval", "+Interval"),
Binding("-", "decrease_interval", "-Interval"),
Binding("up", "cursor_up", "Up", show=False),
Binding("down", "cursor_down", "Down", show=False),
]
CSS = """
.dashboard {
height: 100%;
layout: horizontal;
}
.sidebar {
width: 30;
height: 100%;
border: solid $primary;
padding: 0;
}
.sidebar-title {
text-style: bold;
padding: 1;
background: $primary-darken-2;
}
.countdown-container {
height: 3;
padding: 0 1;
border-top: solid $primary;
background: $surface;
}
.countdown-text {
text-align: center;
}
.main-panel {
width: 1fr;
height: 100%;
padding: 0;
}
.task-header {
height: 5;
padding: 1;
border-bottom: solid $primary;
}
.task-name {
text-style: bold;
}
.progress-row {
height: 3;
padding: 0 1;
}
.log-container {
height: 1fr;
border: solid $primary;
padding: 0;
}
.log-title {
padding: 0 1;
background: $primary-darken-2;
}
ListView {
height: 1fr;
}
ListItem {
padding: 0 1;
}
ListItem:hover {
background: $primary-darken-1;
}
Log {
height: 1fr;
border: none;
}
ProgressBar {
width: 1fr;
padding: 0 1;
}
"""
selected_task: reactive[str] = reactive("archive")
sync_interval: reactive[int] = reactive(DEFAULT_SYNC_INTERVAL)
next_sync_time: reactive[float] = reactive(0.0)
def __init__(self, sync_interval: int = DEFAULT_SYNC_INTERVAL):
super().__init__()
self._mounted: asyncio.Event = asyncio.Event()
self._task_logs: Dict[str, List[str]] = {}
self._task_items: Dict[str, TaskListItem] = {}
self._sync_callback: Optional[Callable] = None
self._countdown_task: Optional[asyncio.Task] = None
self._spinner_task: Optional[asyncio.Task] = None
self._initial_sync_interval = sync_interval
def compose(self) -> ComposeResult:
"""Compose the dashboard layout."""
yield Header()
with Horizontal(classes="dashboard"):
# Sidebar with task list
with Vertical(classes="sidebar"):
yield Static("Tasks", classes="sidebar-title")
yield ListView(
# Stage 1: Sync local changes to server
TaskListItem("archive", "Archive Mail", id="task-archive"),
TaskListItem("outbox", "Outbox Send", id="task-outbox"),
# Stage 2: Fetch from server
TaskListItem("inbox", "Inbox Sync", id="task-inbox"),
TaskListItem("calendar", "Calendar Sync", id="task-calendar"),
# Stage 3: Task management
TaskListItem("godspeed", "Godspeed Sync", id="task-godspeed"),
TaskListItem("sweep", "Task Sweep", id="task-sweep"),
id="task-list",
)
# Countdown timer at bottom of sidebar
with Vertical(classes="countdown-container"):
yield Static(
"Next sync: --:--", id="countdown", classes="countdown-text"
)
# Main panel with selected task details
with Vertical(classes="main-panel"):
# Task header with name and progress
with Vertical(classes="task-header"):
yield Static(
"Archive Mail", id="selected-task-name", classes="task-name"
)
with Horizontal(classes="progress-row"):
yield Static("Progress:", id="progress-label")
yield ProgressBar(total=100, id="task-progress")
yield Static("0%", id="progress-percent")
# Log for selected task
with Vertical(classes="log-container"):
yield Static("Activity Log", classes="log-title")
yield Log(id="task-log")
yield Footer()
def on_mount(self) -> None:
"""Initialize the dashboard."""
# Store references to task items
task_list = self.query_one("#task-list", ListView)
for item in task_list.children:
if isinstance(item, TaskListItem):
self._task_items[item.task_id] = item
self._task_logs[item.task_id] = []
# Initialize sync interval
self.sync_interval = self._initial_sync_interval
self.schedule_next_sync()
# Start countdown timer and spinner animation
self._countdown_task = asyncio.create_task(self._update_countdown())
self._spinner_task = asyncio.create_task(self._animate_spinners())
self._log_to_task("archive", "Dashboard initialized. Waiting to start sync...")
self._mounted.set()
def on_list_view_selected(self, event: ListView.Selected) -> None:
"""Handle task selection from the list."""
if isinstance(event.item, TaskListItem):
self.selected_task = event.item.task_id
self._update_main_panel()
def on_list_view_highlighted(self, event: ListView.Highlighted) -> None:
"""Handle task highlight from the list."""
if isinstance(event.item, TaskListItem):
self.selected_task = event.item.task_id
self._update_main_panel()
def _update_main_panel(self) -> None:
"""Update the main panel to show selected task details."""
task_item = self._task_items.get(self.selected_task)
if not task_item:
return
# Update task name
try:
name_widget = self.query_one("#selected-task-name", Static)
name_widget.update(Text(task_item.task_name, style="bold"))
except Exception:
pass
# Update progress bar
try:
progress_bar = self.query_one("#task-progress", ProgressBar)
progress_bar.total = task_item.total
progress_bar.progress = task_item.progress
percent_widget = self.query_one("#progress-percent", Static)
pct = (
int((task_item.progress / task_item.total) * 100)
if task_item.total > 0
else 0
)
percent_widget.update(f"{pct}%")
except Exception:
pass
# Update log with task-specific logs
try:
log_widget = self.query_one("#task-log", Log)
log_widget.clear()
for entry in self._task_logs.get(self.selected_task, []):
log_widget.write_line(entry)
except Exception:
pass
def _log_to_task(self, task_id: str, message: str, level: str = "INFO") -> None:
"""Add a log entry to a specific task."""
timestamp = datetime.now().strftime("%H:%M:%S")
formatted = f"[{timestamp}] {level}: {message}"
if task_id not in self._task_logs:
self._task_logs[task_id] = []
self._task_logs[task_id].append(formatted)
# If this is the selected task, also write to the visible log
if task_id == self.selected_task:
try:
log_widget = self.query_one("#task-log", Log)
log_widget.write_line(formatted)
except Exception:
pass
def start_task(self, task_id: str, total: int = 100) -> None:
"""Start a task."""
if task_id in self._task_items:
item = self._task_items[task_id]
item.status = TaskStatus.RUNNING
item.progress = 0
item.total = total
item.update_display()
self._log_to_task(task_id, f"Starting {item.task_name}...")
if task_id == self.selected_task:
self._update_main_panel()
def update_task(self, task_id: str, progress: int, message: str = "") -> None:
"""Update task progress."""
if task_id in self._task_items:
item = self._task_items[task_id]
item.progress = progress
item.update_display()
if message:
self._log_to_task(task_id, message)
if task_id == self.selected_task:
self._update_main_panel()
def complete_task(self, task_id: str, message: str = "") -> None:
"""Mark a task as complete."""
if task_id in self._task_items:
item = self._task_items[task_id]
item.status = TaskStatus.COMPLETED
item.progress = item.total
item.update_display()
self._log_to_task(
task_id,
f"Completed: {message}" if message else "Completed successfully",
)
if task_id == self.selected_task:
self._update_main_panel()
def error_task(self, task_id: str, error: str) -> None:
"""Mark a task as errored."""
if task_id in self._task_items:
item = self._task_items[task_id]
item.status = TaskStatus.ERROR
item.update_display()
self._log_to_task(task_id, f"ERROR: {error}", "ERROR")
if task_id == self.selected_task:
self._update_main_panel()
def skip_task(self, task_id: str, reason: str = "") -> None:
"""Mark a task as skipped (completed with no work)."""
if task_id in self._task_items:
item = self._task_items[task_id]
item.status = TaskStatus.COMPLETED
item.update_display()
self._log_to_task(task_id, f"Skipped: {reason}" if reason else "Skipped")
if task_id == self.selected_task:
self._update_main_panel()
def action_refresh(self) -> None:
"""Refresh the dashboard."""
self._update_main_panel()
def action_cursor_up(self) -> None:
"""Move cursor up in task list."""
task_list = self.query_one("#task-list", ListView)
task_list.action_cursor_up()
def action_cursor_down(self) -> None:
"""Move cursor down in task list."""
task_list = self.query_one("#task-list", ListView)
task_list.action_cursor_down()
def action_sync_now(self) -> None:
"""Trigger an immediate sync."""
if self._sync_callback:
asyncio.create_task(self._run_sync_callback())
else:
self._log_to_task("archive", "No sync callback configured")
async def _run_sync_callback(self) -> None:
"""Run the sync callback if set."""
if self._sync_callback:
if asyncio.iscoroutinefunction(self._sync_callback):
await self._sync_callback()
else:
self._sync_callback()
def action_increase_interval(self) -> None:
"""Increase sync interval by 1 minute."""
self.sync_interval = min(self.sync_interval + 60, 3600) # Max 1 hour
self._update_countdown_display()
self._log_to_task(
self.selected_task,
f"Sync interval: {self.sync_interval // 60} min",
)
def action_decrease_interval(self) -> None:
"""Decrease sync interval by 1 minute."""
self.sync_interval = max(self.sync_interval - 60, 60) # Min 1 minute
self._update_countdown_display()
self._log_to_task(
self.selected_task,
f"Sync interval: {self.sync_interval // 60} min",
)
def set_sync_callback(self, callback: Callable) -> None:
"""Set the callback to run when sync is triggered."""
self._sync_callback = callback
def schedule_next_sync(self) -> None:
"""Schedule the next sync time."""
import time
self.next_sync_time = time.time() + self.sync_interval
def reset_all_tasks(self) -> None:
"""Reset all tasks to pending state."""
for task_id, item in self._task_items.items():
item.status = TaskStatus.PENDING
item.progress = 0
item.update_display()
self._update_main_panel()
async def _update_countdown(self) -> None:
"""Update the countdown timer every second."""
import time
while True:
try:
self._update_countdown_display()
await asyncio.sleep(1)
except asyncio.CancelledError:
break
except Exception:
await asyncio.sleep(1)
def _update_countdown_display(self) -> None:
"""Update the countdown display widget."""
import time
try:
countdown_widget = self.query_one("#countdown", Static)
remaining = max(0, self.next_sync_time - time.time())
if remaining <= 0:
countdown_widget.update(f"Syncing... ({self.sync_interval // 60}m)")
else:
minutes = int(remaining // 60)
seconds = int(remaining % 60)
countdown_widget.update(
f"Next: {minutes:02d}:{seconds:02d} ({self.sync_interval // 60}m)"
)
except Exception:
pass
async def _animate_spinners(self) -> None:
"""Animate spinners for running tasks."""
while True:
try:
# Update all running task spinners
for task_id, item in self._task_items.items():
if item.status == TaskStatus.RUNNING:
item.advance_spinner()
item.update_display()
await asyncio.sleep(0.08) # ~12 FPS for smooth animation
except asyncio.CancelledError:
break
except Exception:
await asyncio.sleep(0.08)
class SyncProgressTracker:
"""Track sync progress and update the dashboard."""
def __init__(self, dashboard: SyncDashboard):
self.dashboard = dashboard
def start_task(self, task_id: str, total: int = 100) -> None:
"""Start tracking a task."""
self.dashboard.start_task(task_id, total)
def update_task(self, task_id: str, progress: int, message: str = "") -> None:
"""Update task progress."""
self.dashboard.update_task(task_id, progress, message)
def complete_task(self, task_id: str, message: str = "") -> None:
"""Mark a task as complete."""
self.dashboard.complete_task(task_id, message)
def error_task(self, task_id: str, error: str) -> None:
"""Mark a task as failed."""
self.dashboard.error_task(task_id, error)
def skip_task(self, task_id: str, reason: str = "") -> None:
"""Mark a task as skipped."""
self.dashboard.skip_task(task_id, reason)
# Global dashboard instance
_dashboard_instance: Optional[SyncDashboard] = None
_progress_tracker: Optional[SyncProgressTracker] = None
def get_dashboard() -> Optional[SyncDashboard]:
"""Get the global dashboard instance."""
global _dashboard_instance
return _dashboard_instance
def get_progress_tracker() -> Optional[SyncProgressTracker]:
"""Get the global progress_tracker"""
global _progress_tracker
return _progress_tracker
async def run_dashboard_sync():
"""Run sync with dashboard UI."""
global _dashboard_instance, _progress_tracker
dashboard = SyncDashboard()
tracker = SyncProgressTracker(dashboard)
_dashboard_instance = dashboard
_progress_tracker = tracker
async def do_sync():
"""Run the actual sync process."""
try:
# Reset all tasks before starting
dashboard.reset_all_tasks()
# Simulate sync progress for demo (replace with actual sync calls)
# Stage 1: Sync local changes to server
# Archive mail
tracker.start_task("archive", 100)
tracker.update_task("archive", 50, "Scanning for archived messages...")
await asyncio.sleep(0.3)
tracker.update_task("archive", 100, "Moving 3 messages to archive...")
await asyncio.sleep(0.2)
tracker.complete_task("archive", "3 messages archived")
# Outbox
tracker.start_task("outbox", 100)
tracker.update_task("outbox", 50, "Checking outbox...")
await asyncio.sleep(0.2)
tracker.complete_task("outbox", "No pending emails")
# Stage 2: Fetch from server
# Inbox sync
tracker.start_task("inbox", 100)
for i in range(0, 101, 20):
tracker.update_task("inbox", i, f"Fetching emails... {i}%")
await asyncio.sleep(0.3)
tracker.complete_task("inbox", "150 emails processed")
# Calendar sync
tracker.start_task("calendar", 100)
for i in range(0, 101, 25):
tracker.update_task("calendar", i, f"Syncing events... {i}%")
await asyncio.sleep(0.3)
tracker.complete_task("calendar", "25 events synced")
# Stage 3: Task management
# Godspeed sync
tracker.start_task("godspeed", 100)
for i in range(0, 101, 33):
tracker.update_task(
"godspeed", min(i, 100), f"Syncing tasks... {min(i, 100)}%"
)
await asyncio.sleep(0.3)
tracker.complete_task("godspeed", "42 tasks synced")
# Task sweep
tracker.start_task("sweep")
tracker.update_task("sweep", 50, "Scanning notes directory...")
await asyncio.sleep(0.2)
tracker.skip_task("sweep", "Before 6 PM, skipping daily sweep")
# Schedule next sync
dashboard.schedule_next_sync()
except Exception as e:
tracker.error_task("archive", str(e))
# Set the sync callback so 's' key triggers it
dashboard.set_sync_callback(do_sync)
async def sync_loop():
"""Run sync on interval."""
import time
# Wait for the dashboard to be mounted before updating widgets
await dashboard._mounted.wait()
# Run initial sync
await do_sync()
# Then loop waiting for next sync time
while True:
try:
remaining = dashboard.next_sync_time - time.time()
if remaining <= 0:
await do_sync()
else:
await asyncio.sleep(1)
except asyncio.CancelledError:
break
except Exception:
await asyncio.sleep(1)
# Run dashboard and sync loop concurrently
await asyncio.gather(dashboard.run_async(), sync_loop())