feat(cockpit): Video Panel

dimos/agents/skills/person_follow.py

@@ -19,7 +19,6 @@
 
 import numpy as np
 from reactivex.disposable import Disposable
-from turbojpeg import TurboJPEG
 
 from dimos.agents.annotation import skill
 from dimos.agents.capabilities import CAP_MOVEMENT
@@ -40,6 +39,7 @@
 from dimos.navigation.visual_servoing.detection_navigation import DetectionNavigation
 from dimos.navigation.visual_servoing.visual_servoing_2d import VisualServoing2D
 from dimos.utils.logging_config import setup_logger
+from dimos.utils.turbojpeg import get_turbojpeg
 
 logger = setup_logger()
 
@@ -331,5 +331,5 @@ def _send_stop_reason(self, query: str, reason: str) -> None:
 
 
 def _decode_base64_image(b64: str) -> Image:
-    bgr_array = TurboJPEG().decode(base64.b64decode(b64))
+    bgr_array = get_turbojpeg().decode(base64.b64decode(b64))
     return Image(data=bgr_array, format=ImageFormat.BGR)

dimos/e2e_tests/test_cockpit_browser.py

@@ -12,15 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""Cockpit browser e2e (the T3 acceptance demo, in CI).
+"""Cockpit browser e2e (the T3 + T5 acceptance demos, in CI).
 
 Starts `dimos --robot-ip fake run unitree-go2-basic --local-relay` (the go2
 blueprint on the go2_short replay dataset: real recorded odom + camera, no
 MuJoCo and no hardware) and drives the served Cockpit with a real headless
-browser: live odom must tick, the session must stay connected (Firefox's
-tight QUIC stream credit found the relay's stream-per-frame overflow the
-first time), and killing + restarting the dimos process must take the page
-through reconnecting and back to live data without a reload.
+browser: live odom must tick, the video panel must decode real camera frames
+at rate, the session must stay connected (Firefox's tight QUIC stream credit
+found the relay's stream-per-frame overflow the first time; T5's video load
+is exactly the traffic that pressures it), and killing + restarting the
+dimos process must take the page through reconnecting and back to live data
+without a reload.
 
 Runs in both Chromium and Firefox: their WebTransport implementations differ
 enough that one browser staying green says little about the other.
@@ -111,6 +113,38 @@ def _assert_odom_ticks(page: Page) -> None:
     expect(page.get_by_test_id("ch-odom-value")).to_contain_text("yaw")
 
 
+def _assert_video_plays(page: Page) -> None:
+    # The manifest-driven video panel is live: the badge reporting an fps
+    # proves frames arrive, the canvas leaving its 300 px HTML default width
+    # proves one frame decoded, and two successive pixel changes prove frames
+    # keep drawing (the replay dataset has genuinely varying content).
+    expect(page.get_by_test_id("video-color_image-badge")).to_contain_text("fps", timeout=60_000)
+    page.wait_for_function(
+        """() => {
+          const canvas = document.querySelector('[data-testid="video-color_image-canvas"]');
+          return canvas !== null && canvas.width !== 300;
+        }""",
+        timeout=30_000,
+    )
+
+    def wait_pixels_change() -> None:
+        baseline = page.evaluate(
+            """() =>
+              document.querySelector('[data-testid="video-color_image-canvas"]').toDataURL()"""
+        )
+        page.wait_for_function(
+            """(baseline) => {
+              const canvas = document.querySelector('[data-testid="video-color_image-canvas"]');
+              return canvas.toDataURL() !== baseline;
+            }""",
+            arg=baseline,
+            timeout=30_000,
+        )
+
+    wait_pixels_change()
+    wait_pixels_change()
+
+
 def test_cockpit_live_data_and_reconnect(
     start_go2_replay: Callable[[], DimosCliCall], page: Page
 ) -> None:
@@ -122,6 +156,7 @@ def test_cockpit_live_data_and_reconnect(
     expect(status).to_have_attribute("data-phase", "connected", timeout=120_000)
     expect(page.get_by_test_id("robot")).not_to_have_text("no robot", timeout=30_000)
     _assert_odom_ticks(page)
+    _assert_video_plays(page)
 
     # Stability: record every phase change and require none for the window.
     page.evaluate("""() => {
@@ -134,6 +169,7 @@ def test_cockpit_live_data_and_reconnect(
     flaps = page.evaluate("() => window.__phases")
     assert flaps == [], f"session flapped during the stability window: {flaps}"
     _assert_odom_ticks(page)
+    _assert_video_plays(page)
 
     # Kill dimos (relay dies with it): the page must notice on its own.
     call.stop()
@@ -144,3 +180,4 @@ def test_cockpit_live_data_and_reconnect(
     _wait_for_relay(restarted, START_TIMEOUT_S)
     expect(status).to_have_attribute("data-phase", "connected", timeout=120_000)
     _assert_odom_ticks(page)
+    _assert_video_plays(page)

dimos/msgs/sensor_msgs/Image.py

@@ -26,9 +26,11 @@
 import numpy as np
 import reactivex as rx
 from reactivex import operators as ops
+from turbojpeg import TJPF_RGB
 
 from dimos.types.timestamped import Timestamped, TimestampedBufferCollection, to_human_readable
 from dimos.utils.reactive import quality_barrier
+from dimos.utils.turbojpeg import get_turbojpeg
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -542,12 +544,9 @@ def to_jpeg_bytes(self, quality: int = 75) -> bytes:
         Returns:
             Raw JPEG bytes.
         """
-        from turbojpeg import TJPF_RGB, TurboJPEG
-
-        jpeg = TurboJPEG()
         # Canonicalize to RGB so JPEG bytes are deterministic regardless of input format.
         rgb_array = self.to_rgb().data
-        return jpeg.encode(rgb_array, quality=quality, pixel_format=TJPF_RGB)  # type: ignore[no-any-return]
+        return get_turbojpeg().encode(rgb_array, quality=quality, pixel_format=TJPF_RGB)  # type: ignore[no-any-return]
 
     def lcm_jpeg_encode(self, quality: int = 75, frame_id: str | None = None) -> bytes:
         """Convert to LCM Image message with JPEG-compressed data.
@@ -599,15 +598,12 @@ def lcm_jpeg_decode(cls, data: bytes, **kwargs: Any) -> Image:
         Returns:
             Image instance
         """
-        from turbojpeg import TJPF_RGB, TurboJPEG
-
-        jpeg = TurboJPEG()
         msg = LCMImage.lcm_decode(data)
 
         if msg.encoding != "jpeg":
             raise ValueError(f"Expected JPEG encoding, got {msg.encoding}")
 
-        rgb_array = jpeg.decode(msg.data, pixel_format=TJPF_RGB)
+        rgb_array = get_turbojpeg().decode(msg.data, pixel_format=TJPF_RGB)
 
         return cls(
             data=rgb_array,

dimos/protocol/pubsub/impl/jpeg_shm.py

@@ -14,28 +14,26 @@
 
 from typing import Any
 
-from turbojpeg import TurboJPEG
-
 from dimos.msgs.sensor_msgs.Image import Image, ImageFormat
 from dimos.protocol.pubsub.encoders import PubSubEncoderMixin
 from dimos.protocol.pubsub.impl.shmpubsub import SharedMemoryPubSubBase
+from dimos.utils.turbojpeg import get_turbojpeg
 
 
 class JpegSharedMemoryEncoderMixin(PubSubEncoderMixin[str, Image, bytes]):
     def __init__(self, quality: int = 75, **kwargs) -> None:  # type: ignore[no-untyped-def]
         super().__init__(**kwargs)
-        self.jpeg = TurboJPEG()
         self.quality = quality
 
     def encode(self, msg: Any, _topic: str) -> bytes:
         if not isinstance(msg, Image):
             raise ValueError("Can only encode images.")
 
         bgr_image = msg.to_bgr().to_opencv()
-        return self.jpeg.encode(bgr_image, quality=self.quality)  # type: ignore[no-any-return]
+        return get_turbojpeg().encode(bgr_image, quality=self.quality)  # type: ignore[no-any-return]
 
     def decode(self, msg: bytes, _topic: str) -> Image:
-        bgr_array = self.jpeg.decode(msg)
+        bgr_array = get_turbojpeg().decode(msg)
         return Image(data=bgr_array, format=ImageFormat.BGR)
 
 

dimos/robot/unitree/mujoco_connection.py

@@ -92,6 +92,7 @@ def __init__(self, global_config: GlobalConfig) -> None:
         self._stop_timer: threading.Timer | None = None
 
         self._stream_threads: list[threading.Thread] = []
+        self._output_thread: threading.Thread | None = None
         self._stop_events: list[threading.Event] = []
         self._is_cleaned_up = False
 
@@ -126,14 +127,26 @@ def start(self) -> None:
 
             self.process = subprocess.Popen(
                 [executable, str(LAUNCHER_PATH), config_pickle, shm_names_json],
-                stderr=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
                 env=env,
             )
 
         except Exception as e:
             self.shm_data.cleanup()
             raise RuntimeError(f"Failed to start MuJoCo subprocess: {e}") from e
 
+        # A captured pipe must always be drained. MuJoCo can emit a sustained
+        # stream of physics warnings; once an unread OS pipe fills, the child
+        # blocks in write(2) and silently stops updating every shared-memory
+        # sensor while this parent (and the relay) remain healthy.
+        self._output_thread = threading.Thread(
+            target=self._pump_subprocess_output,
+            name="mujoco-output-pump",
+            daemon=True,
+        )
+        self._output_thread.start()
+
         # Wait for process to be ready
         ready_timeout = 300.0
         start_time = time.time()
@@ -164,19 +177,37 @@ def cleanup_on_exit(
         self.stop()
         raise RuntimeError("MuJoCo process failed to start (timeout)")
 
+    def _pump_subprocess_output(self) -> None:
+        """Drain MuJoCo output into DimOS logs so the child cannot pipe-block."""
+        process = self.process
+        if process is None or process.stdout is None:
+            return
+        try:
+            for raw in process.stdout:
+                line = raw.decode(errors="replace").rstrip()
+                if not line:
+                    continue
+                lowered = line.lower()
+                if "warning" in lowered or "error" in lowered or "fatal" in lowered:
+                    logger.warning(f"[mujoco] {line}")
+                else:
+                    logger.info(f"[mujoco] {line}")
+
+            return_code = process.poll()
+            if return_code not in (None, 0) and not self._is_cleaned_up:
+                logger.error(f"[mujoco] subprocess exited unexpectedly with code {return_code}")
+        except (OSError, ValueError) as e:
+            # Teardown can race the final descriptor close; only an unexpected
+            # read failure while live is actionable.
+            if not self._is_cleaned_up:
+                logger.warning(f"MuJoCo output pump terminated: {e}")
+
     def stop(self) -> None:
         if self._is_cleaned_up:
             return
 
         self._is_cleaned_up = True
 
-        # clean up open file descriptors
-        if self.process:
-            if self.process.stderr:
-                self.process.stderr.close()
-            if self.process.stdout:
-                self.process.stdout.close()
-
         # Cancel any pending timers
         if self._stop_timer:
             self._stop_timer.cancel()
@@ -197,7 +228,9 @@ def stop(self) -> None:
         if self.shm_data:
             self.shm_data.signal_stop()
 
-        # Wait for process to finish
+        # Stop the child before joining/closing its output. BufferedReader.close()
+        # can wait on a pump thread blocked in read(), so closing the pipe while
+        # the child is still alive can deadlock shutdown.
         if self.process:
             try:
                 self.process.terminate()
@@ -210,7 +243,22 @@ def stop(self) -> None:
             except Exception as e:
                 logger.error(f"Error stopping MuJoCo process: {e}")
 
+            if self._output_thread is not None and self._output_thread.is_alive():
+                self._output_thread.join(timeout=DEFAULT_THREAD_JOIN_TIMEOUT)
+                if self._output_thread.is_alive():
+                    logger.warning("MuJoCo output thread did not stop gracefully")
+
+            # The exited child has closed its pipe end, so these closes cannot
+            # contend with a reader indefinitely.
+            if self.process.stderr:
+                self.process.stderr.close()
+            if self.process.stdout and not (
+                self._output_thread is not None and self._output_thread.is_alive()
+            ):
+                self.process.stdout.close()
+
             self.process = None
+        self._output_thread = None
 
         # Clean up shared memory
         if self.shm_data: