Add move_through_joint_positions and get_3d_models to Arm (BREAKING)

docs/examples/example.ipynb

@@ -505,14 +505,15 @@
     "# modular-arm/src/my_modular_arm.py\n",
     "import asyncio\n",
     "import os\n",
-    "from typing import Any, ClassVar, Dict, Mapping, Optional, Tuple, Union\n",
+    "from typing import Any, ClassVar, Dict, List, Mapping, Optional, Tuple, Union\n",
     "from typing_extensions import Self\n",
     "\n",
     "from viam.components.arm import Arm, JointPositions, KinematicsFileFormat, Pose\n",
     "from viam.module.module import Module\n",
     "from viam.operations import run_with_operation\n",
     "from viam.proto.app.robot import ComponentConfig\n",
     "from viam.proto.common import Mesh, ResourceName\n",
+    "from viam.proto.component.arm import MoveOptions\n",
     "from viam.resource.base import ResourceBase\n",
     "from viam.resource.registry import Registry, ResourceCreatorRegistration\n",
     "from viam.resource.types import Model, ModelFamily\n",
@@ -567,10 +568,23 @@
     "            if await operation.is_cancelled():\n",
     "                await self.stop()\n",
     "                break\n",
-    "        \n",
+    "\n",
     "        self.joint_positions = positions\n",
     "        self.is_stopped = True\n",
     "\n",
+    "    async def move_through_joint_positions(\n",
+    "        self,\n",
+    "        positions: List[JointPositions],\n",
+    "        options: Optional[MoveOptions] = None,\n",
+    "        extra: Optional[Dict[str, Any]] = None,\n",
+    "        **kwargs,\n",
+    "    ):\n",
+    "        for position in positions:\n",
+    "            self.joint_positions = position\n",
+    "\n",
+    "    async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Mapping[str, Mesh]:\n",
+    "        raise NotImplementedError()\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

@@ -7,6 +7,7 @@
 from viam.components.arm import Arm, JointPositions, KinematicsFileFormat, Pose
 from viam.operations import run_with_operation
 from viam.proto.common import Capsule, Geometry, Mesh, Sphere
+from viam.proto.component.arm import MoveOptions
 
 
 class MyCoolArm(Arm):
@@ -91,6 +92,36 @@ async def move_to_joint_positions(self, positions: JointPositions, extra: Option
 
         self.is_stopped = True
 
+    @run_with_operation
+    async def move_through_joint_positions(
+        self,
+        positions: List[JointPositions],
+        options: Optional[MoveOptions] = None,
+        extra: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ):
+        operation = self.get_operation(kwargs)
+
+        self.is_stopped = False
+
+        # Move through each waypoint in order, honoring cancellation between them.
+        # A real driver would check options.HasField("max_vel_degs_per_sec") etc. and
+        # clamp its motion accordingly; this example moves at a fixed rate.
+        for position in positions:
+            await asyncio.sleep(1)
+
+            if await operation.is_cancelled():
+                await self.stop()
+                break
+
+            self.joint_positions = position
+
+        self.is_stopped = True
+
+    async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Mapping[str, Mesh]:
+        # Return the 3D meshes for this arm, keyed by name. This arm has none.
+        return {}
+
     async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):
         self.is_stopped = True
 

examples/complex_module/src/arm/my_arm.py

@@ -8,7 +8,8 @@
 from viam.logging import getLogger
 from viam.operations import run_with_operation
 from viam.proto.app.robot import ComponentConfig
-from viam.proto.common import Capsule, Geometry, ResourceName, Sphere
+from viam.proto.common import Capsule, Geometry, Mesh, ResourceName, Sphere
+from viam.proto.component.arm import MoveOptions
 from viam.resource.base import ResourceBase
 from viam.resource.registry import Registry, ResourceCreatorRegistration
 from viam.resource.types import Model, ModelFamily
@@ -93,6 +94,38 @@ async def move_to_joint_positions(self, positions: JointPositions, extra: Option
         self.joint_positions = positions
         self.is_stopped = True
 
+    @run_with_operation
+    async def move_through_joint_positions(
+        self,
+        positions: List[JointPositions],
+        options: Optional[MoveOptions] = None,
+        extra: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ):
+        operation = self.get_operation(kwargs)
+
+        self.is_stopped = False
+
+        # Move through each waypoint in order, honoring cancellation between them.
+        # A real driver is expected to honor the velocity/acceleration ceilings in
+        # `options`, checking e.g. options.HasField("max_vel_degs_per_sec") before
+        # applying it (an unset field reads as 0.0); this example ignores them and
+        # just sleeps for a fixed interval between waypoints.
+        for position in positions:
+            await asyncio.sleep(1)
+
+            if await operation.is_cancelled():
+                await self.stop()
+                break
+
+            self.joint_positions = position
+
+        self.is_stopped = True
+
+    async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Mapping[str, Mesh]:
+        # This arm has no meshes to report.
+        return {}
+
     async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):
         self.is_stopped = True
 

examples/server/v1/components.py

@@ -49,7 +49,7 @@
     Vector3,
     Mesh,
 )
-from viam.proto.component.arm import JointPositions
+from viam.proto.component.arm import JointPositions, MoveOptions
 from viam.proto.component.encoder import PositionType
 from viam.streams import StreamWithIterator
 from viam.utils import SensorReading, ValueTypes
@@ -95,6 +95,20 @@ async def move_to_joint_positions(self, positions: JointPositions, extra: Option
         self.is_stopped = False
         self.joint_positions = positions
 
+    async def move_through_joint_positions(
+        self,
+        positions: List[JointPositions],
+        options: Optional[MoveOptions] = None,
+        extra: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ):
+        self.is_stopped = False
+        for position in positions:
+            self.joint_positions = position
+
+    async def get_3d_models(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Mapping[str, Mesh]:
+        return {}
+
     async def stop(self, extra: Optional[Dict[str, Any]] = None, **kwargs):
         self.is_stopped = True
 

src/viam/components/arm/__init__.py

@@ -1,6 +1,6 @@
 from viam.components import KinematicsReturn
-from viam.proto.common import KinematicsFileFormat, Pose
-from viam.proto.component.arm import JointPositions
+from viam.proto.common import KinematicsFileFormat, Mesh, Pose
+from viam.proto.component.arm import JointPositions, MoveOptions
 from viam.resource.registry import Registry, ResourceRegistration
 
 from .arm import Arm
@@ -12,6 +12,8 @@
     "JointPositions",
     "KinematicsFileFormat",
     "KinematicsReturn",
+    "Mesh",
+    "MoveOptions",
     "Pose",
 ]
 

src/viam/components/arm/arm.py

@@ -1,11 +1,11 @@
 import abc
-from typing import Any, Dict, Final, Optional
+from typing import Any, Dict, Final, List, Mapping, Optional
 
 from viam.components import KinematicsReturn
 from viam.components.component_base import ComponentBase
 from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
 
-from . import JointPositions, Pose
+from . import JointPositions, Mesh, MoveOptions, Pose
 
 
 class Arm(ComponentBase):
@@ -20,9 +20,13 @@ class Arm(ComponentBase):
 
         from viam.components.arm import Arm
         # To use move_to_position:
-        from viam.proto.common import Pose
-        # To use move_to_joint_positions:
-        from viam.proto.component.arm import JointPositions
+        from viam.components.arm import Pose
+        # To use move_to_joint_positions and move_through_joint_positions:
+        from viam.components.arm import JointPositions
+        # To use move_through_joint_positions:
+        from viam.components.arm import MoveOptions
+        # To use get_3d_models:
+        from viam.components.arm import Mesh
 
     For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/>`_.
     """
@@ -122,6 +126,63 @@ async def move_to_joint_positions(
         """
         ...
 
+    @abc.abstractmethod
+    async def move_through_joint_positions(
+        self,
+        positions: List[JointPositions],
+        options: Optional[MoveOptions] = None,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        """
+        Move the arm through the given joint positions in the order they are specified,
+        obeying the velocity and acceleration limits in ``options``.
+
+        ::
+
+            my_arm = Arm.from_robot(robot=machine, name="my_arm")
+
+            # Move through two waypoints, capping joint speed and acceleration.
+            await my_arm.move_through_joint_positions(
+                positions=[
+                    JointPositions(values=[0, 45, 0, 0, 0, 0]),
+                    JointPositions(values=[0, 0, 0, 0, 0, 0]),
+                ],
+                options=MoveOptions(max_vel_degs_per_sec=15.0, max_acc_degs_per_sec2=30.0),
+            )
+
+        Args:
+            positions (List[JointPositions]): The waypoints to move through, in order.
+            options (Optional[MoveOptions]): Optional kinematic ceilings obeyed at every
+                point along the trajectory. ``None`` means no limits are requested.
+
+        Note:
+            Unlike the Go SDK, this method does not validate the requested positions
+            against the arm's joint limits before sending them, because the Python SDK
+            cannot yet parse a kinematics model. Implementations are responsible for
+            their own limit checking.
+
+            Every scalar field on ``MoveOptions`` (``max_vel_degs_per_sec``,
+            ``max_acc_degs_per_sec2``, ``max_tcp_speed``) also has explicit presence: an
+            unset field reads back as ``0.0``, indistinguishable from an explicitly-set
+            zero. Implementations must check ``options.HasField("max_vel_degs_per_sec")``
+            (and likewise for the other scalar fields) before applying it as a ceiling —
+            reading an unset field's ``0.0`` directly would misread "no limit requested"
+            as "do not move". Per the proto definition, ``max_vel_degs_per_sec`` is
+            ignored whenever ``max_vel_degs_per_sec_joints`` is set, and likewise
+            ``max_acc_degs_per_sec2`` is ignored whenever ``max_acc_degs_per_sec2_joints``
+            is set; implementations should honor only the per-joint limit in that case,
+            not both.
+
+            An empty ``positions`` list is passed through to the implementation
+            unchanged; implementations must handle it, typically as a no-op.
+
+        For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#movethroughjointpositions>`_.
+        """
+        ...
+
     @abc.abstractmethod
     async def get_joint_positions(
         self,
@@ -219,7 +280,38 @@ async def get_kinematics(
             Viam's kinematic parameter format (spatial vector algebra) (``KinematicsFileFormat.KINEMATICS_FILE_FORMAT_SVA``),
             and the second [1] value represents the byte contents of the file.
             If available, a third [2] value provides meshes keyed by URDF filepath.
+            See ``get_3d_models`` for meshes keyed by model name instead.
 
         For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#getkinematics>`_.
         """
         ...
+
+    @abc.abstractmethod
+    async def get_3d_models(
+        self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
+    ) -> Mapping[str, Mesh]:
+        """
+        Get the 3D models associated with the arm, keyed by name.
+
+        ::
+
+            my_arm = Arm.from_robot(robot=machine, name="my_arm")
+
+            # Get the arm's 3D models.
+            models = await my_arm.get_3d_models()
+
+            for name, mesh in models.items():
+                print(name, mesh.content_type, len(mesh.mesh))
+
+        Returns:
+            Mapping[str, Mesh]: The arm's 3D models keyed by name. Each ``Mesh`` carries a
+            ``content_type`` (for example ``"ply"``) and the raw ``mesh`` bytes in that format.
+            This is distinct from ``get_kinematics``'s third return value, which keys meshes
+            by URDF filepath rather than by model name.
+
+        Note:
+            Implementations with no models must return an empty mapping, not ``None``.
+
+        For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#get3dmodels>`_.
+        """
+        ...

src/viam/components/arm/client.py

@@ -7,10 +7,13 @@
     DoCommandRequest,
     DoCommandResponse,
     Geometry,
+    Get3DModelsRequest,
+    Get3DModelsResponse,
     GetKinematicsRequest,
     GetKinematicsResponse,
     GetStatusRequest,
     GetStatusResponse,
+    Mesh,
 )
 from viam.proto.component.arm import (
     ArmServiceStub,
@@ -21,6 +24,8 @@
     IsMovingRequest,
     IsMovingResponse,
     JointPositions,
+    MoveOptions,
+    MoveThroughJointPositionsRequest,
     MoveToJointPositionsRequest,
     MoveToPositionRequest,
     StopRequest,
@@ -91,6 +96,20 @@ async def move_to_joint_positions(
         request = MoveToJointPositionsRequest(name=self.name, positions=positions, extra=dict_to_struct(extra))
         await self.client.MoveToJointPositions(request, timeout=timeout, metadata=md)
 
+    async def move_through_joint_positions(
+        self,
+        positions: List[JointPositions],
+        options: Optional[MoveOptions] = None,
+        *,
+        extra: Optional[Dict[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ):
+        md = kwargs.get("metadata", self.Metadata()).proto
+        # Passing options=None leaves the optional field genuinely unset.
+        request = MoveThroughJointPositionsRequest(name=self.name, positions=positions, options=options, extra=dict_to_struct(extra))
+        await self.client.MoveThroughJointPositions(request, timeout=timeout, metadata=md)
+
     async def stop(
         self,
         *,
@@ -137,8 +156,20 @@ async def get_kinematics(
         md = kwargs.get("metadata", self.Metadata()).proto
         request = GetKinematicsRequest(name=self.name, extra=dict_to_struct(extra))
         response: GetKinematicsResponse = await self.client.GetKinematics(request, timeout=timeout, metadata=md)
+        # TODO: handle empty meshes in the response to prevent silent mapping
         return (response.format, response.kinematics_data, response.meshes_by_urdf_filepath)
 
+    async def get_3d_models(
+        self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
+    ) -> Mapping[str, Mesh]:
+        md = kwargs.get("metadata", self.Metadata()).proto
+        request = Get3DModelsRequest(name=self.name, extra=dict_to_struct(extra))
+        response: Get3DModelsResponse = await self.client.Get3DModels(request, timeout=timeout, metadata=md)
+        # Copy out of the protobuf map container: `__getitem__` on an absent key would
+        # otherwise create and insert a default-constructed value instead of raising
+        # KeyError, silently violating the Mapping contract.
+        return dict(response.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)