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

feat: update robot setup to use ruamel.yaml for preserving YAML comments and formatting

Browse Source 
feat: enhance run command with Docker volume for Webots asset caching and GPU detection

feat: improve setup command to sequentially run sub-commands for documentation, environment, and robot parameters

fix: update update command to provide clearer logging during git operations and installation

refactor: enhance TUI components with better threading and logging for long-running tasks

chore: improve install script with better error handling, logging, and interactive setup wizard
This commit is contained in:
Даниил Грабарь
2026-05-21 12:46:54 +10:00
parent e18ff53532
commit 6be6032838
11 changed files with 647 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cobot/cli.py
+8
View File
@@ -1,6 +1,8 @@
import argparse
import sys
# Import each command module so we can register its subparser.
# Импортируем каждый модуль команды, чтобы зарегистрировать его подпарсер.
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
@@ -12,6 +14,8 @@ from cobot.commands import update as cmd_update
# Command groups shown in --help output.
# Add new commands here when introducing other categories.
# Группы команд, отображаемые в --help.
# Добавляйте новые команды сюда при создании новых категорий.
_GROUPS = [
("Setup", [
("setup", "first-time setup: docs, build environment, robot config"),
@@ -32,6 +36,8 @@ _GROUPS = [
_DESCRIPTION = "Lightweight Cobot"
# Custom --help action that prints commands grouped by category instead of a flat list.
# Кастомный обработчик --help, который выводит команды по категориям, а не одним списком.
class _GroupedHelpAction(argparse.Action):
def __init__(self, option_strings, dest, default=None, required=False, help=None):
super().__init__(
@@ -79,6 +85,8 @@ def main():
def _register_commands(subparsers):
# Each module registers its own subparser and sets args.func to its run() function.
# Каждый модуль регистрирует свой подпарсер и устанавливает args.func на свою функцию run().
cmd_setup.register(subparsers)
cmd_local_setup.register(subparsers)
cmd_docker_setup.register(subparsers)
cobot/commands/delete.py
+121 -33
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
import argparse
import os
import shutil
import subprocess
from pathlib import Path
@@ -14,6 +13,8 @@ from cobot.tui import SCREEN_CSS, LogScreen, PickScreen
_PROJECT_DIR = Path(__file__).parent.parent.parent
# Stop and remove all Docker containers whose name contains "lwc".
# Останавливаем и удаляем все Docker-контейнеры, чьё имя содержит "lwc".
def _stop_docker_containers(write) -> None:
write("[cyan][*][/cyan] Stopping Docker containers...")
result = subprocess.run(
@@ -29,6 +30,8 @@ def _stop_docker_containers(write) -> None:
write(f"[green][ok][/green] Removed container: {name}")
# Remove all Docker images whose repository or tag contains "lwc".
# Удаляем все Docker-образы, репозиторий или тег которых содержит "lwc".
def _remove_docker_images(write) -> None:
write("[cyan][*][/cyan] Removing Docker images...")
result = subprocess.run(
@@ -47,14 +50,50 @@ def _remove_docker_images(write) -> None:
write(f"[green][ok][/green] Removed image: {img}")
def _remove_ros2(write) -> None:
write("[cyan][*][/cyan] Removing ROS2 Jazzy...")
if Path("/opt/ros/jazzy").exists():
subprocess.run(["sudo", "rm", "-rf", "/opt/ros/jazzy"])
write("[green][ok][/green] Removed /opt/ros/jazzy")
else:
write("[dim]ROS2 Jazzy not found, skipping.[/dim]")
# Remove the Docker volume that stores the Webots asset cache.
# Удаляем Docker volume с кэшем ассетов Webots.
def _remove_webots_volume(write) -> None:
result = subprocess.run(
["docker", "volume", "inspect", "lwc-webots-cache"],
capture_output=True,
)
if result.returncode != 0:
write("[dim]Webots cache volume not found, skipping.[/dim]")
return
subprocess.run(["docker", "volume", "rm", "lwc-webots-cache"], capture_output=True)
write("[green][ok][/green] Removed Docker volume: lwc-webots-cache")
# Remove ROS2 Jazzy packages via apt and clean up the source line from shell configs.
# Uses the official removal commands to also unregister the ROS2 apt repository.
# Удаляем пакеты ROS2 Jazzy через apt и очищаем строку source из конфигов оболочки.
# Используем официальные команды удаления, которые также снимают регистрацию apt-репозитория ROS2.
def _remove_ros2(write) -> None:
write("[cyan][*][/cyan] Removing ROS2 Jazzy packages...")
if not Path("/opt/ros/jazzy").exists():
write("[dim]ROS2 Jazzy not found, skipping.[/dim]")
else:
# Remove all ros-jazzy-* packages matched by the apt regex pattern ~n<name>.
# Удаляем все пакеты ros-jazzy-* по regex-паттерну apt ~n<имя>.
subprocess.run(
["sudo", "apt", "remove", "-y", "~nros-jazzy-*"],
capture_output=True,
)
subprocess.run(["sudo", "apt", "autoremove", "-y"], capture_output=True)
write("[green][ok][/green] ROS2 Jazzy packages removed")
# Remove the ROS2 apt source package that added the repository.
# Удаляем пакет apt-источника ROS2, который добавил репозиторий.
subprocess.run(
["sudo", "apt", "remove", "-y", "ros2-apt-source"],
capture_output=True,
)
subprocess.run(["sudo", "apt", "update", "-qq"], capture_output=True)
subprocess.run(["sudo", "apt", "autoremove", "-y"], capture_output=True)
write("[green][ok][/green] ROS2 apt repository removed")
# Clean up the source line that local-setup added to the shell config.
# Очищаем строку source, добавленную local-setup в конфиг оболочки.
source_line = "source /opt/ros/jazzy/setup.bash"
for rc_name in [".bashrc", ".zshrc"]:
rc = Path.home() / rc_name
@@ -63,12 +102,28 @@ def _remove_ros2(write) -> None:
content = rc.read_text()
if source_line not in content:
continue
# Remove the whole block that was added by local-setup, not just the single line.
# Удаляем весь блок добавленный local-setup, а не только одну строку.
new_content = content.replace(f"\n# ROS2 Jazzy\n{source_line}\n", "\n")
new_content = new_content.replace(source_line, "")
rc.write_text(new_content)
write(f"[green][ok][/green] Cleaned up ~/{rc_name}")
# Remove Webots from the system via apt.
# Удаляем Webots из системы через apt.
def _remove_webots(write) -> None:
write("[cyan][*][/cyan] Removing Webots...")
if not shutil.which("webots"):
write("[dim]Webots not found, skipping.[/dim]")
return
subprocess.run(["sudo", "apt", "remove", "-y", "webots"], capture_output=True)
subprocess.run(["sudo", "apt", "autoremove", "-y"], capture_output=True)
write("[green][ok][/green] Webots removed")
# Uninstall the cobot CLI from the uv tool store.
# Удаляем cobot CLI из хранилища инструментов uv.
def _uninstall_cobot(write) -> None:
write("[cyan][*][/cyan] Uninstalling cobot CLI...")
result = subprocess.run(
@@ -81,6 +136,8 @@ def _uninstall_cobot(write) -> None:
write(f"[yellow]Warning:[/yellow] {result.stderr.strip() or 'could not uninstall cobot'}")
# Delete the entire project directory from disk.
# Удаляем всю директорию проекта с диска.
def _remove_project_dir(write) -> None:
write(f"[cyan][*][/cyan] Removing project directory...")
try:
@@ -91,31 +148,35 @@ def _remove_project_dir(write) -> None:
raise
def _task_delete(screen: LogScreen, remove_ros: bool) -> None:
# Run all deletion steps in order.
# Progress ranges are split evenly across the active steps so the bar always reaches 100%.
# Выполняем все шаги удаления по порядку.
# Диапазоны прогресса делятся равномерно между активными шагами, чтобы бар всегда доходил до 100%.
def _task_delete(screen: LogScreen, remove_ros: bool, remove_webots: bool) -> None:
try:
screen.set_progress(0, "Stopping containers...")
_stop_docker_containers(screen.write)
_remove_webots_volume(screen.write)
screen.set_progress(20, "Removing Docker images...")
_remove_docker_images(screen.write)
pct = 40
if remove_ros:
# stop(0-20) images(20-45) ros2(45-70) cobot(70-85) dir(85-100)
screen.set_progress(0, "Stopping containers...")
_stop_docker_containers(screen.write)
screen.set_progress(20, "Removing Docker images...")
_remove_docker_images(screen.write)
screen.set_progress(45, "Removing ROS2 Jazzy...")
screen.set_progress(pct, "Removing ROS2 Jazzy...")
_remove_ros2(screen.write)
screen.set_progress(70, "Uninstalling cobot CLI...")
_uninstall_cobot(screen.write)
screen.set_progress(85, "Removing project directory...")
_remove_project_dir(screen.write)
else:
# stop(0-25) images(25-60) cobot(60-85) dir(85-100)
screen.set_progress(0, "Stopping containers...")
_stop_docker_containers(screen.write)
screen.set_progress(25, "Removing Docker images...")
_remove_docker_images(screen.write)
screen.set_progress(60, "Uninstalling cobot CLI...")
_uninstall_cobot(screen.write)
screen.set_progress(85, "Removing project directory...")
_remove_project_dir(screen.write)
pct = 65
if remove_webots:
screen.set_progress(pct, "Removing Webots...")
_remove_webots(screen.write)
pct = 75
screen.set_progress(pct, "Uninstalling cobot CLI...")
_uninstall_cobot(screen.write)
screen.set_progress(88, "Removing project directory...")
_remove_project_dir(screen.write)
screen.set_progress(100, "Done")
screen.write("\n[green]Project fully removed.[/green]")
@@ -126,7 +187,10 @@ def _task_delete(screen: LogScreen, remove_ros: bool) -> None:
screen.finish(False)
# Multi-step confirmation wizard before anything is deleted.
# Shows extra questions only when the relevant software is actually installed.
# Многошаговый мастер подтверждения перед удалением.
# Дополнительные вопросы показываются только если соответствующее ПО действительно установлено.
class _DeleteApp(App[None]):
CSS = SCREEN_CSS
@@ -148,7 +212,7 @@ class _DeleteApp(App[None]):
self.push_screen(
PickScreen(
"ROS2 Jazzy",
"Also remove ROS2 Jazzy (/opt/ros/jazzy)?",
"Also remove ROS2 Jazzy from the system?",
["No, keep ROS2", "Yes, remove ROS2 Jazzy"],
"No, keep ROS2",
),
@@ -157,8 +221,32 @@ class _DeleteApp(App[None]):
def _on_ros_choice(self, choice: Optional[str]) -> None:
remove_ros = choice is not None and choice.startswith("Yes")
# Only ask about Webots if it is actually installed on this machine.
# Спрашиваем про Webots только если он действительно установлен на этой машине.
if shutil.which("webots"):
self.push_screen(
PickScreen(
"Webots",
"Also remove Webots from the system?",
["No, keep Webots", "Yes, remove Webots"],
"No, keep Webots",
),
lambda c: self._on_webots_choice(c, remove_ros),
)
else:
self._start_deletion(remove_ros, remove_webots=False)
def _on_webots_choice(self, choice: Optional[str], remove_ros: bool) -> None:
remove_webots = choice is not None and choice.startswith("Yes")
self._start_deletion(remove_ros, remove_webots)
def _start_deletion(self, remove_ros: bool, remove_webots: bool) -> None:
self.push_screen(
LogScreen("Deleting project", lambda s: _task_delete(s, remove_ros), show_progress=True),
LogScreen(
"Deleting project",
lambda s: _task_delete(s, remove_ros, remove_webots),
show_progress=True,
),
lambda _: self.exit(),
)
cobot/commands/doc_setup.py
+29
View File
@@ -14,6 +14,11 @@ from textual.app import App
from cobot.tui import SCREEN_CSS, InputScreen, LogScreen
_PROJECT_DIR = Path(__file__).parent.parent.parent
# The documentation source lives inside the project. We mount it into the container so
# MkDocs can pick up live edits without rebuilding the image.
# Исходники документации находятся внутри проекта. Монтируем директорию в контейнер, чтобы
# MkDocs мог подхватывать изменения вживую без пересборки образа.
_DOC_DIR = _PROJECT_DIR / "doc" / "lwc-doc"
_IMAGE_NAME = "lwc-docs"
_CONTAINER_NAME = "lwc-docs"
@@ -22,24 +27,36 @@ _DEFAULT_PORT = "8000"
Write = Callable[[str], None]
# Thin wrapper around docker so we do not repeat ["docker", ...] everywhere.
# Тонкая обёртка вокруг docker, чтобы не повторять ["docker", ...] везде.
def _docker(*args: str, capture: bool = False) -> subprocess.CompletedProcess:
return subprocess.run(["docker", *args], capture_output=capture, text=True)
# Check whether the docs container is currently running.
# Проверяем, запущен ли сейчас контейнер с документацией.
def _is_running() -> bool:
r = _docker("ps", "--filter", f"name={_CONTAINER_NAME}", "--format", "{{.Names}}", capture=True)
return _CONTAINER_NAME in r.stdout
# Check whether the docs Docker image has already been built.
# Проверяем, был ли уже собран Docker-образ для документации.
def _image_exists() -> bool:
return bool(_docker("images", "-q", _IMAGE_NAME, capture=True).stdout.strip())
# Build the MkDocs Docker image. Only needs to run once.
# Progress comes from parsing "Step X/Y" lines in the docker build output.
# Собираем Docker-образ MkDocs. Нужно сделать только один раз.
# Прогресс получаем, парся строки "Step X/Y" из вывода docker build.
def _build_docs_image(
write: Write,
on_progress: Optional[Callable[[float], None]] = None,
) -> bool:
write("[cyan][*][/cyan] Building documentation image (runs once)...")
# DOCKER_BUILDKIT=0 gives us "Step X/Y" lines that we can parse for progress.
# DOCKER_BUILDKIT=0 даёт нам строки "Step X/Y", которые можно парсить для прогресса.
env = {**os.environ, "DOCKER_BUILDKIT": "0"}
proc = subprocess.Popen(
["docker", "build", "-t", _IMAGE_NAME, str(_DOC_DIR)],
@@ -62,6 +79,8 @@ def _build_docs_image(
return False
# Start the docs server. Builds the image first if it does not exist yet.
# Запускаем сервер документации. Сначала собирает образ, если он ещё не существует.
def _task_up(screen: LogScreen, port: str) -> None:
try:
if _is_running():
@@ -91,6 +110,8 @@ def _task_up(screen: LogScreen, port: str) -> None:
result = _docker(
"run", "-d", "--name", _CONTAINER_NAME, "--rm",
"-p", f"{port}:8000",
# Mount the docs directory so edits appear live without restarting the container.
# Монтируем директорию с документацией, чтобы изменения появлялись сразу без перезапуска.
"-v", f"{_DOC_DIR}:/docs",
_IMAGE_NAME, "serve", "--dev-addr=0.0.0.0:8000",
capture=True,
@@ -111,6 +132,8 @@ def _task_up(screen: LogScreen, port: str) -> None:
screen.finish(False)
# Stop the running docs container.
# Останавливаем работающий контейнер с документацией.
def _task_down(screen: LogScreen) -> None:
try:
if not _is_running():
@@ -128,6 +151,8 @@ def _task_down(screen: LogScreen) -> None:
screen.finish(False)
# Stop the container, remove the old image, rebuild it, and start a new container.
# Останавливаем контейнер, удаляем старый образ, пересобираем и запускаем новый контейнер.
def _task_rebuild(screen: LogScreen, port: str) -> None:
try:
if _is_running():
@@ -174,6 +199,8 @@ def _task_rebuild(screen: LogScreen, port: str) -> None:
screen.finish(False)
# One app handles all three actions (up/down/rebuild) by branching in on_mount.
# Одно приложение обрабатывает все три действия (up/down/rebuild), разветвляясь в on_mount.
class _DocApp(App[None]):
CSS = SCREEN_CSS
@@ -202,6 +229,8 @@ class _DocApp(App[None]):
if port is None:
self.exit()
return
# Use the default port if the user cleared the input or typed something that is not a number.
# Используем порт по умолчанию если пользователь очистил ввод или написал не число.
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), show_progress=True),
cobot/commands/docker_setup.py
+37
View File
@@ -16,12 +16,19 @@ from cobot.tui import SCREEN_CSS, InputScreen, LogScreen, PickScreen
_PROJECT_DIR = Path(__file__).parent.parent.parent
_DOCKER_DIR = _PROJECT_DIR / "docker"
# Default Docker Hub repository and local image prefix used when building locally.
# Репозиторий Docker Hub по умолчанию и локальный префикс образов при локальной сборке.
_DEFAULT_HUB_REPO = "evilfisru/lwc"
_DEFAULT_PREFIX = "lwc-local"
# The images must be built in this order because each one is based on the previous.
# Образы должны собираться в этом порядке, потому что каждый основан на предыдущем.
_CONTROLLER_CHAIN = ["ros-core", "ros-base", "ros-iiwa7"]
_WEBOTS_CHAIN = ["ros-core", "ros-base", "ros-iiwa7-webots"]
# Maps each image to the image it is built FROM. None means it starts from scratch (base Ubuntu).
# Сопоставляет каждый образ с тем, на основе которого он собирается. None - начинает с нуля (базовый Ubuntu).
_IMAGE_PARENT: dict[str, str | None] = {
"ros-core": None,
"ros-base": "ros-core",
@@ -29,11 +36,15 @@ _IMAGE_PARENT: dict[str, str | None] = {
"ros-iiwa7-webots": "ros-base",
}
# These images need the full project source as Docker build context because they copy source files.
# Эти образы требуют полный исходный код проекта как контекст сборки, потому что копируют файлы.
_NEEDS_PROJECT_CTX = {"ros-iiwa7", "ros-iiwa7-webots"}
Write = Callable[[str], None]
# All the choices the user makes in the wizard are stored here before we start the actual build.
# Все выборы пользователя в мастере хранятся здесь перед началом фактической сборки.
@dataclass
class _Config:
ros_version: str
@@ -44,6 +55,10 @@ class _Config:
hub_repo: str
# Build one Docker image and stream its output to the log.
# Progress is tracked by parsing "Step X/Y" lines that Docker prints during the build.
# Собирает один Docker-образ и транслирует его вывод в лог.
# Прогресс отслеживается по строкам "Step X/Y", которые Docker печатает во время сборки.
def _build_image(
name: str,
tag: str,
@@ -55,6 +70,8 @@ def _build_image(
build_type: str = "release",
) -> bool:
write(f"[cyan][*][/cyan] Building [bold]{name}[/bold]...")
# DOCKER_BUILDKIT=0 gives us "Step X/Y" lines in the output which we parse for progress.
# DOCKER_BUILDKIT=0 даёт нам строки "Step X/Y" в выводе, которые мы парсим для прогресса.
env = {**os.environ, "DOCKER_BUILDKIT": "0"}
cmd = [
"docker", "build", "-t", tag, "-f", str(dockerfile),
@@ -84,6 +101,8 @@ def _build_image(
return False
# Pull a Docker image from Hub and track progress by counting downloaded layers.
# Скачиваем Docker-образ с Hub и отслеживаем прогресс по количеству скачанных слоёв.
def _pull_image(
name: str,
tag: str,
@@ -104,6 +123,8 @@ def _pull_image(
s = line.rstrip()
if s:
write(s)
# Count layers as they appear and mark them done when Docker confirms they are pulled.
# Считаем слои по мере их появления и отмечаем завершёнными когда Docker подтверждает скачивание.
if "Pulling fs layer" in line or "Waiting" in line:
layers_total += 1
elif "Pull complete" in line or "Already exists" in line:
@@ -120,6 +141,10 @@ def _pull_image(
return False
# The actual work - either build or pull all images depending on what the user chose.
# Each image gets its own slice of the progress bar so the overall bar advances smoothly.
# Основная работа - собираем или скачиваем все образы в зависимости от выбора пользователя.
# Каждый образ получает свой кусок прогресс-бара, чтобы общий бар двигался равномерно.
def _task_execute(screen: LogScreen, cfg: _Config) -> None:
try:
chain = _WEBOTS_CHAIN if cfg.variant == "webots" else _CONTROLLER_CHAIN
@@ -190,15 +215,23 @@ def _task_execute(screen: LogScreen, cfg: _Config) -> None:
screen.finish(False)
# Scan the docker/ directory for subdirectories named after ROS versions (e.g. jazzy).
# If nothing is found we fall back to "jazzy" so the wizard still works.
# Сканируем директорию docker/ на наличие поддиректорий с именами версий ROS (например jazzy).
# Если ничего не найдено, используем "jazzy" по умолчанию, чтобы мастер всё равно работал.
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())
# Put jazzy first so it is the pre-selected default in the wizard.
# Ставим jazzy первым, чтобы он был предвыбранным по умолчанию в мастере.
if "jazzy" in dirs:
dirs = ["jazzy"] + [d for d in dirs if d != "jazzy"]
return dirs or ["jazzy"]
# Multi-step wizard that collects all build options before starting the actual image build.
# Многошаговый мастер, который собирает все параметры сборки перед запуском фактической сборки образа.
class _Wizard(App[None]):
CSS = SCREEN_CSS
@@ -264,6 +297,8 @@ class _Wizard(App[None]):
self.exit()
return
self._state["build_type"] = v or "release"
# Pull needs a Hub repo name, build needs a local image prefix.
# Для pull нужно имя репозитория на Hub, для build - локальный префикс образов.
if self._state["source"] == "pull":
self.push_screen(
InputScreen("Step 5 of 5", "Docker Hub repository:", _DEFAULT_HUB_REPO),
@@ -290,6 +325,8 @@ class _Wizard(App[None]):
self._finish()
def _finish(self) -> None:
# Assemble the config and hand it off to the log screen that does the actual work.
# Собираем конфиг и передаём его экрану лога, который выполняет фактическую работу.
s = self._state
cfg = _Config(
ros_version=s["ros_version"],
cobot/commands/local_setup.py
+167 -15
View File
@@ -17,12 +17,19 @@ from cobot.commands.docker_setup import run as _docker_setup
_PROJECT_DIR = Path(__file__).parent.parent.parent
# Paths used for the ROS2 apt repository signing key and sources list.
# Пути для ключа подписи apt-репозитория ROS2 и файла sources list.
_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"
# Suppress apt interactive prompts such as "restart services?".
# Подавляем интерактивные запросы apt, например "перезапустить службы?".
_APT_ENV = {**os.environ, "DEBIAN_FRONTEND": "noninteractive"}
# Check whether we are running on Ubuntu 24.04, which is required for ROS2 Jazzy.
# Проверяем, запущены ли мы на Ubuntu 24.04, которая требуется для ROS2 Jazzy.
def _detect_ubuntu_2404() -> bool:
path = Path("/etc/os-release")
if not path.exists():
@@ -35,6 +42,8 @@ def _detect_ubuntu_2404() -> bool:
return info.get("ID") == "ubuntu" and info.get("VERSION_ID") == "24.04"
# Check whether ROS2 Jazzy is already installed by looking for its directory.
# Проверяем, установлен ли ROS2 Jazzy, проверяя наличие его директории.
def _detect_ros2_jazzy() -> bool:
return Path("/opt/ros/jazzy").is_dir()
@@ -42,6 +51,8 @@ def _detect_ros2_jazzy() -> bool:
Write = Callable[[str], None]
# Run a command and capture output. Print it to the log only if the command fails.
# Запускаем команду и перехватываем вывод. Выводим в лог только если команда завершилась с ошибкой.
def _run_quiet(cmd: List[str], write: Write | None = None, env: dict | None = None, cwd=None) -> None:
result = subprocess.run(
cmd, capture_output=True, text=True,
@@ -55,6 +66,8 @@ def _run_quiet(cmd: List[str], write: Write | None = None, env: dict | None = No
raise RuntimeError(f"Command failed: {cmd[0]}")
# Run a command and stream every output line to the log in real time.
# Запускаем команду и транслируем каждую строку вывода в лог в реальном времени.
def _run_logged(cmd: List[str], write: Write, env: dict | None = None, cwd=None) -> None:
proc = subprocess.Popen(
cmd,
@@ -80,6 +93,10 @@ def _run_apt_with_progress(
env: dict | None = None,
) -> None:
"""Run an apt command and feed real percentage from APT::Status-Fd to on_progress(0-100)."""
# APT::Status-Fd makes apt write progress lines to a pipe descriptor instead of stdout.
# We read that pipe in a background thread so we can update the progress bar live.
# APT::Status-Fd заставляет apt писать строки прогресса в дескриптор канала, а не в stdout.
# Читаем этот канал в фоновом потоке, чтобы обновлять прогресс-бар в реальном времени.
r_fd, w_fd = os.pipe()
try:
proc = subprocess.Popen(
@@ -91,6 +108,8 @@ def _run_apt_with_progress(
pass_fds=(w_fd,),
)
finally:
# Close the write end in the parent process so the reader thread gets EOF when apt exits.
# Закрываем пишущий конец в родительском процессе, чтобы читающий поток получил EOF при выходе apt.
os.close(w_fd)
def _read_status() -> None:
@@ -116,10 +135,8 @@ def _run_apt_with_progress(
raise RuntimeError(f"Command failed: {cmd[0]}")
# ---------------------------------------------------------------------------
# Installation steps
# ---------------------------------------------------------------------------
# Make sure the system has a UTF-8 locale, which ROS2 requires to work correctly.
# Убеждаемся, что в системе есть локаль UTF-8, которая требуется ROS2 для корректной работы.
def _setup_locale(write: Write) -> None:
write("[cyan][*][/cyan] Checking locale...")
if "UTF-8" in subprocess.run(["locale"], capture_output=True, text=True).stdout:
@@ -133,6 +150,8 @@ def _setup_locale(write: Write) -> None:
write("[green][ok][/green] Locale configured")
# Add the official ROS2 apt repository and its signing key so we can install ROS2 packages.
# Добавляем официальный apt-репозиторий ROS2 и его ключ подписи, чтобы можно было установить пакеты ROS2.
def _add_ros2_repo(write: Write, on_progress: Optional[Callable[[float], None]] = None) -> None:
def _prog(p: float) -> None:
if on_progress:
@@ -159,6 +178,8 @@ def _add_ros2_repo(write: Write, on_progress: Optional[Callable[[float], None]]
try:
urllib.request.urlretrieve(_ROS_KEY_URL, tmp_path)
_prog(60)
# Convert the ASCII-armored key to binary GPG format that apt understands.
# Конвертируем ключ из ASCII-armor формата в бинарный GPG, который понимает apt.
_run_quiet(["sudo", "gpg", "--dearmor", "--yes", "-o", str(_ROS_KEYRING), tmp_path])
finally:
os.unlink(tmp_path)
@@ -192,6 +213,8 @@ def _add_ros2_repo(write: Write, on_progress: Optional[Callable[[float], None]]
_prog(100)
# Install the full ROS2 Jazzy Desktop and the developer tools (colcon, rosdep, etc.).
# Устанавливаем полный ROS2 Jazzy Desktop и инструменты разработчика (colcon, rosdep и т.д.).
def _install_ros2_jazzy(write: Write, on_progress: Optional[Callable[[float], None]] = None) -> None:
write("[cyan][*][/cyan] Installing ros-jazzy-desktop and ros-dev-tools...")
_run_apt_with_progress(
@@ -203,6 +226,8 @@ def _install_ros2_jazzy(write: Write, on_progress: Optional[Callable[[float], No
write("[green][ok][/green] ROS2 Jazzy Desktop installed")
# Install colcon if it is not already available. It is used to build the project packages.
# Устанавливаем colcon если он ещё не доступен. Он используется для сборки пакетов проекта.
def _install_colcon(write: Write) -> None:
if shutil.which("colcon"):
write("[green][ok][/green] colcon already available")
@@ -216,6 +241,10 @@ def _install_colcon(write: Write) -> None:
write("[green][ok][/green] colcon installed")
# Add "source /opt/ros/jazzy/setup.bash" to the user's shell config file.
# This makes ROS2 commands available in every new terminal session.
# Добавляем "source /opt/ros/jazzy/setup.bash" в конфиг оболочки пользователя.
# Это делает команды ROS2 доступными в каждой новой сессии терминала.
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")
@@ -228,10 +257,8 @@ def _setup_shell_rc(write: Write) -> None:
write(f"[green][ok][/green] Added ROS2 setup to ~/{rc.name}")
# ---------------------------------------------------------------------------
# Background tasks (run inside LogScreen worker)
# ---------------------------------------------------------------------------
# Full ROS2 Jazzy installation split into 5 clearly visible steps with individual progress ranges.
# Полная установка ROS2 Jazzy, разбитая на 5 наглядных шагов с отдельными диапазонами прогресса.
def _task_install_jazzy(screen: LogScreen) -> None:
try:
# Step 1 — locale (0 → 5 %)
@@ -275,6 +302,8 @@ def _task_install_jazzy(screen: LogScreen) -> None:
screen.finish(False)
# Build all project packages with colcon and track progress by counting finished packages.
# Собираем все пакеты проекта с помощью colcon и отслеживаем прогресс по количеству завершённых пакетов.
def _task_build(screen: LogScreen) -> None:
try:
if not shutil.which("colcon"):
@@ -283,7 +312,8 @@ def _task_build(screen: LogScreen) -> None:
screen.finish(False)
return
# Count packages so we can show X/total progress
# Count packages first so we can show X/total in the progress label.
# Сначала считаем пакеты, чтобы показывать X/всего в подписи прогресса.
list_result = subprocess.run(
["colcon", "list"], capture_output=True, text=True, cwd=_PROJECT_DIR,
)
@@ -296,6 +326,8 @@ def _task_build(screen: LogScreen) -> None:
def _track(line: str) -> None:
nonlocal built
screen.write(line)
# colcon prints "Finished <<<" or "Failed <<<" when each package is done.
# colcon печатает "Finished <<<" или "Failed <<<" когда каждый пакет готов.
if "Finished <<<" in line or "Failed <<<" in line:
built += 1
screen.set_progress(built / total * 100, f"{built} / {total} packages done")
@@ -310,10 +342,109 @@ def _task_build(screen: LogScreen) -> None:
screen.finish(False)
# ---------------------------------------------------------------------------
# Textual apps
# ---------------------------------------------------------------------------
# Webots version that matches the Docker images used in this project.
# Версия Webots, соответствующая Docker-образам используемым в этом проекте.
_WEBOTS_VERSION = "2025a"
_WEBOTS_DEB_URL = (
f"https://github.com/cyberbotics/webots/releases/download/"
f"R{_WEBOTS_VERSION}/webots_{_WEBOTS_VERSION}_amd64.deb"
)
def webots_installed() -> bool:
"""Return True if Webots is available on PATH."""
return shutil.which("webots") is not None
# Download the Webots .deb from GitHub and install it with apt.
# Progress: download (0-65%), apt install (65-100%).
# Скачиваем .deb Webots с GitHub и устанавливаем через apt.
# Прогресс: скачивание (0-65%), установка apt (65-100%).
def _task_install_webots(screen: LogScreen) -> None:
try:
screen.write(f"[bold]Installing Webots {_WEBOTS_VERSION}[/bold]\n")
with tempfile.TemporaryDirectory() as tmp:
deb_path = Path(tmp) / f"webots_{_WEBOTS_VERSION}_amd64.deb"
screen.write(f"[dim]{_WEBOTS_DEB_URL}[/dim]\n")
screen.set_progress(0, "Downloading Webots...")
# urllib calls this hook periodically with how many bytes have been downloaded.
# urllib вызывает этот обратный вызов периодически с количеством скачанных байт.
def _hook(blocks: int, block_size: int, total: int) -> None:
if total > 0:
pct = min(blocks * block_size / total * 65, 65)
mb = blocks * block_size / 1_048_576
total_mb = total / 1_048_576
screen.set_progress(pct, f"Downloading... {mb:.0f} / {total_mb:.0f} MB")
urllib.request.urlretrieve(_WEBOTS_DEB_URL, deb_path, _hook)
screen.write("[green]Download complete.[/green]")
screen.set_progress(65, "Installing package...")
proc = subprocess.Popen(
["sudo", "apt-get", "install", "-y", str(deb_path)],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True,
)
for line in proc.stdout:
s = line.rstrip()
if s:
screen.write(s)
proc.wait()
if proc.returncode != 0:
screen.write("\n[red]Installation failed.[/red]")
screen.finish(False)
return
screen.set_progress(100, "Done")
screen.write("\n[green]Webots installed successfully.[/green]")
screen.finish(True)
except Exception as exc:
screen.write(f"\n[red]Error:[/red] {exc}")
screen.finish(False)
# Minimal single-question app used between steps where a full wizard is not needed.
# Минимальное приложение с одним вопросом, используемое между шагами где полный мастер не нужен.
class _Ask(App[Optional[str]]):
CSS = SCREEN_CSS
def __init__(self, step: str, question: str, options: list, default: str):
super().__init__()
self._step = step
self._question = question
self._options = options
self._default = default
def on_mount(self) -> None:
self.push_screen(
PickScreen(self._step, self._question, self._options, self._default),
self.exit,
)
def _ask(step: str, question: str, options: list, default: str) -> Optional[str]:
return _Ask(step, question, options, default).run()
# Public app used by run.py to install Webots before launching locally.
# Публичное приложение, используемое run.py для установки Webots перед локальным запуском.
class WebotsInstallApp(App[bool]):
CSS = SCREEN_CSS
def on_mount(self) -> None:
self.push_screen(
LogScreen(f"Installing Webots {_WEBOTS_VERSION}", _task_install_webots, show_progress=True),
self.exit,
)
# Ask the user if they want to install ROS2 Jazzy, then run the installer if they say yes.
# Спрашиваем пользователя хочет ли он установить ROS2 Jazzy, и запускаем установщик если да.
class _InstallJazzyApp(App[None]):
CSS = SCREEN_CSS
@@ -338,6 +469,8 @@ class _InstallJazzyApp(App[None]):
)
# Run the colcon build without asking any questions - used when ROS2 is already installed.
# Запускаем сборку colcon без лишних вопросов - используется когда ROS2 уже установлен.
class _BuildApp(App[None]):
CSS = SCREEN_CSS
@@ -348,6 +481,8 @@ class _BuildApp(App[None]):
)
# Shown when the OS is not Ubuntu 24.04. Offers to fall back to docker-setup instead.
# Показывается когда ОС не Ubuntu 24.04. Предлагает перейти к docker-setup вместо этого.
class _DockerPromptApp(App[bool]):
CSS = SCREEN_CSS
@@ -363,9 +498,6 @@ class _DockerPromptApp(App[bool]):
)
# ---------------------------------------------------------------------------
# CLI registration
# ---------------------------------------------------------------------------
def register(subparsers: argparse._SubParsersAction) -> None:
p = subparsers.add_parser(
@@ -376,13 +508,33 @@ def register(subparsers: argparse._SubParsersAction) -> None:
def run(args: argparse.Namespace) -> None:
# If this is not Ubuntu 24.04 we cannot install ROS2 Jazzy natively - offer Docker instead.
# Если это не Ubuntu 24.04 мы не можем установить ROS2 Jazzy нативно - предлагаем Docker вместо этого.
if not _detect_ubuntu_2404():
if _DockerPromptApp().run():
_docker_setup(args)
return
# ROS2 not installed yet - show the installer.
# After installation the user must restart the terminal, so we stop here.
# ROS2 ещё не установлен - показываем установщик.
# После установки пользователь должен перезапустить терминал, поэтому останавливаемся здесь.
if not _detect_ros2_jazzy():
_InstallJazzyApp().run()
return
# ROS2 is ready - build the project.
# ROS2 готов - собираем проект.
_BuildApp().run()
# Ask about Webots only after a successful build, and only if it is not already installed.
# Спрашиваем про Webots только после успешной сборки и только если он ещё не установлен.
if not webots_installed():
v = _ask(
"Optional: Webots",
f"Install Webots {_WEBOTS_VERSION} simulator? (can also be done later via cobot run)",
[f"Yes, install Webots {_WEBOTS_VERSION}", "No, skip"],
"No, skip",
)
if v and v.startswith("Yes"):
WebotsInstallApp().run()
cobot/commands/robot_setup.py
+38 -8
View File
@@ -17,10 +17,16 @@ from cobot.tui import SCREEN_CSS, InputScreen, PickScreen
_PROJECT_DIR = Path(__file__).parent.parent.parent
_CONFIG_PATH = _PROJECT_DIR / "cobot-setting.yaml"
# Use ruamel.yaml instead of PyYAML so comments and formatting in the config file are preserved.
# Используем ruamel.yaml вместо PyYAML, чтобы комментарии и форматирование в конфиге сохранялись.
_yaml = YAML()
_yaml.preserve_quotes = True
# One question inside a configuration block.
# A field can either show a pick list (options) or a free-text input (no options).
# Один вопрос внутри блока конфигурации.
# Поле может показывать список вариантов (options) или поле для ввода текста (без options).
@dataclass
class _Field:
key: str # dot-separated path within the block, e.g. "webots.world"
@@ -33,6 +39,8 @@ class _Field:
return self.key.split(".")[-1]
# A group of related fields shown together under one "Configure X?" question.
# Группа связанных полей, показываемая вместе под одним вопросом "Настроить X?".
@dataclass
class _Block:
yaml_key: str # top-level key in cobot-setting.yaml
@@ -40,6 +48,8 @@ class _Block:
fields: List[_Field]
# All configuration blocks. Each block maps to a top-level key in cobot-setting.yaml.
# Все блоки конфигурации. Каждый блок соответствует ключу верхнего уровня в cobot-setting.yaml.
_BLOCKS: List[_Block] = [
_Block(
yaml_key="foxglove",
@@ -101,17 +111,20 @@ _BLOCKS: List[_Block] = [
_Field("active_controller", "Active ROS controller:", "jtc",
note="jtc = JointTrajectoryController (MoveIt), forward = ForwardCommandController",
options=["jtc", "forward"]),
_Field("joint_position_tau", "Position EMA filter τ (s):", "0.04",
_Field("joint_position_tau", "Position EMA filter tau (s):", "0.04",
note="Smooths position commands before sending to FRI"),
_Field("joint_velocity_tau", "Velocity EMA filter τ (s):", "0.01",
_Field("joint_velocity_tau", "Velocity EMA filter tau (s):", "0.01",
note="Removes spikes from finite-difference velocity estimation"),
],
),
]
# Try to keep the original YAML type (bool, int, float) when saving a value back.
# Trying to preserve type prevents "true" from becoming a plain string in the YAML file.
# Пытаемся сохранить исходный тип YAML (bool, int, float) при записи значения обратно.
# Сохранение типа предотвращает превращение "true" в обычную строку в YAML-файле.
def _coerce(value: str, original: Any) -> Any:
"""Try to preserve the original YAML scalar type."""
if isinstance(original, bool):
return value.lower() == "true"
if isinstance(original, int):
@@ -127,6 +140,8 @@ def _coerce(value: str, original: Any) -> Any:
return value
# Read a value from a nested YAML mapping using a dot-separated key like "webots.transform".
# Читаем значение из вложенного YAML-словаря по ключу с точками, например "webots.transform".
def _get_nested(mapping: Any, path: str) -> Any:
keys = path.split(".")
cur = mapping
@@ -137,6 +152,8 @@ def _get_nested(mapping: Any, path: str) -> Any:
return cur
# Write a value into a nested YAML mapping using a dot-separated key.
# Записываем значение в вложенный YAML-словарь по ключу с точками.
def _set_nested(mapping: Any, path: str, value: Any) -> None:
keys = path.split(".")
cur = mapping
@@ -146,6 +163,8 @@ def _set_nested(mapping: Any, path: str, value: Any) -> None:
cur[keys[-1]] = _coerce(value, original)
# Shown after all blocks have been configured to confirm the file was saved.
# Показывается после настройки всех блоков для подтверждения сохранения файла.
class _SavedScreen(Screen[None]):
BINDINGS = [Binding("enter,escape", "close", "Close")]
@@ -159,13 +178,17 @@ class _SavedScreen(Screen[None]):
self.dismiss(None)
# The main configuration wizard. Goes through each block in order.
# For each block it first asks "Configure X?" then steps through all its fields.
# Главный мастер конфигурации. Проходит по каждому блоку по порядку.
# Для каждого блока сначала спрашивает "Настроить X?" а затем проходит по всем его полям.
class _Wizard(App[None]):
CSS = SCREEN_CSS
def __init__(self, data: Any):
super().__init__()
self._data = data
self._blocks = list(_BLOCKS) # copy so we can pop
self._blocks = list(_BLOCKS)
self._block_idx = 0
self._field_idx = 0
self._current_block: Optional[_Block] = None
@@ -174,9 +197,10 @@ class _Wizard(App[None]):
def on_mount(self) -> None:
self._next_block()
def _next_block(self) -> None:
if self._block_idx >= len(self._blocks):
# All blocks done - save and show the confirmation screen.
# Все блоки пройдены - сохраняем и показываем экран подтверждения.
_save_config(self._data)
self.push_screen(_SavedScreen(), lambda _: self.exit())
return
@@ -204,9 +228,10 @@ class _Wizard(App[None]):
self._field_idx = 0
self._next_field()
else:
# Skip all fields in this block and jump to the next block.
# Пропускаем все поля этого блока и переходим к следующему.
self._next_block()
def _next_field(self) -> None:
if not self._pending_fields:
self._next_block()
@@ -220,7 +245,7 @@ class _Wizard(App[None]):
field_num = self._field_idx
total_fields = len(block.fields)
step = f"Block {block_num} of {total_blocks} · Field {field_num} of {total_fields}"
step = f"Block {block_num} of {total_blocks} - Field {field_num} of {total_fields}"
# Resolve current value from loaded YAML as the pre-filled default
yaml_val = _get_nested(self._data[block.yaml_key], f.key)
@@ -241,16 +266,21 @@ class _Wizard(App[None]):
return
block = self._current_block
_set_nested(self._data[block.yaml_key], f.key, v)
# Remove the field we just handled and move on to the next one.
# Удаляем только что обработанное поле и переходим к следующему.
self._pending_fields.pop(0)
self._next_field()
# Load the config file preserving all comments and key order.
# Загружаем конфиг-файл, сохраняя все комментарии и порядок ключей.
def _load_config() -> Any:
with open(_CONFIG_PATH, "r", encoding="utf-8") as fh:
return _yaml.load(fh)
# Write the modified config back to disk preserving comments and formatting.
# Записываем изменённый конфиг обратно на диск, сохраняя комментарии и форматирование.
def _save_config(data: Any) -> None:
with open(_CONFIG_PATH, "w", encoding="utf-8") as fh:
_yaml.dump(data, fh)
cobot/commands/run.py
+98 -35
View File
@@ -11,17 +11,32 @@ from typing import Callable, List, Optional
from textual.app import App
from cobot.tui import SCREEN_CSS, LogScreen, PickScreen, RunScreen
from cobot.commands.local_setup import webots_installed, WebotsInstallApp, _WEBOTS_VERSION
_PROJECT_DIR = Path(__file__).parent.parent.parent
_CONFIG_PATH = _PROJECT_DIR / "cobot-setting.yaml"
_INSTALL_DIR = _PROJECT_DIR / "install"
_JAZZY_DIR = Path("/opt/ros/jazzy")
# Path where the config file is mounted inside the Docker container.
# Путь по которому конфиг-файл монтируется внутри Docker-контейнера.
_CONFIG_IN_CONTAINER = "/ros2_ws/cobot-setting.yaml"
# Container names used for docker run and docker kill.
# Имена контейнеров, используемые для docker run и docker kill.
_CONTAINER_CONTROLLER = "lwc-controller"
_CONTAINER_WEBOTS = "lwc-webots"
# Candidates checked in order; for controller the webots image is a valid fallback
# Named Docker volume that stores the Webots asset cache between container runs.
# Without it Webots re-downloads all 3D assets from the internet on every launch.
# Именованный Docker volume для хранения кэша ассетов Webots между запусками контейнера.
# Без него Webots заново скачивает все 3D-ассеты из интернета при каждом запуске.
_WEBOTS_CACHE_VOLUME = "lwc-webots-cache"
# Candidates checked in order - for the controller the webots image is a valid fallback
# because it already contains all controller packages too.
# Кандидаты проверяются по порядку - для контроллера образ webots является допустимым запасным,
# так как он уже содержит все пакеты контроллера.
_CONTROLLER_IMAGES = [
"lwc-local:ros-iiwa7-jazzy",
"evilfisru/lwc:iiwa-jazzy",
@@ -37,10 +52,10 @@ _WEBOTS_IMAGES = [
]
# ---------------------------------------------------------------------------
# Small utilities
# ---------------------------------------------------------------------------
# A minimal app that asks one question and exits immediately with the chosen value.
# We need a full App because Textual screens cannot run outside one.
# Минимальное приложение, которое задаёт один вопрос и сразу выходит с выбранным значением.
# Нам нужен полноценный App, потому что экраны Textual не могут работать вне него.
class _Ask(App[Optional[str]]):
CSS = SCREEN_CSS
@@ -59,9 +74,13 @@ class _Ask(App[Optional[str]]):
def _ask(step: str, question: str, options: List[str], default: str) -> Optional[str]:
# Returns None when the user pressed Escape to cancel.
# Возвращает None когда пользователь нажал Escape для отмены.
return _Ask(step, question, options, default).run()
# Detect the GPU type so we can pass the right flags to docker run for Webots rendering.
# Определяем тип GPU, чтобы передать нужные флаги в docker run для рендеринга Webots.
def _detect_gpu() -> str:
if shutil.which("nvidia-smi"):
if subprocess.run(["nvidia-smi"], capture_output=True).returncode == 0:
@@ -71,6 +90,8 @@ def _detect_gpu() -> str:
return "software"
# List all Docker images currently available on this machine.
# Получаем список всех Docker-образов доступных на этой машине.
def _docker_images() -> set:
r = subprocess.run(
["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"],
@@ -79,6 +100,8 @@ def _docker_images() -> set:
return set(r.stdout.strip().splitlines())
# Return the first image from the candidates list that is already present locally.
# Возвращаем первый образ из списка кандидатов, который уже присутствует локально.
def _find_image(candidates: List[str]) -> Optional[str]:
available = _docker_images()
for img in candidates:
@@ -87,14 +110,16 @@ def _find_image(candidates: List[str]) -> Optional[str]:
return None
# ---------------------------------------------------------------------------
# Build project (colcon build --mixin release)
# ---------------------------------------------------------------------------
# Build the ROS2 project locally with colcon. Used when launching in local mode
# and the install/ directory does not exist yet.
# Собираем ROS2-проект локально с помощью colcon. Используется при запуске в локальном режиме,
# если директория install/ ещё не существует.
def _task_build(screen: LogScreen) -> None:
try:
screen.write("[bold]Building project with colcon[/bold]\n")
# Count packages first so we can show X/total progress.
# Сначала считаем пакеты, чтобы показывать X/всего в прогрессе.
list_proc = subprocess.run(
["bash", "-c", f"source {_JAZZY_DIR}/setup.bash && colcon list"],
capture_output=True, text=True, cwd=_PROJECT_DIR,
@@ -113,6 +138,8 @@ def _task_build(screen: LogScreen) -> None:
s = line.rstrip()
if s:
screen.write(s)
# colcon prints "Finished <<<" or "Failed <<<" when each package is done.
# colcon печатает "Finished <<<" или "Failed <<<" когда каждый пакет готов.
if "Finished <<<" in line or "Failed <<<" in line:
built += 1
screen.set_progress(built / total * 100, f"{built} / {total} packages done")
@@ -141,10 +168,10 @@ class _BuildApp(App[bool]):
)
# ---------------------------------------------------------------------------
# Local launch task
# ---------------------------------------------------------------------------
# Start the ROS2 launch file directly on this machine without Docker.
# Uses start_new_session so we can kill the whole process group with one signal.
# Запускаем launch-файл ROS2 напрямую на этой машине без Docker.
# Используем start_new_session, чтобы можно было убить всю группу процессов одним сигналом.
def _task_run_local(screen: RunScreen, mode: str) -> None:
config = str(_CONFIG_PATH)
ros_cmd = f"ros2 launch iiwa_bringup iiwa.launch.py setting:={config}"
@@ -168,6 +195,8 @@ def _task_run_local(screen: RunScreen, mode: str) -> None:
start_new_session=True,
)
screen.set_proc(proc)
# Kill the entire process group so all child processes (nodes) are terminated together.
# Убиваем всю группу процессов, чтобы все дочерние процессы (узлы) завершились вместе.
screen.set_kill_fn(lambda: os.killpg(os.getpgid(proc.pid), signal.SIGTERM))
for line in proc.stdout:
@@ -179,18 +208,22 @@ def _task_run_local(screen: RunScreen, mode: str) -> None:
screen.finish(stopped=screen._stopped)
# ---------------------------------------------------------------------------
# Docker launch task
# ---------------------------------------------------------------------------
# Start the ROS2 launch file inside a Docker container.
# For Webots mode we also forward X11 and GPU access so the simulator window can appear on screen.
# Запускаем launch-файл ROS2 внутри Docker-контейнера.
# Для режима Webots также пробрасываем X11 и доступ к GPU, чтобы окно симулятора появилось на экране.
def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None:
container = _CONTAINER_WEBOTS if mode == "webots" else _CONTAINER_CONTROLLER
ros_cmd = f"ros2 launch iiwa_bringup iiwa.launch.py setting:={_CONFIG_IN_CONTAINER}"
ros_cmd = (
"source /ros2_ws/install/setup.bash && "
f"ros2 launch iiwa_bringup iiwa.launch.py setting:={_CONFIG_IN_CONTAINER}"
)
if mode == "webots":
ros_cmd += " simulate:=1"
# Remove stale container with the same name
# Remove any stale container with the same name left from a previous run.
# Удаляем устаревший контейнер с таким же именем, оставшийся от предыдущего запуска.
subprocess.run(["docker", "rm", "-f", container], capture_output=True)
cmd = [
@@ -201,11 +234,16 @@ def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None
]
if mode == "webots":
# Allow the container to open windows on the host display.
# Разрешаем контейнеру открывать окна на дисплее хоста.
subprocess.run(["xhost", "+local:docker"], capture_output=True)
cmd += [
"-e", f"DISPLAY={os.environ.get('DISPLAY', ':0')}",
"-e", "QT_X11_NO_MITSHM=1",
"-v", "/tmp/.X11-unix:/tmp/.X11-unix:rw",
# Persist the Webots asset cache so it is not re-downloaded on every launch.
# Сохраняем кэш ассетов Webots, чтобы он не скачивался заново при каждом запуске.
"-v", f"{_WEBOTS_CACHE_VOLUME}:/root/.cache/Cyberbotics/Webots",
]
if gpu == "nvidia":
cmd += [
@@ -214,13 +252,19 @@ def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None
"-e", "NVIDIA_DRIVER_CAPABILITIES=graphics,utility,compute",
]
elif gpu == "mesa":
# Pass through the DRI device for Intel/AMD hardware acceleration.
# Пробрасываем DRI-устройство для аппаратного ускорения Intel/AMD.
cmd += ["--device", "/dev/dri"]
else:
# No GPU found - fall back to software rendering via llvmpipe.
# GPU не найден - используем программный рендеринг через llvmpipe.
cmd += [
"-e", "LIBGL_ALWAYS_SOFTWARE=1",
"-e", "GALLIUM_DRIVER=llvmpipe",
]
# Mount the config file so the container uses our local cobot-setting.yaml.
# Монтируем конфиг-файл, чтобы контейнер использовал наш локальный cobot-setting.yaml.
if _CONFIG_PATH.exists():
cmd += ["-v", f"{_CONFIG_PATH}:{_CONFIG_IN_CONTAINER}:ro"]
@@ -244,6 +288,11 @@ def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None
text=True,
)
screen.set_proc(proc)
# Use docker kill instead of proc.terminate() so the container is stopped immediately.
# Terminating only the docker CLI process leaves the container itself running.
# Используем docker kill вместо proc.terminate(), чтобы контейнер остановился немедленно.
# Завершение только процесса docker CLI оставляет сам контейнер работающим.
screen.set_kill_fn(lambda: subprocess.run(["docker", "kill", container], capture_output=True))
for line in proc.stdout:
s = line.rstrip()
@@ -254,10 +303,8 @@ def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None
screen.finish(stopped=screen._stopped)
# ---------------------------------------------------------------------------
# RunApp wrapper
# ---------------------------------------------------------------------------
# Wraps a RunScreen in an App so it can be launched with .run().
# Оборачивает RunScreen в App, чтобы его можно было запустить через .run().
class _RunApp(App[None]):
CSS = SCREEN_CSS
@@ -270,10 +317,10 @@ class _RunApp(App[None]):
self.push_screen(RunScreen(self._title, self._run_fn), lambda _: self.exit())
# ---------------------------------------------------------------------------
# Local flow
# ---------------------------------------------------------------------------
# Guide the user through launching locally - asks what to run, checks prerequisites,
# installs Webots and builds the project if needed, then launches.
# Ведёт пользователя через локальный запуск - спрашивает что запустить, проверяет
# предварительные условия, устанавливает Webots и собирает проект при необходимости, затем запускает.
def _local_flow(args: argparse.Namespace) -> None:
mode_v = _ask(
"Run local",
@@ -285,6 +332,20 @@ def _local_flow(args: argparse.Namespace) -> None:
return
mode = "webots" if mode_v == "Webots simulator" else "controller"
# Check Webots installed (local mode only)
if mode == "webots" and not webots_installed():
v = _ask(
"Webots not found",
f"Webots {_WEBOTS_VERSION} is not installed. Install it now?",
[f"Yes, install Webots {_WEBOTS_VERSION}", "No, cancel"],
f"Yes, install Webots {_WEBOTS_VERSION}",
)
if v is None or v.startswith("No"):
return
ok = WebotsInstallApp().run()
if not ok:
return
# Check ROS2 Jazzy
if not _JAZZY_DIR.is_dir():
v = _ask(
@@ -316,10 +377,10 @@ def _local_flow(args: argparse.Namespace) -> None:
_RunApp(f"Running {label} — local", lambda s: _task_run_local(s, mode)).run()
# ---------------------------------------------------------------------------
# Docker flow
# ---------------------------------------------------------------------------
# Guide the user through launching in Docker - asks what to run, finds a suitable image,
# detects the GPU for Webots, and launches.
# Ведёт пользователя через запуск в Docker - спрашивает что запустить, ищет подходящий образ,
# определяет GPU для Webots и запускает.
def _docker_flow(args: argparse.Namespace) -> None:
if not shutil.which("docker"):
from rich.console import Console
@@ -340,6 +401,8 @@ def _docker_flow(args: argparse.Namespace) -> None:
image = _find_image(candidates)
if image is None:
# No image available - offer to run docker-setup to get one.
# Образ не найден - предлагаем запустить docker-setup чтобы его получить.
what = "Webots" if mode == "webots" else "controller or Webots"
v = _ask(
"No image found",
@@ -352,6 +415,8 @@ def _docker_flow(args: argparse.Namespace) -> None:
_docker_setup(args)
return
# Only detect GPU for Webots - the controller does not need a display.
# GPU определяем только для Webots - контроллеру дисплей не нужен.
gpu = _detect_gpu() if mode == "webots" else "software"
label = "Webots simulator" if mode == "webots" else "Controller"
@@ -361,10 +426,6 @@ def _docker_flow(args: argparse.Namespace) -> None:
).run()
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def register(subparsers: argparse._SubParsersAction) -> None:
p = subparsers.add_parser(
"run",
@@ -388,6 +449,8 @@ def run(args: argparse.Namespace) -> None:
elif mode == "docker":
_docker_flow(args)
else:
# No mode given - ask the user how they want to run.
# Режим не указан - спрашиваем пользователя как он хочет запустить.
v = _ask(
"Run",
"How do you want to run the project?",
cobot/commands/setup.py
+14 -4
View File
@@ -3,6 +3,8 @@ from typing import List, Optional
from textual.app import App
# Import each sub-command's run() so we can call them in sequence.
# Импортируем run() каждой подкоманды, чтобы вызывать их по порядку.
from cobot.commands.doc_setup import run as _doc_setup
from cobot.commands.docker_setup import run as _docker_setup
from cobot.commands.local_setup import run as _local_setup
@@ -10,8 +12,11 @@ from cobot.commands.robot_setup import run as _robot_setup
from cobot.tui import SCREEN_CSS, PickScreen
# A minimal Textual app that asks a single question and exits with the chosen value.
# We need this because Textual screens cannot run outside of an App context.
# Минимальное Textual-приложение, которое задаёт один вопрос и выходит с выбранным значением.
# Нам это нужно, потому что экраны Textual не могут работать вне контекста приложения.
class _Ask(App[Optional[str]]):
"""Single-question picker that exits immediately with the chosen value."""
CSS = SCREEN_CSS
def __init__(self, step: str, question: str, options: List[str], default: str):
@@ -29,6 +34,8 @@ class _Ask(App[Optional[str]]):
def _ask(step: str, question: str, options: List[str], default: str) -> Optional[str]:
# Returns None if the user pressed Escape to cancel the whole wizard.
# Возвращает None если пользователь нажал Escape для отмены всего мастера.
return _Ask(step, question, options, default).run()
@@ -38,14 +45,16 @@ def register(subparsers):
def run(args: argparse.Namespace) -> None:
# Step 1 — documentation
# Step 1 - documentation server.
# Шаг 1 - сервер документации.
v = _ask("Step 1 of 3", "Set up the documentation server?", ["Yes", "No"], "Yes")
if v is None:
return
if v == "Yes":
_doc_setup(args)
# Step 2 — build environment
# Step 2 - build environment: local ROS2 or Docker.
# Шаг 2 - среда сборки: локальный ROS2 или Docker.
v = _ask(
"Step 2 of 3",
"How do you want to set up the build environment?",
@@ -62,7 +71,8 @@ def run(args: argparse.Namespace) -> None:
else:
_docker_setup(args)
# Step 3 — robot parameters
# Step 3 - robot parameters in cobot-setting.yaml.
# Шаг 3 - параметры робота в cobot-setting.yaml.
v = _ask(
"Step 3 of 3",
"Configure robot parameters (cobot-setting.yaml)?",
cobot/commands/update.py
+12 -3
View File
@@ -11,9 +11,14 @@ from cobot.tui import SCREEN_CSS, LogScreen
_PROJECT_DIR = Path(__file__).parent.parent.parent
# Pull the latest commits from the remote and reinstall the cobot CLI in one go.
# Progress bar: fetch (0-30%), pull (30-80%), reinstall (80-100%).
# Скачиваем последние коммиты с удалённого репозитория и переустанавливаем cobot CLI за один раз.
# Прогресс-бар: fetch (0-30%), pull (30-80%), переустановка (80-100%).
def _task_update(screen: LogScreen) -> None:
try:
# Current branch
# Find out which branch we are on so we can fetch and pull the right one.
# Определяем на какой ветке мы находимся, чтобы делать fetch и pull нужной ветки.
branch = subprocess.check_output(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
cwd=_PROJECT_DIR, text=True,
@@ -33,7 +38,8 @@ def _task_update(screen: LogScreen) -> None:
return
screen.set_progress(30)
# Check how many commits behind
# Count how many commits the remote is ahead of us.
# Считаем сколько коммитов нас опережает удалённый репозиторий.
behind = subprocess.check_output(
["git", "rev-list", f"HEAD..origin/{branch}", "--count"],
cwd=_PROJECT_DIR, text=True,
@@ -45,7 +51,8 @@ def _task_update(screen: LogScreen) -> None:
screen.finish(True)
return
# Show incoming commits
# Show which commits are coming in so the user knows what changed.
# Показываем какие коммиты приходят, чтобы пользователь знал что изменилось.
screen.write(f"\n[bold]{behind} new commit(s):[/bold]")
log_lines = subprocess.check_output(
["git", "log", f"HEAD..origin/{branch}", "--oneline"],
@@ -71,6 +78,8 @@ def _task_update(screen: LogScreen) -> None:
screen.set_progress(80)
# Reinstall (80 → 100 %)
# Reinstall so the cobot binary picks up any new dependencies from pyproject.toml.
# Переустанавливаем, чтобы бинарник cobot подхватил новые зависимости из pyproject.toml.
screen.set_progress(80, "Reinstalling cobot CLI...")
screen.write("\n[cyan][*][/cyan] Reinstalling cobot CLI...")
reinstall = subprocess.run(
cobot/tui.py
+53 -12
View File
@@ -8,6 +8,8 @@ from textual.binding import Binding
from textual.screen import Screen
from textual.widgets import Footer, Input, LoadingIndicator, ProgressBar, RadioButton, RadioSet, RichLog, Static
# Shared CSS applied to every screen in the app.
# Общий CSS, применяемый ко всем экранам приложения.
SCREEN_CSS = """
Screen {
padding: 2 4;
@@ -77,6 +79,10 @@ RunScreen #hint {
"""
# A screen that shows a question and a list of radio button options.
# The user picks one and presses Enter - the chosen string is returned as the result.
# Экран с вопросом и списком вариантов в виде радио-кнопок.
# Пользователь выбирает один и нажимает Enter - выбранная строка возвращается как результат.
class PickScreen(Screen[Optional[str]]):
BINDINGS = [
Binding("enter", "submit", "Confirm", priority=True),
@@ -98,6 +104,8 @@ class PickScreen(Screen[Optional[str]]):
yield Static(self._note, id="note")
with RadioSet(id="choices"):
for opt in self._options:
# Pre-select the default option so the user can just press Enter to accept it.
# Заранее выделяем вариант по умолчанию, чтобы пользователь мог просто нажать Enter.
yield RadioButton(opt, value=(opt == self._default))
yield Footer()
@@ -115,9 +123,15 @@ class PickScreen(Screen[Optional[str]]):
self.dismiss(str(btn.label) if btn else self._default)
def action_abort(self) -> None:
# Exit the whole app, not just this screen, so the calling code knows the user cancelled.
# Выходим из всего приложения, а не только из этого экрана, чтобы вызывающий код знал об отмене.
self.app.exit(None)
# A screen that shows a question with a free-text input field.
# The user types a value, presses Enter, and the text is returned as the result.
# Экран с вопросом и полем для ввода произвольного текста.
# Пользователь вводит значение, нажимает Enter, и текст возвращается как результат.
class InputScreen(Screen[Optional[str]]):
BINDINGS = [
Binding("enter", "submit", "Confirm", priority=True),
@@ -154,9 +168,13 @@ class InputScreen(Screen[Optional[str]]):
self.app.exit(None)
# A screen that streams output from a background task into a scrollable log.
# Used for long-running operations like installs and builds.
# Press Enter or Escape to close once the task finishes.
# Экран, который транслирует вывод фоновой задачи в прокручиваемый лог.
# Используется для долгих операций, таких как установка и сборка.
# После завершения задачи закрывается по нажатию Enter или Escape.
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], show_progress: bool = False):
@@ -179,10 +197,13 @@ class LogScreen(Screen[bool]):
def on_mount(self) -> None:
self.query_one(RichLog).focus()
# Run the task in a worker thread so the UI stays responsive.
# Запускаем задачу в отдельном потоке, чтобы интерфейс не зависал.
self.app.run_worker(lambda: self._run_fn(self), thread=True)
def set_progress(self, pct: float, label: str = "") -> None:
"""Thread-safe: update the progress bar and optional step label."""
# Thread-safe - this is called from the worker thread, not the UI thread.
# Потокобезопасно - вызывается из рабочего потока, а не из потока интерфейса.
if self._show_progress:
self.app.call_from_thread(self._do_set_progress, pct, label)
@@ -192,14 +213,16 @@ class LogScreen(Screen[bool]):
self.query_one("#step-label", Static).update(label)
def write(self, line: str) -> None:
"""Thread-safe: append a line to the log."""
# Thread-safe - append a line to the log from a worker thread.
# Потокобезопасно - добавляет строку в лог из рабочего потока.
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."""
# Thread-safe - called by the task when it is done to show the close hint.
# Потокобезопасно - вызывается задачей по завершении, чтобы показать подсказку о закрытии.
self.app.call_from_thread(self._do_finish, success)
def _do_finish(self, success: bool) -> None:
@@ -214,13 +237,17 @@ class LogScreen(Screen[bool]):
self.query_one("#hint", Static).update(msg)
def action_close(self) -> None:
# Only allow closing after the task has finished, not while it is still running.
# Разрешаем закрытие только после завершения задачи, а не во время её работы.
if self._finished:
self.dismiss(self._success)
# A screen for a long-running process that the user can stop at any time.
# Shows a live log and offers S / Enter / Escape to stop or close.
# Экран для долго работающего процесса, который пользователь может остановить в любой момент.
# Показывает живой лог и предлагает S / Enter / Escape для остановки или закрытия.
class RunScreen(Screen[None]):
"""Streams a long-running process. S/Enter/Escape stops or closes."""
BINDINGS = [
Binding("s", "stop_close", "Stop", show=True, priority=True),
Binding("enter", "stop_close", "Close", show=False),
@@ -231,8 +258,8 @@ class RunScreen(Screen[None]):
super().__init__()
self._title = title
self._run_fn = task
self._proc = None # set via set_proc()
self._kill_fn = None # optional custom kill callable
self._proc = None # the subprocess, set via set_proc()
self._kill_fn = None # optional custom kill callable, set via set_kill_fn()
self._finished = False
self._stopped = False
@@ -245,23 +272,33 @@ class RunScreen(Screen[None]):
def on_mount(self) -> None:
self.query_one(RichLog).focus()
# Run the process task in a worker thread so the UI stays responsive.
# Запускаем задачу с процессом в отдельном потоке, чтобы интерфейс не зависал.
self.app.run_worker(lambda: self._run_fn(self), thread=True)
def set_proc(self, proc) -> None:
"""Register the running subprocess so Stop can terminate it."""
# Register the subprocess so the Stop button knows what to terminate.
# Регистрируем subprocess, чтобы кнопка Stop знала что завершать.
self._proc = proc
def set_kill_fn(self, fn: Callable) -> None:
"""Override the default terminate() with a custom kill function."""
# Override the default proc.terminate() with a custom kill function.
# For example, docker kill or os.killpg for process groups.
# Заменяем стандартный proc.terminate() кастомной функцией завершения.
# Например, docker kill или os.killpg для групп процессов.
self._kill_fn = fn
def write(self, line: str) -> None:
# Thread-safe - called from the worker thread to append a log line.
# Потокобезопасно - вызывается из рабочего потока для добавления строки в лог.
self.app.call_from_thread(self._append, line)
def _append(self, line: str) -> None:
self.query_one(RichLog).write(line)
def finish(self, stopped: bool = False) -> None:
# Thread-safe - called by the task when the process exits naturally.
# Потокобезопасно - вызывается задачей когда процесс завершается естественным образом.
self.app.call_from_thread(self._do_finish, stopped)
def _do_finish(self, stopped: bool) -> None:
@@ -274,6 +311,10 @@ class RunScreen(Screen[None]):
self.query_one("#hint", Static).update(msg)
def action_stop_close(self) -> None:
# This runs in the UI thread, so we call _append() directly instead of write()
# because write() uses call_from_thread() which only works from other threads.
# Выполняется в потоке UI, поэтому вызываем _append() напрямую, а не write(),
# потому что write() использует call_from_thread(), который работает только из других потоков.
if self._finished:
self.dismiss(None)
return
@@ -288,4 +329,4 @@ class RunScreen(Screen[None]):
self._proc.terminate()
except Exception:
pass
self.write("\n[yellow]Stopping process...[/yellow]")
self._append("\n[yellow]Stopping process...[/yellow]")