grabardm/lightweight-cobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Refactor Docker and Local Setup Commands

Browse Source 
- 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.
This commit is contained in:
Даниил Грабарь
2026-05-20 17:41:11 +03:00
parent bbfec17876
commit 5f16825cf7
6 changed files with 507 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cobot/commands/doc_setup.py
+135 -111
View File
@@ -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)
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:
def _task_up(screen: LogScreen, port: str) -> None:
try:
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]")
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)
port = _Wizard().run()
if port is None:
screen.write(f"[red]Doc directory not found:[/red] {_DOC_DIR}")
screen.finish(False)
return
_console.print()
if not _image_exists():
_console.print("[bold]Building documentation image...[/bold]")
if not _build_docs_image():
sys.exit(1)
_console.print()
if not _build_docs_image(screen.write):
screen.finish(False)
return
else:
_console.print("[dim]Documentation image already built, skipping...[/dim]\n")
_console.print("[bold]Starting MkDocs server...[/bold]")
_console.print(f"[dim]Mounting {_DOC_DIR} into container on port {port}...[/dim]")
screen.write("[dim]Documentation image already built, skipping.[/dim]")
screen.write("\n[cyan][*][/cyan] Starting MkDocs server...")
result = _docker(
"run", "-d",
"--name", _CONTAINER_NAME,
"--rm",
"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",
_IMAGE_NAME, "serve", "--dev-addr=0.0.0.0:8000",
capture=True,
)
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]")
def _cmd_down() -> None:
if not _is_running():
_console.print("[yellow]Docs container is not running.[/yellow]")
screen.write(f"[red]Failed to start container.[/red]\n{result.stderr}")
screen.finish(False)
return
_console.print("[bold]Stopping documentation server...[/bold]")
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)
except Exception as exc:
screen.write(f"[red]Error:[/red] {exc}")
screen.finish(False)
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)
_console.print("[green]Done.[/green] Container removed.")
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 _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:
_console.print("[yellow]No image to remove, building fresh.[/yellow]")
_cmd_up()
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()
cobot/commands/docker_setup.py
+137 -198
View File
@@ -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}:<name>-{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,
title = (
f"Building Docker images — ROS {cfg.ros_version}"
if cfg.source == "build"
else f"Pulling Docker image — ROS {cfg.ros_version}"
)
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"
self.push_screen(
LogScreen(title, lambda screen: _task_execute(screen, cfg)),
lambda _: self.exit(),
)
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].")
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()
cobot/commands/local_setup.py
+131 -126
View File
@@ -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:
def _task_build(screen: LogScreen) -> None:
try:
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(
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,
)
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]")
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()
cobot/commands/robot_setup.py
+26 -45
View File
@@ -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()
cobot/tui.py
+58 -2
View File
@@ -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)
install.sh
+1 -1
View File
@@ -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