UI API
This page documents the user interface components (Rich-based interactive dashboard).
UI State
Thread-safe state container for the dashboard.
state
SessionErrorEntry
dataclass
SessionErrorEntry(path: Path, size_bytes: Optional[int], width: Optional[int], height: Optional[int], fps: Optional[float], codec: Optional[str], audio_codec: Optional[str], duration_seconds: Optional[float], error_message: str, created_at: datetime)
Snapshot of a failed job captured for current-session Logs tab.
UIState
UIState(activity_feed_max_items: int = 5)
Thread-safe state manager for the interactive UI.
Source code in vbc/ui/state.py
| def __init__(self, activity_feed_max_items: int = 5):
self._lock = threading.RLock()
# Counters
self.completed_count = 0
self.failed_count = 0
self.skipped_count = 0
self.hw_cap_count = 0
self.cam_skipped_count = 0
self.min_ratio_skip_count = 0 # Files copied instead of compressed (ratio too low)
self.interrupted_count = 0 # Files interrupted by Ctrl+C
self.completed_count_at_last_discovery = 0
self.failed_count_at_last_discovery = 0
self.session_completed_base = 0
# Discovery counters (files skipped before processing)
self.files_to_process = 0
self.already_compressed_count = 0
self.ignored_small_count = 0
self.ignored_err_count = 0
self.ignored_av1_count = 0
# Bytes tracking
self.total_input_bytes = 0
self.total_output_bytes = 0
self.throughput_history: deque[Tuple[datetime, int]] = deque()
# Job lists
self.active_jobs: List[CompressionJob] = []
self.recent_jobs = deque(maxlen=activity_feed_max_items)
# Web dashboard can render variable-height Activity Feed,
# so keep a deeper history than TUI's compact feed window.
self.web_recent_jobs = deque(maxlen=max(80, activity_feed_max_items))
self.pending_files: List[Any] = [] # VideoFile objects waiting to be submitted
# Job timing tracking
self.job_start_times: Dict[str, datetime] = {} # filename -> start time
# Global Status
self.discovery_finished = False
self.discovery_finished_time: Optional[datetime] = None
self.total_files_found = 0
self.current_threads = 0
self.source_folders_count = 1
self.shutdown_requested = False
self.interrupt_requested = False
self.error_paused = False
self.error_status_text: Optional[str] = None
self.error_message: Optional[str] = None
self.finished = False
self.waiting_for_input = False
self.strip_unicode_display = True
self.ui_title = "VBC"
# Tabbed overlay state
self.show_overlay = False
self.active_tab = "shortcuts" # "shortcuts" | "settings" | "io" | "dirs" | "tui" | "reference" | "logs"
self.overlay_dim_level = "mid" # "light" | "mid" | "dark"
self.show_info = False
self.info_message = ""
self.config_lines: List[str] = []
self.io_input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]] = []
self.io_output_dir_lines: List[str] = []
self.io_errors_dir_lines: List[str] = []
self.io_suffix_output_dirs: Optional[str] = None
self.io_suffix_errors_dirs: Optional[str] = None
self.io_queue_sort: str = "name"
self.io_queue_seed: Optional[int] = None
self.log_path: Optional[str] = None
self.debug_enabled: bool = False
self.processing_start_time: Optional[datetime] = None
self.last_action: str = ""
self.last_action_time: Optional[datetime] = None
# Dirs tab state
self.dirs_cursor: int = 0
self.dirs_input_mode: bool = False
self.dirs_input_buffer: str = ""
self.dirs_pending_toggle: Dict[str, bool] = {} # path → new enabled state (True=enable, False=disable)
self.dirs_pending_add: List[str] = []
self.dirs_pending_remove: Set[str] = set()
self.dirs_pending_order: List[str] = [] # staged row order (applies on [S])
self.dirs_config_entries: List[Tuple[str, bool]] = [] # ordered config input_dirs: (path, enabled)
self.dirs_error_msg: str = "" # error shown inside Dirs overlay
# GPU Metrics
self.gpu_data: Optional[Dict[str, Any]] = None
# GPU Sparkline
self.gpu_sparkline_metric_idx: int = 0 # Index into active GPU sparkline metric order
self.gpu_sparkline_preset: str = DEFAULT_GPU_SPARKLINE_PRESET
self.gpu_sparkline_palette: str = DEFAULT_GPU_SPARKLINE_PALETTE
self.gpu_sparkline_mode: str = "sparkline" # sparkline | palette
self.gpu_history_temp: deque = deque(maxlen=60)
self.gpu_history_pwr: deque = deque(maxlen=60)
self.gpu_history_gpu: deque = deque(maxlen=60)
self.gpu_history_mem: deque = deque(maxlen=60)
self.gpu_history_fan: deque = deque(maxlen=60)
# Logs tab (current session only)
self.logs_page_size: int = 10
self.logs_page_index: int = 0
self.session_error_logs: List[SessionErrorEntry] = []
self._discovery_error_keys: set[Tuple[Path, str]] = set()
|
add_recent_job
add_recent_job(job: CompressionJob)
Thread-safe wrapper for adding job to recent activity feeds.
Source code in vbc/ui/state.py
| def add_recent_job(self, job: CompressionJob):
"""Thread-safe wrapper for adding job to recent activity feeds."""
with self._lock:
self._add_recent_job_unlocked(job)
|
set_last_action
set_last_action(action: str)
Set last action message with timestamp (like old vbc.py).
Source code in vbc/ui/state.py
| def set_last_action(self, action: str):
"""Set last action message with timestamp (like old vbc.py)."""
with self._lock:
self.last_action = action
self.last_action_time = datetime.now()
|
get_last_action
Get last action message (clears after 60 seconds, like old vbc.py).
Source code in vbc/ui/state.py
| def get_last_action(self) -> str:
"""Get last action message (clears after 60 seconds, like old vbc.py)."""
with self._lock:
if self.last_action and self.last_action_time:
elapsed = (datetime.now() - self.last_action_time).total_seconds()
if elapsed > 60: # Clear after 1 minute
self.last_action = ""
self.last_action_time = None
return self.last_action
|
dirs_effective_order
dirs_effective_order() -> List[str]
Return current row order used by Dirs UI (including staged reorder).
Source code in vbc/ui/state.py
| def dirs_effective_order(self) -> List[str]:
"""Return current row order used by Dirs UI (including staged reorder)."""
with self._lock:
return list(self._dirs_effective_order_unlocked())
|
dirs_set_pending_order
dirs_set_pending_order(ordered_paths: List[str]) -> None
Stage a new Dirs row order; clears staged order when equal to base order.
Source code in vbc/ui/state.py
| def dirs_set_pending_order(self, ordered_paths: List[str]) -> None:
"""Stage a new Dirs row order; clears staged order when equal to base order."""
with self._lock:
base = self._dirs_base_order_unlocked()
base_set = set(base)
normalized: List[str] = []
seen: Set[str] = set()
for path in ordered_paths:
if path in base_set and path not in seen:
normalized.append(path)
seen.add(path)
for path in base:
if path not in seen:
normalized.append(path)
if normalized == base:
self.dirs_pending_order = []
else:
self.dirs_pending_order = normalized
|
dirs_has_pending_changes
dirs_has_pending_changes() -> bool
Return True when Dirs tab contains any staged changes awaiting apply.
Source code in vbc/ui/state.py
| def dirs_has_pending_changes(self) -> bool:
"""Return True when Dirs tab contains any staged changes awaiting apply."""
with self._lock:
if self.dirs_pending_toggle or self.dirs_pending_add or self.dirs_pending_remove:
return True
return self._dirs_effective_order_unlocked() != self._dirs_base_order_unlocked()
|
dirs_get_all_entries
dirs_get_all_entries() -> List[Tuple[str, str, Optional[int], Optional[int], Optional[str]]]
Return combined list of (path, status, file_count, size_bytes, fs_status) for Dirs tab.
Status values:
- "active" → green — currently active dir
- "disabled" → dim — currently disabled dir
- "pending_add" → yellow — staged for adding
- "pending_remove" → red — staged for deletion
- "pending_toggle_off"→ yellow — active, pending disable
- "pending_toggle_on" → yellow — disabled, pending enable
fs_status: "ok" | "missing" | "no_access" | None (for disabled/pending)
Source code in vbc/ui/state.py
| def dirs_get_all_entries(self) -> List[Tuple[str, str, Optional[int], Optional[int], Optional[str]]]:
"""Return combined list of (path, status, file_count, size_bytes, fs_status) for Dirs tab.
Status values:
- "active" → green — currently active dir
- "disabled" → dim — currently disabled dir
- "pending_add" → yellow — staged for adding
- "pending_remove" → red — staged for deletion
- "pending_toggle_off"→ yellow — active, pending disable
- "pending_toggle_on" → yellow — disabled, pending enable
fs_status: "ok" | "missing" | "no_access" | None (for disabled/pending)
"""
with self._lock:
entries: List[Tuple[str, str, Optional[int], Optional[int], Optional[str]]] = []
from vbc.config.input_dirs import STATUS_OK, STATUS_MISSING
stats_by_path = {entry: (fs, count, size) for fs, entry, count, size in self.io_input_dir_stats}
enabled_by_path = {path: enabled for path, enabled in self.dirs_config_entries}
effective_order = self._dirs_effective_order_unlocked()
pending_add_set = set(self.dirs_pending_add)
for path in effective_order:
if path in enabled_by_path:
enabled = enabled_by_path[path]
fs_entry = stats_by_path.get(path)
if fs_entry is not None:
fs, count, size = fs_entry
fs_status = "ok" if fs == STATUS_OK else ("missing" if fs == STATUS_MISSING else "no_access")
else:
import os as _os
p = Path(path)
if not p.exists():
fs_status = "missing"
elif not _os.access(p, _os.R_OK | _os.X_OK):
fs_status = "no_access"
else:
fs_status = "ok"
count, size = None, None
if path in self.dirs_pending_remove:
entries.append((path, "pending_remove", count, size, fs_status))
continue
pending_enabled = self.dirs_pending_toggle.get(path)
if pending_enabled is not None:
if pending_enabled:
entries.append((path, "pending_toggle_on", count, size, fs_status))
else:
entries.append((path, "pending_toggle_off", count, size, fs_status))
continue
entries.append((path, "active" if enabled else "disabled", count, size, fs_status))
continue
if path in pending_add_set:
entries.append((path, "pending_add", None, None, None))
return entries
|
open_overlay
open_overlay(tab: Optional[str] = None) -> None
Open overlay, optionally on a specific tab.
Source code in vbc/ui/state.py
| def open_overlay(self, tab: Optional[str] = None) -> None:
"""Open overlay, optionally on a specific tab."""
with self._lock:
self.show_overlay = True
if tab and tab in self.OVERLAY_TABS:
self.active_tab = tab
|
close_overlay
Close overlay.
Source code in vbc/ui/state.py
| def close_overlay(self) -> None:
"""Close overlay."""
with self._lock:
self.show_overlay = False
|
toggle_overlay
toggle_overlay(tab: Optional[str] = None) -> None
Toggle overlay. If open on different tab, switch tabs.
Source code in vbc/ui/state.py
| def toggle_overlay(self, tab: Optional[str] = None) -> None:
"""Toggle overlay. If open on different tab, switch tabs."""
with self._lock:
if not self.show_overlay:
# Closed → Open
self.show_overlay = True
if tab:
self.active_tab = tab
elif tab and self.active_tab != tab:
# Open on different tab → Switch tab
self.active_tab = tab
else:
# Open on same tab → Close
self.show_overlay = False
|
cycle_tab
cycle_tab(direction: int = 1) -> None
Cycle through tabs. direction: 1=next, -1=previous.
Source code in vbc/ui/state.py
| def cycle_tab(self, direction: int = 1) -> None:
"""Cycle through tabs. direction: 1=next, -1=previous."""
with self._lock:
if not self.show_overlay:
# Closed → Open on first tab
self.show_overlay = True
return
current_idx = self.OVERLAY_TABS.index(self.active_tab)
next_idx = (current_idx + direction) % len(self.OVERLAY_TABS)
self.active_tab = self.OVERLAY_TABS[next_idx]
|
cycle_overlay_dim_level
cycle_overlay_dim_level(direction: int = 1) -> None
Cycle overlay dim level. direction: 1=next, -1=previous.
Source code in vbc/ui/state.py
| def cycle_overlay_dim_level(self, direction: int = 1) -> None:
"""Cycle overlay dim level. direction: 1=next, -1=previous."""
with self._lock:
current_idx = self.OVERLAY_DIM_LEVELS.index(self.overlay_dim_level)
next_idx = (current_idx + direction) % len(self.OVERLAY_DIM_LEVELS)
self.overlay_dim_level = self.OVERLAY_DIM_LEVELS[next_idx]
|
add_session_error
add_session_error(job: CompressionJob, error_message: str) -> None
Store failed job snapshot for current-session Logs tab.
Source code in vbc/ui/state.py
| def add_session_error(self, job: CompressionJob, error_message: str) -> None:
"""Store failed job snapshot for current-session Logs tab."""
with self._lock:
metadata = job.source_file.metadata
self._append_session_error_entry(
path=job.source_file.path,
size_bytes=job.source_file.size_bytes if job.source_file else None,
width=metadata.width if metadata else None,
height=metadata.height if metadata else None,
fps=metadata.fps if metadata else None,
codec=metadata.codec if metadata else None,
audio_codec=metadata.audio_codec if metadata else None,
duration_seconds=metadata.duration if metadata else None,
error_message=error_message or (job.error_message or "Unknown error"),
)
|
add_discovery_error
add_discovery_error(path: Path, size_bytes: Optional[int], error_message: str) -> None
Store discovery-time .err marker entry for current-session Logs tab.
Source code in vbc/ui/state.py
| def add_discovery_error(self, path: Path, size_bytes: Optional[int], error_message: str) -> None:
"""Store discovery-time `.err` marker entry for current-session Logs tab."""
with self._lock:
normalized_error = self._normalize_error_message(error_message)
key = (path, normalized_error)
if key in self._discovery_error_keys:
return
self._discovery_error_keys.add(key)
self._append_session_error_entry(
path=path,
size_bytes=size_bytes,
width=None,
height=None,
fps=None,
codec=None,
audio_codec=None,
duration_seconds=None,
error_message=normalized_error,
)
|
cycle_logs_page
cycle_logs_page(direction: int) -> None
Navigate logs pages. direction: 1=next, -1=prev.
Source code in vbc/ui/state.py
| def cycle_logs_page(self, direction: int) -> None:
"""Navigate logs pages. direction: 1=next, -1=prev."""
with self._lock:
total_pages = self.logs_total_pages()
next_page = self.logs_page_index + direction
if next_page < 0:
next_page = 0
if next_page > total_pages - 1:
next_page = total_pages - 1
self.logs_page_index = next_page
|
get_logs_page
get_logs_page() -> Tuple[List[SessionErrorEntry], int, int, int]
Return (entries, page_index, total_pages, total_entries).
Source code in vbc/ui/state.py
| def get_logs_page(self) -> Tuple[List[SessionErrorEntry], int, int, int]:
"""Return (entries, page_index, total_pages, total_entries)."""
with self._lock:
total_entries = len(self.session_error_logs)
total_pages = self.logs_total_pages()
page_index = min(self.logs_page_index, total_pages - 1)
page_size = self.logs_page_size
start = page_index * page_size
end = start + page_size
entries = list(self.session_error_logs[start:end])
return entries, page_index, total_pages, total_entries
|
UI Manager
Event subscriber that updates UIState based on domain events.
manager
UIManager
UIManager(bus: EventBus, state: UIState, demo_mode: bool = False, config_path: Optional[Path] = None)
Subscribes to EventBus and updates UIState.
Source code in vbc/ui/manager.py
| def __init__(self, bus: EventBus, state: UIState, demo_mode: bool = False, config_path: Optional[Path] = None):
self.bus = bus
self.state = state
self.demo_mode = demo_mode
self.config_path = config_path
self._setup_subscriptions()
|
on_toggle_overlay_tab
on_toggle_overlay_tab(event: ToggleOverlayTab)
Handle overlay toggle with optional tab selection.
Source code in vbc/ui/manager.py
| def on_toggle_overlay_tab(self, event: ToggleOverlayTab):
"""Handle overlay toggle with optional tab selection."""
self.state.toggle_overlay(event.tab)
|
on_cycle_overlay_tab
on_cycle_overlay_tab(event: CycleOverlayTab)
Handle tab cycling.
Source code in vbc/ui/manager.py
| def on_cycle_overlay_tab(self, event: CycleOverlayTab):
"""Handle tab cycling."""
self.state.cycle_tab(event.direction)
|
on_close_overlay
on_close_overlay(event: CloseOverlay)
Handle overlay close.
Source code in vbc/ui/manager.py
| def on_close_overlay(self, event: CloseOverlay):
"""Handle overlay close."""
self.state.close_overlay()
|
on_cycle_overlay_dim
on_cycle_overlay_dim(event: CycleOverlayDim)
Cycle overlay background dimming level (TUI tab only).
Source code in vbc/ui/manager.py
| def on_cycle_overlay_dim(self, event: CycleOverlayDim):
"""Cycle overlay background dimming level (TUI tab only)."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "tui":
return
self.state.cycle_overlay_dim_level(event.direction)
self.state.set_last_action(f"TUI: Overlay dim {self.state.overlay_dim_level}")
|
on_rotate_gpu_metric
on_rotate_gpu_metric(event: RotateGpuMetric)
Rotate GPU sparkline metric.
Source code in vbc/ui/manager.py
| def on_rotate_gpu_metric(self, event: RotateGpuMetric):
"""Rotate GPU sparkline metric."""
spark_cfg = get_gpu_sparkline_config(self.state.gpu_sparkline_preset)
metric_names = [metric.display_label for metric in spark_cfg.metrics]
if not metric_names:
return
with self.state._lock:
self.state.gpu_sparkline_metric_idx = (
self.state.gpu_sparkline_metric_idx + 1
) % len(metric_names)
current_name = metric_names[self.state.gpu_sparkline_metric_idx]
self.state.set_last_action(f"GPU Graph: {current_name}")
|
on_cycle_sparkline_preset
on_cycle_sparkline_preset(event: CycleSparklinePreset)
Cycle GPU sparkline preset (TUI tab only).
Source code in vbc/ui/manager.py
| def on_cycle_sparkline_preset(self, event: CycleSparklinePreset):
"""Cycle GPU sparkline preset (TUI tab only)."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "tui":
return
presets = list_gpu_sparkline_presets()
if not presets:
return
if self.state.gpu_sparkline_mode != "sparkline":
next_preset = presets[0]
self.state.gpu_sparkline_mode = "sparkline"
else:
try:
current_idx = presets.index(self.state.gpu_sparkline_preset)
except ValueError:
current_idx = 0
next_idx = (current_idx + event.direction) % len(presets)
next_preset = presets[next_idx]
self.state.gpu_sparkline_preset = next_preset
spark_cfg = get_gpu_sparkline_config(next_preset)
if spark_cfg.metrics:
self.state.gpu_sparkline_metric_idx %= len(spark_cfg.metrics)
else:
self.state.gpu_sparkline_metric_idx = 0
preset_label = format_preset_label(next_preset, spark_cfg)
self.state.set_last_action(f"Sparkline: {preset_label}")
|
on_cycle_sparkline_palette
on_cycle_sparkline_palette(event: CycleSparklinePalette)
Cycle GPU sparkline palette (TUI tab only).
Source code in vbc/ui/manager.py
| def on_cycle_sparkline_palette(self, event: CycleSparklinePalette):
"""Cycle GPU sparkline palette (TUI tab only)."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "tui":
return
palettes = list_gpu_sparkline_palettes()
if not palettes:
return
if self.state.gpu_sparkline_mode != "palette":
next_palette = palettes[0]
self.state.gpu_sparkline_mode = "palette"
else:
try:
current_idx = palettes.index(self.state.gpu_sparkline_palette)
except ValueError:
current_idx = 0
next_idx = (current_idx + event.direction) % len(palettes)
next_palette = palettes[next_idx]
self.state.gpu_sparkline_palette = next_palette
palette = get_gpu_sparkline_palette(next_palette)
self.state.set_last_action(f"Palette: {palette.display_label}")
|
on_cycle_logs_page
on_cycle_logs_page(event: CycleLogsPage)
Cycle page in Logs tab only.
Source code in vbc/ui/manager.py
| def on_cycle_logs_page(self, event: CycleLogsPage):
"""Cycle page in Logs tab only."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "logs":
return
self.state.cycle_logs_page(event.direction)
|
on_action_message
on_action_message(event: ActionMessage)
Handle user action feedback messages (like old vbc.py).
Source code in vbc/ui/manager.py
| def on_action_message(self, event: ActionMessage):
"""Handle user action feedback messages (like old vbc.py)."""
self.state.set_last_action(event.message)
|
on_dirs_cursor_move
on_dirs_cursor_move(event: DirsCursorMove)
Move cursor in the Dirs tab directory list.
Source code in vbc/ui/manager.py
| def on_dirs_cursor_move(self, event: DirsCursorMove):
"""Move cursor in the Dirs tab directory list."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "dirs":
return
entries = self.state.dirs_get_all_entries()
if not entries:
return
new_pos = self.state.dirs_cursor + event.direction
self.state.dirs_cursor = max(0, min(new_pos, len(entries) - 1))
|
on_dirs_toggle_selected
on_dirs_toggle_selected(event: DirsToggleSelected)
Toggle enabled/disabled state of the dir under cursor.
Source code in vbc/ui/manager.py
| def on_dirs_toggle_selected(self, event: DirsToggleSelected):
"""Toggle enabled/disabled state of the dir under cursor."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "dirs":
return
entries = self.state.dirs_get_all_entries()
if not entries or self.state.dirs_cursor >= len(entries):
return
entry = entries[self.state.dirs_cursor]
path, status = entry[0], entry[1]
fs_status = entry[4] if len(entry) > 4 else None
# Cannot toggle pending_add or pending_remove entries
if status in ("pending_add", "pending_remove"):
return
if status in ("active", "pending_toggle_off"):
# Toggle between active and pending-disable
if status == "pending_toggle_off":
del self.state.dirs_pending_toggle[path]
else:
self.state.dirs_pending_toggle[path] = False
elif status in ("disabled", "pending_toggle_on"):
# Block enabling if path doesn't exist or isn't accessible
from pathlib import Path as _Path
import os as _os
p = _Path(path)
if not p.exists():
self.state.dirs_error_msg = "Cannot enable: path does not exist"
return
if not _os.access(p, _os.R_OK | _os.X_OK):
self.state.dirs_error_msg = "Cannot enable: no read access"
return
self.state.dirs_error_msg = ""
# Toggle between disabled and pending-enable
if status == "pending_toggle_on":
del self.state.dirs_pending_toggle[path]
else:
self.state.dirs_pending_toggle[path] = True
|
on_dirs_swap_selected
on_dirs_swap_selected(event: DirsSwapSelected)
Swap row under cursor with adjacent row in selected direction.
Source code in vbc/ui/manager.py
| def on_dirs_swap_selected(self, event: DirsSwapSelected):
"""Swap row under cursor with adjacent row in selected direction."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "dirs":
return
if self.state.dirs_input_mode:
return
entries = self.state.dirs_get_all_entries()
if not entries or self.state.dirs_cursor >= len(entries):
return
src_idx = self.state.dirs_cursor
dst_idx = src_idx + event.direction
if dst_idx < 0 or dst_idx >= len(entries):
return
ordered_paths = [entry[0] for entry in entries]
ordered_paths[src_idx], ordered_paths[dst_idx] = ordered_paths[dst_idx], ordered_paths[src_idx]
self.state.dirs_set_pending_order(ordered_paths)
self.state.dirs_cursor = dst_idx
|
on_dirs_enter_add_mode
on_dirs_enter_add_mode(event: DirsEnterAddMode)
Enter add-path input mode.
Source code in vbc/ui/manager.py
| def on_dirs_enter_add_mode(self, event: DirsEnterAddMode):
"""Enter add-path input mode."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "dirs":
return
self.state.dirs_input_mode = True
self.state.dirs_input_buffer = ""
|
on_dirs_mark_delete
on_dirs_mark_delete(event: DirsMarkDelete)
Mark the dir under cursor for deletion (or unmark if already marked).
Source code in vbc/ui/manager.py
| def on_dirs_mark_delete(self, event: DirsMarkDelete):
"""Mark the dir under cursor for deletion (or unmark if already marked)."""
with self.state._lock:
if not self.state.show_overlay or self.state.active_tab != "dirs":
return
entries = self.state.dirs_get_all_entries()
if not entries or self.state.dirs_cursor >= len(entries):
return
path, status = entries[self.state.dirs_cursor][:2]
if status == "pending_remove":
# Unmark
self.state.dirs_pending_remove.discard(path)
elif status == "pending_add":
# Cancel the pending add
if path in self.state.dirs_pending_add:
self.state.dirs_pending_add.remove(path)
# Cursor may be out of bounds after removal — clamp
new_max = max(0, len(self.state.dirs_get_all_entries()) - 1)
self.state.dirs_cursor = min(self.state.dirs_cursor, new_max)
else:
self.state.dirs_pending_remove.add(path)
|
on_dirs_input_char(event: DirsInputChar)
Append a character to (or backspace from) the add-path input buffer.
Source code in vbc/ui/manager.py
| def on_dirs_input_char(self, event: DirsInputChar):
"""Append a character to (or backspace from) the add-path input buffer."""
with self.state._lock:
if not self.state.dirs_input_mode:
return
if event.char == '\x7f':
self.state.dirs_input_buffer = self.state.dirs_input_buffer[:-1]
else:
self.state.dirs_input_buffer += event.char
|
on_dirs_confirm_add
on_dirs_confirm_add(event: DirsConfirmAdd)
Validate and stage the input buffer as a pending add.
Source code in vbc/ui/manager.py
| def on_dirs_confirm_add(self, event: DirsConfirmAdd):
"""Validate and stage the input buffer as a pending add."""
with self.state._lock:
if not self.state.dirs_input_mode:
return
path = self.state.dirs_input_buffer.strip()
self.state.dirs_input_mode = False
self.state.dirs_input_buffer = ""
if not path:
return
# Check total directory limit
total = len(self.state.dirs_config_entries) + len(self.state.dirs_pending_add)
if total >= DirsOverlay.MAX_DIRS:
self.state.set_last_action(
f"Dirs limit reached ({DirsOverlay.MAX_DIRS} max). Remove a directory first."
)
return
# Skip duplicates
existing = {entry for entry, _ in self.state.dirs_config_entries} | set(self.state.dirs_pending_add)
if path in existing:
self.state.set_last_action(f"Directory already in list: {path}")
return
self.state.dirs_pending_add.append(path)
|
on_dirs_cancel_input(event: DirsCancelInput)
Cancel add-path input mode.
Source code in vbc/ui/manager.py
| def on_dirs_cancel_input(self, event: DirsCancelInput):
"""Cancel add-path input mode."""
with self.state._lock:
self.state.dirs_input_mode = False
self.state.dirs_input_buffer = ""
|
on_dirs_apply_changes
on_dirs_apply_changes(event: DirsApplyChanges)
Apply all pending Dirs changes, persist to YAML, and trigger re-scan.
Source code in vbc/ui/manager.py
| def on_dirs_apply_changes(self, event: DirsApplyChanges):
"""Apply all pending Dirs changes, persist to YAML, and trigger re-scan."""
with self.state._lock:
current_entries = list(self.state.dirs_config_entries)
pending_toggle = dict(self.state.dirs_pending_toggle)
pending_add = list(self.state.dirs_pending_add)
pending_remove = set(self.state.dirs_pending_remove)
effective_order = self.state.dirs_effective_order()
previous_stats = {
entry: (label, count, size)
for label, entry, count, size in self.state.io_input_dir_stats
}
# Compute new ordered entries
enabled_by_path = {path: enabled for path, enabled in current_entries}
pending_add_set = set(pending_add)
new_entries: list = []
seen_paths: set[str] = set()
for path in effective_order:
if path in seen_paths or path in pending_remove:
continue
seen_paths.add(path)
if path in enabled_by_path:
new_entries.append((path, pending_toggle.get(path, enabled_by_path[path])))
elif path in pending_add_set:
new_entries.append((path, True))
# Include leftovers defensively (should rarely happen, but keeps behavior robust).
for path, enabled in current_entries:
if path in seen_paths or path in pending_remove:
continue
seen_paths.add(path)
new_entries.append((path, pending_toggle.get(path, enabled)))
for path in pending_add:
if path in seen_paths or path in pending_remove:
continue
seen_paths.add(path)
new_entries.append((path, True))
new_active = [path for path, enabled in new_entries if enabled]
# Persist to YAML
if self.config_path:
try:
from vbc.config.loader import save_dirs_config
serialized_entries = [
{"path": path, "enabled": enabled}
for path, enabled in new_entries
]
save_dirs_config(self.config_path, serialized_entries)
except Exception as exc:
self.state.set_last_action(f"ERROR saving config: {exc}")
return
# Update state
rebuilt_active_stats = []
for path, enabled in new_entries:
if not enabled:
continue
prev = previous_stats.get(path)
if prev is None:
rebuilt_active_stats.append((STATUS_OK, path, None, None))
else:
label, count, size = prev
rebuilt_active_stats.append((label, path, count, size))
with self.state._lock:
self.state.dirs_config_entries = new_entries
self.state.io_input_dir_stats = rebuilt_active_stats
self.state.dirs_pending_toggle.clear()
self.state.dirs_pending_add.clear()
self.state.dirs_pending_remove.clear()
self.state.dirs_pending_order.clear()
self.state.dirs_input_mode = False
self.state.dirs_input_buffer = ""
n_active = len(new_active)
n_disabled = len(new_entries) - n_active
self.state.set_last_action(
f"Dirs saved: {n_active} active, {n_disabled} disabled"
)
self.bus.publish(InputDirsChanged(active_dirs=new_active))
self.bus.publish(RefreshRequested())
|
Keyboard Listener
Non-blocking keyboard input handler.
keyboard
ToggleConfig
Bases: Event
DEPRECATED: Use ToggleOverlayTab instead. Event emitted when user toggles config display (Key 'C').
ToggleLegend
Bases: Event
DEPRECATED: Use ToggleOverlayTab instead. Event emitted when user toggles legend display (Key 'L').
Bases: Event
DEPRECATED: Use ToggleOverlayTab instead. Event emitted when user toggles menu display (Key 'M').
HideConfig
Bases: Event
DEPRECATED: Use CloseOverlay instead. Event emitted when user closes config display (Esc).
ToggleOverlayTab
Bases: Event
Event emitted to toggle overlay with optional tab selection.
CycleOverlayTab
Bases: Event
Event emitted to cycle through overlay tabs.
CycleLogsPage
Bases: Event
Event emitted to cycle logs page in Logs tab.
CloseOverlay
Bases: Event
Event emitted to close the overlay.
CycleOverlayDim
Bases: Event
Event emitted to cycle overlay background dimming level.
RotateGpuMetric
Bases: Event
Event emitted when user rotates GPU sparkline metric (Key 'G').
CycleSparklinePreset
Bases: Event
Event emitted to cycle GPU sparkline preset (Key 'W').
CycleSparklinePalette
Bases: Event
Event emitted to cycle GPU sparkline palette (Key 'P').
KeyboardListener
KeyboardListener(event_bus: EventBus, state: Optional[UIState] = None)
Listens for keyboard input in a background thread.
Source code in vbc/ui/keyboard.py
| def __init__(self, event_bus: EventBus, state: Optional["UIState"] = None):
self.event_bus = event_bus
self.state = state
self._stop_event = threading.Event()
self._thread: Optional[threading.Thread] = None
|
start
Starts the listener thread.
Source code in vbc/ui/keyboard.py
| def start(self):
"""Starts the listener thread."""
self._thread = threading.Thread(target=self._run, daemon=True)
self._thread.start()
|
stop
Stops the listener thread.
Source code in vbc/ui/keyboard.py
| def stop(self):
"""Stops the listener thread."""
self._stop_event.set()
if self._thread:
self._thread.join(timeout=1.0)
|
Dashboard
Rich Live dashboard with status, progress, activity, queue, active job, and overlay surfaces.
dashboard
Dashboard
Dashboard(state: UIState, panel_height_scale: float = 0.7, max_active_jobs: int = 8)
Adaptive UI implementation with dynamic density control.
Source code in vbc/ui/dashboard.py
| def __init__(self, state: UIState, panel_height_scale: float = 0.7, max_active_jobs: int = 8):
self.state = state
self.panel_height_scale = panel_height_scale # UI scale factor
self.max_active_jobs = max_active_jobs # Max jobs to reserve space for
self.console = Console()
self._live: Optional[Live] = None
self._refresh_thread: Optional[threading.Thread] = None
self._stop_refresh = threading.Event()
self._ui_lock = threading.Lock()
self._spinner_frame = 0
|
format_size(size: int) -> str
Format size: 123B, 1.2KB, 45.1MB, 3.2GB.
Source code in vbc/ui/dashboard.py
| def format_size(self, size: int) -> str:
"""Format size: 123B, 1.2KB, 45.1MB, 3.2GB."""
if size == 0:
return "0B"
units = ['B', 'KB', 'MB', 'GB', 'TB']
idx = 0
val = float(size)
while val >= 1024.0 and idx < len(units) - 1:
val /= 1024.0
idx += 1
if idx < 2: # B, KB -> no decimal usually, but let's stick to spec
if idx == 0:
return f"{int(val)}B"
return f"{val:.1f}KB"
return f"{val:.1f}{units[idx]}"
|
format_time(seconds: float) -> str
Format time: 59s, 01m 01s, 1h 01m.
Source code in vbc/ui/dashboard.py
| def format_time(self, seconds: float) -> str:
"""Format time: 59s, 01m 01s, 1h 01m."""
if seconds is None:
return "--:--"
if seconds < 60:
return f"{int(seconds)}s"
if seconds < 3600:
return f"{int(seconds // 60):02d}m {int(seconds % 60):02d}s"
return f"{int(seconds // 3600)}h {int((seconds % 3600) // 60):02d}m"
|
format_global_eta(seconds: float) -> str
Format global ETA: hh:mm or mm:ss.
Source code in vbc/ui/dashboard.py
| def format_global_eta(self, seconds: float) -> str:
"""Format global ETA: hh:mm or mm:ss."""
if seconds is None:
return "--:--"
if seconds < 60:
return f"{int(seconds):02d}s"
elif seconds < 3600:
return f"{int(seconds // 60):02d}m {int(seconds % 60):02d}s"
else:
return f"{int(seconds // 3600):02d}h {int((seconds % 3600) // 60):02d}m"
|
is_wide_char
is_wide_char(char: str) -> bool
Check if Unicode character is wide (takes 2 terminal columns).
Source code in vbc/ui/dashboard.py
| def is_wide_char(char: str) -> bool:
"""Check if Unicode character is wide (takes 2 terminal columns)."""
if not char:
return False
# East Asian Width categories: F(ull), W(ide) = 2 cols, others = 1 col
width = unicodedata.east_asian_width(char[0])
return width in ('F', 'W')
|
format_icon(icon: str) -> str
Format icon with appropriate spacing (wide chars don't need trailing space).
Source code in vbc/ui/dashboard.py
| def format_icon(icon: str) -> str:
"""Format icon with appropriate spacing (wide chars don't need trailing space)."""
if is_wide_char(icon):
return icon # No space needed (e.g., ⚡)
else:
return f"{icon} " # Add space (e.g., ✓ )
|
Modern Overlays
Tabbed overlay renderers for Settings, Shortcuts, I/O, Dirs, TUI, Reference, and Logs content.
modern_overlays
VBC Modernized Overlays
Nowoczesny, estetyczny design dla paneli CONFIG, REFERENCE (dawniej LEGEND), SHORTCUTS (dawniej MENU), I/O i TUI.
Używa Rich library z kartami, tabelami i hierarchiczną strukturą.
Koncepcja:
- Prefs (C) - konfiguracja sesji w kartach tematycznych
- Ref (E) - legenda statusów i symboli
- Keys (M) - skróty klawiszowe z podziałem funkcjonalnym
- I/O (F) - foldery i ustawienia kolejki
- TUI (T) - ustawienia interfejsu terminalowego
Wszystkie panele zachowują 100% obecnej funkcjonalności, ale prezentują
ją w bardziej przejrzysty i nowoczesny sposób.
SettingsOverlay
SettingsOverlay(config_lines: List[str], spinner_frame: int = 0, log_path: Optional[str] = None, debug_enabled: bool = False)
Panel ustawień sesji - wyświetla konfigurację w kartach tematycznych.
Karty:
- ENCODING: encoder, preset, quality, audio, fallback
- PROCESSING: threads, prefetch, queue sort, cpu threads
- QUALITY & FILTERS: dynamic quality, camera filter, skip AV1, rotation
- LOGGING: log path, debug flags
- METADATA & CLEANUP: exiftool, analysis, autorotate, cleanup flags
Source code in vbc/ui/modern_overlays.py
| def __init__(
self,
config_lines: List[str],
spinner_frame: int = 0,
log_path: Optional[str] = None,
debug_enabled: bool = False,
):
self.config_lines = config_lines
self.spinner_frame = spinner_frame
self.log_path = log_path
self.debug_enabled = debug_enabled
self._parsed = parse_config_lines(config_lines)
|
render
Returns complete Panel with footer (for backward compatibility).
Source code in vbc/ui/modern_overlays.py
| def render(self) -> Panel:
"""Returns complete Panel with footer (for backward compatibility)."""
footer = Text.from_markup(
f"[{COLORS['dim']}]Press [white on {COLORS['border']}] Esc [/] close • "
f"[white on {COLORS['border']}] E [/] Ref • "
f"[white on {COLORS['border']}] M [/] Keys[/]",
justify="center"
)
content_with_footer = Group(
self._render_content(),
"",
footer
)
return Panel(
content_with_footer,
title="[bold white]⚙ PREFS[/]",
subtitle=f"[{COLORS['dim']}][C] to toggle[/]",
border_style=COLORS['accent_green'],
box=ROUNDED,
padding=(1, 2),
)
|
IoOverlay
IoOverlay(config_lines: List[str], input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]], output_dir_lines: List[str], errors_dir_lines: List[str], suffix_output_dirs: Optional[str], suffix_errors_dirs: Optional[str], queue_sort: str, queue_seed: Optional[int])
Panel I/O - foldery i ustawienia kolejkowania.
Source code in vbc/ui/modern_overlays.py
| def __init__(
self,
config_lines: List[str],
input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]],
output_dir_lines: List[str],
errors_dir_lines: List[str],
suffix_output_dirs: Optional[str],
suffix_errors_dirs: Optional[str],
queue_sort: str,
queue_seed: Optional[int],
):
self.config_lines = config_lines
self.input_dir_stats = input_dir_stats
self.output_dir_lines = output_dir_lines
self.errors_dir_lines = errors_dir_lines
self.suffix_output_dirs = suffix_output_dirs
self.suffix_errors_dirs = suffix_errors_dirs
self.queue_sort = queue_sort
self.queue_seed = queue_seed
self._parsed = parse_config_lines(config_lines)
|
render
Returns complete Panel with footer (for backward compatibility).
Source code in vbc/ui/modern_overlays.py
| def render(self) -> Panel:
"""Returns complete Panel with footer (for backward compatibility)."""
footer = Text.from_markup(
f"[{COLORS['dim']}]Press [white on {COLORS['border']}] Esc [/] close • "
f"[white on {COLORS['border']}] C [/] Prefs • "
f"[white on {COLORS['border']}] E [/] Ref[/]",
justify="center"
)
content_with_footer = Group(
self._render_content(),
"",
footer
)
return Panel(
content_with_footer,
title="[bold white]📁 I/O[/]",
subtitle=f"[{COLORS['dim']}][F] to toggle[/]",
border_style=COLORS['accent_blue'],
box=ROUNDED,
padding=(1, 2),
)
|
ReferenceOverlay
ReferenceOverlay(spinner_frame: int = 0, sparkline_preset: Optional[str] = None, sparkline_palette: Optional[str] = None, sparkline_mode: str = 'sparkline')
Panel referencyjny - legenda statusów, spinnerów i GPU graph.
Sekcje:
- STATUS CODES: fail, err, hw_cap, skip, kept, small, av1, cam + symbole wyniku
- ACTIVE JOB INDICATORS: animowane spinnery (normalny vs rotation)
- GPU GRAPH: metryki, skale, symbole sparkline
Source code in vbc/ui/modern_overlays.py
| def __init__(
self,
spinner_frame: int = 0,
sparkline_preset: Optional[str] = None,
sparkline_palette: Optional[str] = None,
sparkline_mode: str = "sparkline",
):
self.spinner_frame = spinner_frame
self.sparkline_preset = sparkline_preset
self.sparkline_palette = sparkline_palette
self.sparkline_mode = sparkline_mode
|
render
Returns complete Panel with footer (for backward compatibility).
Source code in vbc/ui/modern_overlays.py
| def render(self) -> Panel:
"""Returns complete Panel with footer (for backward compatibility)."""
footer = Text.from_markup(
f"[{COLORS['dim']}]Press [white on {COLORS['border']}] Esc [/] close • "
f"[white on {COLORS['border']}] C [/] Prefs • "
f"[white on {COLORS['border']}] M [/] Keys[/]",
justify="center"
)
content_with_footer = Group(
self._render_content(),
"",
footer
)
return Panel(
content_with_footer,
title="[bold white]📖 REF[/]",
subtitle=f"[{COLORS['dim']}][E] to toggle[/]",
border_style=COLORS['accent_orange'],
box=ROUNDED,
padding=(1, 2),
)
|
ShortcutsOverlay
Panel skrótów klawiszowych - pogrupowane tematycznie.
Grupy:
- NAVIGATION: M, Esc, Ctrl+C
- PANELS: C, E, L, G
- JOB CONTROL: S, R, </>, </>
+ Quick Reference z kolorowymi badge'ami
render
Returns complete Panel with footer (for backward compatibility).
Source code in vbc/ui/modern_overlays.py
| def render(self) -> Panel:
"""Returns complete Panel with footer (for backward compatibility)."""
footer = Text.from_markup(
f"[{COLORS['dim']}]Press [white on {COLORS['border']}] Esc [/] close • "
f"[white on {COLORS['border']}] C [/] Prefs • "
f"[white on {COLORS['border']}] E [/] Ref[/]",
justify="center"
)
content_with_footer = Group(
self._render_content(),
"",
footer
)
return Panel(
content_with_footer,
title="[bold white]⌨ KEYS[/]",
subtitle=f"[{COLORS['dim']}][M] to toggle[/]",
border_style=COLORS['accent_cyan'],
box=ROUNDED,
padding=(1, 2),
)
|
TuiOverlay
TuiOverlay(dim_level: str = 'mid', sparkline_preset: Optional[str] = None, sparkline_palette: Optional[str] = None, sparkline_mode: str = 'sparkline')
Panel ustawien TUI - wyglad i zachowanie interfejsu.
Source code in vbc/ui/modern_overlays.py
| def __init__(
self,
dim_level: str = "mid",
sparkline_preset: Optional[str] = None,
sparkline_palette: Optional[str] = None,
sparkline_mode: str = "sparkline",
):
self.dim_level = dim_level
self.sparkline_preset = sparkline_preset
self.sparkline_palette = sparkline_palette
self.sparkline_mode = sparkline_mode
|
render
Returns complete Panel with footer (for backward compatibility).
Source code in vbc/ui/modern_overlays.py
| def render(self) -> Panel:
"""Returns complete Panel with footer (for backward compatibility)."""
footer = Text.from_markup(
f"[{COLORS['dim']}]Press [white on {COLORS['border']}] Esc [/] close • "
f"[white on {COLORS['border']}] I [/] Dim level • "
f"[white on {COLORS['border']}] W [/] Sparkline • "
f"[white on {COLORS['border']}] P [/] Palette[/]",
justify="center",
)
content_with_footer = Group(
self._render_content(),
"",
footer,
)
return Panel(
content_with_footer,
title=f"[bold white]{ICONS['tui']} TUI[/]",
subtitle=f"[{COLORS['dim']}][T] to toggle[/]",
border_style=COLORS['accent_purple'],
box=ROUNDED,
padding=(1, 2),
)
|
DirsOverlay
DirsOverlay(entries: List[Tuple[str, str, Optional[int], Optional[int]]], cursor: int, input_mode: bool, input_buffer: str, has_pending_changes: bool, suffix_output_dirs: Optional[str], suffix_errors_dirs: Optional[str], output_dir_lines: List[str], errors_dir_lines: List[str], error_msg: str = '')
Interactive directory manager panel (Dirs [D] tab).
Displays all input directories (active + disabled + pending changes) with
cursor navigation and status indicators. Output/Errors dir suffixes shown
read-only at the bottom.
Entry status labels
[ON] green — currently active directory
[OFF] dim — currently disabled directory
[ON*] yellow — pending add or pending enable
[DEL] red — pending deletion
[~] yellow — pending toggle (active→disabled or disabled→active)
Source code in vbc/ui/modern_overlays.py
| def __init__(
self,
entries: List[Tuple[str, str, Optional[int], Optional[int]]],
cursor: int,
input_mode: bool,
input_buffer: str,
has_pending_changes: bool,
suffix_output_dirs: Optional[str],
suffix_errors_dirs: Optional[str],
output_dir_lines: List[str],
errors_dir_lines: List[str],
error_msg: str = "",
):
self.entries = entries # (path, status, file_count, size_bytes)
self.cursor = max(0, min(cursor, max(0, len(entries) - 1)))
self.input_mode = input_mode
self.input_buffer = input_buffer
self.has_pending_changes = has_pending_changes
self.suffix_output_dirs = suffix_output_dirs
self.suffix_errors_dirs = suffix_errors_dirs
self.output_dir_lines = output_dir_lines
self.errors_dir_lines = errors_dir_lines
self.error_msg = error_msg
|
render
Returns complete Panel with footer (for backward compatibility).
Source code in vbc/ui/modern_overlays.py
| def render(self) -> Panel:
"""Returns complete Panel with footer (for backward compatibility)."""
return Panel(
self._render_content(),
title="[bold white]📁 DIRS[/]",
subtitle=f"[{COLORS['dim']}][D] to toggle[/]",
border_style=COLORS['accent_green'],
box=ROUNDED,
padding=(1, 2),
)
|
make_card
make_card(title: str, content: RenderableType, icon: str = '', title_color: str = 'cyan', width: Optional[int] = None) -> Panel
Tworzy estetyczną kartę z tytułem i zawartością.
Source code in vbc/ui/modern_overlays.py
| def make_card(title: str, content: RenderableType, icon: str = "",
title_color: str = "cyan", width: Optional[int] = None) -> Panel:
"""Tworzy estetyczną kartę z tytułem i zawartością."""
title_text = f"{icon} {title}" if icon else title
return Panel(
content,
title=f"[bold {title_color}]{title_text}[/]",
title_align="left",
border_style=COLORS['border'],
box=ROUNDED,
padding=(0, 1),
width=width,
)
|
make_kv_table
make_kv_table(rows: List[tuple], highlight_keys: set = None) -> Table
Tworzy tabelę klucz-wartość dla sekcji konfiguracji.
Source code in vbc/ui/modern_overlays.py
| def make_kv_table(rows: List[tuple], highlight_keys: set = None) -> Table:
"""Tworzy tabelę klucz-wartość dla sekcji konfiguracji."""
highlight_keys = highlight_keys or set()
table = Table(
show_header=False,
box=None,
padding=(0, 1),
expand=True,
)
table.add_column("Key", style=COLORS['muted'], no_wrap=True)
table.add_column("Value", justify="right", overflow="fold")
for key, value in rows:
if key in highlight_keys:
value_style = f"bold {COLORS['accent_green']}"
elif value in ("None", "False", "0", "—"):
value_style = COLORS['dim']
else:
value_style = "white"
table.add_row(key, f"[{value_style}]{value}[/]")
return table
|
make_two_column_layout
make_two_column_layout(left: RenderableType, right: RenderableType) -> Table
Tworzy layout dwukolumnowy z równymi kolumnami.
Source code in vbc/ui/modern_overlays.py
| def make_two_column_layout(left: RenderableType, right: RenderableType) -> Table:
"""Tworzy layout dwukolumnowy z równymi kolumnami."""
table = Table(show_header=False, box=None, expand=True, padding=0)
table.add_column(ratio=1)
table.add_column(width=1) # spacer
table.add_column(ratio=1)
table.add_row(left, "", right)
return table
|
make_shortcut_row
make_shortcut_row(key: str, description: str, key_color: str = 'white') -> Table
Tworzy wiersz skrótu klawiszowego.
Source code in vbc/ui/modern_overlays.py
| def make_shortcut_row(key: str, description: str, key_color: str = "white") -> Table:
"""Tworzy wiersz skrótu klawiszowego."""
table = Table(show_header=False, box=None, padding=0, expand=True)
table.add_column(width=12)
table.add_column()
key_badge = f"[bold {key_color} on {COLORS['border']}] {key} [/]"
table.add_row(key_badge, description)
return table
|
parse_config_lines
parse_config_lines(lines: List[str]) -> dict
Parsuje config_lines do słownika.
Source code in vbc/ui/modern_overlays.py
| def parse_config_lines(lines: List[str]) -> dict:
"""Parsuje config_lines do słownika."""
result = {}
for line in lines:
if ": " in line:
parts = line.split(": ", 1)
key = parts[0].strip()
value = parts[1].strip() if len(parts) > 1 else ""
result[key.lower().replace(" ", "_")] = value
return result
|
format_size(size_bytes: Optional[int]) -> str
Format size: 123B, 1.2KB, 45.1MB, 3.2GB.
Source code in vbc/ui/modern_overlays.py
| def format_size(size_bytes: Optional[int]) -> str:
"""Format size: 123B, 1.2KB, 45.1MB, 3.2GB."""
if size_bytes is None:
return "—"
if size_bytes == 0:
return "0B"
size = float(size_bytes)
for unit in ("B", "KB", "MB", "GB", "TB"):
if size < 1024.0:
return f"{size:.1f}{unit}"
size /= 1024.0
return f"{size:.1f}PB"
|
generate_settings_overlay
generate_settings_overlay(config_lines: List[str], spinner_frame: int = 0, log_path: Optional[str] = None, debug_enabled: bool = False) -> Panel
Generuje overlay Settings (dawniej Config) dla Dashboard.
Source code in vbc/ui/modern_overlays.py
| def generate_settings_overlay(
config_lines: List[str],
spinner_frame: int = 0,
log_path: Optional[str] = None,
debug_enabled: bool = False,
) -> Panel:
"""Generuje overlay Settings (dawniej Config) dla Dashboard."""
return SettingsOverlay(config_lines, spinner_frame, log_path, debug_enabled).render()
|
generate_io_overlay
generate_io_overlay(config_lines: List[str], input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]], output_dir_lines: List[str], errors_dir_lines: List[str], suffix_output_dirs: Optional[str], suffix_errors_dirs: Optional[str], queue_sort: str, queue_seed: Optional[int]) -> Panel
Generuje overlay I/O dla Dashboard.
Source code in vbc/ui/modern_overlays.py
| def generate_io_overlay(
config_lines: List[str],
input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]],
output_dir_lines: List[str],
errors_dir_lines: List[str],
suffix_output_dirs: Optional[str],
suffix_errors_dirs: Optional[str],
queue_sort: str,
queue_seed: Optional[int],
) -> Panel:
"""Generuje overlay I/O dla Dashboard."""
return IoOverlay(
config_lines,
input_dir_stats,
output_dir_lines,
errors_dir_lines,
suffix_output_dirs,
suffix_errors_dirs,
queue_sort,
queue_seed,
).render()
|
generate_reference_overlay
generate_reference_overlay(spinner_frame: int = 0, sparkline_preset: Optional[str] = None, sparkline_palette: Optional[str] = None, sparkline_mode: str = 'sparkline') -> Panel
Generuje overlay Reference (dawniej Legend) dla Dashboard.
Source code in vbc/ui/modern_overlays.py
| def generate_reference_overlay(
spinner_frame: int = 0,
sparkline_preset: Optional[str] = None,
sparkline_palette: Optional[str] = None,
sparkline_mode: str = "sparkline",
) -> Panel:
"""Generuje overlay Reference (dawniej Legend) dla Dashboard."""
return ReferenceOverlay(
spinner_frame,
sparkline_preset,
sparkline_palette,
sparkline_mode,
).render()
|
generate_shortcuts_overlay
generate_shortcuts_overlay() -> Panel
Generuje overlay Shortcuts (dawniej Menu) dla Dashboard.
Source code in vbc/ui/modern_overlays.py
| def generate_shortcuts_overlay() -> Panel:
"""Generuje overlay Shortcuts (dawniej Menu) dla Dashboard."""
return ShortcutsOverlay().render()
|
generate_tui_overlay
generate_tui_overlay(dim_level: str = 'mid', sparkline_preset: Optional[str] = None, sparkline_palette: Optional[str] = None, sparkline_mode: str = 'sparkline') -> Panel
Generuje overlay TUI dla Dashboard.
Source code in vbc/ui/modern_overlays.py
| def generate_tui_overlay(
dim_level: str = "mid",
sparkline_preset: Optional[str] = None,
sparkline_palette: Optional[str] = None,
sparkline_mode: str = "sparkline",
) -> Panel:
"""Generuje overlay TUI dla Dashboard."""
return TuiOverlay(dim_level, sparkline_preset, sparkline_palette, sparkline_mode).render()
|
render_settings_content
render_settings_content(config_lines: List[str], spinner_frame: int = 0, log_path: Optional[str] = None, debug_enabled: bool = False) -> RenderableType
Render Settings tab content (without outer Panel or footer).
Source code in vbc/ui/modern_overlays.py
| def render_settings_content(
config_lines: List[str],
spinner_frame: int = 0,
log_path: Optional[str] = None,
debug_enabled: bool = False,
) -> RenderableType:
"""Render Settings tab content (without outer Panel or footer)."""
return SettingsOverlay(config_lines, spinner_frame, log_path, debug_enabled)._render_content()
|
render_reference_content
render_reference_content(spinner_frame: int = 0, sparkline_preset: Optional[str] = None, sparkline_palette: Optional[str] = None, sparkline_mode: str = 'sparkline') -> RenderableType
Render Reference tab content (without outer Panel or footer).
Source code in vbc/ui/modern_overlays.py
| def render_reference_content(
spinner_frame: int = 0,
sparkline_preset: Optional[str] = None,
sparkline_palette: Optional[str] = None,
sparkline_mode: str = "sparkline",
) -> RenderableType:
"""Render Reference tab content (without outer Panel or footer)."""
return ReferenceOverlay(
spinner_frame,
sparkline_preset,
sparkline_palette,
sparkline_mode,
)._render_content()
|
render_shortcuts_content
render_shortcuts_content() -> RenderableType
Render Shortcuts tab content (without outer Panel or footer).
Source code in vbc/ui/modern_overlays.py
| def render_shortcuts_content() -> RenderableType:
"""Render Shortcuts tab content (without outer Panel or footer)."""
return ShortcutsOverlay()._render_content()
|
render_io_content
render_io_content(config_lines: List[str], input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]], output_dir_lines: List[str], errors_dir_lines: List[str], suffix_output_dirs: Optional[str], suffix_errors_dirs: Optional[str], queue_sort: str, queue_seed: Optional[int]) -> RenderableType
Render I/O tab content (without outer Panel or footer).
Source code in vbc/ui/modern_overlays.py
| def render_io_content(
config_lines: List[str],
input_dir_stats: List[Tuple[str, str, Optional[int], Optional[int]]],
output_dir_lines: List[str],
errors_dir_lines: List[str],
suffix_output_dirs: Optional[str],
suffix_errors_dirs: Optional[str],
queue_sort: str,
queue_seed: Optional[int],
) -> RenderableType:
"""Render I/O tab content (without outer Panel or footer)."""
return IoOverlay(
config_lines,
input_dir_stats,
output_dir_lines,
errors_dir_lines,
suffix_output_dirs,
suffix_errors_dirs,
queue_sort,
queue_seed,
)._render_content()
|
render_tui_content
render_tui_content(dim_level: str = 'mid', sparkline_preset: Optional[str] = None, sparkline_palette: Optional[str] = None, sparkline_mode: str = 'sparkline') -> RenderableType
Render TUI tab content (without outer Panel or footer).
Source code in vbc/ui/modern_overlays.py
| def render_tui_content(
dim_level: str = "mid",
sparkline_preset: Optional[str] = None,
sparkline_palette: Optional[str] = None,
sparkline_mode: str = "sparkline",
) -> RenderableType:
"""Render TUI tab content (without outer Panel or footer)."""
return TuiOverlay(dim_level, sparkline_preset, sparkline_palette, sparkline_mode)._render_content()
|
render_dirs_content
render_dirs_content(entries: List[Tuple[str, str, Optional[int], Optional[int]]], cursor: int, input_mode: bool, input_buffer: str, has_pending_changes: bool, suffix_output_dirs: Optional[str], suffix_errors_dirs: Optional[str], output_dir_lines: List[str], errors_dir_lines: List[str], error_msg: str = '') -> RenderableType
Render Dirs tab content (without outer Panel or footer).
Parameters:
| Name |
Type |
Description |
Default |
entries
|
List[Tuple[str, str, Optional[int], Optional[int]]]
|
List of (path, status, file_count, size_bytes) from UIState.dirs_get_all_entries().
|
required
|
cursor
|
int
|
Current cursor position (0-based).
|
required
|
input_mode
|
bool
|
True when add-path input mode is active.
|
required
|
input_buffer
|
str
|
Current add-path input text.
|
required
|
has_pending_changes
|
bool
|
True when staged Dirs changes require [S] apply.
|
required
|
suffix_output_dirs
|
Optional[str]
|
Output directory suffix (e.g. "_out").
|
required
|
suffix_errors_dirs
|
Optional[str]
|
Errors directory suffix (e.g. "_err").
|
required
|
output_dir_lines
|
List[str]
|
Explicit output directory paths (for non-suffix mode).
|
required
|
errors_dir_lines
|
List[str]
|
Explicit errors directory paths (for non-suffix mode).
|
required
|
Source code in vbc/ui/modern_overlays.py
| def render_dirs_content(
entries: List[Tuple[str, str, Optional[int], Optional[int]]],
cursor: int,
input_mode: bool,
input_buffer: str,
has_pending_changes: bool,
suffix_output_dirs: Optional[str],
suffix_errors_dirs: Optional[str],
output_dir_lines: List[str],
errors_dir_lines: List[str],
error_msg: str = "",
) -> RenderableType:
"""Render Dirs tab content (without outer Panel or footer).
Args:
entries: List of (path, status, file_count, size_bytes) from UIState.dirs_get_all_entries().
cursor: Current cursor position (0-based).
input_mode: True when add-path input mode is active.
input_buffer: Current add-path input text.
has_pending_changes: True when staged Dirs changes require [S] apply.
suffix_output_dirs: Output directory suffix (e.g. "_out").
suffix_errors_dirs: Errors directory suffix (e.g. "_err").
output_dir_lines: Explicit output directory paths (for non-suffix mode).
errors_dir_lines: Explicit errors directory paths (for non-suffix mode).
"""
return DirsOverlay(
entries,
cursor,
input_mode,
input_buffer,
has_pending_changes,
suffix_output_dirs,
suffix_errors_dirs,
output_dir_lines,
errors_dir_lines,
error_msg=error_msg,
)._render_content()
|
GPU Sparklines
Sparkline presets, palettes, and rendering helpers for GPU metrics.
gpu_sparkline
bin_value
bin_value(val: Optional[float], min_val: float, max_val: float, num_bins: int) -> int
Map value to 0..(num_bins-1) for sparkline block char. -1 for None.
Source code in vbc/ui/gpu_sparkline.py
| def bin_value(val: Optional[float], min_val: float, max_val: float, num_bins: int) -> int:
"""Map value to 0..(num_bins-1) for sparkline block char. -1 for None."""
if val is None:
return -1
if num_bins <= 1 or max_val <= min_val:
return 0
if val <= min_val:
return 0
if val >= max_val:
return num_bins - 1
ratio = (val - min_val) / (max_val - min_val)
return min(num_bins - 1, int(ratio * num_bins))
|
render_sparkline
render_sparkline(history: Iterable[Optional[float]], spark_len: int, min_val: float, max_val: float, style: SparklineStyle, palette: Optional[Sequence[str]] = None, glyph: Optional[str] = None) -> str
Render sparkline with newest on right, missing as style.missing.
Source code in vbc/ui/gpu_sparkline.py
| def render_sparkline(
history: Iterable[Optional[float]],
spark_len: int,
min_val: float,
max_val: float,
style: SparklineStyle,
palette: Optional[Sequence[str]] = None,
glyph: Optional[str] = None,
) -> str:
"""Render sparkline with newest on right, missing as style.missing."""
if spark_len <= 0:
return ""
samples = list(history)[-spark_len:] # Last N samples (oldest -> newest)
chars: List[str] = []
visible_len = 0
for val in samples:
bin_idx = bin_value(val, min_val, max_val, style.num_bins)
if bin_idx < 0 or not style.blocks:
char = style.missing
if palette:
chars.append(f"[dim]{char}[/]")
else:
chars.append(char)
else:
char = glyph or style.blocks[bin_idx]
if palette:
color = _palette_color_for_value(val, min_val, max_val, palette)
chars.append(f"[{color}]{char}[/]")
else:
chars.append(char)
visible_len += 1
if visible_len < spark_len:
chars.append(" " * (spark_len - visible_len))
return "".join(chars)
|
Usage Examples
Complete UI Setup
from pathlib import Path
from vbc.config.loader import load_config
from vbc.infrastructure.event_bus import EventBus
from vbc.ui.state import UIState
from vbc.ui.manager import UIManager
from vbc.ui.keyboard import KeyboardListener
from vbc.ui.dashboard import Dashboard
# Load config
config = load_config(Path("conf/vbc.yaml"))
# Create UI components
bus = EventBus()
ui_state = UIState()
ui_state.current_threads = config.general.threads
ui_state.strip_unicode_display = config.general.strip_unicode_display
# Create UI manager (subscribes to events)
ui_manager = UIManager(bus, ui_state)
# Create keyboard listener
keyboard = KeyboardListener(bus)
# Create dashboard
dashboard = Dashboard(ui_state)
# Start keyboard listener
keyboard.start()
try:
# Run dashboard
with dashboard:
# Main processing loop here
orchestrator.run(Path("/videos"))
finally:
keyboard.stop()
UIState Operations
from vbc.ui.state import UIState
from vbc.domain.models import CompressionJob, VideoFile, JobStatus
from pathlib import Path
state = UIState()
# Add active job
job = CompressionJob(
source_file=VideoFile(path=Path("video.mp4"), size_bytes=1000000)
)
state.add_active_job(job)
# Complete job
state.add_completed_job(job, output_size=450000)
# Check stats
print(f"Completed: {state.completed_count}")
print(f"Compression ratio: {state.compression_ratio:.2f}")
print(f"Space saved: {state.space_saved_bytes} bytes")
# Get last action (with 60s timeout)
action = state.get_last_action()
if action:
print(f"Last action: {action}")
Event-Driven UI Updates
from vbc.infrastructure.event_bus import EventBus
from vbc.ui.state import UIState
from vbc.ui.manager import UIManager
from vbc.domain.events import JobStarted, JobCompleted
# Setup
bus = EventBus()
state = UIState()
manager = UIManager(bus, state)
# Publish events - UI updates automatically
job = CompressionJob(...)
bus.publish(JobStarted(job=job))
# UIManager.on_job_started() updates state.active_jobs
bus.publish(JobCompleted(job=job))
# UIManager.on_job_completed() updates counters and recent_jobs
Keyboard Controls
from vbc.infrastructure.event_bus import EventBus
from vbc.ui.keyboard import KeyboardListener, ThreadControlEvent, RequestShutdown
bus = EventBus()
# Subscribe to keyboard events
def on_thread_control(event: ThreadControlEvent):
print(f"Thread change: {event.change:+d}")
def on_shutdown(event: RequestShutdown):
print("Shutdown requested!")
bus.subscribe(ThreadControlEvent, on_thread_control)
bus.subscribe(RequestShutdown, on_shutdown)
# Start listener
keyboard = KeyboardListener(bus)
keyboard.start()
# User presses '>' → ThreadControlEvent(change=+1)
# User presses '<' → ThreadControlEvent(change=-1)
# User presses 'S' → RequestShutdown()
# Cleanup
keyboard.stop()
Dashboard Surfaces
The dashboard is rendered as a compact Rich layout rather than a fixed list of legacy panels:
- Top status/KPI area with runtime counters, hints, and optional GPU sparkline metrics.
- Size-based progress area with total/session counters and ETA.
- Active jobs area with per-file progress.
- Activity and completed-job history.
- Queue preview for pending files.
- Tabbed overlay system for Settings, Shortcuts, I/O, Dirs, TUI, Reference, and Logs.
Dashboard Rendering
from vbc.ui.dashboard import Dashboard
from vbc.ui.state import UIState
state = UIState()
# Set configuration lines (shown in config overlay)
state.config_lines = [
"Video Batch Compression - NVENC AV1 (GPU)",
"Start: 2025-12-21 15:30:00",
"Input: /videos",
"Output: /videos_out",
"Threads: 8 (Prefetch: 1x)",
"Encoder: NVENC AV1 | Preset: p7 (Slow/HQ)",
"Quality: CQ45 (Global Default)",
"Dynamic Quality: ILCE-7RM5:38, DC-GH7:40",
]
# Create dashboard
dashboard = Dashboard(state)
# Render in context
with dashboard:
# Your processing loop
# Dashboard updates automatically every 1s
pass
Thread Safety
All UI state updates use locks:
class UIState:
def __init__(self):
self._lock = threading.RLock()
def add_completed_job(self, job, output_size):
with self._lock:
self.completed_count += 1
self.total_input_bytes += job.source_file.size_bytes
self.total_output_bytes += output_size
# ...
Unicode Handling
from vbc.ui.state import UIState
state = UIState()
state.strip_unicode_display = True # Default
# Sanitize filename for display
def sanitize(filename: str) -> str:
if not state.strip_unicode_display:
return filename
return ''.join(c if ord(c) < 128 else '?' for c in filename)
# Example
filename = "video_🎬_final.mp4"
display = sanitize(filename)
# Output: "video_?_final.mp4" (prevents table alignment issues)
Keyboard Event Reference
| Key |
Event |
Handler |
< or , |
ThreadControlEvent(change=-1) |
Orchestrator decreases threads |
> or . |
ThreadControlEvent(change=+1) |
Orchestrator increases threads |
S or s |
RequestShutdown() |
Orchestrator graceful shutdown |
R or r |
RefreshRequested() |
Orchestrator re-scans directory |
C or c |
ToggleOverlayTab(tab="settings") |
UIManager toggles Prefs tab |
F or f |
ToggleOverlayTab(tab="io") |
UIManager toggles I/O tab |
M or m |
ToggleOverlayTab(tab="shortcuts") |
UIManager toggles Keys tab |
D or d |
ToggleOverlayTab(tab="dirs") |
UIManager toggles Dirs tab |
T or t |
ToggleOverlayTab(tab="tui") |
UIManager toggles TUI tab |
E or e |
ToggleOverlayTab(tab="reference") |
UIManager toggles Ref tab |
L or l |
ToggleOverlayTab(tab="logs") |
UIManager toggles Logs tab |
I or i |
CycleOverlayDim(direction=+1) |
UIManager cycles overlay dim level |
G or g |
RotateGpuMetric() |
UIManager rotates the GPU metric shown in the top bar |
W or w |
CycleSparklinePreset(direction=+1) |
UIManager cycles sparkline presets |
P or p |
CycleSparklinePalette(direction=+1) |
UIManager cycles sparkline palettes |
[ / ] |
CycleLogsPage(direction=-1/+1) |
UIManager pages Logs tab |
Tab |
CycleOverlayTab(direction=+1) |
UIManager cycles tabs |
Esc |
CloseOverlay() |
UIManager closes active overlay |
Ctrl+C |
InterruptRequested() |
Orchestrator terminates active jobs |
Prefs Overlay
Toggle with C key:
┌─ CONFIGURATION ────────────────────────────────┐
│ Video Batch Compression - NVENC AV1 (GPU) │
│ Start: 2025-12-21 15:30:00 │
│ Input: /videos │
│ Output: /videos_out │
│ Threads: 8 (Prefetch: 1x) │
│ Encoder: NVENC AV1 | Preset: p7 (Slow/HQ) │
│ Audio: Auto (lossless->AAC 256k, AAC/MP3 copy, other->AAC 192k) │
│ Quality: CQ45 (Global Default) │
│ Dynamic Quality: ILCE-7RM5:38, DC-GH7:40 │
│ Camera Filter: None │
│ Metadata: Deep (ExifTool + XMP) │
│ Autorotate: 3 rules loaded │
│ Manual Rotation: None │
│ Extensions: .mp4, .mov, .avi → .mp4 │
│ Min size: 1.0MB | Skip AV1: false │
│ Clean errors: false | Strip Unicode: true │
│ Debug logging: false │
│ │
│ Press Esc to close │
└────────────────────────────────────────────────┘