fix: add docstrings to improve code documentation and clarity

This commit is contained in:
Даниил Грабарь
2026-05-22 18:16:59 +03:00
parent a7590cc93f
commit 463a1e1423
9 changed files with 209 additions and 0 deletions
+14
View File
@@ -39,6 +39,12 @@ _DESCRIPTION = "Lightweight Cobot"
# Custom --help action that prints commands grouped by category instead of a flat list. # Custom --help action that prints commands grouped by category instead of a flat list.
# Кастомный обработчик --help, который выводит команды по категориям, а не одним списком. # Кастомный обработчик --help, который выводит команды по категориям, а не одним списком.
class _GroupedHelpAction(argparse.Action): class _GroupedHelpAction(argparse.Action):
"""Custom argparse action that replaces the default --help output with a grouped
command listing organized by category (Setup, Run, Management).
Кастомный обработчик argparse, заменяющий стандартный вывод --help на сгруппированный
список команд по категориям (Setup, Run, Management).
"""
def __init__(self, option_strings, dest, default=None, required=False, help=None): def __init__(self, option_strings, dest, default=None, required=False, help=None):
super().__init__( super().__init__(
option_strings=option_strings, option_strings=option_strings,
@@ -63,6 +69,9 @@ class _GroupedHelpAction(argparse.Action):
def main(): def main():
"""Entry point for the cobot CLI. Parses arguments and dispatches to the correct command.
Точка входа CLI cobot. Разбирает аргументы и вызывает нужную команду.
"""
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="cobot", prog="cobot",
description=_DESCRIPTION, description=_DESCRIPTION,
@@ -85,6 +94,11 @@ def main():
def _register_commands(subparsers): def _register_commands(subparsers):
"""Register all command subparsers. Each command module calls register() which adds its
own subparser and sets args.func to its run() function.
Регистрирует все подпарсеры команд. Каждый модуль вызывает register(), добавляет свой
подпарсер и устанавливает args.func на свою функцию run().
"""
# Each module registers its own subparser and sets args.func to its run() function. # Each module registers its own subparser and sets args.func to its run() function.
# Каждый модуль регистрирует свой подпарсер и устанавливает args.func на свою функцию run(). # Каждый модуль регистрирует свой подпарсер и устанавливает args.func на свою функцию run().
cmd_setup.register(subparsers) cmd_setup.register(subparsers)
+33
View File
@@ -16,6 +16,9 @@ _PROJECT_DIR = Path(__file__).parent.parent.parent
# Stop and remove all Docker containers whose name contains "lwc". # Stop and remove all Docker containers whose name contains "lwc".
# Останавливаем и удаляем все Docker-контейнеры, чьё имя содержит "lwc". # Останавливаем и удаляем все Docker-контейнеры, чьё имя содержит "lwc".
def _stop_docker_containers(write) -> None: def _stop_docker_containers(write) -> None:
"""Stop and force-remove all Docker containers whose name contains "lwc".
Останавливает и принудительно удаляет все Docker-контейнеры с "lwc" в имени.
"""
write("[cyan][*][/cyan] Stopping Docker containers...") write("[cyan][*][/cyan] Stopping Docker containers...")
result = subprocess.run( result = subprocess.run(
["docker", "ps", "-a", "--filter", "name=lwc", "--format", "{{.Names}}"], ["docker", "ps", "-a", "--filter", "name=lwc", "--format", "{{.Names}}"],
@@ -33,6 +36,9 @@ def _stop_docker_containers(write) -> None:
# Remove all Docker images whose repository or tag contains "lwc". # Remove all Docker images whose repository or tag contains "lwc".
# Удаляем все Docker-образы, репозиторий или тег которых содержит "lwc". # Удаляем все Docker-образы, репозиторий или тег которых содержит "lwc".
def _remove_docker_images(write) -> None: def _remove_docker_images(write) -> None:
"""Force-remove all local Docker images whose name or tag contains "lwc".
Принудительно удаляет все локальные Docker-образы с "lwc" в имени или теге.
"""
write("[cyan][*][/cyan] Removing Docker images...") write("[cyan][*][/cyan] Removing Docker images...")
result = subprocess.run( result = subprocess.run(
["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"], ["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"],
@@ -53,6 +59,9 @@ def _remove_docker_images(write) -> None:
# Remove the Docker volume that stores the Webots asset cache. # Remove the Docker volume that stores the Webots asset cache.
# Удаляем Docker volume с кэшем ассетов Webots. # Удаляем Docker volume с кэшем ассетов Webots.
def _remove_webots_volume(write) -> None: def _remove_webots_volume(write) -> None:
"""Remove the lwc-webots-cache Docker volume if it exists. Skips silently if absent.
Удаляет Docker volume lwc-webots-cache если он существует. Молча пропускает если отсутствует.
"""
result = subprocess.run( result = subprocess.run(
["docker", "volume", "inspect", "lwc-webots-cache"], ["docker", "volume", "inspect", "lwc-webots-cache"],
capture_output=True, capture_output=True,
@@ -69,6 +78,11 @@ def _remove_webots_volume(write) -> None:
# Удаляем пакеты ROS2 Jazzy через apt и очищаем строку source из конфигов оболочки. # Удаляем пакеты ROS2 Jazzy через apt и очищаем строку source из конфигов оболочки.
# Используем официальные команды удаления, которые также снимают регистрацию apt-репозитория ROS2. # Используем официальные команды удаления, которые также снимают регистрацию apt-репозитория ROS2.
def _remove_ros2(write) -> None: def _remove_ros2(write) -> None:
"""Remove all ros-jazzy-* packages, the ros2-apt-source package, and the ROS2 source
line from .bashrc / .zshrc. Does nothing if /opt/ros/jazzy is not present.
Удаляет все пакеты ros-jazzy-*, пакет ros2-apt-source и строку source ROS2 из
.bashrc / .zshrc. Ничего не делает если /opt/ros/jazzy отсутствует.
"""
write("[cyan][*][/cyan] Removing ROS2 Jazzy packages...") write("[cyan][*][/cyan] Removing ROS2 Jazzy packages...")
if not Path("/opt/ros/jazzy").exists(): if not Path("/opt/ros/jazzy").exists():
write("[dim]ROS2 Jazzy not found, skipping.[/dim]") write("[dim]ROS2 Jazzy not found, skipping.[/dim]")
@@ -113,6 +127,9 @@ def _remove_ros2(write) -> None:
# Remove Webots from the system via apt. # Remove Webots from the system via apt.
# Удаляем Webots из системы через apt. # Удаляем Webots из системы через apt.
def _remove_webots(write) -> None: def _remove_webots(write) -> None:
"""Remove the webots package via apt and run autoremove. Skips if webots is not found on PATH.
Удаляет пакет webots через apt и запускает autoremove. Пропускает если webots не найден в PATH.
"""
write("[cyan][*][/cyan] Removing Webots...") write("[cyan][*][/cyan] Removing Webots...")
if not shutil.which("webots"): if not shutil.which("webots"):
write("[dim]Webots not found, skipping.[/dim]") write("[dim]Webots not found, skipping.[/dim]")
@@ -125,6 +142,9 @@ def _remove_webots(write) -> None:
# Uninstall the cobot CLI from the uv tool store. # Uninstall the cobot CLI from the uv tool store.
# Удаляем cobot CLI из хранилища инструментов uv. # Удаляем cobot CLI из хранилища инструментов uv.
def _uninstall_cobot(write) -> None: def _uninstall_cobot(write) -> None:
"""Uninstall the lightweight-cobot package from the uv tool store.
Удаляет пакет lightweight-cobot из хранилища инструментов uv.
"""
write("[cyan][*][/cyan] Uninstalling cobot CLI...") write("[cyan][*][/cyan] Uninstalling cobot CLI...")
result = subprocess.run( result = subprocess.run(
["uv", "tool", "uninstall", "lightweight-cobot"], ["uv", "tool", "uninstall", "lightweight-cobot"],
@@ -139,6 +159,9 @@ def _uninstall_cobot(write) -> None:
# Delete the entire project directory from disk. # Delete the entire project directory from disk.
# Удаляем всю директорию проекта с диска. # Удаляем всю директорию проекта с диска.
def _remove_project_dir(write) -> None: def _remove_project_dir(write) -> None:
"""Recursively delete the entire project directory (_PROJECT_DIR) from disk.
Рекурсивно удаляет всю директорию проекта (_PROJECT_DIR) с диска.
"""
write(f"[cyan][*][/cyan] Removing project directory...") write(f"[cyan][*][/cyan] Removing project directory...")
try: try:
shutil.rmtree(_PROJECT_DIR) shutil.rmtree(_PROJECT_DIR)
@@ -153,6 +176,11 @@ def _remove_project_dir(write) -> None:
# Выполняем все шаги удаления по порядку. # Выполняем все шаги удаления по порядку.
# Диапазоны прогресса делятся равномерно между активными шагами, чтобы бар всегда доходил до 100%. # Диапазоны прогресса делятся равномерно между активными шагами, чтобы бар всегда доходил до 100%.
def _task_delete(screen: LogScreen, remove_ros: bool, remove_webots: bool) -> None: def _task_delete(screen: LogScreen, remove_ros: bool, remove_webots: bool) -> None:
"""Worker function that runs inside LogScreen. Runs all deletion steps in order:
containers -> images -> ROS2 (optional) -> Webots (optional) -> cobot CLI -> project dir.
Рабочая функция внутри LogScreen. Выполняет все шаги удаления по порядку:
контейнеры -> образы -> ROS2 (опционально) -> Webots (опционально) -> cobot CLI -> директория.
"""
try: try:
screen.set_progress(0, "Stopping containers...") screen.set_progress(0, "Stopping containers...")
_stop_docker_containers(screen.write) _stop_docker_containers(screen.write)
@@ -194,6 +222,11 @@ def _task_delete(screen: LogScreen, remove_ros: bool, remove_webots: bool) -> No
# Многошаговый мастер подтверждения перед удалением. # Многошаговый мастер подтверждения перед удалением.
# Дополнительные вопросы показываются только если соответствующее ПО действительно установлено. # Дополнительные вопросы показываются только если соответствующее ПО действительно установлено.
class _DeleteApp(App[None]): class _DeleteApp(App[None]):
"""Deletion wizard that asks for confirmation, then optionally asks about ROS2 and Webots,
then launches LogScreen running _task_delete.
Мастер удаления: просит подтверждение, затем опционально спрашивает про ROS2 и Webots,
затем запускает LogScreen с _task_delete.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def on_mount(self) -> None: def on_mount(self) -> None:
+28
View File
@@ -30,12 +30,18 @@ Write = Callable[[str], None]
# Thin wrapper around docker so we do not repeat ["docker", ...] everywhere. # Thin wrapper around docker so we do not repeat ["docker", ...] everywhere.
# Тонкая обёртка вокруг docker, чтобы не повторять ["docker", ...] везде. # Тонкая обёртка вокруг docker, чтобы не повторять ["docker", ...] везде.
def _docker(*args: str, capture: bool = False) -> subprocess.CompletedProcess: def _docker(*args: str, capture: bool = False) -> subprocess.CompletedProcess:
"""Run a docker subcommand. Pass capture=True to capture stdout/stderr instead of printing.
Запускает подкоманду docker. capture=True перехватывает stdout/stderr вместо вывода на экран.
"""
return subprocess.run(["docker", *args], capture_output=capture, text=True) return subprocess.run(["docker", *args], capture_output=capture, text=True)
# Check whether the docs container is currently running. # Check whether the docs container is currently running.
# Проверяем, запущен ли сейчас контейнер с документацией. # Проверяем, запущен ли сейчас контейнер с документацией.
def _is_running() -> bool: def _is_running() -> bool:
"""Return True if the lwc-docs container is currently running.
Возвращает True если контейнер lwc-docs в данный момент запущен.
"""
r = _docker("ps", "--filter", f"name={_CONTAINER_NAME}", "--format", "{{.Names}}", capture=True) r = _docker("ps", "--filter", f"name={_CONTAINER_NAME}", "--format", "{{.Names}}", capture=True)
return _CONTAINER_NAME in r.stdout return _CONTAINER_NAME in r.stdout
@@ -43,6 +49,9 @@ def _is_running() -> bool:
# Check whether the docs Docker image has already been built. # Check whether the docs Docker image has already been built.
# Проверяем, был ли уже собран Docker-образ для документации. # Проверяем, был ли уже собран Docker-образ для документации.
def _image_exists() -> bool: def _image_exists() -> bool:
"""Return True if the lwc-docs Docker image exists locally.
Возвращает True если Docker-образ lwc-docs существует локально.
"""
return bool(_docker("images", "-q", _IMAGE_NAME, capture=True).stdout.strip()) return bool(_docker("images", "-q", _IMAGE_NAME, capture=True).stdout.strip())
@@ -55,6 +64,9 @@ def _build_docs_image(
on_progress: Optional[Callable[[float], None]] = None, on_progress: Optional[Callable[[float], None]] = None,
register_proc: Optional[Callable] = None, register_proc: Optional[Callable] = None,
) -> bool: ) -> bool:
"""Build the lwc-docs Docker image from the doc/lwc-doc directory. Returns True on success.
Собирает Docker-образ lwc-docs из директории doc/lwc-doc. Возвращает True при успехе.
"""
write("[cyan][*][/cyan] Building documentation image (runs once)...") 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 gives us "Step X/Y" lines that we can parse for progress.
# DOCKER_BUILDKIT=0 даёт нам строки "Step X/Y", которые можно парсить для прогресса. # DOCKER_BUILDKIT=0 даёт нам строки "Step X/Y", которые можно парсить для прогресса.
@@ -87,6 +99,9 @@ def _build_docs_image(
# Start the docs server. Builds the image first if it does not exist yet. # Start the docs server. Builds the image first if it does not exist yet.
# Запускаем сервер документации. Сначала собирает образ, если он ещё не существует. # Запускаем сервер документации. Сначала собирает образ, если он ещё не существует.
def _task_up(screen: LogScreen, port: str) -> None: def _task_up(screen: LogScreen, port: str) -> None:
"""Worker function for the "up" action. Builds the image if missing, then starts the container.
Рабочая функция для действия "up". Собирает образ если отсутствует, затем запускает контейнер.
"""
try: try:
if _is_running(): if _is_running():
screen.write(f"[green]Docs already running at:[/green] http://localhost:{port}") screen.write(f"[green]Docs already running at:[/green] http://localhost:{port}")
@@ -153,6 +168,9 @@ def _task_up(screen: LogScreen, port: str) -> None:
# Stop the running docs container. # Stop the running docs container.
# Останавливаем работающий контейнер с документацией. # Останавливаем работающий контейнер с документацией.
def _task_down(screen: LogScreen) -> None: def _task_down(screen: LogScreen) -> None:
"""Worker function for the "down" action. Stops the lwc-docs container if it is running.
Рабочая функция для действия "down". Останавливает контейнер lwc-docs если он запущен.
"""
try: try:
if not _is_running(): if not _is_running():
screen.write("[yellow]Docs container is not running.[/yellow]") screen.write("[yellow]Docs container is not running.[/yellow]")
@@ -176,6 +194,11 @@ def _task_down(screen: LogScreen) -> None:
# Stop the container, remove the old image, rebuild it, and start a new container. # Stop the container, remove the old image, rebuild it, and start a new container.
# Останавливаем контейнер, удаляем старый образ, пересобираем и запускаем новый контейнер. # Останавливаем контейнер, удаляем старый образ, пересобираем и запускаем новый контейнер.
def _task_rebuild(screen: LogScreen, port: str) -> None: def _task_rebuild(screen: LogScreen, port: str) -> None:
"""Worker function for the "rebuild" action. Stops the container, removes the old image,
rebuilds it, and starts a fresh container on the given port.
Рабочая функция для действия "rebuild". Останавливает контейнер, удаляет старый образ,
пересобирает его и запускает новый контейнер на указанном порту.
"""
try: try:
if _is_running(): if _is_running():
screen.set_progress(5, "Stopping container...") screen.set_progress(5, "Stopping container...")
@@ -236,6 +259,11 @@ def _task_rebuild(screen: LogScreen, port: str) -> None:
# One app handles all three actions (up/down/rebuild) by branching in on_mount. # One app handles all three actions (up/down/rebuild) by branching in on_mount.
# Одно приложение обрабатывает все три действия (up/down/rebuild), разветвляясь в on_mount. # Одно приложение обрабатывает все три действия (up/down/rebuild), разветвляясь в on_mount.
class _DocApp(App[None]): class _DocApp(App[None]):
"""Documentation server app. Handles "up", "down", and "rebuild" actions by branching
in on_mount to the appropriate LogScreen task.
Приложение сервера документации. Обрабатывает действия "up", "down" и "rebuild",
разветвляясь в on_mount к соответствующей задаче LogScreen.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def __init__(self, action: str): def __init__(self, action: str):
+25
View File
@@ -70,6 +70,11 @@ def _build_image(
build_type: str = "release", build_type: str = "release",
register_proc: Optional[Callable] = None, register_proc: Optional[Callable] = None,
) -> bool: ) -> bool:
"""Build a single Docker image from a Dockerfile and stream its output line by line.
Returns True on success, False if the build failed or was cancelled.
Собирает один Docker-образ из Dockerfile и транслирует вывод построчно.
Возвращает True при успехе, False если сборка завершилась ошибкой или была отменена.
"""
write(f"[cyan][*][/cyan] Building [bold]{name}[/bold]...") 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 gives us "Step X/Y" lines in the output which we parse for progress.
# DOCKER_BUILDKIT=0 даёт нам строки "Step X/Y" в выводе, которые мы парсим для прогресса. # DOCKER_BUILDKIT=0 даёт нам строки "Step X/Y" в выводе, которые мы парсим для прогресса.
@@ -115,6 +120,11 @@ def _pull_image(
on_progress: Optional[Callable[[float], None]] = None, on_progress: Optional[Callable[[float], None]] = None,
register_proc: Optional[Callable] = None, register_proc: Optional[Callable] = None,
) -> bool: ) -> bool:
"""Pull a Docker image from Docker Hub and report layer-by-layer progress.
Returns True on success, False if the pull failed or was cancelled.
Скачивает Docker-образ с Docker Hub и сообщает о прогрессе по слоям.
Возвращает True при успехе, False если скачивание завершилось ошибкой или было отменено.
"""
write(f"[cyan][*][/cyan] Pulling [bold]{name}[/bold] ({tag})...") write(f"[cyan][*][/cyan] Pulling [bold]{name}[/bold] ({tag})...")
if on_progress: if on_progress:
on_progress(5) on_progress(5)
@@ -156,6 +166,11 @@ def _pull_image(
# Основная работа - собираем или скачиваем все образы в зависимости от выбора пользователя. # Основная работа - собираем или скачиваем все образы в зависимости от выбора пользователя.
# Каждый образ получает свой кусок прогресс-бара, чтобы общий бар двигался равномерно. # Каждый образ получает свой кусок прогресс-бара, чтобы общий бар двигался равномерно.
def _task_execute(screen: LogScreen, cfg: _Config) -> None: def _task_execute(screen: LogScreen, cfg: _Config) -> None:
"""Worker function that runs inside LogScreen. Builds or pulls all images in the chain
defined by the user's choices and updates the progress bar after each image.
Рабочая функция, выполняемая внутри LogScreen. Собирает или скачивает все образы из цепочки
согласно выбору пользователя и обновляет прогресс-бар после каждого образа.
"""
try: try:
chain = _WEBOTS_CHAIN if cfg.variant == "webots" else _CONTROLLER_CHAIN chain = _WEBOTS_CHAIN if cfg.variant == "webots" else _CONTROLLER_CHAIN
n = len(chain) n = len(chain)
@@ -244,6 +259,11 @@ def _task_execute(screen: LogScreen, cfg: _Config) -> None:
# Сканируем директорию docker/ на наличие поддиректорий с именами версий ROS (например jazzy). # Сканируем директорию docker/ на наличие поддиректорий с именами версий ROS (например jazzy).
# Если ничего не найдено, используем "jazzy" по умолчанию, чтобы мастер всё равно работал. # Если ничего не найдено, используем "jazzy" по умолчанию, чтобы мастер всё равно работал.
def _discover_versions() -> List[str]: def _discover_versions() -> List[str]:
"""Return a sorted list of ROS versions found in docker/. Jazzy is placed first.
Falls back to ["jazzy"] if the directory does not exist or is empty.
Возвращает отсортированный список версий ROS найденных в docker/. Jazzy идёт первым.
Возвращает ["jazzy"] если директория не существует или пуста.
"""
if not _DOCKER_DIR.exists(): if not _DOCKER_DIR.exists():
return ["jazzy"] return ["jazzy"]
dirs = sorted(d.name for d in _DOCKER_DIR.iterdir() if d.is_dir()) dirs = sorted(d.name for d in _DOCKER_DIR.iterdir() if d.is_dir())
@@ -257,6 +277,11 @@ def _discover_versions() -> List[str]:
# Multi-step wizard that collects all build options before starting the actual image build. # Multi-step wizard that collects all build options before starting the actual image build.
# Многошаговый мастер, который собирает все параметры сборки перед запуском фактической сборки образа. # Многошаговый мастер, который собирает все параметры сборки перед запуском фактической сборки образа.
class _Wizard(App[None]): class _Wizard(App[None]):
"""Five-step wizard: ROS version -> source (pull/build) -> variant -> build type -> repo/prefix.
Collects all options, then hands off to LogScreen which runs _task_execute.
Пятишаговый мастер: версия ROS -> источник (pull/build) -> вариант -> тип сборки -> репо/префикс.
Собирает все параметры, затем передаёт управление LogScreen который запускает _task_execute.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def __init__(self, versions: List[str], default_version: str = "jazzy"): def __init__(self, versions: List[str], default_version: str = "jazzy"):
+25
View File
@@ -125,6 +125,9 @@ _BLOCKS: List[_Block] = [
# Пытаемся сохранить исходный тип YAML (bool, int, float) при записи значения обратно. # Пытаемся сохранить исходный тип YAML (bool, int, float) при записи значения обратно.
# Сохранение типа предотвращает превращение "true" в обычную строку в YAML-файле. # Сохранение типа предотвращает превращение "true" в обычную строку в YAML-файле.
def _coerce(value: str, original: Any) -> Any: def _coerce(value: str, original: Any) -> Any:
"""Convert a string value to match the type of the original YAML value (bool, int, float, str).
Преобразует строковое значение к типу исходного значения YAML (bool, int, float, str).
"""
if isinstance(original, bool): if isinstance(original, bool):
return value.lower() == "true" return value.lower() == "true"
if isinstance(original, int): if isinstance(original, int):
@@ -143,6 +146,9 @@ def _coerce(value: str, original: Any) -> Any:
# Read a value from a nested YAML mapping using a dot-separated key like "webots.transform". # Read a value from a nested YAML mapping using a dot-separated key like "webots.transform".
# Читаем значение из вложенного YAML-словаря по ключу с точками, например "webots.transform". # Читаем значение из вложенного YAML-словаря по ключу с точками, например "webots.transform".
def _get_nested(mapping: Any, path: str) -> Any: def _get_nested(mapping: Any, path: str) -> Any:
"""Return the value at a dot-separated path inside a nested YAML mapping, or None if missing.
Возвращает значение по пути с точками внутри вложенного YAML-словаря, или None если отсутствует.
"""
keys = path.split(".") keys = path.split(".")
cur = mapping cur = mapping
for k in keys: for k in keys:
@@ -155,6 +161,9 @@ def _get_nested(mapping: Any, path: str) -> Any:
# Write a value into a nested YAML mapping using a dot-separated key. # Write a value into a nested YAML mapping using a dot-separated key.
# Записываем значение в вложенный YAML-словарь по ключу с точками. # Записываем значение в вложенный YAML-словарь по ключу с точками.
def _set_nested(mapping: Any, path: str, value: Any) -> None: def _set_nested(mapping: Any, path: str, value: Any) -> None:
"""Set the value at a dot-separated path inside a nested YAML mapping, coercing type to match.
Устанавливает значение по пути с точками во вложенном YAML-словаре, приводя тип к исходному.
"""
keys = path.split(".") keys = path.split(".")
cur = mapping cur = mapping
for k in keys[:-1]: for k in keys[:-1]:
@@ -166,6 +175,9 @@ def _set_nested(mapping: Any, path: str, value: Any) -> None:
# Shown after all blocks have been configured to confirm the file was saved. # Shown after all blocks have been configured to confirm the file was saved.
# Показывается после настройки всех блоков для подтверждения сохранения файла. # Показывается после настройки всех блоков для подтверждения сохранения файла.
class _SavedScreen(Screen[None]): class _SavedScreen(Screen[None]):
"""Confirmation screen shown after all configuration blocks are saved. Press Enter to close.
Экран подтверждения, показываемый после сохранения всех блоков конфигурации. Enter для закрытия.
"""
BINDINGS = [Binding("enter,escape", "close", "Close")] BINDINGS = [Binding("enter,escape", "close", "Close")]
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
@@ -183,6 +195,13 @@ class _SavedScreen(Screen[None]):
# Главный мастер конфигурации. Проходит по каждому блоку по порядку. # Главный мастер конфигурации. Проходит по каждому блоку по порядку.
# Для каждого блока сначала спрашивает "Настроить X?" а затем проходит по всем его полям. # Для каждого блока сначала спрашивает "Настроить X?" а затем проходит по всем его полям.
class _Wizard(App[None]): class _Wizard(App[None]):
"""Configuration wizard that iterates over all _BLOCKS. For each block it asks
"Configure X?" and if confirmed steps through every field with PickScreen or InputScreen.
Saves to cobot-setting.yaml when all blocks are done and shows _SavedScreen.
Мастер конфигурации, проходящий по всем _BLOCKS. Для каждого блока спрашивает
"Настроить X?" и при подтверждении проходит по всем полям через PickScreen или InputScreen.
Сохраняет в cobot-setting.yaml по завершении и показывает _SavedScreen.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def __init__(self, data: Any): def __init__(self, data: Any):
@@ -275,6 +294,9 @@ class _Wizard(App[None]):
# Load the config file preserving all comments and key order. # Load the config file preserving all comments and key order.
# Загружаем конфиг-файл, сохраняя все комментарии и порядок ключей. # Загружаем конфиг-файл, сохраняя все комментарии и порядок ключей.
def _load_config() -> Any: def _load_config() -> Any:
"""Load cobot-setting.yaml with ruamel.yaml, preserving comments and key order.
Загружает cobot-setting.yaml с помощью ruamel.yaml, сохраняя комментарии и порядок ключей.
"""
with open(_CONFIG_PATH, "r", encoding="utf-8") as fh: with open(_CONFIG_PATH, "r", encoding="utf-8") as fh:
return _yaml.load(fh) return _yaml.load(fh)
@@ -282,6 +304,9 @@ def _load_config() -> Any:
# Write the modified config back to disk preserving comments and formatting. # Write the modified config back to disk preserving comments and formatting.
# Записываем изменённый конфиг обратно на диск, сохраняя комментарии и форматирование. # Записываем изменённый конфиг обратно на диск, сохраняя комментарии и форматирование.
def _save_config(data: Any) -> None: def _save_config(data: Any) -> None:
"""Write the modified YAML data back to cobot-setting.yaml, preserving comments.
Записывает изменённые данные YAML обратно в cobot-setting.yaml, сохраняя комментарии.
"""
with open(_CONFIG_PATH, "w", encoding="utf-8") as fh: with open(_CONFIG_PATH, "w", encoding="utf-8") as fh:
_yaml.dump(data, fh) _yaml.dump(data, fh)
+47
View File
@@ -58,6 +58,9 @@ _WEBOTS_IMAGES = [
# Минимальное приложение, которое задаёт один вопрос и сразу выходит с выбранным значением. # Минимальное приложение, которое задаёт один вопрос и сразу выходит с выбранным значением.
# Нам нужен полноценный App, потому что экраны Textual не могут работать вне него. # Нам нужен полноценный App, потому что экраны Textual не могут работать вне него.
class _Ask(App[Optional[str]]): class _Ask(App[Optional[str]]):
"""Minimal one-question Textual app. Pushes a PickScreen and exits with the chosen value.
Минимальное однвопросное Textual-приложение. Открывает PickScreen и завершается с выбранным значением.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def __init__(self, step: str, question: str, options: List[str], default: str): def __init__(self, step: str, question: str, options: List[str], default: str):
@@ -75,6 +78,9 @@ class _Ask(App[Optional[str]]):
def _ask(step: str, question: str, options: List[str], default: str) -> Optional[str]: def _ask(step: str, question: str, options: List[str], default: str) -> Optional[str]:
"""Show a single-choice PickScreen and return the selected value, or None on Escape.
Показывает PickScreen с одним выбором и возвращает выбранное значение или None при Escape.
"""
# Returns None when the user pressed Escape to cancel. # Returns None when the user pressed Escape to cancel.
# Возвращает None когда пользователь нажал Escape для отмены. # Возвращает None когда пользователь нажал Escape для отмены.
return _Ask(step, question, options, default).run() return _Ask(step, question, options, default).run()
@@ -83,6 +89,9 @@ def _ask(step: str, question: str, options: List[str], default: str) -> Optional
# Detect the GPU type so we can pass the right flags to docker run for Webots rendering. # Detect the GPU type so we can pass the right flags to docker run for Webots rendering.
# Определяем тип GPU, чтобы передать нужные флаги в docker run для рендеринга Webots. # Определяем тип GPU, чтобы передать нужные флаги в docker run для рендеринга Webots.
def _detect_gpu() -> str: def _detect_gpu() -> str:
"""Return "nvidia", "mesa", or "software" based on what GPU drivers are available.
Возвращает "nvidia", "mesa" или "software" в зависимости от доступных драйверов GPU.
"""
if shutil.which("nvidia-smi"): if shutil.which("nvidia-smi"):
if subprocess.run(["nvidia-smi"], capture_output=True).returncode == 0: if subprocess.run(["nvidia-smi"], capture_output=True).returncode == 0:
return "nvidia" return "nvidia"
@@ -94,6 +103,9 @@ def _detect_gpu() -> str:
# List all Docker images currently available on this machine. # List all Docker images currently available on this machine.
# Получаем список всех Docker-образов доступных на этой машине. # Получаем список всех Docker-образов доступных на этой машине.
def _docker_images() -> set: def _docker_images() -> set:
"""Return the set of "repository:tag" strings for all locally available Docker images.
Возвращает множество строк "репозиторий:тег" для всех локально доступных Docker-образов.
"""
r = subprocess.run( r = subprocess.run(
["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"], ["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"],
capture_output=True, text=True, capture_output=True, text=True,
@@ -104,6 +116,9 @@ def _docker_images() -> set:
# Return the first image from the candidates list that is already present locally. # Return the first image from the candidates list that is already present locally.
# Возвращаем первый образ из списка кандидатов, который уже присутствует локально. # Возвращаем первый образ из списка кандидатов, который уже присутствует локально.
def _find_image(candidates: List[str]) -> Optional[str]: def _find_image(candidates: List[str]) -> Optional[str]:
"""Return the first candidate image that exists locally, or None if none are available.
Возвращает первый образ-кандидат, присутствующий локально, или None если ни один не найден.
"""
available = _docker_images() available = _docker_images()
for img in candidates: for img in candidates:
if img in available: if img in available:
@@ -116,6 +131,11 @@ def _find_image(candidates: List[str]) -> Optional[str]:
# Собираем ROS2-проект локально с помощью colcon. Используется при запуске в локальном режиме, # Собираем ROS2-проект локально с помощью colcon. Используется при запуске в локальном режиме,
# если директория install/ ещё не существует. # если директория install/ ещё не существует.
def _task_build(screen: LogScreen) -> None: def _task_build(screen: LogScreen) -> None:
"""Worker function that runs inside LogScreen. Counts packages, then runs colcon build
with release mixin and updates progress as each package finishes.
Рабочая функция внутри LogScreen. Подсчитывает пакеты, запускает colcon build с mixin release
и обновляет прогресс по мере завершения каждого пакета.
"""
try: try:
screen.write("[bold]Building project with colcon[/bold]\n") screen.write("[bold]Building project with colcon[/bold]\n")
@@ -160,6 +180,10 @@ def _task_build(screen: LogScreen) -> None:
class _BuildApp(App[bool]): class _BuildApp(App[bool]):
"""Minimal app that opens a LogScreen running _task_build and exits with the build result.
Минимальное приложение, открывающее LogScreen с _task_build и завершающееся с результатом сборки.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def on_mount(self) -> None: def on_mount(self) -> None:
@@ -174,6 +198,11 @@ class _BuildApp(App[bool]):
# Запускаем launch-файл ROS2 напрямую на этой машине без Docker. # Запускаем launch-файл ROS2 напрямую на этой машине без Docker.
# Используем start_new_session, чтобы можно было убить всю группу процессов одним сигналом. # Используем start_new_session, чтобы можно было убить всю группу процессов одним сигналом.
def _task_run_local(screen: RunScreen, mode: str) -> None: def _task_run_local(screen: RunScreen, mode: str) -> None:
"""Worker function that runs inside RunScreen. Launches iiwa.launch.py locally by sourcing
ROS2 and install/setup.bash, then streams output until the process exits or is stopped.
Рабочая функция внутри RunScreen. Запускает iiwa.launch.py локально через source ROS2 и
install/setup.bash, затем транслирует вывод до завершения процесса или его остановки.
"""
config = str(_CONFIG_PATH) config = str(_CONFIG_PATH)
ros_cmd = f"ros2 launch iiwa_bringup iiwa.launch.py setting:={config}" ros_cmd = f"ros2 launch iiwa_bringup iiwa.launch.py setting:={config}"
if mode == "webots": if mode == "webots":
@@ -223,6 +252,11 @@ def _task_run_local(screen: RunScreen, mode: str) -> None:
# Запускаем launch-файл ROS2 внутри Docker-контейнера. # Запускаем launch-файл ROS2 внутри Docker-контейнера.
# Для режима Webots также пробрасываем X11 и доступ к GPU, чтобы окно симулятора появилось на экране. # Для режима Webots также пробрасываем X11 и доступ к GPU, чтобы окно симулятора появилось на экране.
def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None: def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None:
"""Worker function that runs inside RunScreen. Builds the docker run command with the
appropriate GPU/X11 flags for Webots, then streams container output until stopped or exited.
Рабочая функция внутри RunScreen. Формирует команду docker run с нужными флагами GPU/X11
для Webots, затем транслирует вывод контейнера до остановки или завершения.
"""
container = _CONTAINER_WEBOTS if mode == "webots" else _CONTAINER_CONTROLLER container = _CONTAINER_WEBOTS if mode == "webots" else _CONTAINER_CONTROLLER
ros_cmd = ( ros_cmd = (
@@ -317,6 +351,9 @@ def _task_run_docker(screen: RunScreen, image: str, mode: str, gpu: str) -> None
# Wraps a RunScreen in an App so it can be launched with .run(). # Wraps a RunScreen in an App so it can be launched with .run().
# Оборачивает RunScreen в App, чтобы его можно было запустить через .run(). # Оборачивает RunScreen в App, чтобы его можно было запустить через .run().
class _RunApp(App[None]): class _RunApp(App[None]):
"""Minimal app that wraps a RunScreen so it can be started with .run().
Минимальное приложение, оборачивающее RunScreen чтобы его можно было запустить через .run().
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def __init__(self, title: str, task: Callable): def __init__(self, title: str, task: Callable):
@@ -333,6 +370,11 @@ class _RunApp(App[None]):
# Ведёт пользователя через локальный запуск - спрашивает что запустить, проверяет # Ведёт пользователя через локальный запуск - спрашивает что запустить, проверяет
# предварительные условия, устанавливает Webots и собирает проект при необходимости, затем запускает. # предварительные условия, устанавливает Webots и собирает проект при необходимости, затем запускает.
def _local_flow(args: argparse.Namespace) -> None: def _local_flow(args: argparse.Namespace) -> None:
"""Interactive flow for local (non-Docker) launch. Checks Webots, ROS2, and build state,
offers to install/build missing pieces, then starts RunScreen.
Интерактивный сценарий для локального (не Docker) запуска. Проверяет Webots, ROS2 и состояние
сборки, предлагает установить/собрать недостающее, затем запускает RunScreen.
"""
mode_v = _ask( mode_v = _ask(
"Run local", "Run local",
"What do you want to launch?", "What do you want to launch?",
@@ -393,6 +435,11 @@ def _local_flow(args: argparse.Namespace) -> None:
# Ведёт пользователя через запуск в Docker - спрашивает что запустить, ищет подходящий образ, # Ведёт пользователя через запуск в Docker - спрашивает что запустить, ищет подходящий образ,
# определяет GPU для Webots и запускает. # определяет GPU для Webots и запускает.
def _docker_flow(args: argparse.Namespace) -> None: def _docker_flow(args: argparse.Namespace) -> None:
"""Interactive flow for Docker launch. Finds the best available image, detects the GPU
for Webots mode, then starts RunScreen with the docker run task.
Интерактивный сценарий для запуска в Docker. Находит лучший доступный образ, определяет GPU
для режима Webots, затем запускает RunScreen с задачей docker run.
"""
if not shutil.which("docker"): if not shutil.which("docker"):
from rich.console import Console from rich.console import Console
Console().print("[red]Error:[/red] Docker is not installed or not on PATH.") Console().print("[red]Error:[/red] Docker is not installed or not on PATH.")
+12
View File
@@ -17,6 +17,9 @@ from cobot.tui import SCREEN_CSS, PickScreen
# Минимальное Textual-приложение, которое задаёт один вопрос и выходит с выбранным значением. # Минимальное Textual-приложение, которое задаёт один вопрос и выходит с выбранным значением.
# Нам это нужно, потому что экраны Textual не могут работать вне контекста приложения. # Нам это нужно, потому что экраны Textual не могут работать вне контекста приложения.
class _Ask(App[Optional[str]]): class _Ask(App[Optional[str]]):
"""Minimal one-question Textual app. Pushes a PickScreen and exits with the chosen value.
Минимальное однвопросное Textual-приложение. Открывает PickScreen и завершается с выбранным значением.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def __init__(self, step: str, question: str, options: List[str], default: str): def __init__(self, step: str, question: str, options: List[str], default: str):
@@ -34,17 +37,26 @@ class _Ask(App[Optional[str]]):
def _ask(step: str, question: str, options: List[str], default: str) -> Optional[str]: def _ask(step: str, question: str, options: List[str], default: str) -> Optional[str]:
"""Show a PickScreen and return the selected value, or None if the user pressed Escape.
Показывает PickScreen и возвращает выбранное значение или None если пользователь нажал Escape.
"""
# Returns None if the user pressed Escape to cancel the whole wizard. # Returns None if the user pressed Escape to cancel the whole wizard.
# Возвращает None если пользователь нажал Escape для отмены всего мастера. # Возвращает None если пользователь нажал Escape для отмены всего мастера.
return _Ask(step, question, options, default).run() return _Ask(step, question, options, default).run()
def register(subparsers): def register(subparsers):
"""Register the "setup" subparser.
Регистрирует подпарсер "setup".
"""
p = subparsers.add_parser("setup", help="First-time project setup") p = subparsers.add_parser("setup", help="First-time project setup")
p.set_defaults(func=run) p.set_defaults(func=run)
def run(args: argparse.Namespace) -> None: def run(args: argparse.Namespace) -> None:
"""Run the three-step first-time setup wizard: doc server -> build env -> robot config.
Запускает трёхшаговый мастер первоначальной настройки: сервер документации -> среда сборки -> конфиг.
"""
# Step 1 - documentation server. # Step 1 - documentation server.
# Шаг 1 - сервер документации. # Шаг 1 - сервер документации.
v = _ask("Step 1 of 3", "Set up the documentation server?", ["Yes", "No"], "Yes") v = _ask("Step 1 of 3", "Set up the documentation server?", ["Yes", "No"], "Yes")
+9
View File
@@ -16,6 +16,11 @@ _PROJECT_DIR = Path(__file__).parent.parent.parent
# Скачиваем последние коммиты с удалённого репозитория и переустанавливаем cobot CLI за один раз. # Скачиваем последние коммиты с удалённого репозитория и переустанавливаем cobot CLI за один раз.
# Прогресс-бар: fetch (0-30%), pull (30-80%), переустановка (80-100%). # Прогресс-бар: fetch (0-30%), pull (30-80%), переустановка (80-100%).
def _task_update(screen: LogScreen) -> None: def _task_update(screen: LogScreen) -> None:
"""Worker function that runs inside LogScreen. Fetches the current branch, shows incoming
commits, pulls changes, then reinstalls the cobot CLI via uv tool install --editable.
Рабочая функция, выполняемая внутри LogScreen. Получает текущую ветку, показывает входящие
коммиты, вытягивает изменения, затем переустанавливает cobot CLI через uv tool install --editable.
"""
try: try:
# Find out which branch we are on so we can fetch and pull the right one. # Find out which branch we are on so we can fetch and pull the right one.
# Определяем на какой ветке мы находимся, чтобы делать fetch и pull нужной ветки. # Определяем на какой ветке мы находимся, чтобы делать fetch и pull нужной ветки.
@@ -110,6 +115,10 @@ def _task_update(screen: LogScreen) -> None:
class _UpdateApp(App[None]): class _UpdateApp(App[None]):
"""Minimal Textual app that opens a LogScreen running _task_update and exits when it closes.
Минимальное Textual-приложение, открывающее LogScreen с _task_update и завершающееся при закрытии.
"""
CSS = SCREEN_CSS CSS = SCREEN_CSS
def on_mount(self) -> None: def on_mount(self) -> None:
+16
View File
@@ -84,6 +84,9 @@ RunScreen #hint {
# Экран с вопросом и списком вариантов в виде радио-кнопок. # Экран с вопросом и списком вариантов в виде радио-кнопок.
# Пользователь выбирает один и нажимает Enter - выбранная строка возвращается как результат. # Пользователь выбирает один и нажимает Enter - выбранная строка возвращается как результат.
class PickScreen(Screen[Optional[str]]): class PickScreen(Screen[Optional[str]]):
"""Single-choice radio button screen. Returns the selected option string, or None on Escape.
Экран выбора одного варианта с радио-кнопками. Возвращает выбранную строку или None при Escape.
"""
BINDINGS = [ BINDINGS = [
Binding("enter", "submit", "Confirm", priority=True), Binding("enter", "submit", "Confirm", priority=True),
Binding("escape", "abort", "Cancel"), Binding("escape", "abort", "Cancel"),
@@ -133,6 +136,9 @@ class PickScreen(Screen[Optional[str]]):
# Экран с вопросом и полем для ввода произвольного текста. # Экран с вопросом и полем для ввода произвольного текста.
# Пользователь вводит значение, нажимает Enter, и текст возвращается как результат. # Пользователь вводит значение, нажимает Enter, и текст возвращается как результат.
class InputScreen(Screen[Optional[str]]): class InputScreen(Screen[Optional[str]]):
"""Free-text input screen. Returns the trimmed value on Enter, or None on Escape.
Экран свободного ввода текста. Возвращает обрезанное значение при Enter или None при Escape.
"""
BINDINGS = [ BINDINGS = [
Binding("enter", "submit", "Confirm", priority=True), Binding("enter", "submit", "Confirm", priority=True),
Binding("escape", "abort", "Cancel"), Binding("escape", "abort", "Cancel"),
@@ -175,6 +181,11 @@ class InputScreen(Screen[Optional[str]]):
# Используется для долгих операций, таких как установка и сборка. # Используется для долгих операций, таких как установка и сборка.
# После завершения задачи закрывается по нажатию Enter или Escape. # После завершения задачи закрывается по нажатию Enter или Escape.
class LogScreen(Screen[bool]): class LogScreen(Screen[bool]):
"""Log screen for long-running background tasks. Shows a scrollable log and optional
progress bar. Returns True on success, False on failure after the task finishes.
Экран лога для долгих фоновых задач. Показывает прокручиваемый лог и опциональный
прогресс-бар. Возвращает True при успехе, False при ошибке после завершения задачи.
"""
BINDINGS = [Binding("enter,escape", "close", "Close", show=False)] BINDINGS = [Binding("enter,escape", "close", "Close", show=False)]
def __init__(self, title: str, task: Callable[[LogScreen], None], show_progress: bool = False): def __init__(self, title: str, task: Callable[[LogScreen], None], show_progress: bool = False):
@@ -275,6 +286,11 @@ class LogScreen(Screen[bool]):
# Экран для долго работающего процесса, который пользователь может остановить в любой момент. # Экран для долго работающего процесса, который пользователь может остановить в любой момент.
# Показывает живой лог и предлагает S / Enter / Escape для остановки или закрытия. # Показывает живой лог и предлагает S / Enter / Escape для остановки или закрытия.
class RunScreen(Screen[None]): class RunScreen(Screen[None]):
"""Run screen for a persistent process (e.g. ROS2 launch). Shows a live log and allows
the user to stop the process with S or close after it exits with Enter/Escape.
Экран запуска для постоянно работающего процесса (например ros2 launch). Показывает живой
лог и позволяет остановить процесс клавишей S или закрыть после завершения через Enter/Escape.
"""
BINDINGS = [ BINDINGS = [
Binding("s", "stop_close", "Stop", show=True, priority=True), Binding("s", "stop_close", "Stop", show=True, priority=True),
Binding("enter", "stop_close", "Close", show=False), Binding("enter", "stop_close", "Close", show=False),