no longer a breaking change, using union type

Author: nfranczak

docs/examples/example.ipynb

@@ -2,11 +2,11 @@
 
 import asyncio
 import json
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Mapping, Optional, Tuple, Union
 
 from viam.components.arm import Arm, JointPositions, KinematicsFileFormat, Pose
 from viam.operations import run_with_operation
-from viam.proto.common import Capsule, Geometry, Sphere
+from viam.proto.common import Capsule, Geometry, Mesh, Sphere
 
 
 class MyCoolArm(Arm):
@@ -100,7 +100,9 @@ async def is_moving(self) -> bool:
     async def get_geometries(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> List[Geometry]:
         return self.geometries
 
-    async def get_kinematics(self, extra: Optional[Dict[str, Any]] = None, **kwargs) -> Tuple[KinematicsFileFormat.ValueType, bytes]:
+    async def get_kinematics(
+        self, extra: Optional[Dict[str, Any]] = None, **kwargs
+    ) -> Union[Tuple[KinematicsFileFormat.ValueType, bytes], Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]]]:
         return KinematicsFileFormat.KINEMATICS_FILE_FORMAT_SVA, self.kinematics
 
     async def close(self):

docs/examples/my_cool_arm.py

@@ -1,12 +1,17 @@
 import abc
-from typing import Any, Dict, Final, Optional, Tuple, Mapping
+from typing import Any, Dict, Final, Optional, Tuple, Mapping, Union
 
 from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
 
 from ..component_base import ComponentBase
 from . import JointPositions, KinematicsFileFormat, Pose
 from viam.proto.common import Mesh
 
+KinematicsReturn = Union[
+    Tuple[KinematicsFileFormat.ValueType, bytes],
+    Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]],
+]
+
 
 class Arm(ComponentBase):
     """
@@ -196,7 +201,7 @@ async def is_moving(self) -> bool:
     @abc.abstractmethod
     async def get_kinematics(
         self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
-    ) -> Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]]:
+    ) -> KinematicsReturn:
         """
         Get the kinematics information associated with the arm.
 
@@ -218,6 +223,7 @@ async def get_kinematics(
             file, either in URDF format (``KinematicsFileFormat.KINEMATICS_FILE_FORMAT_URDF``) or
             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.
 
         For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#getkinematics>`_.
         """

src/viam/components/arm/arm.py

@@ -109,7 +109,12 @@ async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematics
         assert request is not None
         arm = self.get_resource(request.name)
         timeout = stream.deadline.time_remaining() if stream.deadline else None
-        format, kinematics_data, meshes = await arm.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        kinematics = await arm.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        if len(kinematics) == 2:
+            format, kinematics_data = kinematics
+            meshes = {}
+        else:
+            format, kinematics_data, meshes = kinematics
         response = GetKinematicsResponse(format=format, kinematics_data=kinematics_data, meshes_by_urdf_filepath=meshes)
         await stream.send_message(response)
 

src/viam/components/arm/service.py

@@ -1,12 +1,17 @@
 import abc
-from typing import Any, Dict, Final, List, Optional, Tuple, Mapping
+from typing import Any, Dict, Final, List, Optional, Tuple, Mapping, Union
 
 from viam.components.arm import KinematicsFileFormat
 from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
 from viam.proto.common import Mesh
 
 from ..component_base import ComponentBase
 
+KinematicsReturn = Union[
+    Tuple[KinematicsFileFormat.ValueType, bytes],
+    Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]],
+]
+
 
 class Gantry(ComponentBase):
     """
@@ -160,7 +165,7 @@ async def is_moving(self) -> bool:
     @abc.abstractmethod
     async def get_kinematics(
         self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
-    ) -> Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]]:
+    ) -> KinematicsReturn:
         """
         Get the kinematics information associated with the gantry.
 
@@ -182,6 +187,7 @@ async def get_kinematics(
             file, either in URDF format (``KinematicsFileFormat.KINEMATICS_FILE_FORMAT_URDF``) or
             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.
 
         For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#getkinematics>`_.
         """

src/viam/components/gantry/gantry.py

@@ -115,7 +115,12 @@ async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematics
         assert request is not None
         gantry = self.get_resource(request.name)
         timeout = stream.deadline.time_remaining() if stream.deadline else None
-        format, data, meshes = await gantry.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        kinematics = await gantry.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        if len(kinematics) == 2:
+            format, data = kinematics
+            meshes = {}
+        else:
+            format, data, meshes = kinematics
         response = GetKinematicsResponse(format=format, kinematics_data=data, meshes_by_urdf_filepath=meshes)
         await stream.send_message(response)
 

src/viam/components/gantry/service.py

@@ -1,13 +1,18 @@
 import abc
 from dataclasses import dataclass
-from typing import Any, Dict, Final, Optional, Tuple, Mapping
+from typing import Any, Dict, Final, Optional, Tuple, Mapping, Union
 
 from viam.components.component_base import ComponentBase
 from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
 from viam.proto.common import Mesh
 
 from . import KinematicsFileFormat
 
+KinematicsReturn = Union[
+    Tuple[KinematicsFileFormat.ValueType, bytes],
+    Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]],
+]
+
 
 class Gripper(ComponentBase):
     """
@@ -162,7 +167,7 @@ async def get_kinematics(
         extra: Optional[Dict[str, Any]] = None,
         timeout: Optional[float] = None,
         **kwargs,
-    ) -> Tuple[KinematicsFileFormat.ValueType, bytes, Mapping[str, Mesh]]:
+    ) -> KinematicsReturn:
         """
         Get the kinematics information associated with the gripper.
 
@@ -184,6 +189,7 @@ async def get_kinematics(
             file, either in URDF format (``KinematicsFileFormat.KINEMATICS_FILE_FORMAT_URDF``) or
             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.
 
         For more information, see `Gripper component <https://docs.viam.com/dev/reference/apis/components/gripper/#getkinematics>`_.
         """

src/viam/components/gripper/gripper.py

@@ -105,6 +105,11 @@ async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematics
         assert request is not None
         gripper = self.get_resource(request.name)
         timeout = stream.deadline.time_remaining() if stream.deadline else None
-        format, kinematics_data, meshes = await gripper.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout)
+        kinematics = await gripper.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout)
+        if len(kinematics) == 2:
+            format, kinematics_data = kinematics
+            meshes = {}
+        else:
+            format, kinematics_data, meshes = kinematics
         response = GetKinematicsResponse(format=format, kinematics_data=kinematics_data, meshes_by_urdf_filepath=meshes)
         await stream.send_message(response)