Skip to content

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() -> str

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() -> None

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
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
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').

ToggleMenu

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
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
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
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
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
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

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
render() -> Panel

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
render() -> Panel

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
render() -> Panel

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
render() -> Panel

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
render() -> Panel

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
render() -> Panel

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

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                            │
└────────────────────────────────────────────────┘