Benchmark replay

.github/workflows/codspeed.yml

@@ -0,0 +1,132 @@
+name: codspeed
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
+  cancel-in-progress: true
+
+permissions: {}
+
+env:
+  UV_NO_SYNC: "1"
+  # Walltime benchmarks die without a traceback otherwise.
+  PYTHONFAULTHANDLER: "1"
+
+jobs:
+  compute-ros-pin:
+    # Same as ci.yml.
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read  # For checkout
+    outputs:
+      digest: ${{ steps.read.outputs.digest }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v7
+      - id: read
+        run: |
+          DIGEST=$(grep -oE 'sha256:[a-f0-9]{64}' docker/ros-dev-pin/Dockerfile)
+          if [ -z "$DIGEST" ]; then
+            echo "::error::No sha256 digest found in docker/ros-dev-pin/Dockerfile"
+            exit 1
+          fi
+          echo "digest=$DIGEST" >> "$GITHUB_OUTPUT"
+
+  benchmarks:
+    # Same-repo PRs and pushes only: fork PRs must not run on the self-hosted
+    # runner (mirrors the self-hosted-tests gate in ci.yml).
+    if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name == github.repository
+    needs: compute-ros-pin
+    timeout-minutes: 30
+    # codspeed-macro doesn't support ARMv8.2+ needed for ML dependencies.
+    runs-on:
+      - self-hosted
+      - Linux
+      - base
+    container:
+      image: ghcr.io/dimensionalos/ros-dev@${{ needs.compute-ros-pin.outputs.digest }}
+      options: --memory=6g --memory-swap=6g --cap-add PERFMON
+      volumes:
+        - /var/cache/dimos-nix:/nix
+        - /var/cache/dimos-root-cache:/root/.cache
+    permissions:
+      contents: read
+      id-token: write  # OIDC upload to CodSpeed (tokenless for public repos)
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v7
+      - name: Fetch the replay dataset from LFS if not cached
+        run: |
+          git config --global --add safe.directory "$GITHUB_WORKSPACE"
+          if git lfs pointer --check --file data/.lfs/go2_hongkong_office.db.tar.gz; then
+            git lfs pull --include="data/.lfs/go2_hongkong_office.db.tar.gz" --exclude=""
+          else
+            echo "archive already smudged"
+          fi
+      - name: Install system dependencies
+        run: |
+          apt-get update
+          apt-get install -y libturbojpeg
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.3.2
+        with:
+          # /root/.cache is a persisted volume
+          enable-cache: false
+      # uv-managed rather than the container's python for sqlite (>=3.45).
+      # TODO: Can be dropped after upgrading to 26.04.
+      - name: Setup Python
+        run: uv python install 3.12
+      - name: Install dependencies
+        run: uv sync --group tests --frozen
+      - name: Pre-extract the replay database
+        run: uv run python -c "from dimos.utils.data import get_data; print(get_data('go2_hongkong_office.db'))"
+      - name: Run benchmarks
+        uses: CodSpeedHQ/action@v5
+        env:
+          # In this rootful container the runner would otherwise autodetect
+          # elevation and wrap the benchmark in systemd-run, which does not
+          # exist here. No cpuset hooks either, so run unisolated — the job
+          # already has the machine to itself.
+          CODSPEED_ISOLATION: "false"
+        with:
+          mode: walltime
+          run: uv run pytest --codspeed -m '' --no-cov dimos/robot/unitree/go2/test_replay_benchmark.py
+
+  simulation:
+    # Single-process instruction-count benchmark (Valgrind) of the replay
+    # compute pipeline; runs on hosted runners — no self-hosted, no workers.
+    if: github.event_name == 'push' || github.event.pull_request.head.repo.full_name == github.repository
+    timeout-minutes: 45
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      id-token: write  # OIDC upload to CodSpeed
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v7
+      - name: Fetch the replay dataset
+        run: git lfs pull --include="data/.lfs/go2_short.db.tar.gz" --exclude=""
+      # pyaudio build needs portaudio; the recording's JPEG frames need turbojpeg.
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y portaudio19-dev libturbojpeg
+      - name: Install uv
+        uses: astral-sh/setup-uv@v8.3.2
+        with:
+          enable-cache: true
+      - name: Setup Python
+        run: uv python install 3.12
+      - name: Install dependencies
+        run: uv sync --group tests --frozen
+      - name: Pre-extract the replay database
+        run: uv run python -c "from dimos.utils.data import get_data; print(get_data('go2_short.db'))"
+      - name: Run benchmarks
+        uses: CodSpeedHQ/action@v5
+        with:
+          mode: simulation
+          run: uv run pytest --codspeed -m '' --no-cov dimos/robot/unitree/go2/test_replay_pipeline_benchmark.py

dimos/core/global_config.py

@@ -55,6 +55,13 @@ class GlobalConfig(BaseSettings):
     simulation: str = ""
     replay: bool = False
     replay_db: str = "go2_short"
+    # Playback rate multiplier for --replay (1.0 = realtime; large values drain
+    # as fast as the pipeline allows since late frames are never delayed).
+    replay_speed: float = 1.0
+    # Seconds into the recording to start replay at (None = the beginning).
+    replay_seek: float | None = None
+    # Seconds of recording to play before the streams complete (None = all).
+    replay_duration: float | None = None
     new_memory: bool = False
     # Discover zenoh peers across the network.
     # Toggling off drops back to loopback-only discovery:

dimos/robot/unitree/go2/connection.py

@@ -143,8 +143,12 @@ def make_connection(
     connection_type = cfg.unitree_connection_type.lower()
 
     if ip in ("fake", "mock", "replay") or connection_type == "replay":
-        dataset = cfg.replay_db
-        return ReplayConnection(dataset=dataset)
+        return ReplayConnection(
+            dataset=cfg.replay_db,
+            speed=cfg.replay_speed,
+            seek=cfg.replay_seek,
+            duration=cfg.replay_duration,
+        )
     elif ip == "mujoco" or connection_type in ("mujoco", "true"):
         from dimos.robot.unitree.mujoco_connection import MujocoConnection
 
@@ -165,15 +169,20 @@ def make_connection(
 
 
 class ReplayConnection(UnitreeWebRTCConnection, CompositeResource):
-    def __init__(  # type: ignore[no-untyped-def]
+    def __init__(
         self,
         dataset: str = "go2_china_office",
-        **kwargs,
+        *,
+        speed: float = 1.0,
+        seek: float | None = None,
+        duration: float | None = None,
+        loop: bool = False,
     ) -> None:
         self.dataset = dataset
-        self._loop = kwargs.get("loop", False)
-        self._seek = kwargs.get("seek")
-        self._duration = kwargs.get("duration")
+        self._speed = speed
+        self._seek = seek
+        self._duration = duration
+        self._loop = loop
 
     @cached_property
     def replay(self) -> Replay:
@@ -183,7 +192,9 @@ def replay(self) -> Replay:
             SqliteStore(path=str(resolve_db_path(self.dataset)), must_exist=True)
         )
         store.start()
-        return store.replay(loop=self._loop, seek=self._seek, duration=self._duration)
+        return store.replay(
+            speed=self._speed, seek=self._seek, duration=self._duration, loop=self._loop
+        )
 
     def connect(self) -> None:
         pass

dimos/robot/unitree/go2/test_connection.py

@@ -50,6 +50,30 @@ def test_make_connection_webrtc_forwards_aes_128_key(stub_webrtc: MagicMock) ->
     )
 
 
+def test_make_connection_replay_forwards_speed_and_window() -> None:
+    """--replay-speed/--replay-seek/--replay-duration reach ReplayConnection.
+
+    The `replay` property is lazy, so no database is opened here.
+    """
+    cfg = GlobalConfig(
+        replay=True,
+        replay_db="some_db",
+        replay_speed=25.0,
+        replay_seek=3.5,
+        replay_duration=60.0,
+    )
+    conn = go2_conn.make_connection(None, cfg)
+    assert isinstance(conn, go2_conn.ReplayConnection)
+    assert conn.dataset == "some_db"
+    assert (conn._speed, conn._seek, conn._duration) == (25.0, 3.5, 60.0)
+
+
+def test_make_connection_replay_defaults_realtime() -> None:
+    conn = go2_conn.make_connection("replay", GlobalConfig(replay=True))
+    assert isinstance(conn, go2_conn.ReplayConnection)
+    assert (conn._speed, conn._seek, conn._duration) == (1.0, None, None)
+
+
 def test_connection_config_aes_key_defaults_from_global_config() -> None:
     """ConnectionConfig.aes_128_key defaults from GlobalConfig.unitree_aes_128_key."""
     g = GlobalConfig(robot_ip="127.0.0.1", unitree_aes_128_key="dd" * 16)