From 3da290a4f38c84d8b51cf6842e53c5677c715d18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=B0=D0=BD=D0=B8=D0=B8=D0=BB=20=D0=93=D1=80=D0=B0?= =?UTF-8?q?=D0=B1=D0=B0=D1=80=D1=8C?= Date: Sat, 23 May 2026 11:07:01 +0300 Subject: [PATCH] feat: implement MultiPickScreen for multi-selection in the TUI --- cobot/commands/clean.py | 190 +++++++++++----------------------------- cobot/tui.py | 56 +++++++++++- 2 files changed, 106 insertions(+), 140 deletions(-) diff --git a/cobot/commands/clean.py b/cobot/commands/clean.py index 4afa8c1..c9f09cc 100644 --- a/cobot/commands/clean.py +++ b/cobot/commands/clean.py @@ -3,104 +3,44 @@ from __future__ import annotations import argparse import shutil from pathlib import Path -from typing import Optional +from typing import List, Optional from textual.app import App -from cobot.tui import SCREEN_CSS, InputScreen, LogScreen, PickScreen +from cobot.tui import SCREEN_CSS, LogScreen, MultiPickScreen _PROJECT_DIR = Path(__file__).parent.parent.parent -_BUILD_DIR = _PROJECT_DIR / "build" -_INSTALL_DIR = _PROJECT_DIR / "install" -_LOG_DIR = _PROJECT_DIR / "log" + +_DIR_OPTIONS = ["build/", "install/", "log/"] +_DIR_MAP = { + "build/": _PROJECT_DIR / "build", + "install/": _PROJECT_DIR / "install", + "log/": _PROJECT_DIR / "log", +} -def _remove_dir(path: Path, write) -> None: - """Remove a directory tree if it exists, logging the result. - Рекурсивно удаляет директорию если она существует, выводя результат в лог. - """ - if path.exists(): - shutil.rmtree(path) - write(f"[green][ok][/green] Removed {path.relative_to(_PROJECT_DIR)}") - else: - write(f"[dim]Not found: {path.relative_to(_PROJECT_DIR)}[/dim]") - - -def _task_clean_all(screen: LogScreen) -> None: - """Delete build/, install/, and log/ directories entirely. - Полностью удаляет директории build/, install/ и log/. +def _task_clean(screen: LogScreen, dirs: List[str]) -> None: + """Delete the selected top-level directories. + Удаляет выбранные директории верхнего уровня. """ try: - screen.set_progress(0, "Cleaning build/...") - screen.write("[bold]Cleaning all build artifacts[/bold]\n") + screen.write("[bold]Cleaning build artifacts[/bold]\n") + total = len(dirs) + for i, label in enumerate(dirs): + if screen.is_stopped(): + return + screen.set_progress(i / total * 100, f"Removing {label}...") + path = _DIR_MAP[label] + if path.exists(): + shutil.rmtree(path) + screen.write(f"[green][ok][/green] Removed {label}") + else: + screen.write(f"[dim]Not found: {label}[/dim]") - _remove_dir(_BUILD_DIR, screen.write) - if screen.is_stopped(): - return - - screen.set_progress(40, "Cleaning install/...") - _remove_dir(_INSTALL_DIR, screen.write) - if screen.is_stopped(): - return - - screen.set_progress(75, "Cleaning log/...") - _remove_dir(_LOG_DIR, screen.write) - if screen.is_stopped(): - return - - screen.set_progress(100, "Done") - screen.write("\n[green]All build artifacts removed.[/green]") - screen.finish(True) - - except Exception as exc: if not screen.is_stopped(): - screen.write(f"\n[red]Error:[/red] {exc}") - screen.finish(False) - - -def _task_clean_package(screen: LogScreen, package: str) -> None: - """Delete build/, install/, and all log/*/ directories. - Удаляет build/, install/ и все log/*/ директории. - """ - try: - screen.set_progress(0, f"Cleaning {package}...") - screen.write(f"[bold]Cleaning package: {package}[/bold]\n") - - _remove_dir(_BUILD_DIR / package, screen.write) - if screen.is_stopped(): - return - - screen.set_progress(40, f"Cleaning install/{package}...") - _remove_dir(_INSTALL_DIR / package, screen.write) - if screen.is_stopped(): - return - - screen.set_progress(70, f"Cleaning log entries for {package}...") - # colcon creates one subdirectory per package inside each timestamped log run. - # colcon создаёт по одной поддиректории на пакет внутри каждого лога с временной меткой. - if _LOG_DIR.exists(): - removed = 0 - for log_run in _LOG_DIR.iterdir(): - if not log_run.is_dir(): - continue - pkg_log = log_run / package - if pkg_log.exists(): - shutil.rmtree(pkg_log) - screen.write( - f"[green][ok][/green] Removed log/{log_run.name}/{package}" - ) - removed += 1 - if removed == 0: - screen.write(f"[dim]No log entries found for {package}[/dim]") - else: - screen.write("[dim]Not found: log/[/dim]") - - if screen.is_stopped(): - return - - screen.set_progress(100, "Done") - screen.write(f"\n[green]Package '{package}' artifacts removed.[/green]") - screen.finish(True) + screen.set_progress(100, "Done") + screen.write("\n[green]Done.[/green]") + screen.finish(True) except Exception as exc: if not screen.is_stopped(): @@ -109,74 +49,46 @@ def _task_clean_package(screen: LogScreen, package: str) -> None: class _CleanApp(App[None]): - """Clean wizard: asks what to clean (all / specific package) then runs deletion. - Мастер очистки: спрашивает что удалить (всё / конкретный пакет), затем выполняет удаление. + """Clean wizard: lets the user pick which directories to delete, then removes them. + Мастер очистки: позволяет выбрать директории для удаления, затем удаляет их. """ CSS = SCREEN_CSS - def __init__(self, target: Optional[str]): + def __init__(self, all_dirs: bool): super().__init__() - # None = ask, "all" = delete everything, anything else = package name. - # None = спросить, "all" = удалить всё, иначе = имя пакета. - self._target = target + # True = skip the question and delete everything right away. + # True = пропустить вопрос и сразу удалить всё. + self._all_dirs = all_dirs def on_mount(self) -> None: - if self._target is None: - self._ask_target() - elif self._target == "all": - self._start_all() + if self._all_dirs: + self._start(_DIR_OPTIONS) else: - self._start_package(self._target) + self._ask_dirs() - def _ask_target(self) -> None: + def _ask_dirs(self) -> None: self.push_screen( - PickScreen( + MultiPickScreen( "clean", - "What do you want to clean?", - ["All (build/ install/ log/)", "Specific package"], - "All (build/ install/ log/)", + "Which directories to delete?", + _DIR_OPTIONS, + note="Space — toggle · Enter — confirm", ), - self._got_target_choice, + self._got_dirs, ) - def _got_target_choice(self, value: Optional[str]) -> None: - if value is None: + def _got_dirs(self, dirs: Optional[List[str]]) -> None: + if not dirs: self.exit() return - if value.startswith("All"): - self._start_all() - else: - self._ask_package() + self._start(dirs) - def _ask_package(self) -> None: - self.push_screen( - InputScreen( - "clean", - "Package name to clean:", - "", - note="Example: iiwa_controller", - ), - self._got_package, - ) - - def _got_package(self, value: Optional[str]) -> None: - if value is None or not value.strip(): - self.exit() - return - self._start_package(value.strip()) - - def _start_all(self) -> None: - self.push_screen( - LogScreen("Cleaning all artifacts", _task_clean_all, show_progress=True), - lambda _: self.exit(), - ) - - def _start_package(self, package: str) -> None: + def _start(self, dirs: List[str]) -> None: self.push_screen( LogScreen( - f"Cleaning package: {package}", - lambda s: _task_clean_package(s, package), + "Cleaning", + lambda s: _task_clean(s, dirs), show_progress=True, ), lambda _: self.exit(), @@ -191,9 +103,9 @@ def register(subparsers: argparse._SubParsersAction) -> None: p.add_argument( "target", nargs="?", - metavar="all | PACKAGE", + metavar="all", default=None, - help="'all' to remove everything, or a package name to clean only that package", + help="'all' to skip the prompt and delete all three directories at once", ) p.set_defaults(func=run) @@ -202,4 +114,4 @@ def run(args: argparse.Namespace) -> None: """Entry point for the clean command. Точка входа для команды clean. """ - _CleanApp(getattr(args, "target", None)).run() + _CleanApp(all_dirs=(getattr(args, "target", None) == "all")).run() diff --git a/cobot/tui.py b/cobot/tui.py index 706cfb9..86cab36 100644 --- a/cobot/tui.py +++ b/cobot/tui.py @@ -8,7 +8,7 @@ from textual import on from textual.app import ComposeResult from textual.binding import Binding from textual.screen import Screen -from textual.widgets import Footer, Input, LoadingIndicator, ProgressBar, RadioButton, RadioSet, RichLog, Static +from textual.widgets import Checkbox, Footer, Input, LoadingIndicator, ProgressBar, RadioButton, RadioSet, RichLog, Static # Shared CSS applied to every screen in the app. # Общий CSS, применяемый ко всем экранам приложения. @@ -78,6 +78,13 @@ RunScreen #hint { color: $text-muted; text-style: dim; } +MultiPickScreen #choices { + height: auto; + margin-bottom: 1; +} +MultiPickScreen Checkbox { + margin: 0; +} """ @@ -294,6 +301,53 @@ class LogScreen(Screen[bool]): self.dismiss(self._success) +# A screen that shows a list of checkboxes for multi-selection. +# The user toggles items with Space, confirms with Enter, cancels with Escape. +# Экран с чекбоксами для множественного выбора. +# Пользователь переключает пункты пробелом, подтверждает Enter, отменяет Escape. +class MultiPickScreen(Screen[Optional[List[str]]]): + """Multi-choice checkbox screen. Returns the list of selected option strings, or None on Escape. + Экран множественного выбора с чекбоксами. Возвращает список выбранных строк или None при Escape. + """ + BINDINGS = [ + Binding("enter", "submit", "Confirm", priority=True), + Binding("escape", "abort", "Cancel"), + ] + + def __init__(self, step: str, question: str, options: List[str], + defaults: Optional[List[str]] = None, note: str = ""): + super().__init__() + self._step = step + self._question = question + self._options = options + # All options are checked by default when defaults is None. + # Все пункты отмечены по умолчанию если defaults не передан. + self._defaults = set(defaults) if defaults is not None else set(options) + self._note = note + + def compose(self) -> ComposeResult: + yield Static(self._step, id="step") + yield Static(self._question, id="question") + if self._note: + yield Static(self._note, id="note") + with Static(id="choices"): + for opt in self._options: + yield Checkbox(opt, value=(opt in self._defaults)) + yield Footer() + + def on_mount(self) -> None: + checkboxes = list(self.query(Checkbox)) + if checkboxes: + checkboxes[0].focus() + + def action_submit(self) -> None: + selected = [str(cb.label) for cb in self.query(Checkbox) if cb.value] + self.dismiss(selected) + + def action_abort(self) -> None: + self.app.exit(None) + + # A screen for a long-running process that the user can stop at any time. # Shows a live log and offers S / Enter / Escape to stop or close. # Экран для долго работающего процесса, который пользователь может остановить в любой момент.