RSDK-14459: Add manual mode arm API

doc/dependency_decisions.yml

@@ -94,7 +94,7 @@
     :when: 2026-08-12 00:00:00.000000000 Z
 - - :approve
   - googleapis-common-protos
-  - :who: njooma
-    :why: Apache 2.0
-    :versions: []
-    :when: 2026-09-10 22:09:58.590402000 Z
+  - :who: npmenard
+    :why: Apache-2.0
+    :versions: ['1.75.3']
+    :when: 2026-09-04 00:00:00.000000000 Z

docs/examples/example.ipynb

@@ -585,6 +585,15 @@
     "    async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Mapping[str, Mesh]:\n",
     "        raise NotImplementedError()\n",
     "\n",
+    "    async def set_manual_mode(self, manual_mode: bool, enabled_for: int = 0, extra: Optional[Dict[str, Any]] = None, **kwargs):\n",
+    "        raise NotImplementedError()\n",
+    "\n",
+    "    async def get_manual_mode(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> bool:\n",
+    "        raise NotImplementedError()\n",
+    "\n",
+    "    async def get_properties(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Arm.Properties:\n",
+    "        return Arm.Properties()\n",
+    "\n",
     "    async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):\n",
     "        self.is_stopped = True\n",
     "\n",

docs/examples/my_cool_arm.py

@@ -122,6 +122,17 @@ async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs)
         # Return the 3D meshes for this arm, keyed by name. This arm has none.
         return {}
 
+    async def set_manual_mode(self, manual_mode: bool, enabled_for: int = 0, extra: Optional[Dict[str, Any]] = None, **kwargs):
+        # This arm does not support manual mode.
+        raise NotImplementedError()
+
+    async def get_manual_mode(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> bool:
+        # This arm does not support manual mode.
+        raise NotImplementedError()
+
+    async def get_properties(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Arm.Properties:
+        return Arm.Properties(support_manual_mode=False, support_cartesian_commands=True)
+
     async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):
         self.is_stopped = True
 

examples/complex_module/src/arm/my_arm.py

@@ -126,6 +126,17 @@ async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs)
         # This arm has no meshes to report.
         return {}
 
+    async def set_manual_mode(self, manual_mode: bool, enabled_for: int = 0, extra: Optional[Dict[str, Any]] = None, **kwargs):
+        # This arm does not support manual mode.
+        raise NotImplementedError()
+
+    async def get_manual_mode(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> bool:
+        # This arm does not support manual mode.
+        raise NotImplementedError()
+
+    async def get_properties(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Arm.Properties:
+        return Arm.Properties(support_manual_mode=False, support_cartesian_commands=True)
+
     async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):
         self.is_stopped = True
 

examples/server/v1/components.py

@@ -109,6 +109,15 @@ async def move_through_joint_positions(
     async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Mapping[str, Mesh]:
         return {}
 
+    async def set_manual_mode(self, manual_mode: bool, enabled_for: int = 0, extra: Optional[Dict[str, Any]] = None, **kwargs):
+        raise NotImplementedError()
+
+    async def get_manual_mode(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> bool:
+        raise NotImplementedError()
+
+    async def get_properties(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Arm.Properties:
+        return Arm.Properties(support_manual_mode=False, support_cartesian_commands=True)
+
     async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):
         self.is_stopped = True
 

src/viam/components/arm/arm.py

@@ -1,8 +1,9 @@
 import abc
-from typing import Any, Dict, Final, List, Mapping, Optional
+from typing import Any, Dict, Final, List, Mapping, Optional, TypeAlias
 
 from viam.components import KinematicsReturn
 from viam.components.component_base import ComponentBase
+from viam.proto.component.arm import GetPropertiesResponse
 from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
 
 from . import JointPositions, Mesh, MoveOptions, Pose
@@ -31,6 +32,8 @@ class Arm(ComponentBase):
     For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/>`_.
     """
 
+    Properties: "TypeAlias" = GetPropertiesResponse
+
     API: Final = API(RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT, "arm")  # pyright: ignore [reportIncompatibleVariableOverride]
 
     @abc.abstractmethod
@@ -315,3 +318,85 @@ async def get_3d_models(
         For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#get3dmodels>`_.
         """
         ...
+
+    @abc.abstractmethod
+    async def set_manual_mode(
+        self,
+        manual_mode: bool,
+        enabled_for: int = 0,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        """
+        Enter or exit manual mode for an arm that supports it.
+
+        ::
+
+            my_arm = Arm.from_robot(robot=machine, name="my_arm")
+
+            # Enter manual mode for at most 30 seconds.
+            await my_arm.set_manual_mode(manual_mode=True, enabled_for=30)
+
+            # Exit manual mode.
+            await my_arm.set_manual_mode(manual_mode=False)
+
+        Args:
+            manual_mode (bool): Whether to enter (``True``) or exit (``False``) manual mode.
+            enabled_for (int): How long to stay in manual mode, in seconds. ``0`` means no time limit.
+
+        For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#setmanualmode>`_.
+        """
+        ...
+
+    @abc.abstractmethod
+    async def get_manual_mode(
+        self,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ) -> bool:
+        """
+        Get whether the arm is currently in manual mode.
+
+        ::
+
+            my_arm = Arm.from_robot(robot=machine, name="my_arm")
+
+            # Print whether the arm is currently in manual mode.
+            print(await my_arm.get_manual_mode())
+
+        Returns:
+            bool: Whether the arm is in manual mode.
+
+        For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#getmanualmode>`_.
+        """
+        ...
+
+    @abc.abstractmethod
+    async def get_properties(
+        self,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ) -> Properties:
+        """
+        Get a mapping of each optional feature to whether it is supported by this arm.
+
+        ::
+
+            my_arm = Arm.from_robot(robot=machine, name="my_arm")
+
+            # Get the properties of the arm.
+            properties = await my_arm.get_properties()
+
+        Returns:
+            Properties: The arm's properties; whether it supports software-enabled manual mode
+            and whether it supports direct cartesian commands (``move_to_position``).
+
+        For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#getproperties>`_.
+        """
+        ...

src/viam/components/arm/client.py

@@ -21,13 +21,17 @@
     GetEndPositionResponse,
     GetJointPositionsRequest,
     GetJointPositionsResponse,
+    GetManualModeRequest,
+    GetManualModeResponse,
+    GetPropertiesRequest,
     IsMovingRequest,
     IsMovingResponse,
     JointPositions,
     MoveOptions,
     MoveThroughJointPositionsRequest,
     MoveToJointPositionsRequest,
     MoveToPositionRequest,
+    SetManualModeRequest,
     StopRequest,
 )
 from viam.resource.rpc_client_base import ReconfigurableResourceRPCClientBase
@@ -173,3 +177,39 @@ async def get_3d_models(
     async def get_geometries(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[Geometry]:
         md = kwargs.get("metadata", self.Metadata())
         return await get_geometries(self.client, self.name, extra, timeout, md)
+
+    async def set_manual_mode(
+        self,
+        manual_mode: bool,
+        enabled_for: int = 0,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        md = kwargs.get("metadata", self.Metadata()).proto
+        request = SetManualModeRequest(name=self.name, manual_mode=manual_mode, enabled_for=enabled_for, extra=dict_to_struct(extra))
+        await self.client.SetManualMode(request, timeout=timeout, metadata=md)
+
+    async def get_manual_mode(
+        self,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ) -> bool:
+        md = kwargs.get("metadata", self.Metadata()).proto
+        request = GetManualModeRequest(name=self.name, extra=dict_to_struct(extra))
+        response: GetManualModeResponse = await self.client.GetManualMode(request, timeout=timeout, metadata=md)
+        return response.manual_mode
+
+    async def get_properties(
+        self,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ) -> Arm.Properties:
+        md = kwargs.get("metadata", self.Metadata()).proto
+        request = GetPropertiesRequest(name=self.name, extra=dict_to_struct(extra))
+        return await self.client.GetProperties(request, timeout=timeout, metadata=md)

src/viam/components/arm/service.py

@@ -17,6 +17,10 @@
     GetEndPositionResponse,
     GetJointPositionsRequest,
     GetJointPositionsResponse,
+    GetManualModeRequest,
+    GetManualModeResponse,
+    GetPropertiesRequest,
+    GetPropertiesResponse,
     IsMovingRequest,
     IsMovingResponse,
     MoveThroughJointPositionsRequest,
@@ -25,6 +29,8 @@
     MoveToJointPositionsResponse,
     MoveToPositionRequest,
     MoveToPositionResponse,
+    SetManualModeRequest,
+    SetManualModeResponse,
     StopRequest,
     StopResponse,
     UnimplementedArmServiceBase,
@@ -169,3 +175,35 @@ async def GetGeometries(self, stream: Stream[GetGeometriesRequest, GetGeometries
         geometries = await arm.get_geometries(extra=struct_to_dict(request.extra), timeout=timeout)
         response = GetGeometriesResponse(geometries=geometries)
         await stream.send_message(response)
+
+    async def SetManualMode(self, stream: Stream[SetManualModeRequest, SetManualModeResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        arm = self.get_resource(request.name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        await arm.set_manual_mode(
+            request.manual_mode,
+            enabled_for=request.enabled_for,
+            extra=struct_to_dict(request.extra),
+            timeout=timeout,
+            metadata=stream.metadata,
+        )
+        response = SetManualModeResponse()
+        await stream.send_message(response)
+
+    async def GetManualMode(self, stream: Stream[GetManualModeRequest, GetManualModeResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        arm = self.get_resource(request.name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        manual_mode = await arm.get_manual_mode(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        response = GetManualModeResponse(manual_mode=manual_mode)
+        await stream.send_message(response)
+
+    async def GetProperties(self, stream: Stream[GetPropertiesRequest, GetPropertiesResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        arm = self.get_resource(request.name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        response = await arm.get_properties(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        await stream.send_message(response)