From 5f16825cf7a51174d8c63f9711bb326d3fa748d1 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: Wed, 20 May 2026 17:41:11 +0300 Subject: [PATCH] Refactor Docker and Local Setup Commands - Enhanced the docker_setup.py to streamline image building and pulling processes with improved logging and error handling. - Introduced a new LogScreen for better user feedback during long-running tasks. - Updated local_setup.py to utilize logging for installation steps and improved error handling. - Refactored robot_setup.py to simplify configuration saving and user interaction. - Added a new LogScreen class in tui.py for consistent logging across different setup processes. - Modified install.sh to adjust the installation directory for better organization. --- cobot/commands/doc_setup.py | 270 ++++++++++++++------------ cobot/commands/docker_setup.py | 339 ++++++++++++++------------------- cobot/commands/local_setup.py | 267 +++++++++++++------------- cobot/commands/robot_setup.py | 71 +++---- cobot/tui.py | 60 +++++- install.sh | 2 +- 6 files changed, 507 insertions(+), 502 deletions(-) diff --git a/cobot/commands/doc_setup.py b/cobot/commands/doc_setup.py index 6aa1db2..2f74368 100644 --- a/cobot/commands/doc_setup.py +++ b/cobot/commands/doc_setup.py @@ -2,20 +2,15 @@ from __future__ import annotations import argparse import os -import re import shutil import subprocess import sys from pathlib import Path -from typing import List, Optional +from typing import Callable, Optional -from rich.console import Console -from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn from textual.app import App -from cobot.tui import SCREEN_CSS, InputScreen - -_console = Console() +from cobot.tui import SCREEN_CSS, InputScreen, LogScreen _PROJECT_DIR = Path(__file__).parent.parent.parent _DOC_DIR = _PROJECT_DIR / "doc" / "lwc-doc" @@ -23,150 +18,183 @@ _IMAGE_NAME = "lwc-docs" _CONTAINER_NAME = "lwc-docs" _DEFAULT_PORT = "8000" - - -class _Wizard(App[Optional[str]]): - CSS = SCREEN_CSS - - def on_mount(self) -> None: - self.push_screen( - InputScreen("Step 1 of 1", "Port to serve documentation on:", _DEFAULT_PORT), - self._got_port, - ) - - def _got_port(self, v: Optional[str]) -> None: - if v is None: - self.exit(None) - return - port = v.strip() or _DEFAULT_PORT - if not port.isdigit(): - self.exit(_DEFAULT_PORT) - else: - self.exit(port) +Write = Callable[[str], None] def _docker(*args: str, capture: bool = False) -> subprocess.CompletedProcess: - return subprocess.run( - ["docker", *args], - capture_output=capture, - text=True, - ) + return subprocess.run(["docker", *args], capture_output=capture, text=True) def _is_running() -> bool: - result = _docker( - "ps", "--filter", f"name={_CONTAINER_NAME}", "--format", "{{.Names}}", - capture=True, - ) - return _CONTAINER_NAME in result.stdout + r = _docker("ps", "--filter", f"name={_CONTAINER_NAME}", "--format", "{{.Names}}", capture=True) + return _CONTAINER_NAME in r.stdout def _image_exists() -> bool: - result = _docker("images", "-q", _IMAGE_NAME, capture=True) - return bool(result.stdout.strip()) + return bool(_docker("images", "-q", _IMAGE_NAME, capture=True).stdout.strip()) -def _build_docs_image() -> bool: - _console.print("[dim]Installing MkDocs plugins into the image (this runs once)...[/dim]") +def _build_docs_image(write: Write) -> bool: + write("[cyan][*][/cyan] Building documentation image (runs once)...") env = {**os.environ, "DOCKER_BUILDKIT": "0"} proc = subprocess.Popen( ["docker", "build", "-t", _IMAGE_NAME, str(_DOC_DIR)], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env, ) - - captured: List[str] = [] - - with Progress( - SpinnerColumn(), - TextColumn(" [bold cyan]lwc-docs[/bold cyan]"), - BarColumn(bar_width=32), - TaskProgressColumn(), - console=_console, - transient=False, - ) as prog: - task = prog.add_task("", total=100) - total = 1 - for line in proc.stdout: - captured.append(line) - m = re.match(r"Step (\d+)/(\d+) :", line) - if m: - step, total = int(m.group(1)), int(m.group(2)) - prog.update(task, completed=step / total * 100) - prog.update(task, completed=100) - + for line in proc.stdout: + s = line.rstrip() + if s: + write(s) proc.wait() if proc.returncode != 0: - _console.print("\n[red]Image build failed:[/red]") - _console.print("".join(captured), highlight=False) + write("[red]Image build failed.[/red]") return False + write("[green][ok][/green] Documentation image ready") return True -def _cmd_up() -> None: - if _is_running(): - _console.print(f"[green]Docs are already running at:[/green] http://localhost:{_DEFAULT_PORT}") - _console.print(" Stop with: [bold]cobot doc-setup down[/bold]") - return +def _task_up(screen: LogScreen, port: str) -> None: + try: + if _is_running(): + screen.write(f"[green]Docs already running at:[/green] http://localhost:{port}") + screen.write(" Stop with: [bold]cobot doc-setup down[/bold]") + screen.finish(True) + return - if not _DOC_DIR.exists(): - _console.print(f"[red]Doc directory not found:[/red] {_DOC_DIR}") - sys.exit(1) + if not _DOC_DIR.exists(): + screen.write(f"[red]Doc directory not found:[/red] {_DOC_DIR}") + screen.finish(False) + return - port = _Wizard().run() - if port is None: - return + if not _image_exists(): + if not _build_docs_image(screen.write): + screen.finish(False) + return + else: + screen.write("[dim]Documentation image already built, skipping.[/dim]") - _console.print() + screen.write("\n[cyan][*][/cyan] Starting MkDocs server...") + result = _docker( + "run", "-d", "--name", _CONTAINER_NAME, "--rm", + "-p", f"{port}:8000", + "-v", f"{_DOC_DIR}:/docs", + _IMAGE_NAME, "serve", "--dev-addr=0.0.0.0:8000", + capture=True, + ) + if result.returncode != 0: + screen.write(f"[red]Failed to start container.[/red]\n{result.stderr}") + screen.finish(False) + return - if not _image_exists(): - _console.print("[bold]Building documentation image...[/bold]") - if not _build_docs_image(): - sys.exit(1) - _console.print() - else: - _console.print("[dim]Documentation image already built, skipping...[/dim]\n") + screen.write(f"\n[green]Docs running at:[/green] http://localhost:{port}") + screen.write(" Edit files in [bold]doc/lwc-doc/docs/[/bold] — reloads automatically.") + screen.write(" Stop with: [bold]cobot doc-setup down[/bold]") + screen.finish(True) - _console.print("[bold]Starting MkDocs server...[/bold]") - _console.print(f"[dim]Mounting {_DOC_DIR} into container on port {port}...[/dim]") - - result = _docker( - "run", "-d", - "--name", _CONTAINER_NAME, - "--rm", - "-p", f"{port}:8000", - "-v", f"{_DOC_DIR}:/docs", - _IMAGE_NAME, - "serve", "--dev-addr=0.0.0.0:8000", - ) - if result.returncode != 0: - _console.print("[red]Failed to start container.[/red]") - sys.exit(1) - - _console.print(f"\n[green]Docs running at:[/green] http://localhost:{port}") - _console.print(" Edit files in [bold]doc/lwc-doc/docs/[/bold] — the site reloads automatically.") - _console.print(" Stop with: [bold]cobot doc-setup down[/bold]") + except Exception as exc: + screen.write(f"[red]Error:[/red] {exc}") + screen.finish(False) -def _cmd_down() -> None: - if not _is_running(): - _console.print("[yellow]Docs container is not running.[/yellow]") - return - - _console.print("[bold]Stopping documentation server...[/bold]") - _docker("stop", _CONTAINER_NAME) - _console.print("[green]Done.[/green] Container removed.") +def _task_down(screen: LogScreen) -> None: + try: + if not _is_running(): + screen.write("[yellow]Docs container is not running.[/yellow]") + screen.finish(True) + return + screen.write("[cyan][*][/cyan] Stopping documentation server...") + _docker("stop", _CONTAINER_NAME) + screen.write("[green][ok][/green] Container stopped.") + screen.finish(True) + except Exception as exc: + screen.write(f"[red]Error:[/red] {exc}") + screen.finish(False) +def _task_rebuild(screen: LogScreen, port: str) -> None: + try: + if _is_running(): + screen.write("[cyan][*][/cyan] Stopping existing container...") + _docker("stop", _CONTAINER_NAME) + screen.write("[green][ok][/green] Stopped.") -def _cmd_rebuild() -> None: - _cmd_down() - result = _docker("rmi", "-f", _IMAGE_NAME, capture=True) - if result.returncode != 0: - _console.print("[yellow]No image to remove, building fresh.[/yellow]") - _cmd_up() + if _image_exists(): + screen.write("[cyan][*][/cyan] Removing old image...") + _docker("rmi", "-f", _IMAGE_NAME) + screen.write("[green][ok][/green] Image removed.") + + if not _build_docs_image(screen.write): + screen.finish(False) + return + + screen.write("\n[cyan][*][/cyan] Starting MkDocs server...") + result = _docker( + "run", "-d", "--name", _CONTAINER_NAME, "--rm", + "-p", f"{port}:8000", + "-v", f"{_DOC_DIR}:/docs", + _IMAGE_NAME, "serve", "--dev-addr=0.0.0.0:8000", + capture=True, + ) + if result.returncode != 0: + screen.write(f"[red]Failed to start container.[/red]\n{result.stderr}") + screen.finish(False) + return + + screen.write(f"\n[green]Docs running at:[/green] http://localhost:{port}") + screen.write(" Stop with: [bold]cobot doc-setup down[/bold]") + screen.finish(True) + + except Exception as exc: + screen.write(f"[red]Error:[/red] {exc}") + screen.finish(False) + + +class _DocApp(App[None]): + CSS = SCREEN_CSS + + def __init__(self, action: str): + super().__init__() + self._action = action + + def on_mount(self) -> None: + if self._action == "down": + self.push_screen( + LogScreen("Documentation server", _task_down), + lambda _: self.exit(), + ) + elif self._action == "rebuild": + self.push_screen( + InputScreen("Step 1 of 1", "Port to serve documentation on:", _DEFAULT_PORT), + self._got_port_rebuild, + ) + else: + self.push_screen( + InputScreen("Step 1 of 1", "Port to serve documentation on:", _DEFAULT_PORT), + self._got_port_up, + ) + + def _got_port_up(self, port: Optional[str]) -> None: + if port is None: + self.exit() + return + p = (port.strip() or _DEFAULT_PORT) if port.isdigit() or not port.strip() else _DEFAULT_PORT + self.push_screen( + LogScreen("Documentation server", lambda s: _task_up(s, p)), + lambda _: self.exit(), + ) + + def _got_port_rebuild(self, port: Optional[str]) -> None: + if port is None: + self.exit() + return + p = (port.strip() or _DEFAULT_PORT) if port.isdigit() or not port.strip() else _DEFAULT_PORT + self.push_screen( + LogScreen("Documentation server — rebuild", lambda s: _task_rebuild(s, p)), + lambda _: self.exit(), + ) def register(subparsers: argparse._SubParsersAction) -> None: @@ -183,13 +211,9 @@ def register(subparsers: argparse._SubParsersAction) -> None: def run(args: argparse.Namespace) -> None: if not shutil.which("docker"): - _console.print("[red]Error:[/red] Docker is not installed or not on PATH.") + from rich.console import Console + Console().print("[red]Error:[/red] Docker is not installed or not on PATH.") sys.exit(1) action = getattr(args, "action", "up") - if action == "down": - _cmd_down() - elif action == "rebuild": - _cmd_rebuild() - else: - _cmd_up() + _DocApp(action).run() diff --git a/cobot/commands/docker_setup.py b/cobot/commands/docker_setup.py index 246088e..72f9dbe 100644 --- a/cobot/commands/docker_setup.py +++ b/cobot/commands/docker_setup.py @@ -8,15 +8,11 @@ import subprocess import sys from dataclasses import dataclass from pathlib import Path -from typing import List, Optional +from typing import Callable, List, Optional -from rich.console import Console -from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn from textual.app import App -from cobot.tui import SCREEN_CSS, InputScreen, PickScreen - -_console = Console() +from cobot.tui import SCREEN_CSS, InputScreen, LogScreen, PickScreen _PROJECT_DIR = Path(__file__).parent.parent.parent _DOCKER_DIR = _PROJECT_DIR / "docker" @@ -33,22 +29,128 @@ _IMAGE_PARENT: dict[str, str | None] = { "ros-iiwa7-webots": "ros-base", } -# Images that COPY from src/ — need project root as build context. -# Others use their Dockerfile's own directory. _NEEDS_PROJECT_CTX = {"ros-iiwa7", "ros-iiwa7-webots"} +Write = Callable[[str], None] + @dataclass class _Config: ros_version: str - variant: str # "controller" | "webots" - source: str # "build" | "pull" - build_type: str # "release" | "dev" + variant: str + source: str + build_type: str image_prefix: str hub_repo: str -class _Wizard(App[Optional[_Config]]): +def _stream(cmd: List[str], write: Write, env=None) -> bool: + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, env=env or os.environ, + ) + for line in proc.stdout: + s = line.rstrip() + if s: + write(s) + proc.wait() + return proc.returncode == 0 + + +def _build_image( + name: str, tag: str, dockerfile: Path, ctx: Path, + write: Write, + parent_tag: Optional[str] = None, + build_type: str = "release", +) -> bool: + write(f"[cyan][*][/cyan] Building [bold]{name}[/bold]...") + env = {**os.environ, "DOCKER_BUILDKIT": "0"} + cmd = [ + "docker", "build", "-t", tag, "-f", str(dockerfile), + "--build-arg", f"BUILD_TYPE={build_type}", + ] + if parent_tag: + cmd += ["--build-arg", f"IMAGE={parent_tag}"] + cmd.append(str(ctx)) + ok = _stream(cmd, write, env) + if ok: + write(f"[green][ok][/green] {name}") + else: + write(f"[red]Build failed:[/red] {tag}") + return ok + + +def _pull_image(name: str, tag: str, write: Write) -> bool: + write(f"[cyan][*][/cyan] Pulling [bold]{name}[/bold] ({tag})...") + ok = _stream(["docker", "pull", tag], write) + if ok: + write(f"[green][ok][/green] {name}") + else: + write(f"[red]Pull failed:[/red] {tag}") + return ok + + +def _task_execute(screen: LogScreen, cfg: _Config) -> None: + try: + chain = _WEBOTS_CHAIN if cfg.variant == "webots" else _CONTROLLER_CHAIN + + if cfg.source == "build": + screen.write( + f"[bold]Building {len(chain)} image(s) — " + f"ROS {cfg.ros_version} — {cfg.build_type}[/bold]\n" + ) + for name in chain: + tag = f"{cfg.image_prefix}:{name}-{cfg.ros_version}" + dockerfile = _DOCKER_DIR / cfg.ros_version / name / "Dockerfile" + if not dockerfile.exists(): + screen.write(f"[red]Dockerfile not found:[/red] {dockerfile}") + screen.finish(False) + return + ctx = _PROJECT_DIR if name in _NEEDS_PROJECT_CTX else dockerfile.parent + parent_name = _IMAGE_PARENT.get(name) + parent_tag = ( + f"{cfg.image_prefix}:{parent_name}-{cfg.ros_version}" + if parent_name else None + ) + if not _build_image(name, tag, dockerfile, ctx, screen.write, parent_tag, cfg.build_type): + screen.finish(False) + return + screen.write( + f"\n[green]Done.[/green] " + f"Images tagged [bold]{cfg.image_prefix}:-{cfg.ros_version}[/bold]." + ) + + else: + short = "webots" if cfg.variant == "webots" else "iiwa" + suffix = "-dev" if cfg.build_type == "dev" else "" + full_ref = f"{cfg.hub_repo}:{short}-{cfg.ros_version}{suffix}" + screen.write( + f"[bold]Pulling from {cfg.hub_repo} — " + f"ROS {cfg.ros_version} — {cfg.build_type}[/bold]\n" + ) + if not _pull_image(short, full_ref, screen.write): + screen.finish(False) + return + screen.write(f"\n[green]Done.[/green] Image ready: [bold]{full_ref}[/bold].") + + screen.finish(True) + + except Exception as exc: + screen.write(f"\n[red]Error:[/red] {exc}") + screen.finish(False) + + +def _discover_versions() -> List[str]: + if not _DOCKER_DIR.exists(): + return ["jazzy"] + dirs = sorted(d.name for d in _DOCKER_DIR.iterdir() if d.is_dir()) + if "jazzy" in dirs: + dirs = ["jazzy"] + [d for d in dirs if d != "jazzy"] + return dirs or ["jazzy"] + + +class _Wizard(App[None]): CSS = SCREEN_CSS def __init__(self, versions: List[str], default_version: str = "jazzy"): @@ -60,8 +162,6 @@ class _Wizard(App[Optional[_Config]]): def on_mount(self) -> None: self._ask_version() - # ── step helpers ────────────────────────────────────────────────────────── - def _ask_version(self) -> None: self.push_screen( PickScreen("Step 1 of 5", "Select ROS version:", self.versions, self.default_version), @@ -70,17 +170,21 @@ class _Wizard(App[Optional[_Config]]): def _got_version(self, v: Optional[str]) -> None: if v is None: - self.exit(None) + self.exit() return self._state["ros_version"] = v self.push_screen( - PickScreen("Step 2 of 5", "Source:", ["Pull from Docker Hub", "Build locally"], "Pull from Docker Hub"), + PickScreen( + "Step 2 of 5", "Source:", + ["Pull from Docker Hub", "Build locally"], + "Pull from Docker Hub", + ), self._got_source, ) def _got_source(self, v: Optional[str]) -> None: if v is None: - self.exit(None) + self.exit() return self._state["source"] = "build" if v == "Build locally" else "pull" self.push_screen( @@ -98,7 +202,7 @@ class _Wizard(App[Optional[_Config]]): def _got_variant(self, v: Optional[str]) -> None: if v is None: - self.exit(None) + self.exit() return self._state["variant"] = "webots" if v.startswith("Controller with Webots") else "controller" self.push_screen( @@ -108,7 +212,7 @@ class _Wizard(App[Optional[_Config]]): def _got_build_type(self, v: Optional[str]) -> None: if v is None: - self.exit(None) + self.exit() return self._state["build_type"] = v or "release" if self._state["source"] == "pull": @@ -124,215 +228,50 @@ class _Wizard(App[Optional[_Config]]): def _got_hub_repo(self, v: Optional[str]) -> None: if v is None: - self.exit(None) + self.exit() return self._state["hub_repo"] = v self._finish() def _got_prefix(self, v: Optional[str]) -> None: if v is None: - self.exit(None) + self.exit() return self._state["image_prefix"] = v self._finish() def _finish(self) -> None: s = self._state - self.exit(_Config( + cfg = _Config( ros_version=s["ros_version"], variant=s["variant"], source=s["source"], build_type=s["build_type"], image_prefix=s.get("image_prefix", _DEFAULT_PREFIX), hub_repo=s.get("hub_repo", _DEFAULT_HUB_REPO), - )) - - -def _discover_versions() -> List[str]: - if not _DOCKER_DIR.exists(): - return ["jazzy"] - dirs = sorted(d.name for d in _DOCKER_DIR.iterdir() if d.is_dir()) - if "jazzy" in dirs: - dirs = ["jazzy"] + [d for d in dirs if d != "jazzy"] - return dirs or ["jazzy"] - - -def _build_image( - name: str, tag: str, dockerfile: Path, ctx: Path, - parent_tag: Optional[str] = None, - build_type: str = "release", -) -> bool: - env = {**os.environ, "DOCKER_BUILDKIT": "0"} - cmd = [ - "docker", "build", - "-t", tag, - "-f", str(dockerfile), - "--build-arg", f"BUILD_TYPE={build_type}", - ] - if parent_tag: - cmd += ["--build-arg", f"IMAGE={parent_tag}"] - cmd.append(str(ctx)) - proc = subprocess.Popen( - cmd, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, env=env, - ) - - captured: List[str] = [] - - with Progress( - SpinnerColumn(), - TextColumn(f" [bold cyan]{name:<24}[/bold cyan]"), - BarColumn(bar_width=32), - TaskProgressColumn(), - console=_console, - transient=False, - ) as prog: - task = prog.add_task("", total=100) - total = 1 - for line in proc.stdout: - captured.append(line) - m = re.match(r"Step (\d+)/(\d+) :", line) - if m: - step, total = int(m.group(1)), int(m.group(2)) - prog.update(task, completed=step / total * 100) - prog.update(task, completed=100) - - proc.wait() - if proc.returncode != 0: - _console.print(f"\n[red]Build failed:[/red] {tag}\n") - _console.print("".join(captured), highlight=False) - return False - return True - - -def _parse_docker_size(s: str) -> float: - s = s.strip() - for suffix, mult in [("GB", 1e9), ("MB", 1e6), ("kB", 1e3), ("B", 1.0)]: - if s.endswith(suffix): - try: - return float(s[: -len(suffix)]) * mult - except ValueError: - return 0.0 - try: - return float(s) - except ValueError: - return 0.0 - - -_RE_PULLING = re.compile(r"^([a-f0-9]+): Pulling fs layer") -_RE_DONE = re.compile(r"^([a-f0-9]+): (?:Pull complete|Already exists|Layer already exists)") -_RE_DL = re.compile(r"^([a-f0-9]+): Downloading(?:\s+\[.*?\])?\s+([\d.]+\s*\w+)/([\d.]+\s*\w+)") - - -def _pull_image(name: str, tag: str) -> bool: - proc = subprocess.Popen( - ["docker", "pull", tag], - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, - ) - - captured: List[str] = [] - layers_pulling: set[str] = set() - layers_done: set[str] = set() - layers_total: dict[str, float] = {} - layers_current: dict[str, float] = {} - has_bytes = False - - with Progress( - SpinnerColumn(), - TextColumn(f" [bold cyan]{name:<24}[/bold cyan]"), - BarColumn(bar_width=32), - TaskProgressColumn(), - console=_console, - transient=False, - ) as prog: - task = prog.add_task("", total=100) - - for line in proc.stdout: - captured.append(line) - line = line.strip() - - if m := _RE_PULLING.match(line): - layers_pulling.add(m.group(1)) - - elif m := _RE_DL.match(line): - lid, cur, tot = m.group(1), _parse_docker_size(m.group(2)), _parse_docker_size(m.group(3)) - if tot > 0: - has_bytes = True - layers_current[lid] = cur - layers_total[lid] = tot - - elif m := _RE_DONE.match(line): - lid = m.group(1) - layers_done.add(lid) - if lid in layers_total: - layers_current[lid] = layers_total[lid] - - if has_bytes and layers_total: - total_b = sum(layers_total.values()) - done_b = sum(layers_current.get(lid, 0.0) for lid in layers_total) - prog.update(task, completed=done_b / total_b * 100) - elif layers_pulling: - prog.update(task, completed=len(layers_done) / len(layers_pulling) * 100) - - prog.update(task, completed=100) - - proc.wait() - if proc.returncode != 0: - _console.print(f"\n[red]Pull failed:[/red] {tag}\n") - _console.print("".join(captured), highlight=False) - return False - return True - - -def _execute(cfg: _Config) -> None: - chain = _WEBOTS_CHAIN if cfg.variant == "webots" else _CONTROLLER_CHAIN - _console.print() - - if cfg.source == "build": - _console.print( - f"[bold]Building {len(chain)} image(s) • ROS {cfg.ros_version} • {cfg.build_type}[/bold]\n" ) - for name in chain: - tag = f"{cfg.image_prefix}:{name}-{cfg.ros_version}" - dockerfile = _DOCKER_DIR / cfg.ros_version / name / "Dockerfile" - if not dockerfile.exists(): - _console.print(f"[red]Dockerfile not found:[/red] {dockerfile}") - sys.exit(1) - ctx = _PROJECT_DIR if name in _NEEDS_PROJECT_CTX else dockerfile.parent - parent_name = _IMAGE_PARENT.get(name) - parent_tag = f"{cfg.image_prefix}:{parent_name}-{cfg.ros_version}" if parent_name else None - if not _build_image(name, tag, dockerfile, ctx, parent_tag, cfg.build_type): - sys.exit(1) - _console.print(f"\n[green]Done.[/green] Images: [bold]{cfg.image_prefix}:{{name}}-{cfg.ros_version}[/bold].") - - else: - short = "webots" if cfg.variant == "webots" else "iiwa" - suffix = "-dev" if cfg.build_type == "dev" else "" - image_tag = f"{short}-{cfg.ros_version}{suffix}" - full_ref = f"{cfg.hub_repo}:{image_tag}" - _console.print(f"[bold]Pulling from {cfg.hub_repo} • ROS {cfg.ros_version} • {cfg.build_type}[/bold]\n") - if not _pull_image(short, full_ref): - sys.exit(1) - _console.print(f"\n[green]Done.[/green] Image ready: [bold]{full_ref}[/bold].") + title = ( + f"Building Docker images — ROS {cfg.ros_version}" + if cfg.source == "build" + else f"Pulling Docker image — ROS {cfg.ros_version}" + ) + self.push_screen( + LogScreen(title, lambda screen: _task_execute(screen, cfg)), + lambda _: self.exit(), + ) def register(subparsers: argparse._SubParsersAction) -> None: - p = subparsers.add_parser("docker-setup", help="Set up Docker images for KUKA iiwa7") + p = subparsers.add_parser("docker-setup", help="Build or pull Docker images for KUKA iiwa7") p.set_defaults(func=run) def run(args: argparse.Namespace) -> None: if not shutil.which("docker"): - _console.print("[red]Error:[/red] Docker is not installed or not on PATH.") + from rich.console import Console + Console().print("[red]Error:[/red] Docker is not installed or not on PATH.") sys.exit(1) versions = _discover_versions() default = "jazzy" if "jazzy" in versions else versions[0] - cfg = _Wizard(versions=versions, default_version=default).run() - - if cfg is None: - return - - _execute(cfg) + _Wizard(versions=versions, default_version=default).run() diff --git a/cobot/commands/local_setup.py b/cobot/commands/local_setup.py index 8110ce4..0ba9329 100644 --- a/cobot/commands/local_setup.py +++ b/cobot/commands/local_setup.py @@ -4,32 +4,30 @@ import argparse import os import shutil import subprocess -import sys import tempfile import urllib.request from pathlib import Path -from typing import List, Optional +from typing import Callable, List, Optional -from rich.console import Console from textual.app import App -from cobot.tui import SCREEN_CSS, PickScreen +from cobot.tui import SCREEN_CSS, LogScreen, PickScreen from cobot.commands.docker_setup import run as _docker_setup -_console = Console() _PROJECT_DIR = Path(__file__).parent.parent.parent _ROS_KEYRING = Path("/usr/share/keyrings/ros-archive-keyring.gpg") _ROS_SOURCES = Path("/etc/apt/sources.list.d/ros2.list") _ROS_KEY_URL = "https://raw.githubusercontent.com/ros/rosdistro/master/ros.key" +_APT_ENV = {**os.environ, "DEBIAN_FRONTEND": "noninteractive"} def _detect_ubuntu_2404() -> bool: - os_release = Path("/etc/os-release") - if not os_release.exists(): + path = Path("/etc/os-release") + if not path.exists(): return False info: dict[str, str] = {} - for line in os_release.read_text().splitlines(): + for line in path.read_text().splitlines(): if "=" in line: k, _, v = line.partition("=") info[k.strip()] = v.strip().strip('"') @@ -40,142 +38,159 @@ def _detect_ros2_jazzy() -> bool: return Path("/opt/ros/jazzy").is_dir() -def _run(cmd: List[str]) -> None: - result = subprocess.run(cmd) +Write = Callable[[str], None] + + +def _run_quiet(cmd: List[str]) -> None: + result = subprocess.run(cmd, capture_output=True) if result.returncode != 0: - raise RuntimeError(f"Command failed: {' '.join(cmd)}") + raise RuntimeError(f"Exit {result.returncode}: {cmd[0]}") -def _check_output(cmd: List[str]) -> str: - return subprocess.check_output(cmd, text=True).strip() +def _run_logged(cmd: List[str], write: Write, env: dict | None = None, cwd=None) -> None: + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + env=env or os.environ, + cwd=cwd, + ) + for line in proc.stdout: + s = line.rstrip() + if s: + write(s) + proc.wait() + if proc.returncode != 0: + raise RuntimeError(f"Exit {proc.returncode}: {cmd[0]}") -def _step(msg: str) -> None: - _console.print(f"[cyan][*][/cyan] {msg}") - - -def _ok(msg: str) -> None: - _console.print(f"[green][ok][/green] {msg}") - - -def _apt_install(*packages: str) -> None: - _run(["sudo", "apt-get", "install", "-y", "--no-install-recommends", *packages]) - - -def _setup_locale() -> None: - _step("Checking locale...") - result = subprocess.run(["locale"], capture_output=True, text=True) - if "UTF-8" in result.stdout: - _ok("UTF-8 locale active") +def _setup_locale(write: Write) -> None: + write("[cyan][*][/cyan] Checking locale...") + if "UTF-8" in subprocess.run(["locale"], capture_output=True, text=True).stdout: + write("[green][ok][/green] UTF-8 locale active") return - _run(["sudo", "apt-get", "update", "-qq"]) - _apt_install("locales") - _run(["sudo", "locale-gen", "en_US.UTF-8"]) - _run(["sudo", "update-locale", "LC_ALL=en_US.UTF-8", "LANG=en_US.UTF-8"]) - _ok("Locale configured") + write("[cyan][*][/cyan] Configuring UTF-8 locale...") + _run_quiet(["sudo", "apt-get", "update", "-qq"]) + _run_logged(["sudo", "apt-get", "install", "-y", "--no-install-recommends", "locales"], write, _APT_ENV) + _run_logged(["sudo", "locale-gen", "en_US.UTF-8"], write) + _run_quiet(["sudo", "update-locale", "LC_ALL=en_US.UTF-8", "LANG=en_US.UTF-8"]) + write("[green][ok][/green] Locale configured") -def _add_ros2_repo() -> None: - _step("Adding ROS2 apt repository...") - _run(["sudo", "apt-get", "update", "-qq"]) - _apt_install("software-properties-common", "curl") - _run(["sudo", "add-apt-repository", "-y", "universe"]) +def _add_ros2_repo(write: Write) -> None: + write("[cyan][*][/cyan] Adding ROS2 apt repository...") + _run_quiet(["sudo", "apt-get", "update", "-qq"]) + _run_logged( + ["sudo", "apt-get", "install", "-y", "--no-install-recommends", + "software-properties-common", "curl"], + write, _APT_ENV, + ) + _run_quiet(["sudo", "add-apt-repository", "-y", "universe"]) if not _ROS_KEYRING.exists(): - _step("Downloading ROS2 signing key...") + write("[cyan][*][/cyan] Downloading ROS2 signing key...") with tempfile.NamedTemporaryFile(delete=False, suffix=".gpg") as tmp: tmp_path = tmp.name try: urllib.request.urlretrieve(_ROS_KEY_URL, tmp_path) - _run(["sudo", "cp", tmp_path, str(_ROS_KEYRING)]) + _run_quiet(["sudo", "cp", tmp_path, str(_ROS_KEYRING)]) finally: os.unlink(tmp_path) if not _ROS_SOURCES.exists(): - arch = _check_output(["dpkg", "--print-architecture"]) - codename = _check_output( - ["bash", "-c", ". /etc/os-release && echo $UBUNTU_CODENAME"] - ) + arch = subprocess.check_output(["dpkg", "--print-architecture"], text=True).strip() + codename = subprocess.check_output( + ["bash", "-c", ". /etc/os-release && echo $UBUNTU_CODENAME"], text=True + ).strip() sources_line = ( f"deb [arch={arch} signed-by={_ROS_KEYRING}] " f"http://packages.ros.org/ros2/ubuntu {codename} main\n" ) proc = subprocess.run( ["sudo", "tee", str(_ROS_SOURCES)], - input=sources_line, - capture_output=True, - text=True, + input=sources_line, capture_output=True, text=True, ) if proc.returncode != 0: raise RuntimeError(f"Failed to write {_ROS_SOURCES}") - _run(["sudo", "apt-get", "update", "-qq"]) - _ok("ROS2 repository ready") + _run_quiet(["sudo", "apt-get", "update", "-qq"]) + write("[green][ok][/green] ROS2 repository ready") -def _install_ros2_jazzy() -> None: - _step("Installing ros-jazzy-ros-base + ros-dev-tools...") - _apt_install("ros-jazzy-ros-base", "ros-dev-tools") - _ok("ROS2 Jazzy installed") +def _install_ros2_jazzy(write: Write) -> None: + write("[cyan][*][/cyan] Installing ros-jazzy-ros-base and ros-dev-tools...") + _run_logged( + ["sudo", "apt-get", "install", "-y", "--no-install-recommends", + "ros-jazzy-ros-base", "ros-dev-tools"], + write, _APT_ENV, + ) + write("[green][ok][/green] ROS2 Jazzy installed") -def _install_colcon() -> None: +def _install_colcon(write: Write) -> None: if shutil.which("colcon"): - _ok("colcon already available") + write("[green][ok][/green] colcon already available") return - _step("Installing colcon...") - _apt_install("python3-colcon-common-extensions") - _ok("colcon installed") + write("[cyan][*][/cyan] Installing colcon...") + _run_logged( + ["sudo", "apt-get", "install", "-y", "--no-install-recommends", + "python3-colcon-common-extensions"], + write, _APT_ENV, + ) + write("[green][ok][/green] colcon installed") -def _setup_shell_rc() -> None: +def _setup_shell_rc(write: Write) -> None: shell_name = Path(os.environ.get("SHELL", "/bin/bash")).name rc = Path.home() / (".zshrc" if shell_name == "zsh" else ".bashrc") source_line = "source /opt/ros/jazzy/setup.bash" if rc.exists() and source_line in rc.read_text(): - _ok(f"ROS2 setup already in {rc.name}") + write(f"[green][ok][/green] ROS2 setup already in {rc.name}") return with rc.open("a") as f: f.write(f"\n# ROS2 Jazzy\n{source_line}\n") - _ok(f"Added ROS2 setup to ~/{rc.name}") + write(f"[green][ok][/green] Added ROS2 setup to ~/{rc.name}") -def _install_jazzy() -> None: - _console.print("\n[bold]Installing ROS2 Jazzy...[/bold]\n") +def _task_install_jazzy(screen: LogScreen) -> None: try: - _setup_locale() - _add_ros2_repo() - _install_ros2_jazzy() - _install_colcon() - _setup_shell_rc() - except (subprocess.CalledProcessError, RuntimeError) as exc: - _console.print(f"\n[red]Installation failed:[/red] {exc}") - sys.exit(1) - _console.print( - "\n[green]ROS2 Jazzy installed.[/green] " - "Restart the terminal, then run [bold]cobot local-setup[/bold] again to build." - ) + screen.write("[bold]Installing ROS2 Jazzy[/bold]\n") + _setup_locale(screen.write) + _add_ros2_repo(screen.write) + _install_ros2_jazzy(screen.write) + _install_colcon(screen.write) + _setup_shell_rc(screen.write) + screen.write( + "\nRestart the terminal, then run [bold]cobot local-setup[/bold] again to build." + ) + screen.finish(True) + except Exception as exc: + screen.write(f"\n[red]Error:[/red] {exc}") + screen.finish(False) -def _build_project() -> None: - if not shutil.which("colcon"): - _console.print("[red]Error:[/red] colcon not found. Source ROS2 first:") - _console.print(" [bold]source /opt/ros/jazzy/setup.bash[/bold]") - sys.exit(1) - - _console.print("\n[bold]Building project with colcon...[/bold]\n") - result = subprocess.run( - ["colcon", "build", "--symlink-install"], - cwd=_PROJECT_DIR, - ) - if result.returncode != 0: - _console.print("\n[red]Build failed.[/red]") - sys.exit(result.returncode) - _console.print("\n[green]Build complete.[/green]") - _console.print(" Activate workspace: [bold]source install/setup.bash[/bold]") +def _task_build(screen: LogScreen) -> None: + try: + if not shutil.which("colcon"): + screen.write("[red]colcon not found.[/red]") + screen.write("Source ROS2 first: [bold]source /opt/ros/jazzy/setup.bash[/bold]") + screen.finish(False) + return + screen.write("[bold]Building project with colcon[/bold]\n") + _run_logged( + ["colcon", "build", "--symlink-install"], + screen.write, + cwd=_PROJECT_DIR, + ) + screen.write("\nActivate workspace: [bold]source install/setup.bash[/bold]") + screen.finish(True) + except Exception as exc: + screen.write(f"\n[red]Error:[/red] {exc}") + screen.finish(False) -class _AskInstallJazzy(App[Optional[str]]): +class _InstallJazzyApp(App[None]): CSS = SCREEN_CSS def on_mount(self) -> None: @@ -186,11 +201,30 @@ class _AskInstallJazzy(App[Optional[str]]): ["Yes, install ROS2 Jazzy", "No, skip"], "Yes, install ROS2 Jazzy", ), - self.exit, + self._on_choice, + ) + + def _on_choice(self, choice: Optional[str]) -> None: + if choice is None or choice.startswith("No"): + self.exit() + return + self.push_screen( + LogScreen("Installing ROS2 Jazzy", _task_install_jazzy), + lambda _: self.exit(), ) -class _AskDockerSetup(App[Optional[str]]): +class _BuildApp(App[None]): + CSS = SCREEN_CSS + + def on_mount(self) -> None: + self.push_screen( + LogScreen("Building project", _task_build), + lambda _: self.exit(), + ) + + +class _DockerPromptApp(App[bool]): CSS = SCREEN_CSS def on_mount(self) -> None: @@ -201,55 +235,26 @@ class _AskDockerSetup(App[Optional[str]]): ["Yes, run docker-setup", "No, exit"], "Yes, run docker-setup", ), - self.exit, + lambda v: self.exit(v is not None and v.startswith("Yes")), ) -def _flow_ubuntu_without_ros() -> None: - _console.print( - "\n[yellow]Warning:[/yellow] ROS2 Jazzy is not installed " - "(/opt/ros/jazzy not found).\n" - ) - answer = _AskInstallJazzy().run() - if answer is None or answer.startswith("No"): - _console.print("[yellow]Skipped ROS2 installation.[/yellow]") - return - _install_jazzy() - - -def _flow_not_ubuntu(args: argparse.Namespace) -> None: - _console.print( - "\n[yellow]Warning:[/yellow] Ubuntu 24.04 not detected. " - "Native build is not supported on this OS.\n" - ) - answer = _AskDockerSetup().run() - if answer is None or answer.startswith("No"): - _console.print("[yellow]Exiting without changes.[/yellow]") - return - - _docker_setup(args) - - def register(subparsers: argparse._SubParsersAction) -> None: p = subparsers.add_parser( "local-setup", - help="Build the project locally (requires Ubuntu 24.04 and ROS2 Jazzy)", + help="Install ROS2 Jazzy natively and build the project with colcon", ) p.set_defaults(func=run) def run(args: argparse.Namespace) -> None: - _console.print("[bold]Checking environment...[/bold]") - if not _detect_ubuntu_2404(): - _flow_not_ubuntu(args) + if _DockerPromptApp().run(): + _docker_setup(args) return - _console.print("[green] Ubuntu 24.04[/green] ✓") - if not _detect_ros2_jazzy(): - _flow_ubuntu_without_ros() + _InstallJazzyApp().run() return - _console.print("[green] ROS2 Jazzy[/green] ✓\n") - _build_project() + _BuildApp().run() diff --git a/cobot/commands/robot_setup.py b/cobot/commands/robot_setup.py index 7ed4c6b..7db29b4 100644 --- a/cobot/commands/robot_setup.py +++ b/cobot/commands/robot_setup.py @@ -6,13 +6,14 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable, List, Optional, Tuple -from rich.console import Console from ruamel.yaml import YAML -from textual.app import App +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.screen import Screen +from textual.widgets import Footer, Static from cobot.tui import SCREEN_CSS, InputScreen, PickScreen -_console = Console() _PROJECT_DIR = Path(__file__).parent.parent.parent _CONFIG_PATH = _PROJECT_DIR / "cobot-setting.yaml" @@ -20,17 +21,13 @@ _yaml = YAML() _yaml.preserve_quotes = True -# --------------------------------------------------------------------------- -# Field descriptors -# --------------------------------------------------------------------------- - @dataclass class _Field: key: str # dot-separated path within the block, e.g. "webots.world" question: str default: Any note: str = "" - options: Optional[List[str]] = None # if set → PickScreen, else → InputScreen + options: Optional[List[str]] = None # if set - PickScreen, else - InputScreen def label(self) -> str: return self.key.split(".")[-1] @@ -43,10 +40,6 @@ class _Block: fields: List[_Field] -# --------------------------------------------------------------------------- -# Block definitions — bottom to top order -# --------------------------------------------------------------------------- - _BLOCKS: List[_Block] = [ _Block( yaml_key="foxglove", @@ -117,10 +110,6 @@ _BLOCKS: List[_Block] = [ ] -# --------------------------------------------------------------------------- -# Wizard app -# --------------------------------------------------------------------------- - def _coerce(value: str, original: Any) -> Any: """Try to preserve the original YAML scalar type.""" if isinstance(original, bool): @@ -157,7 +146,20 @@ def _set_nested(mapping: Any, path: str, value: Any) -> None: cur[keys[-1]] = _coerce(value, original) -class _Wizard(App[bool]): +class _SavedScreen(Screen[None]): + BINDINGS = [Binding("enter,escape", "close", "Close")] + + def compose(self) -> ComposeResult: + yield Static("Done", id="step") + yield Static(f"Configuration saved to {_CONFIG_PATH.name}", id="question") + yield Static("Press Enter to close.", id="note") + yield Footer() + + def action_close(self) -> None: + self.dismiss(None) + + +class _Wizard(App[None]): CSS = SCREEN_CSS def __init__(self, data: Any): @@ -172,13 +174,11 @@ class _Wizard(App[bool]): def on_mount(self) -> None: self._next_block() - # ------------------------------------------------------------------ - # Block-level flow - # ------------------------------------------------------------------ def _next_block(self) -> None: if self._block_idx >= len(self._blocks): - self.exit(True) + _save_config(self._data) + self.push_screen(_SavedScreen(), lambda _: self.exit()) return block = self._blocks[self._block_idx] total = len(self._blocks) @@ -195,7 +195,7 @@ class _Wizard(App[bool]): def _got_block_choice(self, v: Optional[str], block: _Block) -> None: if v is None: - self.exit(False) + self.exit() return self._block_idx += 1 if v == "Yes": @@ -206,9 +206,6 @@ class _Wizard(App[bool]): else: self._next_block() - # ------------------------------------------------------------------ - # Field-level flow - # ------------------------------------------------------------------ def _next_field(self) -> None: if not self._pending_fields: @@ -240,7 +237,7 @@ class _Wizard(App[bool]): def _got_field(self, v: Optional[str], f: _Field) -> None: if v is None: - self.exit(False) + self.exit() return block = self._current_block _set_nested(self._data[block.yaml_key], f.key, v) @@ -248,9 +245,6 @@ class _Wizard(App[bool]): self._next_field() -# --------------------------------------------------------------------------- -# YAML read / write -# --------------------------------------------------------------------------- def _load_config() -> Any: with open(_CONFIG_PATH, "r", encoding="utf-8") as fh: @@ -262,10 +256,6 @@ def _save_config(data: Any) -> None: _yaml.dump(data, fh) -# --------------------------------------------------------------------------- -# CLI entry points -# --------------------------------------------------------------------------- - def register(subparsers: argparse._SubParsersAction) -> None: p = subparsers.add_parser("robot-setup", help="Configure cobot-setting.yaml interactively") p.set_defaults(func=run) @@ -273,18 +263,9 @@ def register(subparsers: argparse._SubParsersAction) -> None: def run(args: argparse.Namespace) -> None: if not _CONFIG_PATH.exists(): - _console.print(f"[red]Config not found:[/red] {_CONFIG_PATH}") + from rich.console import Console + Console().print(f"[red]Config not found:[/red] {_CONFIG_PATH}") sys.exit(1) data = _load_config() - ok = _Wizard(data).run() - if not ok: - _console.print("[yellow]Setup cancelled.[/yellow]") - return - - _save_config(data) - _console.print(f"\n[green]Configuration saved:[/green] {_CONFIG_PATH}") - _console.print( - " Start the robot container with: [bold]cobot robot-setup[/bold] " - "then run [bold]./docker/jazzy/ros-iiwa7-webots/run.sh[/bold]" - ) + _Wizard(data).run() diff --git a/cobot/tui.py b/cobot/tui.py index f86c3ab..79977c5 100644 --- a/cobot/tui.py +++ b/cobot/tui.py @@ -1,12 +1,12 @@ from __future__ import annotations -from typing import List, Optional +from typing import Callable, List, Optional 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, RadioButton, RadioSet, Static +from textual.widgets import Footer, Input, RadioButton, RadioSet, RichLog, Static SCREEN_CSS = """ Screen { @@ -36,6 +36,16 @@ RadioSet { Input { margin-bottom: 1; } +LogScreen #log { + height: 1fr; + border: none; + padding: 0 1; + margin-top: 1; +} +LogScreen #hint { + margin-top: 1; + color: $text; +} """ @@ -114,3 +124,49 @@ class InputScreen(Screen[Optional[str]]): def action_abort(self) -> None: self.app.exit(None) + + +class LogScreen(Screen[bool]): + """Streams task output into a scrollable log; press Enter to close when done.""" + + BINDINGS = [Binding("enter,escape", "close", "Close", show=False)] + + def __init__(self, title: str, task: Callable[[LogScreen], None]): + super().__init__() + self._title = title + self._run_fn = task + self._finished = False + + def compose(self) -> ComposeResult: + yield Static(self._title, id="step") + yield RichLog(id="log", highlight=True, markup=True, wrap=True) + yield Static("", id="hint") + yield Footer() + + def on_mount(self) -> None: + self.query_one(RichLog).focus() + self.app.run_worker(lambda: self._run_fn(self), thread=True) + + def write(self, line: str) -> None: + """Thread-safe: append a line to the log.""" + self.app.call_from_thread(self._append, line) + + def _append(self, line: str) -> None: + self.query_one(RichLog).write(line) + + def finish(self, success: bool) -> None: + """Thread-safe: mark task done and prompt the user to close.""" + self.app.call_from_thread(self._do_finish, success) + + def _do_finish(self, success: bool) -> None: + self._finished = True + msg = ( + "[green]Done![/green] Press [bold]Enter[/bold] to close." + if success + else "[red]Failed.[/red] Press [bold]Enter[/bold] to close." + ) + self.query_one("#hint", Static).update(msg) + + def action_close(self) -> None: + if self._finished: + self.dismiss(True) diff --git a/install.sh b/install.sh index 01c9641..bcaeda7 100644 --- a/install.sh +++ b/install.sh @@ -13,7 +13,7 @@ BOLD='\033[1m' PYTHON_VERSION="3.11" REPO_URL="https://gitverse.ru/daniel-robotics/lightweight-cobot.git" -INSTALL_DIR="${COBOT_INSTALL_DIR:-$HOME/.lwc/ros2_iiwa7}" +INSTALL_DIR="${COBOT_INSTALL_DIR:-$HOME/.lwc}" # Определяем интерактивный режим: при запуске через curl | bash stdin не является терминалом if [ -t 0 ]; then IS_INTERACTIVE=true; else IS_INTERACTIVE=false; fi