665 lines
28 KiB
Python
665 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
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
|
|
|
|
# GitHub repository that publishes the official ROS2 apt source package.
|
|
# Репозиторий GitHub, публикующий официальный пакет источника apt для ROS2.
|
|
_ROS_APT_SOURCE_API = (
|
|
"https://api.github.com/repos/ros-infrastructure/ros-apt-source/releases/latest"
|
|
)
|
|
_ROS_APT_SOURCE_DEB = (
|
|
"https://github.com/ros-infrastructure/ros-apt-source/releases/download"
|
|
"/{version}/ros2-apt-source_{version}.{codename}_all.deb"
|
|
)
|
|
|
|
# Suppress apt interactive prompts such as "restart services?".
|
|
# Подавляем интерактивные запросы apt, например "перезапустить службы?".
|
|
_APT_ENV = {**os.environ, "DEBIAN_FRONTEND": "noninteractive"}
|
|
|
|
# HTTP/HTTPS timeouts for apt so a stalled server does not hang the process forever.
|
|
# 60 seconds per connection attempt is generous enough for any healthy mirror.
|
|
# HTTP/HTTPS таймауты для apt, чтобы зависший сервер не блокировал процесс бесконечно.
|
|
# 60 секунд на попытку подключения достаточно для любого нормального зеркала.
|
|
_APT_TIMEOUTS = [
|
|
"-o", "Acquire::http::Timeout=60",
|
|
"-o", "Acquire::https::Timeout=60",
|
|
"-o", "Acquire::Retries=3",
|
|
]
|
|
|
|
|
|
# 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,
|
|
register_proc: Callable | None = None,
|
|
) -> None:
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
env=env or os.environ,
|
|
cwd=cwd,
|
|
)
|
|
if register_proc:
|
|
register_proc(proc)
|
|
for line in proc.stdout:
|
|
s = line.rstrip()
|
|
if s:
|
|
write(s)
|
|
proc.wait()
|
|
if proc.returncode not in (0, -9):
|
|
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,
|
|
register_proc: Callable | 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 + ["-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)
|
|
|
|
# Tell the caller about this process so it can be killed if the user cancels.
|
|
# Сообщаем вызывающему о процессе, чтобы его можно было завершить при отмене пользователем.
|
|
if register_proc:
|
|
register_proc(proc)
|
|
|
|
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 not in (0, -9):
|
|
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 using the ros2-apt-source package.
|
|
# Добавляем официальный репозиторий ROS2 через пакет ros2-apt-source.
|
|
def _add_ros2_repo(
|
|
write: Write,
|
|
on_progress: Optional[Callable[[float], None]] = None,
|
|
register_proc: Callable | None = None,
|
|
) -> None:
|
|
def _prog(p: float) -> None:
|
|
if on_progress:
|
|
on_progress(p)
|
|
|
|
write("[cyan][*][/cyan] Configuring ROS2 apt repository...")
|
|
|
|
for leftover in [
|
|
"/etc/apt/sources.list.d/ros2.list",
|
|
"/usr/share/keyrings/ros-archive-keyring.gpg",
|
|
]:
|
|
if Path(leftover).exists():
|
|
subprocess.run(["sudo", "rm", "-f", leftover], capture_output=True)
|
|
write(f"[dim]Removed leftover: {leftover}[/dim]")
|
|
|
|
# Update apt cache so we can install curl and software-properties-common.
|
|
# Обновляем кеш apt перед установкой curl и software-properties-common.
|
|
subprocess.run(["sudo", "apt-get", "update"] + _APT_TIMEOUTS, capture_output=True, timeout=120)
|
|
_prog(10)
|
|
|
|
_run_quiet(
|
|
["sudo", "apt-get", "install", "-y", "--no-install-recommends",
|
|
"software-properties-common", "curl"],
|
|
write, _APT_ENV,
|
|
)
|
|
_prog(20)
|
|
_run_quiet(["sudo", "add-apt-repository", "-y", "universe"], write)
|
|
_prog(30)
|
|
|
|
# Fetch the latest ros2-apt-source release version from the GitHub API.
|
|
# Получаем последнюю версию ros2-apt-source через GitHub API.
|
|
write("[cyan][*][/cyan] Fetching ros2-apt-source package...")
|
|
req = urllib.request.Request(
|
|
_ROS_APT_SOURCE_API,
|
|
headers={"User-Agent": "cobot-installer"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
version = json.loads(resp.read())["tag_name"]
|
|
|
|
codename = subprocess.check_output(
|
|
["bash", "-c", ". /etc/os-release && echo ${UBUNTU_CODENAME:-${VERSION_CODENAME}}"],
|
|
text=True,
|
|
).strip()
|
|
deb_url = _ROS_APT_SOURCE_DEB.format(version=version, codename=codename)
|
|
_prog(40)
|
|
|
|
# Download the .deb and install it. The package sets up the GPG key and sources list automatically.
|
|
# Скачиваем .deb и устанавливаем его. Пакет сам настраивает GPG-ключ и список источников.
|
|
with tempfile.NamedTemporaryFile(suffix=".deb", delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
try:
|
|
with urllib.request.urlopen(deb_url, timeout=60) as resp:
|
|
Path(tmp_path).write_bytes(resp.read())
|
|
_prog(55)
|
|
|
|
# apt install resolves dependencies automatically unlike dpkg -i.
|
|
# The "./" prefix tells apt this is a local file, not a package name from a repo.
|
|
# apt install автоматически разрешает зависимости в отличие от dpkg -i.
|
|
# Префикс "./" говорит apt что это локальный файл, а не имя пакета из репозитория.
|
|
_run_quiet(["sudo", "apt-get", "install", "-y", tmp_path], write, _APT_ENV)
|
|
finally:
|
|
os.unlink(tmp_path)
|
|
|
|
write("[green][ok][/green] ROS2 apt source installed")
|
|
_prog(60)
|
|
|
|
# Ask dpkg which files the package installed and find the apt sources entry.
|
|
# This is more reliable than comparing directory snapshots: dpkg knows exactly what it placed.
|
|
# Спрашиваем dpkg какие файлы установил пакет и находим файл источников apt.
|
|
# Это надёжнее сравнения снимков директории: dpkg точно знает что он положил.
|
|
dpkg_files = subprocess.run(
|
|
["dpkg", "-L", "ros2-apt-source"],
|
|
capture_output=True, text=True,
|
|
).stdout.splitlines()
|
|
sources_file = next(
|
|
(f for f in dpkg_files
|
|
if f.startswith("/etc/apt/") and f.endswith((".list", ".sources"))),
|
|
None,
|
|
)
|
|
|
|
if sources_file:
|
|
write(f"[dim]Sources file: {sources_file}[/dim]")
|
|
update_cmd = [
|
|
"sudo", "apt-get", "update",
|
|
"-o", f"Dir::Etc::sourcelist={sources_file}",
|
|
"-o", "Dir::Etc::sourceparts=-",
|
|
"-o", "APT::Get::List-Cleanup=0",
|
|
] + _APT_TIMEOUTS
|
|
else:
|
|
# Fall back to full update if dpkg -L does not reveal the sources file.
|
|
# Запасной вариант: полный update если dpkg -L не раскрыл файл источников.
|
|
write("[dim]Sources file not found via dpkg -L, running full apt-get update[/dim]")
|
|
update_cmd = ["sudo", "apt-get", "update"] + _APT_TIMEOUTS
|
|
|
|
write("[cyan][*][/cyan] Updating ROS2 package list...")
|
|
_run_apt_with_progress(
|
|
update_cmd, write,
|
|
lambda p: _prog(60 + p * 0.40),
|
|
_APT_ENV,
|
|
register_proc=register_proc,
|
|
)
|
|
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,
|
|
register_proc: Callable | 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"] + _APT_TIMEOUTS,
|
|
write,
|
|
on_progress or (lambda _: None),
|
|
_APT_ENV,
|
|
register_proc=register_proc,
|
|
)
|
|
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),
|
|
register_proc=screen.set_proc,
|
|
)
|
|
|
|
if screen.is_stopped():
|
|
return
|
|
|
|
# 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),
|
|
register_proc=screen.set_proc,
|
|
)
|
|
|
|
if screen.is_stopped():
|
|
return
|
|
|
|
# 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")
|
|
|
|
if not screen.is_stopped():
|
|
screen.write(
|
|
"\nRestart the terminal, then run [bold]cobot local-setup[/bold] again to build."
|
|
)
|
|
screen.finish(True)
|
|
except Exception as exc:
|
|
if not screen.is_stopped():
|
|
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, register_proc=screen.set_proc)
|
|
|
|
if not screen.is_stopped():
|
|
screen.set_progress(100, "Build complete")
|
|
screen.write("\nActivate workspace: [bold]source install/setup.bash[/bold]")
|
|
screen.finish(True)
|
|
except Exception as exc:
|
|
if not screen.is_stopped():
|
|
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...")
|
|
|
|
# Download in 64 KB chunks so we can update the progress bar and bail out if the user
|
|
# cancels midway through instead of blocking in urlretrieve until the full file arrives.
|
|
# Скачиваем по 64 КБ, чтобы обновлять прогресс-бар и прерваться при отмене пользователем,
|
|
# а не блокироваться в urlretrieve до получения всего файла.
|
|
with urllib.request.urlopen(_WEBOTS_DEB_URL, timeout=30) as resp:
|
|
total = int(resp.headers.get("Content-Length", 0))
|
|
downloaded = 0
|
|
with open(deb_path, "wb") as f:
|
|
while True:
|
|
if screen.is_stopped():
|
|
return
|
|
chunk = resp.read(65536)
|
|
if not chunk:
|
|
break
|
|
f.write(chunk)
|
|
downloaded += len(chunk)
|
|
if total > 0:
|
|
pct = min(downloaded / total * 65, 65)
|
|
mb = downloaded / 1_048_576
|
|
total_mb = total / 1_048_576
|
|
screen.set_progress(pct, f"Downloading... {mb:.0f} / {total_mb:.0f} MB")
|
|
|
|
if screen.is_stopped():
|
|
return
|
|
|
|
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,
|
|
)
|
|
screen.set_proc(proc)
|
|
for line in proc.stdout:
|
|
s = line.rstrip()
|
|
if s:
|
|
screen.write(s)
|
|
proc.wait()
|
|
|
|
if proc.returncode not in (0, -9):
|
|
screen.write("\n[red]Installation failed.[/red]")
|
|
screen.finish(False)
|
|
return
|
|
|
|
if screen.is_stopped():
|
|
return
|
|
|
|
screen.set_progress(100, "Done")
|
|
screen.write("\n[green]Webots installed successfully.[/green]")
|
|
screen.finish(True)
|
|
except Exception as exc:
|
|
if not screen.is_stopped():
|
|
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():
|
|
# Cache the sudo token now, while the terminal is in normal mode and the password
|
|
# prompt is visible. Once Textual takes over the screen, sudo prompts become invisible
|
|
# and the process hangs silently waiting for input that never arrives.
|
|
# Кешируем sudo-токен сейчас, пока терминал работает в обычном режиме и запрос пароля
|
|
# виден пользователю. После запуска Textual sudo не может показать запрос и процесс
|
|
# зависает молча, ожидая ввод который никогда не придёт.
|
|
subprocess.run(["sudo", "-v"], check=False)
|
|
_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"):
|
|
# Same sudo pre-cache before launching the Webots installer TUI.
|
|
# Тот же предварительный кеш sudo перед запуском TUI установщика Webots.
|
|
subprocess.run(["sudo", "-v"], check=False)
|
|
WebotsInstallApp().run()
|