Files
lightweight-cobot/cobot/commands/local_setup.py
T

546 lines
23 KiB
Python

from __future__ import annotations
import argparse
import os
import shutil
import subprocess
import tempfile
import threading
import urllib.request
from pathlib import Path
from typing import Callable, List, Optional
from textual.app import App
from cobot.tui import SCREEN_CSS, LogScreen, PickScreen
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():
return False
info: dict[str, str] = {}
for line in path.read_text().splitlines():
if "=" in line:
k, _, v = line.partition("=")
info[k.strip()] = v.strip().strip('"')
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()
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,
env=env or os.environ, cwd=cwd,
)
if result.returncode != 0:
if write:
for line in (result.stdout + result.stderr).splitlines():
if line.strip():
write(line)
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,
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"Command failed: {cmd[0]}")
def _run_apt_with_progress(
cmd: List[str],
write: Write,
on_progress: Callable[[float], None],
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(
cmd + [f"-o", f"APT::Status-Fd={w_fd}"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
env=env or os.environ,
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:
with os.fdopen(r_fd, "r") as f:
for line in f:
# Format: dlstatus:N:PCT:MSG or pmstatus:NAME:PCT:MSG
parts = line.strip().split(":", 3)
if len(parts) >= 3:
try:
on_progress(float(parts[2]))
except ValueError:
pass
t = threading.Thread(target=_read_status, daemon=True)
t.start()
for line in proc.stdout:
s = line.rstrip()
if s:
write(s)
proc.wait()
t.join()
if proc.returncode != 0:
raise RuntimeError(f"Command failed: {cmd[0]}")
# 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:
write("[green][ok][/green] UTF-8 locale active")
return
write("[cyan][*][/cyan] Configuring UTF-8 locale...")
_run_quiet(["sudo", "apt-get", "update", "-qq"], write)
_run_quiet(["sudo", "apt-get", "install", "-y", "--no-install-recommends", "locales"], write, _APT_ENV)
_run_quiet(["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)
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:
on_progress(p)
write("[cyan][*][/cyan] Adding ROS2 apt repository...")
# Best-effort update - 60 second timeout so a bad mirror doesn't hang forever.
# Фоновое обновление с таймаутом 60 секунд, чтобы зависший зеркальный сервер не блокировал процесс.
subprocess.run(["sudo", "apt-get", "update", "-qq"], capture_output=True, timeout=60)
_prog(15)
_run_quiet(
["sudo", "apt-get", "install", "-y", "--no-install-recommends",
"software-properties-common", "curl", "gnupg"],
write, _APT_ENV,
)
_prog(35)
_run_quiet(["sudo", "add-apt-repository", "-y", "universe"], write)
_prog(45)
write("[cyan][*][/cyan] Downloading ROS2 signing key...")
with tempfile.NamedTemporaryFile(delete=False, suffix=".key") as tmp:
tmp_path = tmp.name
try:
# 30 second timeout - if GitHub is unreachable we fail fast instead of hanging forever.
# Таймаут 30 секунд - если GitHub недоступен, падаем быстро вместо бесконечного зависания.
with urllib.request.urlopen(_ROS_KEY_URL, timeout=30) as resp:
with open(tmp_path, "wb") as f:
f.write(resp.read())
_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)
write("[green][ok][/green] Signing key installed")
_prog(65)
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,
)
if proc.returncode != 0:
raise RuntimeError(f"Failed to write {_ROS_SOURCES}")
_prog(70)
write("[cyan][*][/cyan] Updating apt cache...")
_run_apt_with_progress(
["sudo", "apt-get", "update", "-q"],
write,
lambda p: _prog(70 + p * 0.30),
_APT_ENV,
)
write("[green][ok][/green] ROS2 repository ready")
_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(
["sudo", "apt-get", "install", "-y", "ros-jazzy-desktop", "ros-dev-tools"],
write,
on_progress or (lambda _: None),
_APT_ENV,
)
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")
return
write("[cyan][*][/cyan] Installing colcon...")
_run_quiet(
["sudo", "apt-get", "install", "-y", "--no-install-recommends",
"python3-colcon-common-extensions"],
write, _APT_ENV,
)
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")
source_line = "source /opt/ros/jazzy/setup.bash"
if rc.exists() and source_line in rc.read_text():
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")
write(f"[green][ok][/green] Added ROS2 setup to ~/{rc.name}")
# 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 %)
screen.set_progress(0, "Setting up locale...")
screen.write("[bold]Step 1 / 5 — Locale[/bold]")
_setup_locale(screen.write)
# Step 2 — ROS2 repo (5 → 20 %)
screen.set_progress(5, "Adding ROS2 repository...")
screen.write("\n[bold]Step 2 / 5 — ROS2 repository[/bold]")
_add_ros2_repo(
screen.write,
on_progress=lambda p: screen.set_progress(5 + p * 0.15),
)
# Step 3 — ROS2 Jazzy (20 → 85 %)
screen.set_progress(20, "Installing ROS2 Jazzy Desktop...")
screen.write("\n[bold]Step 3 / 5 — ROS2 Jazzy Desktop[/bold]")
_install_ros2_jazzy(
screen.write,
on_progress=lambda p: screen.set_progress(20 + p * 0.65),
)
# Step 4 — colcon (85 → 92 %)
screen.set_progress(85, "Installing colcon...")
screen.write("\n[bold]Step 4 / 5 — colcon[/bold]")
_install_colcon(screen.write)
# Step 5 — shell rc (92 → 100 %)
screen.set_progress(92, "Configuring shell...")
screen.write("\n[bold]Step 5 / 5 — Shell configuration[/bold]")
_setup_shell_rc(screen.write)
screen.set_progress(100, "Done")
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)
# 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"):
screen.write("[red]colcon not found.[/red]")
screen.write("Source ROS2 first: [bold]source /opt/ros/jazzy/setup.bash[/bold]")
screen.finish(False)
return
# 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,
)
total = max(len([l for l in list_result.stdout.splitlines() if l.strip()]), 1)
screen.write(f"[bold]Building {total} package(s) with colcon[/bold]\n")
screen.set_progress(0, f"0 / {total} packages done")
built = 0
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")
_run_logged(["colcon", "build", "--symlink-install"], _track, cwd=_PROJECT_DIR)
screen.set_progress(100, "Build complete")
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)
# 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
def on_mount(self) -> None:
self.push_screen(
PickScreen(
"ROS2 not found",
"ROS2 Jazzy is not installed. Install it now?",
["Yes, install ROS2 Jazzy", "No, skip"],
"Yes, install ROS2 Jazzy",
),
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, show_progress=True),
lambda _: self.exit(),
)
# Run the colcon build without asking any questions - used when ROS2 is already installed.
# Запускаем сборку colcon без лишних вопросов - используется когда ROS2 уже установлен.
class _BuildApp(App[None]):
CSS = SCREEN_CSS
def on_mount(self) -> None:
self.push_screen(
LogScreen("Building project", _task_build, show_progress=True),
lambda _: self.exit(),
)
# 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
def on_mount(self) -> None:
self.push_screen(
PickScreen(
"Unsupported OS",
"Ubuntu 24.04 not detected. Build a Docker image for development?",
["Yes, run docker-setup", "No, exit"],
"Yes, run docker-setup",
),
lambda v: self.exit(v is not None and v.startswith("Yes")),
)
def register(subparsers: argparse._SubParsersAction) -> None:
p = subparsers.add_parser(
"local-setup",
help="Install ROS2 Jazzy natively and build the project with colcon",
)
p.set_defaults(func=run)
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()