diff --git a/docs/examples/example.ipynb b/docs/examples/example.ipynb
index 090358396c..05e6ab2048 100644
--- a/docs/examples/example.ipynb
+++ b/docs/examples/example.ipynb
@@ -505,7 +505,7 @@
"# 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",
@@ -513,6 +513,7 @@
"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",
diff --git a/docs/examples/my_cool_arm.py b/docs/examples/my_cool_arm.py
index 889b3cb667..b944859924 100644
--- a/docs/examples/my_cool_arm.py
+++ b/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
diff --git a/examples/complex_module/src/arm/my_arm.py b/examples/complex_module/src/arm/my_arm.py
index 00b3000df4..ee9a5d28c4 100644
--- a/examples/complex_module/src/arm/my_arm.py
+++ b/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
diff --git a/examples/server/v1/components.py b/examples/server/v1/components.py
index 50ef07aee1..679ed35719 100644
--- a/examples/server/v1/components.py
+++ b/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
diff --git a/src/viam/components/arm/__init__.py b/src/viam/components/arm/__init__.py
index fcc9ce871a..ffad555cb6 100644
--- a/src/viam/components/arm/__init__.py
+++ b/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",
]
diff --git a/src/viam/components/arm/arm.py b/src/viam/components/arm/arm.py
index b3f65595f7..501c16c7c2 100644
--- a/src/viam/components/arm/arm.py
+++ b/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 `_.
"""
@@ -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 `_.
+ """
+ ...
+
@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 `_.
"""
...
+
+ @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 `_.
+ """
+ ...
diff --git a/src/viam/components/arm/client.py b/src/viam/components/arm/client.py
index 7113fa269e..a602d6bdaf 100644
--- a/src/viam/components/arm/client.py
+++ b/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)
diff --git a/src/viam/components/arm/service.py b/src/viam/components/arm/service.py
index f7bba88dde..8b4884eb6d 100644
--- a/src/viam/components/arm/service.py
+++ b/src/viam/components/arm/service.py
@@ -3,6 +3,8 @@
from viam.proto.common import (
DoCommandRequest,
DoCommandResponse,
+ Get3DModelsRequest,
+ Get3DModelsResponse,
GetGeometriesRequest,
GetGeometriesResponse,
GetKinematicsRequest,
@@ -17,6 +19,8 @@
GetJointPositionsResponse,
IsMovingRequest,
IsMovingResponse,
+ MoveThroughJointPositionsRequest,
+ MoveThroughJointPositionsResponse,
MoveToJointPositionsRequest,
MoveToJointPositionsResponse,
MoveToPositionRequest,
@@ -78,6 +82,25 @@ async def MoveToJointPositions(self, stream: Stream[MoveToJointPositionsRequest,
response = MoveToJointPositionsResponse()
await stream.send_message(response)
+ async def MoveThroughJointPositions(self, stream: Stream[MoveThroughJointPositionsRequest, MoveThroughJointPositionsResponse]) -> None:
+ request = await stream.recv_message()
+ assert request is not None
+ name = request.name
+ arm = self.get_resource(name)
+ timeout = stream.deadline.time_remaining() if stream.deadline else None
+ # `options` has explicit presence; passing a zeroed message when the caller sent
+ # nothing would be indistinguishable from a real zero limit.
+ options = request.options if request.HasField("options") else None
+ await arm.move_through_joint_positions(
+ list(request.positions),
+ options=options,
+ extra=struct_to_dict(request.extra),
+ timeout=timeout,
+ metadata=stream.metadata,
+ )
+ response = MoveThroughJointPositionsResponse()
+ await stream.send_message(response)
+
async def Stop(self, stream: Stream[StopRequest, StopResponse]) -> None:
request = await stream.recv_message()
assert request is not None
@@ -129,6 +152,15 @@ async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematics
response = GetKinematicsResponse(format=format, kinematics_data=kinematics_data, meshes_by_urdf_filepath=meshes)
await stream.send_message(response)
+ async def Get3DModels(self, stream: Stream[Get3DModelsRequest, Get3DModelsResponse]) -> 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
+ models = await arm.get_3d_models(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+ response = Get3DModelsResponse(models=models)
+ await stream.send_message(response)
+
async def GetGeometries(self, stream: Stream[GetGeometriesRequest, GetGeometriesResponse]) -> None:
request = await stream.recv_message()
assert request is not None
diff --git a/tests/mocks/components.py b/tests/mocks/components.py
index 3746bc33c8..e5acfca5d0 100644
--- a/tests/mocks/components.py
+++ b/tests/mocks/components.py
@@ -46,6 +46,7 @@
Sphere,
Vector3,
)
+from viam.proto.component.arm import MoveOptions
from viam.proto.component.audioin import AudioChunk, GetAudioResponse
from viam.proto.component.board import PowerMode
from viam.proto.component.encoder import PositionType
@@ -57,6 +58,11 @@
Geometry(center=Pose(x=1, y=2, z=3, o_x=2, o_y=3, o_z=4, theta=20), capsule=Capsule(radius_mm=3, length_mm=8)),
]
+MODELS_3D = {
+ "base_link": Mesh(content_type="ply", mesh=b"\x00\x01"),
+ "link_1": Mesh(content_type="ply", mesh=b"\x02\x03"),
+}
+
class MockArm(Arm):
def __init__(self, name: str):
@@ -67,6 +73,9 @@ def __init__(self, name: str):
self.geometries = GEOMETRIES
self.extra = None
self.timeout: Optional[float] = None
+ self.waypoints: List[JointPositions] = []
+ self.move_options: Optional[MoveOptions] = None
+ self.models_3d = MODELS_3D
super().__init__(name)
async def get_end_position(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Pose:
@@ -102,6 +111,30 @@ async def move_to_joint_positions(
self.extra = extra
self.timeout = timeout
+ 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,
+ ):
+ self.waypoints = list(positions)
+ self.move_options = options
+ if self.waypoints:
+ self.joint_positions = self.waypoints[-1]
+ self.is_stopped = False
+ self.extra = extra
+ self.timeout = timeout
+
+ async def get_3d_models(
+ self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
+ ) -> Mapping[str, Mesh]:
+ self.extra = extra
+ self.timeout = timeout
+ return self.models_3d
+
async def stop(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs):
self.is_stopped = True
self.extra = extra
diff --git a/tests/test_arm.py b/tests/test_arm.py
index 1100f7d4f3..b451554ca7 100644
--- a/tests/test_arm.py
+++ b/tests/test_arm.py
@@ -1,3 +1,4 @@
+import pytest
from grpclib.testing import ChannelFor
from viam.components.arm import ArmClient, KinematicsFileFormat
@@ -5,6 +6,8 @@
from viam.proto.common import (
DoCommandRequest,
DoCommandResponse,
+ Get3DModelsRequest,
+ Get3DModelsResponse,
GetGeometriesRequest,
GetGeometriesResponse,
GetKinematicsRequest,
@@ -22,6 +25,8 @@
IsMovingRequest,
IsMovingResponse,
JointPositions,
+ MoveOptions,
+ MoveThroughJointPositionsRequest,
MoveToJointPositionsRequest,
MoveToPositionRequest,
StopRequest,
@@ -30,7 +35,7 @@
from viam.utils import dict_to_struct, struct_to_dict
from . import expected_grpc_timeout
-from .mocks.components import GEOMETRIES, MockArm
+from .mocks.components import GEOMETRIES, MODELS_3D, MockArm
class TestArm:
@@ -55,6 +60,14 @@ async def test_get_joint_positions(self):
jp = await self.arm.get_joint_positions()
assert jp == self.joint_pos
+ async def test_move_through_joint_positions(self):
+ waypoints = [JointPositions(values=[1, 2, 3]), JointPositions(values=[4, 5, 6])]
+ options = MoveOptions(max_vel_degs_per_sec=15.0, max_acc_degs_per_sec2=30.0)
+ await self.arm.move_through_joint_positions(waypoints, options)
+ assert self.arm.waypoints == waypoints
+ assert self.arm.move_options == options
+ assert self.arm.joint_positions == waypoints[-1]
+
async def test_stop(self):
assert self.arm.is_stopped is False
await self.arm.stop()
@@ -75,6 +88,11 @@ async def test_get_geometries(self):
geometries = await self.arm.get_geometries()
assert geometries == GEOMETRIES
+ async def test_get_3d_models(self):
+ models = await self.arm.get_3d_models(extra={"1": "2"})
+ assert models == MODELS_3D
+ assert self.arm.extra == {"1": "2"}
+
async def test_do(self):
command = {"command": "args"}
resp = await self.arm.do_command(command)
@@ -128,6 +146,40 @@ async def test_get_joint_positions(self):
response: GetJointPositionsResponse = await client.GetJointPositions(request)
assert response.positions == self.joint_pos
+ async def test_move_through_joint_positions(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmServiceStub(channel)
+ waypoints = [JointPositions(values=[1, 2, 3]), JointPositions(values=[4, 5, 6])]
+ options = MoveOptions(
+ max_vel_degs_per_sec=15.0,
+ max_acc_degs_per_sec2=30.0,
+ max_vel_degs_per_sec_joints=[1.0, 2.0, 3.0],
+ max_acc_degs_per_sec2_joints=[4.0, 5.0, 6.0],
+ max_tcp_speed=0.25,
+ )
+ request = MoveThroughJointPositionsRequest(name=self.name, positions=waypoints, options=options)
+ await client.MoveThroughJointPositions(request)
+ assert self.arm.waypoints == waypoints
+ assert self.arm.move_options == options
+
+ async def test_move_through_joint_positions_without_options(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmServiceStub(channel)
+ waypoints = [JointPositions(values=[7, 8, 9])]
+ request = MoveThroughJointPositionsRequest(name=self.name, positions=waypoints)
+ await client.MoveThroughJointPositions(request)
+ assert self.arm.waypoints == waypoints
+ # Must be None, NOT a zeroed MoveOptions: a driver reading
+ # max_vel_degs_per_sec == 0.0 could interpret that as "do not move".
+ assert self.arm.move_options is None
+
+ async def test_move_through_joint_positions_empty(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmServiceStub(channel)
+ request = MoveThroughJointPositionsRequest(name=self.name, positions=[])
+ await client.MoveThroughJointPositions(request)
+ assert self.arm.waypoints == []
+
async def test_stop(self):
async with ChannelFor([self.service]) as channel:
assert self.arm.is_stopped is False
@@ -177,6 +229,26 @@ async def test_get_geometries(self):
response: GetGeometriesResponse = await client.GetGeometries(request)
assert [geometry for geometry in response.geometries] == GEOMETRIES
+ async def test_get_3d_models(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmServiceStub(channel)
+ request = Get3DModelsRequest(name=self.name)
+ response: Get3DModelsResponse = await client.Get3DModels(request)
+ assert dict(response.models) == MODELS_3D
+ assert response.models["base_link"].content_type == "ply"
+ assert response.models["base_link"].mesh == b"\x00\x01"
+
+ async def test_get_3d_models_empty(self):
+ async with ChannelFor([self.service]) as channel:
+ self.arm.models_3d = {}
+ try:
+ client = ArmServiceStub(channel)
+ request = Get3DModelsRequest(name=self.name)
+ response: Get3DModelsResponse = await client.Get3DModels(request)
+ assert dict(response.models) == {}
+ finally:
+ self.arm.models_3d = MODELS_3D
+
async def test_extra(self):
async with ChannelFor([self.service]) as channel:
client = ArmServiceStub(channel)
@@ -221,6 +293,25 @@ async def test_get_joint_positions(self):
jp = await client.get_joint_positions()
assert jp == self.joint_pos
+ async def test_move_through_joint_positions(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmClient(self.name, channel)
+ waypoints = [JointPositions(values=[1, 2, 3]), JointPositions(values=[4, 5, 6])]
+ options = MoveOptions(max_vel_degs_per_sec=15.0, max_tcp_speed=0.25)
+ await client.move_through_joint_positions(waypoints, options=options, extra={"foo": "bar"}, timeout=1.23)
+ assert self.arm.waypoints == waypoints
+ assert self.arm.move_options == options
+ assert self.arm.extra == {"foo": "bar"}
+ assert self.arm.timeout == expected_grpc_timeout(1.23)
+
+ async def test_move_through_joint_positions_without_options(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmClient(self.name, channel)
+ waypoints = [JointPositions(values=[7, 8, 9])]
+ await client.move_through_joint_positions(waypoints)
+ assert self.arm.waypoints == waypoints
+ assert self.arm.move_options is None
+
async def test_stop(self):
async with ChannelFor([self.service]) as channel:
assert self.arm.is_stopped is False
@@ -250,6 +341,16 @@ async def test_get_geometries(self):
geometries = await client.get_geometries()
assert geometries == GEOMETRIES
+ async def test_get_3d_models(self):
+ async with ChannelFor([self.service]) as channel:
+ client = ArmClient(self.name, channel)
+ models = await client.get_3d_models(extra={"1": "2"}, timeout=1.23)
+ assert models == MODELS_3D
+ with pytest.raises(KeyError):
+ models["does_not_exist"]
+ assert self.arm.extra == {"1": "2"}
+ assert self.arm.timeout == expected_grpc_timeout(1.23)
+
async def test_do(self):
async with ChannelFor([self.service]) as channel:
client = ArmClient(self.name, channel)
diff --git a/tests/test_robot.py b/tests/test_robot.py
index a09aeeff79..5acd8a3382 100644
--- a/tests/test_robot.py
+++ b/tests/test_robot.py
@@ -1,5 +1,5 @@
import asyncio
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Mapping, Optional, Tuple
from unittest import mock
import pytest
@@ -14,8 +14,8 @@
from viam.components.movement_sensor import MovementSensor
from viam.errors import ResourceNotFoundError
from viam.proto.app.datasync import UploadMetadata
-from viam.proto.common import Geometry, GeoPoint, Orientation, Pose, PoseInFrame, ResourceName, Transform, Vector3
-from viam.proto.component.arm import JointPositions
+from viam.proto.common import Geometry, GeoPoint, Mesh, Orientation, Pose, PoseInFrame, ResourceName, Transform, Vector3
+from viam.proto.component.arm import JointPositions, MoveOptions
from viam.proto.robot import (
BlockForOperationRequest,
BlockForOperationResponse,
@@ -497,6 +497,24 @@ async def move_to_joint_positions(
):
return await self.actual_client.move_to_joint_positions(positions, extra=extra, timeout=timeout)
+ async def move_through_joint_positions(
+ self,
+ positions: List[JointPositions],
+ options: Optional[MoveOptions] = None,
+ *,
+ extra: Optional[Dict[str, Any]] = None,
+ timeout: Optional[float] = None,
+ ):
+ return await self.actual_client.move_through_joint_positions(positions, options=options, extra=extra, timeout=timeout)
+
+ async def get_3d_models(
+ self,
+ *,
+ extra: Optional[Dict[str, Any]] = None,
+ timeout: Optional[float] = None,
+ ) -> Mapping[str, Mesh]:
+ return await self.actual_client.get_3d_models(extra=extra, timeout=timeout)
+
async def get_joint_positions(
self,
*,