Krishna/feat/openarm mini v2

dimos/cli/can.py

@@ -0,0 +1,99 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Linux CAN interface management commands."""
+
+from __future__ import annotations
+
+import os
+import shlex
+import subprocess
+
+import typer
+
+app = typer.Typer(help="Configure and inspect Linux CAN interfaces", no_args_is_help=True)
+
+
+def _run_ip(*args: str, privileged: bool = False) -> subprocess.CompletedProcess[str]:
+    command = ["ip", *args]
+    if privileged and os.geteuid() != 0:
+        command = ["sudo", "--", *command]
+    if privileged:
+        typer.echo(f"Running: {shlex.join(command)}")
+    try:
+        return subprocess.run(
+            command,
+            check=True,
+            capture_output=not privileged,
+            text=True,
+        )
+    except FileNotFoundError as exc:
+        executable = command[0]
+        raise typer.BadParameter(f"the '{executable}' command is not installed") from exc
+    except subprocess.CalledProcessError as exc:
+        stderr = exc.stderr.strip() if exc.stderr else ""
+        stdout = exc.stdout.strip() if exc.stdout else ""
+        detail = stderr or stdout or f"exit code {exc.returncode}"
+        typer.echo(f"CAN interface command failed: {detail}", err=True)
+        raise typer.Exit(1) from exc
+
+
+@app.command("status")
+def status(interface: str = typer.Argument(..., help="Linux CAN interface name")) -> None:
+    """Show detailed CAN interface state and queue statistics."""
+    result = _run_ip("-details", "-statistics", "link", "show", "dev", interface)
+    typer.echo(result.stdout.rstrip())
+
+
+@app.command("down")
+def down(interface: str = typer.Argument(..., help="Linux CAN interface name")) -> None:
+    """Bring a CAN interface down."""
+    _run_ip("link", "set", "dev", interface, "down", privileged=True)
+    typer.echo(f"CAN interface {interface} is down")
+
+
+@app.command("up")
+def up(interface: str = typer.Argument(..., help="Linux CAN interface name")) -> None:
+    """Bring an already configured CAN interface up."""
+    _run_ip("link", "set", "dev", interface, "up", privileged=True)
+    typer.echo(f"CAN interface {interface} is up")
+
+
+@app.command("setup")
+def setup(
+    interface: str = typer.Argument(..., help="Linux CAN interface name"),
+    bitrate: int = typer.Option(
+        1_000_000,
+        min=1,
+        help="Nominal CAN bitrate in bits per second",
+    ),
+    txqueuelen: int = typer.Option(1_000, min=1, help="Kernel transmit queue length"),
+) -> None:
+    """Configure, bring up, and verify a classic CAN interface."""
+    setup_interface(interface, bitrate=bitrate, txqueuelen=txqueuelen)
+
+
+def setup_interface(interface: str, *, bitrate: int, txqueuelen: int = 1_000) -> None:
+    """Configure and verify one classic CAN interface."""
+
+    _run_ip("link", "show", "dev", interface)
+    _run_ip("link", "set", "dev", interface, "down", privileged=True)
+    _run_ip(
+        "link", "set", "dev", interface, "type", "can", "bitrate", str(bitrate), privileged=True
+    )
+    _run_ip("link", "set", "dev", interface, "txqueuelen", str(txqueuelen), privileged=True)
+    _run_ip("link", "set", "dev", interface, "up", privileged=True)
+    result = _run_ip("-details", "-statistics", "link", "show", "dev", interface)
+    typer.echo(result.stdout.rstrip())
+    typer.echo(f"Configured {interface}: bitrate={bitrate}, txqueuelen={txqueuelen}")

dimos/cli/dimos.py

@@ -50,6 +50,7 @@
 
 from dimos.agents.mcp.mcp_adapter import McpAdapter, McpError
 from dimos.cli.cache import app as cache_app
+from dimos.cli.can import app as can_app
 from dimos.cli.shell import shell
 from dimos.constants import CONFIG_DIR, LOG_DIR
 from dimos.core.daemon import daemonize, install_signal_handlers
@@ -60,7 +61,6 @@
 from dimos.mapping.cli.rename import main as _map_rename_main
 from dimos.mapping.cli.replay import main as _map_replay_main
 from dimos.mapping.cli.replay_marker import main as _map_replay_marker_main
-from dimos.robot.manipulators.piper.cli import app as piper_app
 from dimos.robot.unitree.go2.cli.go2tool import app as go2tool_app
 from dimos.utils.cache import cache_usage_locked
 from dimos.utils.logging_config import setup_logger
@@ -172,8 +172,10 @@ def callback(**kwargs) -> None:  # type: ignore[no-untyped-def]
 
 
 main.callback()(create_dynamic_callback())  # type: ignore[no-untyped-call]
+hardware_app = typer.Typer(help="Configure and inspect robot hardware", no_args_is_help=True)
+hardware_app.add_typer(can_app, name="can")
+main.add_typer(hardware_app, name="hardware")
 main.add_typer(go2tool_app, name="go2tool")
-main.add_typer(piper_app, name="piper")
 main.command()(shell)
 main.add_typer(cache_app, name="cache")
 

dimos/cli/test_can.py

@@ -0,0 +1,202 @@
+# Copyright 2026 Dimensional Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import subprocess
+from unittest.mock import Mock
+
+from click.testing import Result
+import pytest
+from pytest_mock import MockerFixture
+from typer.testing import CliRunner
+
+from dimos.cli.dimos import main
+
+
+def _subprocess_argv(run: Mock) -> list[list[str]]:
+    return [call.args[0] for call in run.call_args_list]
+
+
+def _invoke_can(args: list[str]) -> Result:
+    return CliRunner().invoke(main, ["hardware", "can", *args])
+
+
+def test_can_commands_are_nested_under_hardware_scope() -> None:
+    result = CliRunner().invoke(main, ["hardware", "--help"])
+    legacy = CliRunner().invoke(main, ["can", "--help"])
+
+    assert result.exit_code == 0, result.output
+    assert "can" in result.output
+    assert legacy.exit_code == 2
+
+
+def test_setup_valid_options_configures_and_verifies_can_interface(
+    mocker: MockerFixture,
+) -> None:
+    mocker.patch("dimos.cli.can.os.geteuid", return_value=1000)
+    run = mocker.patch(
+        "dimos.cli.can.subprocess.run",
+        return_value=subprocess.CompletedProcess(
+            [],
+            0,
+            stdout="4: follower_l: UP qlen 1000\n",
+            stderr="",
+        ),
+    )
+
+    result = _invoke_can(["setup", "follower_l"])
+
+    assert result.exit_code == 0, result.output
+    assert "Running: sudo -- ip link set dev follower_l down" in result.stdout
+    assert "bitrate=1000000, txqueuelen=1000" in result.stdout
+    assert _subprocess_argv(run) == [
+        ["ip", "link", "show", "dev", "follower_l"],
+        ["sudo", "--", "ip", "link", "set", "dev", "follower_l", "down"],
+        [
+            "sudo",
+            "--",
+            "ip",
+            "link",
+            "set",
+            "dev",
+            "follower_l",
+            "type",
+            "can",
+            "bitrate",
+            "1000000",
+        ],
+        [
+            "sudo",
+            "--",
+            "ip",
+            "link",
+            "set",
+            "dev",
+            "follower_l",
+            "txqueuelen",
+            "1000",
+        ],
+        ["sudo", "--", "ip", "link", "set", "dev", "follower_l", "up"],
+        ["ip", "-details", "-statistics", "link", "show", "dev", "follower_l"],
+    ]
+
+
+def test_setup_nonpositive_queue_length_returns_usage_error() -> None:
+    result = _invoke_can(["setup", "can0", "--txqueuelen", "0"])
+
+    assert result.exit_code == 2
+    assert "x>=1" in result.output
+
+
+def test_setup_nonpositive_bitrate_returns_usage_error() -> None:
+    result = _invoke_can(["setup", "can0", "--bitrate", "0"])
+
+    assert result.exit_code == 2
+    assert "x>=1" in result.output
+
+
+def test_status_existing_interface_prints_detailed_state(mocker: MockerFixture) -> None:
+    run = mocker.patch(
+        "dimos.cli.can.subprocess.run",
+        return_value=subprocess.CompletedProcess([], 0, stdout="can0: UP\n", stderr=""),
+    )
+
+    result = _invoke_can(["status", "can0"])
+
+    assert result.exit_code == 0, result.output
+    assert result.stdout == "can0: UP\n"
+    run.assert_called_once_with(
+        ["ip", "-details", "-statistics", "link", "show", "dev", "can0"],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+
+
+def test_down_nonroot_user_runs_privileged_command_with_sudo(
+    mocker: MockerFixture,
+) -> None:
+    mocker.patch("dimos.cli.can.os.geteuid", return_value=1000)
+    run = mocker.patch(
+        "dimos.cli.can.subprocess.run",
+        return_value=subprocess.CompletedProcess([], 0),
+    )
+
+    result = _invoke_can(["down", "can1"])
+
+    assert result.exit_code == 0, result.output
+    assert "CAN interface can1 is down" in result.stdout
+    run.assert_called_once_with(
+        ["sudo", "--", "ip", "link", "set", "dev", "can1", "down"],
+        check=True,
+        capture_output=False,
+        text=True,
+    )
+
+
+def test_up_root_user_runs_ip_without_sudo(mocker: MockerFixture) -> None:
+    mocker.patch("dimos.cli.can.os.geteuid", return_value=0)
+    run = mocker.patch(
+        "dimos.cli.can.subprocess.run",
+        return_value=subprocess.CompletedProcess([], 0),
+    )
+
+    result = _invoke_can(["up", "can2"])
+
+    assert result.exit_code == 0, result.output
+    assert "CAN interface can2 is up" in result.stdout
+    run.assert_called_once_with(
+        ["ip", "link", "set", "dev", "can2", "up"],
+        check=True,
+        capture_output=False,
+        text=True,
+    )
+
+
+def test_status_missing_ip_command_returns_usage_error(mocker: MockerFixture) -> None:
+    mocker.patch("dimos.cli.can.subprocess.run", side_effect=FileNotFoundError)
+
+    result = _invoke_can(["status", "can0"])
+
+    assert result.exit_code == 2
+    assert "the 'ip' command is not installed" in result.output
+
+
+@pytest.mark.parametrize(
+    ("stderr", "stdout", "expected_detail"),
+    [
+        ("permission denied\n", "ignored\n", "permission denied"),
+        ("", "device not found\n", "device not found"),
+        ("", "", "exit code 7"),
+    ],
+)
+def test_status_failed_ip_command_reports_available_detail(
+    mocker: MockerFixture,
+    stderr: str,
+    stdout: str,
+    expected_detail: str,
+) -> None:
+    mocker.patch(
+        "dimos.cli.can.subprocess.run",
+        side_effect=subprocess.CalledProcessError(
+            7,
+            ["ip"],
+            output=stdout,
+            stderr=stderr,
+        ),
+    )
+
+    result = _invoke_can(["status", "can0"])
+
+    assert result.exit_code == 1
+    assert f"CAN interface command failed: {expected_detail}" in result.output

dimos/control/test_control.py

@@ -970,6 +970,25 @@ def test_tick_loop_calls_compute(self, mock_adapter, wait_until):
 
         assert mock_task.compute.call_count > 0
 
+    def test_write_all_hardware_rejected_command_logs_error(self, mocker):
+        hardware = {"arm": MagicMock()}
+        hardware["arm"].write_command.return_value = False
+        log_error = mocker.patch("dimos.control.tick_loop.logger.error")
+        tick_loop = TickLoop(
+            tick_rate=100.0,
+            hardware=hardware,
+            hardware_lock=threading.Lock(),
+            tasks={},
+            task_lock=threading.Lock(),
+            joint_to_hardware={"arm/joint1": "arm"},
+        )
+
+        tick_loop._write_all_hardware({"arm": ({"arm/joint1": 0.25}, ControlMode.SERVO_POSITION)})
+
+        log_error.assert_called_once_with(
+            "Hardware arm rejected SERVO_POSITION command from control task"
+        )
+
 
 class TestIntegration:
     def test_full_trajectory_execution(self, mock_adapter, wait_until):