Refactor subprocess handling in setup scripts for improved error reporting and output management

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,
+        capture_output=True, text=True, env=env,
-        text=True, env=env,
     )
-    for line in proc.stdout:
+    if result.returncode != 0:
-        s = line.rstrip()
+        for line in (result.stdout + result.stderr).splitlines():
-        if s:
+            if line.strip():
-            write(s)
+                write(line)
-    proc.wait()
-    if proc.returncode != 0:
         write("[red]Image build failed.[/red]")
         return False
     write("[green][ok][/green] Documentation image ready")

cobot/commands/docker_setup.py

@@ -44,18 +44,16 @@ class _Config:
     hub_repo: str
 
 
-def _stream(cmd: List[str], write: Write, env=None) -> bool:
+def _run_quiet(cmd: List[str], write: Write, env=None) -> bool:
-    proc = subprocess.Popen(
+    result = subprocess.run(
-        cmd,
+        cmd, capture_output=True, text=True,
-        stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+        env=env or os.environ,
-        text=True, env=env or os.environ,
     )
-    for line in proc.stdout:
+    if result.returncode != 0:
-        s = line.rstrip()
+        for line in (result.stdout + result.stderr).splitlines():
-        if s:
+            if line.strip():
-            write(s)
+                write(line)
-    proc.wait()
+    return result.returncode == 0
-    return proc.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:

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:
+def _run_quiet(cmd: List[str], write: Write | None = None, env: dict | None = None, cwd=None) -> None:
-    result = subprocess.run(cmd, capture_output=True)
+    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_quiet(["sudo", "apt-get", "update", "-qq"], write)
-    _run_logged(["sudo", "apt-get", "install", "-y", "--no-install-recommends", "locales"], write, _APT_ENV)
+    _run_quiet(["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", "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", "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,

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,
+            capture_output=True, text=True, cwd=_PROJECT_DIR,
-            text=True, cwd=_PROJECT_DIR,
         )
-        for line in proc.stdout:
+        if pull.returncode != 0:
-            s = line.rstrip()
+            for line in (pull.stdout + pull.stderr).splitlines():
-            if s:
+                if line.strip():
-                screen.write(s)
+                    screen.write(line)
-        proc.wait()
-        if proc.returncode != 0:
             screen.write("[red]Pull failed.[/red]")
             screen.finish(False)
             return

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