From 615a955767db67891c766646bb3a0a9906872952 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=94=D0=B0=D0=BD=D0=B8=D0=B8=D0=BB=20=D0=93=D1=80=D0=B0?= =?UTF-8?q?=D0=B1=D0=B0=D1=80=D1=8C?= Date: Wed, 20 May 2026 18:24:09 +0300 Subject: [PATCH] Refactor subprocess handling in setup scripts for improved error reporting and output management --- cobot/commands/doc_setup.py | 15 ++++++--------- cobot/commands/docker_setup.py | 24 +++++++++++------------- cobot/commands/local_setup.py | 33 ++++++++++++++++++++------------- cobot/commands/update.py | 15 ++++++--------- cobot/tui.py | 8 +++++++- 5 files changed, 50 insertions(+), 45 deletions(-) diff --git a/cobot/commands/doc_setup.py b/cobot/commands/doc_setup.py index 2f74368..201b117 100644 --- a/cobot/commands/doc_setup.py +++ b/cobot/commands/doc_setup.py @@ -38,17 +38,14 @@ def _image_exists() -> bool: def _build_docs_image(write: Write) -> bool: write("[cyan][*][/cyan] Building documentation image (runs once)...") env = {**os.environ, "DOCKER_BUILDKIT": "0"} - proc = subprocess.Popen( + result = subprocess.run( ["docker", "build", "-t", _IMAGE_NAME, str(_DOC_DIR)], - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, env=env, + capture_output=True, text=True, env=env, ) - for line in proc.stdout: - s = line.rstrip() - if s: - write(s) - proc.wait() - if proc.returncode != 0: + if result.returncode != 0: + for line in (result.stdout + result.stderr).splitlines(): + if line.strip(): + write(line) write("[red]Image build failed.[/red]") return False write("[green][ok][/green] Documentation image ready") diff --git a/cobot/commands/docker_setup.py b/cobot/commands/docker_setup.py index 72f9dbe..88454f4 100644 --- a/cobot/commands/docker_setup.py +++ b/cobot/commands/docker_setup.py @@ -44,18 +44,16 @@ class _Config: hub_repo: str -def _stream(cmd: List[str], write: Write, env=None) -> bool: - proc = subprocess.Popen( - cmd, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, env=env or os.environ, +def _run_quiet(cmd: List[str], write: Write, env=None) -> bool: + result = subprocess.run( + cmd, capture_output=True, text=True, + env=env or os.environ, ) - for line in proc.stdout: - s = line.rstrip() - if s: - write(s) - proc.wait() - return proc.returncode == 0 + if result.returncode != 0: + for line in (result.stdout + result.stderr).splitlines(): + if line.strip(): + write(line) + return result.returncode == 0 def _build_image( @@ -73,7 +71,7 @@ def _build_image( if parent_tag: cmd += ["--build-arg", f"IMAGE={parent_tag}"] cmd.append(str(ctx)) - ok = _stream(cmd, write, env) + ok = _run_quiet(cmd, write, env) if ok: write(f"[green][ok][/green] {name}") else: @@ -83,7 +81,7 @@ def _build_image( def _pull_image(name: str, tag: str, write: Write) -> bool: write(f"[cyan][*][/cyan] Pulling [bold]{name}[/bold] ({tag})...") - ok = _stream(["docker", "pull", tag], write) + ok = _run_quiet(["docker", "pull", tag], write) if ok: write(f"[green][ok][/green] {name}") else: diff --git a/cobot/commands/local_setup.py b/cobot/commands/local_setup.py index c765058..536bad3 100644 --- a/cobot/commands/local_setup.py +++ b/cobot/commands/local_setup.py @@ -41,10 +41,17 @@ def _detect_ros2_jazzy() -> bool: Write = Callable[[str], None] -def _run_quiet(cmd: List[str]) -> None: - result = subprocess.run(cmd, capture_output=True) +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: - raise RuntimeError(f"Exit {result.returncode}: {cmd[0]}") + if write: + for line in (result.stdout + result.stderr).splitlines(): + if line.strip(): + write(line) + raise RuntimeError(f"Command failed: {cmd[0]}") def _run_logged(cmd: List[str], write: Write, env: dict | None = None, cwd=None) -> None: @@ -62,7 +69,7 @@ def _run_logged(cmd: List[str], write: Write, env: dict | None = None, cwd=None) write(s) proc.wait() if proc.returncode != 0: - raise RuntimeError(f"Exit {proc.returncode}: {cmd[0]}") + raise RuntimeError(f"Command failed: {cmd[0]}") def _setup_locale(write: Write) -> None: @@ -71,10 +78,10 @@ def _setup_locale(write: Write) -> None: write("[green][ok][/green] UTF-8 locale active") return write("[cyan][*][/cyan] Configuring UTF-8 locale...") - _run_quiet(["sudo", "apt-get", "update", "-qq"]) - _run_logged(["sudo", "apt-get", "install", "-y", "--no-install-recommends", "locales"], write, _APT_ENV) - _run_logged(["sudo", "locale-gen", "en_US.UTF-8"], write) - _run_quiet(["sudo", "update-locale", "LC_ALL=en_US.UTF-8", "LANG=en_US.UTF-8"]) + _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") @@ -84,12 +91,12 @@ def _add_ros2_repo(write: Write) -> None: # Best-effort update before installing prereqs (ignore errors from broken repos) subprocess.run(["sudo", "apt-get", "update", "-qq"], capture_output=True) - _run_logged( + _run_quiet( ["sudo", "apt-get", "install", "-y", "--no-install-recommends", "software-properties-common", "curl", "gnupg"], write, _APT_ENV, ) - _run_quiet(["sudo", "add-apt-repository", "-y", "universe"]) + _run_quiet(["sudo", "add-apt-repository", "-y", "universe"], write) # Always re-download and dearmor the key to fix any previous bad install write("[cyan][*][/cyan] Downloading ROS2 signing key...") @@ -118,13 +125,13 @@ def _add_ros2_repo(write: Write) -> None: raise RuntimeError(f"Failed to write {_ROS_SOURCES}") write("[cyan][*][/cyan] Updating apt cache...") - _run_logged(["sudo", "apt-get", "update", "-q"], write, _APT_ENV) + _run_quiet(["sudo", "apt-get", "update", "-q"], write, _APT_ENV) write("[green][ok][/green] ROS2 repository ready") def _install_ros2_jazzy(write: Write) -> None: write("[cyan][*][/cyan] Installing ros-jazzy-desktop and ros-dev-tools...") - _run_logged( + _run_quiet( ["sudo", "apt-get", "install", "-y", "ros-jazzy-desktop", "ros-dev-tools"], write, _APT_ENV, @@ -137,7 +144,7 @@ def _install_colcon(write: Write) -> None: write("[green][ok][/green] colcon already available") return write("[cyan][*][/cyan] Installing colcon...") - _run_logged( + _run_quiet( ["sudo", "apt-get", "install", "-y", "--no-install-recommends", "python3-colcon-common-extensions"], write, _APT_ENV, diff --git a/cobot/commands/update.py b/cobot/commands/update.py index cecc728..c0fee25 100644 --- a/cobot/commands/update.py +++ b/cobot/commands/update.py @@ -53,17 +53,14 @@ def _task_update(screen: LogScreen) -> None: # Pull screen.write("\n[cyan][*][/cyan] Pulling changes...") - proc = subprocess.Popen( + pull = subprocess.run( ["git", "pull", "origin", branch], - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, cwd=_PROJECT_DIR, + capture_output=True, text=True, cwd=_PROJECT_DIR, ) - for line in proc.stdout: - s = line.rstrip() - if s: - screen.write(s) - proc.wait() - if proc.returncode != 0: + if pull.returncode != 0: + for line in (pull.stdout + pull.stderr).splitlines(): + if line.strip(): + screen.write(line) screen.write("[red]Pull failed.[/red]") screen.finish(False) return diff --git a/cobot/tui.py b/cobot/tui.py index 79977c5..e5c43a3 100644 --- a/cobot/tui.py +++ b/cobot/tui.py @@ -6,7 +6,7 @@ from textual import on from textual.app import ComposeResult from textual.binding import Binding from textual.screen import Screen -from textual.widgets import Footer, Input, RadioButton, RadioSet, RichLog, Static +from textual.widgets import Footer, Input, LoadingIndicator, RadioButton, RadioSet, RichLog, Static SCREEN_CSS = """ Screen { @@ -42,6 +42,10 @@ LogScreen #log { padding: 0 1; margin-top: 1; } +LogScreen #loading { + height: 1; + margin-top: 1; +} LogScreen #hint { margin-top: 1; color: $text; @@ -140,6 +144,7 @@ class LogScreen(Screen[bool]): def compose(self) -> ComposeResult: yield Static(self._title, id="step") yield RichLog(id="log", highlight=True, markup=True, wrap=True) + yield LoadingIndicator(id="loading") yield Static("", id="hint") yield Footer() @@ -160,6 +165,7 @@ class LogScreen(Screen[bool]): def _do_finish(self, success: bool) -> None: self._finished = True + self.query_one("#loading", LoadingIndicator).display = False msg = ( "[green]Done![/green] Press [bold]Enter[/bold] to close." if success