Skip to content
Open
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
37 changes: 24 additions & 13 deletions snakesee/tui/monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import time
from collections.abc import Iterable
from datetime import datetime
from datetime import timedelta
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING
Expand Down Expand Up @@ -214,6 +215,7 @@ def __init__(
Defaults to DEFAULT_CONFIG (standard block characters).
"""
self.workflow_dir = workflow_dir
self._resolved_workflow_dir = str(workflow_dir.resolve())
self.refresh_rate = refresh_rate
self.use_estimation = use_estimation
self.profile_path = profile_path
Expand Down Expand Up @@ -1246,10 +1248,8 @@ def _handle_table_navigation_key(self, key: str, num_jobs: int) -> bool: # noqa
Table mode: Navigate between jobs in running/completions tables.
Press Enter to view a job's log, Esc to exit to normal mode.
"""
# Help key works in table mode
if key == "?":
self._show_help = True
self._force_refresh = True
# Toggle keys (?, p, e, w, a, r, Ctrl+r) work in table mode
if self._handle_toggle_key(key):
return False

# Sort keys work in table mode
Expand Down Expand Up @@ -1447,10 +1447,8 @@ def _handle_log_viewing_key(self, key: str) -> bool:
Log mode: Scroll through the selected job's log file.
Press Esc to return to table navigation mode.
"""
# Help key works in log mode
if key == "?":
self._show_help = True
self._force_refresh = True
# Toggle keys (?, p, e, w, a, r, Ctrl+r) work in log mode
if self._handle_toggle_key(key):
return False

# Escape - exit log viewing mode, return to table navigation
Expand Down Expand Up @@ -1633,10 +1631,12 @@ def _make_help_panel(self) -> Panel:
help_text.add_row("Ctrl+f/b", "Scroll down/up full page")
help_text.add_row("Esc", "Return to table navigation")

from snakesee import __version__

return Panel(
help_text,
title="[bold]Keyboard Shortcuts[/bold]",
subtitle="Press any key to close",
subtitle=f"Press any key to close [dim]│ snakesee v{__version__}[/dim]",
border_style="cyan",
)

Expand All @@ -1656,7 +1656,13 @@ def _make_header(self, progress: WorkflowProgress) -> Panel:
header_text.append(" │ ", style="dim")
header_text.append("Snakemake Monitor", style="bold white")
header_text.append(" │ ", style="dim")
header_text.append(str(self.workflow_dir), style="dim")
# Truncate long paths to avoid crowding out status fields
resolved = self._resolved_workflow_dir
max_path_len = max(20, (self.console.width or 80) - 80)
if len(resolved) > max_path_len:
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: self.workflow_dir.resolve() does filesystem I/O (resolving symlinks) and is called on every render cycle via _make_header. Consider resolving once at construction time (e.g. in __init__ or run()) and caching the result, so the per-frame cost is just a string truncation.

# Keep first component + ... + last components that fit
resolved = resolved[: max_path_len // 2 - 1] + "…" + resolved[-(max_path_len // 2) :]
header_text.append(resolved, style="dim")
header_text.append(" │ Status: ")
header_text.append(progress.status.value.upper(), style=style)

Expand Down Expand Up @@ -1750,9 +1756,14 @@ def _make_progress_panel(
eta_parts.append(f"ETA: {estimate.format_eta()}")

if estimate.seconds_remaining < float("inf") and estimate.seconds_remaining > 0:
completion_time = datetime.now().timestamp() + estimate.seconds_remaining
completion_str = datetime.fromtimestamp(completion_time).strftime("%H:%M:%S")
eta_parts.append(f"(completion: {completion_str})")
now = datetime.now().astimezone()
completion_dt = now + timedelta(seconds=estimate.seconds_remaining)
tz_name = completion_dt.strftime("%Z") or "local"
if completion_dt.date() == now.date():
completion_str = completion_dt.strftime("%H:%M:%S")
else:
completion_str = completion_dt.strftime("%Y-%m-%d %H:%M:%S")
eta_parts.append(f"({completion_str} {tz_name})")

# Show estimation method and inferred cores for transparency
method_info = estimate.method
Expand Down
89 changes: 89 additions & 0 deletions tests/test_tui.py
Original file line number Diff line number Diff line change
Expand Up @@ -1639,6 +1639,95 @@ def test_help_closes_before_job_selection_handles_key(self, tui: WorkflowMonitor
# Job index should NOT have changed (key was consumed by help close)
assert tui._selected_job_index == 5

def test_toggle_keys_work_in_table_mode(self, tui: WorkflowMonitorTUI) -> None:
"""Test that toggle keys (p, e, w, a) work in table navigation mode."""
tui._job_selection_mode = True
tui._log_source = "running"

# p toggles pause
assert not tui._paused
tui._handle_key("p")
assert tui._paused
assert tui._job_selection_mode is True # type: ignore[unreachable] # stays in table mode

# e toggles estimation
initial_est = tui.use_estimation
tui._handle_key("e")
assert tui.use_estimation != initial_est
assert tui._job_selection_mode is True

# w toggles wildcard conditioning
initial_wc = tui._use_wildcard_conditioning
tui._handle_key("w")
assert tui._use_wildcard_conditioning != initial_wc
assert tui._job_selection_mode is True

# a toggles accessibility
from snakesee.tui.accessibility import ACCESSIBLE_CONFIG
from snakesee.tui.accessibility import DEFAULT_CONFIG

assert tui._accessibility_config == DEFAULT_CONFIG
tui._handle_key("a")
assert tui._accessibility_config == ACCESSIBLE_CONFIG
assert tui._job_selection_mode is True

# r triggers force refresh without leaving table mode
tui._force_refresh = False
tui._handle_key("r")
assert tui._force_refresh is True
assert tui._job_selection_mode is True

# Ctrl+r triggers hard refresh without leaving table mode
tui._force_refresh = False
tui._handle_key("\x12")
assert tui._force_refresh is True
assert tui._job_selection_mode is True

def test_toggle_keys_work_in_log_viewing_mode(self, tui: WorkflowMonitorTUI) -> None:
"""Test that toggle keys (p, e, w, a) work in log viewing mode."""
tui._job_selection_mode = True
tui._log_viewing_mode = True
tui._log_source = "running"

# p toggles pause
assert not tui._paused
tui._handle_key("p")
assert tui._paused
assert tui._log_viewing_mode is True # type: ignore[unreachable] # stays in log mode

# w toggles wildcard conditioning
initial_wc = tui._use_wildcard_conditioning
tui._handle_key("w")
assert tui._use_wildcard_conditioning != initial_wc
assert tui._log_viewing_mode is True

# e toggles estimation
initial_est = tui.use_estimation
tui._handle_key("e")
assert tui.use_estimation != initial_est
assert tui._log_viewing_mode is True

# a toggles accessibility
from snakesee.tui.accessibility import ACCESSIBLE_CONFIG
from snakesee.tui.accessibility import DEFAULT_CONFIG

assert tui._accessibility_config == DEFAULT_CONFIG
tui._handle_key("a")
assert tui._accessibility_config == ACCESSIBLE_CONFIG
assert tui._log_viewing_mode is True

# r triggers force refresh without leaving log mode
tui._force_refresh = False
tui._handle_key("r")
assert tui._force_refresh is True
assert tui._log_viewing_mode is True

# Ctrl+r triggers hard refresh without leaving log mode
tui._force_refresh = False
tui._handle_key("\x12")
assert tui._force_refresh is True
assert tui._log_viewing_mode is True

def test_g_jumps_to_first_job_in_running(self, tui: WorkflowMonitorTUI) -> None:
"""Test that 'g' jumps to first job in running table."""
tui._job_selection_mode = True
Expand Down