diff --git a/cobot/cli.py b/cobot/cli.py index d42f26e..fc3b7c3 100644 --- a/cobot/cli.py +++ b/cobot/cli.py @@ -3,10 +3,12 @@ import sys # Import each command module so we can register its subparser. # Импортируем каждый модуль команды, чтобы зарегистрировать его подпарсер. +from cobot.commands import clean as cmd_clean from cobot.commands import delete as cmd_delete from cobot.commands import docker_setup as cmd_docker_setup from cobot.commands import doc_setup as cmd_doc_setup from cobot.commands import local_setup as cmd_local_setup +from cobot.commands import rebuild as cmd_rebuild from cobot.commands import robot_setup as cmd_robot_setup from cobot.commands import run as cmd_run from cobot.commands import setup as cmd_setup @@ -27,6 +29,10 @@ _GROUPS = [ ("Run", [ ("run", "launch the robot controller or Webots simulator (local or Docker)"), ]), + ("Build", [ + ("rebuild", "rebuild ROS2 packages in src/ with colcon"), + ("clean", "remove colcon build artifacts (build/ install/ log/)"), + ]), ("Management", [ ("update", "pull latest changes from the remote git branch and reinstall cobot"), ("delete", "remove the project, Docker images, containers, and optionally ROS2"), @@ -107,5 +113,7 @@ def _register_commands(subparsers): cmd_doc_setup.register(subparsers) cmd_robot_setup.register(subparsers) cmd_run.register(subparsers) + cmd_rebuild.register(subparsers) + cmd_clean.register(subparsers) cmd_update.register(subparsers) cmd_delete.register(subparsers) diff --git a/cobot/commands/clean.py b/cobot/commands/clean.py new file mode 100644 index 0000000..4afa8c1 --- /dev/null +++ b/cobot/commands/clean.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import argparse +import shutil +from pathlib import Path +from typing import Optional + +from textual.app import App + +from cobot.tui import SCREEN_CSS, InputScreen, LogScreen, PickScreen + +_PROJECT_DIR = Path(__file__).parent.parent.parent +_BUILD_DIR = _PROJECT_DIR / "build" +_INSTALL_DIR = _PROJECT_DIR / "install" +_LOG_DIR = _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/. + """ + try: + screen.set_progress(0, "Cleaning build/...") + screen.write("[bold]Cleaning all build artifacts[/bold]\n") + + _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) + + except Exception as exc: + if not screen.is_stopped(): + screen.write(f"\n[red]Error:[/red] {exc}") + screen.finish(False) + + +class _CleanApp(App[None]): + """Clean wizard: asks what to clean (all / specific package) then runs deletion. + Мастер очистки: спрашивает что удалить (всё / конкретный пакет), затем выполняет удаление. + """ + + CSS = SCREEN_CSS + + def __init__(self, target: Optional[str]): + super().__init__() + # None = ask, "all" = delete everything, anything else = package name. + # None = спросить, "all" = удалить всё, иначе = имя пакета. + self._target = target + + def on_mount(self) -> None: + if self._target is None: + self._ask_target() + elif self._target == "all": + self._start_all() + else: + self._start_package(self._target) + + def _ask_target(self) -> None: + self.push_screen( + PickScreen( + "clean", + "What do you want to clean?", + ["All (build/ install/ log/)", "Specific package"], + "All (build/ install/ log/)", + ), + self._got_target_choice, + ) + + def _got_target_choice(self, value: Optional[str]) -> None: + if value is None: + self.exit() + return + if value.startswith("All"): + self._start_all() + else: + self._ask_package() + + 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: + self.push_screen( + LogScreen( + f"Cleaning package: {package}", + lambda s: _task_clean_package(s, package), + show_progress=True, + ), + lambda _: self.exit(), + ) + + +def register(subparsers: argparse._SubParsersAction) -> None: + p = subparsers.add_parser( + "clean", + help="Remove colcon build artifacts (build/ install/ log/)", + ) + p.add_argument( + "target", + nargs="?", + metavar="all | PACKAGE", + default=None, + help="'all' to remove everything, or a package name to clean only that package", + ) + p.set_defaults(func=run) + + +def run(args: argparse.Namespace) -> None: + """Entry point for the clean command. + Точка входа для команды clean. + """ + _CleanApp(getattr(args, "target", None)).run() diff --git a/cobot/commands/rebuild.py b/cobot/commands/rebuild.py new file mode 100644 index 0000000..b242f88 --- /dev/null +++ b/cobot/commands/rebuild.py @@ -0,0 +1,192 @@ +from __future__ import annotations + +import argparse +import subprocess +from pathlib import Path +from typing import List, Optional + +from textual.app import App + +from cobot.tui import SCREEN_CSS, InputScreen, LogScreen, PickScreen +from cobot.commands.local_setup import _ros2_env + +_PROJECT_DIR = Path(__file__).parent.parent.parent + + +def _task_rebuild(screen: LogScreen, packages: List[str], symlink: bool) -> None: + """Run colcon build for the selected packages (or all if packages is empty). + Streams output to the log and tracks per-package progress. + + Запускает colcon build для выбранных пакетов (или всех если packages пуст). + Транслирует вывод в лог и отслеживает прогресс по каждому пакету. + """ + try: + env = _ros2_env() + + # Count packages so we can show X / total progress. + # Считаем пакеты чтобы показывать X / всего в прогрессе. + list_cmd = ["colcon", "list", "--base-paths", "src"] + if packages: + list_cmd += ["--packages-select"] + packages + list_result = subprocess.run( + list_cmd, capture_output=True, text=True, + cwd=_PROJECT_DIR, env=env, + ) + total = max(len([l for l in list_result.stdout.splitlines() if l.strip()]), 1) + + pkg_label = " ".join(packages) if packages else "all packages" + symlink_label = " --symlink-install" if symlink else "" + screen.write(f"[bold]colcon build{symlink_label} — {pkg_label}[/bold]\n") + screen.set_progress(0, f"0 / {total} packages done") + built = 0 + + cmd = ["colcon", "build", "--base-paths", "src"] + if symlink: + cmd.append("--symlink-install") + if packages: + cmd += ["--packages-select"] + packages + + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, cwd=_PROJECT_DIR, env=env, + ) + screen.set_proc(proc) + + for line in proc.stdout: + s = line.rstrip() + if s: + screen.write(s) + if "Finished <<<" in line or "Failed <<<" in line: + built += 1 + screen.set_progress( + built / total * 100, + f"{built} / {total} packages done", + ) + proc.wait() + + if screen.is_stopped(): + return + + if proc.returncode not in (0, -9): + screen.write("\n[red]Build failed.[/red]") + screen.finish(False) + return + + screen.set_progress(100, "Done") + screen.write("\n[green]Build complete.[/green]") + screen.finish(True) + + except Exception as exc: + if not screen.is_stopped(): + screen.write(f"\n[red]Error:[/red] {exc}") + screen.finish(False) + + +class _RebuildApp(App[None]): + """Rebuild wizard: optionally asks for packages and symlink flag, then runs colcon. + Мастер пересборки: опционально спрашивает пакеты и флаг symlink, затем запускает colcon. + """ + + CSS = SCREEN_CSS + + def __init__(self, packages: Optional[List[str]], symlink: Optional[bool]): + super().__init__() + # None means "ask the user interactively". + # None означает "спросить пользователя интерактивно". + self._packages = packages + self._symlink = symlink + + def on_mount(self) -> None: + if self._packages is None: + self._ask_packages() + elif self._symlink is None: + self._ask_symlink() + else: + self._start() + + def _ask_packages(self) -> None: + self.push_screen( + InputScreen( + "rebuild", + "Packages to rebuild (space-separated, leave empty for all):", + "", + note="Example: iiwa_controller iiwa_bringup", + ), + self._got_packages, + ) + + def _got_packages(self, value: Optional[str]) -> None: + if value is None: + self.exit() + return + self._packages = value.split() if value.strip() else [] + if self._symlink is None: + self._ask_symlink() + else: + self._start() + + def _ask_symlink(self) -> None: + self.push_screen( + PickScreen( + "rebuild", + "Use --symlink-install?", + ["Yes", "No"], + "Yes", + ), + self._got_symlink, + ) + + def _got_symlink(self, value: Optional[str]) -> None: + if value is None: + self.exit() + return + self._symlink = value == "Yes" + self._start() + + def _start(self) -> None: + self.push_screen( + LogScreen( + "Rebuilding packages", + lambda s: _task_rebuild(s, self._packages, self._symlink), + show_progress=True, + ), + lambda _: self.exit(), + ) + + +def register(subparsers: argparse._SubParsersAction) -> None: + p = subparsers.add_parser( + "rebuild", + help="Rebuild ROS2 packages in src/ with colcon", + ) + p.add_argument( + "packages", + nargs="*", + metavar="PACKAGE", + help="Packages to rebuild (omit to be asked, or leave empty for all)", + ) + p.set_defaults(symlink=None) + p.add_argument( + "--symlink-install", + dest="symlink", + action="store_true", + help="Pass --symlink-install to colcon build", + ) + p.add_argument( + "--no-symlink-install", + dest="symlink", + action="store_false", + help="Do not pass --symlink-install to colcon build", + ) + p.set_defaults(func=run) + + +def run(args: argparse.Namespace) -> None: + """Entry point for the rebuild command. + Точка входа для команды rebuild. + """ + # Convert empty list to None so the TUI asks interactively. + # Преобразуем пустой список в None, чтобы TUI спросил интерактивно. + packages = args.packages if args.packages else None + _RebuildApp(packages, args.symlink).run()