Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions snakesee/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ def watch(
weighting_strategy: Literal["index", "time"] = "index",
half_life_logs: int = 10,
half_life_days: float = 7.0,
colorblind: bool = False,
) -> None:
"""
Watch a Snakemake workflow in real-time with a TUI dashboard.
Expand All @@ -88,6 +89,8 @@ def watch(
half_life_days: Half-life in days for time-based weighting.
After this many days, a run's weight is halved. Default: 7.0.
Only used when weighting_strategy="time".
colorblind: Use colorblind-accessible mode with distinct characters in
progress bars. Can also be toggled with 'a' key in TUI.
"""
try:
workflow_dir = _validate_workflow_dir(workflow_dir)
Expand Down Expand Up @@ -115,6 +118,9 @@ def watch(
profile_path = profile or find_profile(workflow_dir)

from snakesee.tui import WorkflowMonitorTUI
from snakesee.tui.accessibility import ACCESSIBLE_CONFIG

accessibility_config = ACCESSIBLE_CONFIG if colorblind else None

tui = WorkflowMonitorTUI(
workflow_dir=workflow_dir,
Expand All @@ -125,6 +131,7 @@ def watch(
weighting_strategy=weighting_strategy,
half_life_logs=half_life_logs,
half_life_days=half_life_days,
accessibility_config=accessibility_config,
)
tui.run()

Expand Down
6 changes: 6 additions & 0 deletions snakesee/tui/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@
"""

# Re-export public API for backward compatibility
from snakesee.tui.accessibility import ACCESSIBLE_CONFIG
from snakesee.tui.accessibility import DEFAULT_CONFIG
from snakesee.tui.accessibility import AccessibilityConfig
from snakesee.tui.monitor import DEFAULT_REFRESH_RATE
from snakesee.tui.monitor import FG_BLUE
from snakesee.tui.monitor import FG_GREEN
Expand All @@ -37,6 +40,9 @@
from snakesee.tui.monitor import WorkflowMonitorTUI

__all__ = [
"ACCESSIBLE_CONFIG",
"AccessibilityConfig",
"DEFAULT_CONFIG",
"DEFAULT_REFRESH_RATE",
"FG_BLUE",
"FG_GREEN",
Expand Down
59 changes: 59 additions & 0 deletions snakesee/tui/accessibility.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
"""Accessibility configuration for colorblind-friendly rendering.

Provides alternative visual encodings so that progress bar status
can be distinguished without relying on color perception alone.
"""

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class BarStyle:
"""Character and label for a single progress bar segment.

Attributes:
char: The character used to fill the segment.
label: Human-readable label for the legend.
"""

char: str
label: str


@dataclass(frozen=True, slots=True)
class AccessibilityConfig:
"""Visual encoding configuration for progress bar rendering.

Controls which characters are used for each segment of the progress bar
and whether the legend is always displayed.

Attributes:
succeeded: Style for completed/succeeded jobs.
failed: Style for failed jobs.
remaining: Style for remaining/pending jobs.
incomplete: Style for incomplete jobs (workflow interrupted).
show_legend: If True, always show the legend (not just on failure).
"""

succeeded: BarStyle
failed: BarStyle
remaining: BarStyle
incomplete: BarStyle
show_legend: bool


DEFAULT_CONFIG = AccessibilityConfig(
succeeded=BarStyle(char="\u2588", label="succeeded"),
failed=BarStyle(char="\u2588", label="failed"),
remaining=BarStyle(char="\u2591", label="remaining"),
incomplete=BarStyle(char="\u2591", label="incomplete"),
show_legend=False,
)

ACCESSIBLE_CONFIG = AccessibilityConfig(
succeeded=BarStyle(char="=", label="succeeded"),
failed=BarStyle(char="X", label="failed"),
remaining=BarStyle(char="\u00b7", label="remaining"),
incomplete=BarStyle(char="?", label="incomplete"),
show_legend=True,
)
76 changes: 61 additions & 15 deletions snakesee/tui/monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@
from snakesee.state.clock import get_clock
from snakesee.state.paths import WorkflowPaths
from snakesee.state.workflow_state import WorkflowState
from snakesee.tui.accessibility import ACCESSIBLE_CONFIG
from snakesee.tui.accessibility import DEFAULT_CONFIG
from snakesee.tui.accessibility import AccessibilityConfig
from snakesee.validation import EventAccumulator
from snakesee.validation import ValidationLogger
from snakesee.validation import compare_states
Expand Down Expand Up @@ -145,6 +148,7 @@ class WorkflowMonitorTUI:
p: Pause/resume auto-refresh
e: Toggle time estimation
w: Toggle wildcard conditioning (estimate per sample/batch)
a: Toggle colorblind-accessible mode
r: Force refresh
Ctrl+r: Hard refresh (reload historical data)

Expand Down Expand Up @@ -192,6 +196,7 @@ def __init__(
weighting_strategy: WeightingStrategy = "index",
half_life_logs: int = 10,
half_life_days: float = 7.0,
accessibility_config: AccessibilityConfig | None = None,
) -> None:
"""
Initialize the TUI.
Expand All @@ -205,6 +210,8 @@ def __init__(
weighting_strategy: Strategy for weighting historical data ("index" or "time").
half_life_logs: Half-life in run count for index-based weighting.
half_life_days: Half-life in days for time-based weighting.
accessibility_config: Visual encoding config for colorblind accessibility.
Defaults to DEFAULT_CONFIG (standard block characters).
"""
self.workflow_dir = workflow_dir
self.refresh_rate = refresh_rate
Expand All @@ -213,6 +220,7 @@ def __init__(
self.weighting_strategy = weighting_strategy
self.half_life_logs = half_life_logs
self.half_life_days = half_life_days
self._accessibility_config = accessibility_config or DEFAULT_CONFIG
self.console = Console()
self._running = True
self._estimator: TimeEstimator | None = None
Expand Down Expand Up @@ -1019,7 +1027,7 @@ def _handle_key(self, key: str) -> bool:
return False

def _handle_toggle_key(self, key: str) -> bool:
"""Handle toggle keys (?, p, e, w, r, Ctrl+r). Returns True if key was handled."""
"""Handle toggle keys (?, p, e, w, a, r, Ctrl+r). Returns True if handled."""
if key == "?":
self._show_help = True
self._force_refresh = True
Expand All @@ -1042,6 +1050,14 @@ def _handle_toggle_key(self, key: str) -> bool:
if key.lower() == "r":
self._force_refresh = True
return True
if key.lower() == "a":
# Toggle colorblind-accessible mode
if self._accessibility_config == DEFAULT_CONFIG:
self._accessibility_config = ACCESSIBLE_CONFIG
else:
self._accessibility_config = DEFAULT_CONFIG
self._force_refresh = True
return True
if key == "\x12": # Ctrl+r - hard refresh
self._init_estimator()
self._force_refresh = True
Expand Down Expand Up @@ -1559,6 +1575,7 @@ def _make_help_panel(self) -> Panel:
help_text.add_row("p", "Pause/resume auto-refresh")
help_text.add_row("e", "Toggle time estimation")
help_text.add_row("w", "Toggle wildcard conditioning")
help_text.add_row("a", "Toggle colorblind-accessible mode")
help_text.add_row("r", "Force refresh")
help_text.add_row("Ctrl+r", "Hard refresh (reload historical data)")
help_text.add_row("", "")
Expand Down Expand Up @@ -1648,20 +1665,28 @@ def _make_progress_bar(self, progress: WorkflowProgress, width: int = 40) -> Tex
total = max(1, progress.total_jobs)
succeeded = progress.completed_jobs
failed = progress.failed_jobs
config = self._accessibility_config

# Calculate widths for each segment
succeeded_width = int((succeeded / total) * width)
failed_width = int((failed / total) * width)
remaining_width = width - succeeded_width - failed_width
unfinished = max(0, progress.total_jobs - progress.completed_jobs - progress.failed_jobs)
incomplete = (
min(len(progress.incomplete_jobs_list), unfinished)
if progress.status == WorkflowStatus.INCOMPLETE
else 0
)
incomplete_width = int((incomplete / total) * width)
remaining_width = width - succeeded_width - failed_width - incomplete_width

# Build the bar with colored segments
bar = Text()
bar.append("█" * succeeded_width, style="green")
bar.append("█" * failed_width, style="red")
if progress.status == WorkflowStatus.INCOMPLETE:
bar.append("░" * remaining_width, style="yellow") # Incomplete = yellow
else:
bar.append("░" * remaining_width, style="dim")
bar.append(config.succeeded.char * succeeded_width, style="green")
bar.append(config.failed.char * failed_width, style="red")
if incomplete_width > 0:
bar.append(config.incomplete.char * incomplete_width, style="yellow")
if remaining_width > 0:
bar.append(config.remaining.char * remaining_width, style="dim")
Comment thread
nh13 marked this conversation as resolved.

return bar

Expand Down Expand Up @@ -1719,14 +1744,35 @@ def _make_progress_panel(

eta_text = Text.from_markup(" ".join(eta_parts)) if eta_parts else Text("")

# Legend for the progress bar when there are failures
# Legend for the progress bar
config = self._accessibility_config
legend = Text()
if progress.failed_jobs > 0:
show_legend = progress.failed_jobs > 0 or config.show_legend
if show_legend:
legend.append(" (", style="dim")
legend.append("█", style="green")
legend.append(f"={progress.completed_jobs} succeeded ", style="dim")
legend.append("█", style="red")
legend.append(f"={progress.failed_jobs} failed", style="dim")
legend.append(config.succeeded.char, style="green")
legend.append(f"={progress.completed_jobs} {config.succeeded.label}", style="dim")
if progress.failed_jobs > 0:
legend.append(" ", style="dim")
legend.append(config.failed.char, style="red")
legend.append(f"={progress.failed_jobs} {config.failed.label}", style="dim")
unfinished = max(
0, progress.total_jobs - progress.completed_jobs - progress.failed_jobs
)
incomplete = (
min(len(progress.incomplete_jobs_list), unfinished)
if progress.status == WorkflowStatus.INCOMPLETE
else 0
)
remaining = unfinished - incomplete
if incomplete > 0:
legend.append(" ", style="dim")
legend.append(config.incomplete.char, style="yellow")
legend.append(f"={incomplete} {config.incomplete.label}", style="dim")
if remaining > 0:
legend.append(" ", style="dim")
legend.append(config.remaining.char, style="dim")
legend.append(f"={remaining} {config.remaining.label}", style="dim")
legend.append(")", style="dim")

# Border color based on status (use FG colors for normal states)
Expand All @@ -1740,7 +1786,7 @@ def _make_progress_panel(
border_style = border_colors.get(progress.status, FG_BLUE)

# Combine progress line with legend if present
if progress.failed_jobs > 0:
if show_legend:
full_progress = Text()
full_progress.append(progress_line)
full_progress.append(legend)
Expand Down
22 changes: 22 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,28 @@ def test_watch_calls_tui(self, snakemake_dir: Path, tmp_path: Path) -> None:
weighting_strategy="index",
half_life_logs=10,
half_life_days=7.0,
accessibility_config=None,
)
mock_instance.run.assert_called_once()

def test_watch_calls_tui_in_accessible_mode(self, snakemake_dir: Path, tmp_path: Path) -> None:
"""Test that watch enables accessible rendering when requested."""
from snakesee.tui.accessibility import ACCESSIBLE_CONFIG

with patch("snakesee.tui.WorkflowMonitorTUI") as mock_tui:
mock_instance = mock_tui.return_value
watch(tmp_path, refresh=2.0, no_estimate=True, colorblind=True)

mock_tui.assert_called_once_with(
workflow_dir=tmp_path,
refresh_rate=2.0,
use_estimation=False,
profile_path=None,
use_wildcard_conditioning=True,
weighting_strategy="index",
half_life_logs=10,
half_life_days=7.0,
accessibility_config=ACCESSIBLE_CONFIG,
)
mock_instance.run.assert_called_once()

Expand Down
Loading
Loading