Fix AI

Author: njooma

src/viam/components/arm/service.py

@@ -109,7 +109,7 @@ 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 = await arm.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout)
+        format, kinematics_data = await arm.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
         response = GetKinematicsResponse(format=format, kinematics_data=kinematics_data)
         await stream.send_message(response)
 

src/viam/components/gantry/client.py

@@ -2,7 +2,14 @@
 
 from grpclib.client import Channel
 
-from viam.proto.common import DoCommandRequest, DoCommandResponse, Geometry, GetKinematicsRequest, GetKinematicsResponse, KinematicsFileFormat
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+    Geometry,
+    GetKinematicsRequest,
+    GetKinematicsResponse,
+    KinematicsFileFormat,
+)
 from viam.proto.component.gantry import (
     GantryServiceStub,
     GetLengthsRequest,
@@ -111,16 +118,12 @@ async def do_command(
         return struct_to_dict(response.result)
 
     async def get_kinematics(
-        self,
-        *,
-        extra: Optional[Dict[str, Any]] = None,
-        timeout: Optional[float] = None,
-        **kwargs,
+        self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
     ) -> Tuple[KinematicsFileFormat.ValueType, bytes]:
         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)
-        return response.format, response.kinematics_data
+        return (response.format, response.kinematics_data)
 
     async def get_geometries(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[Geometry]:
         md = kwargs.get("metadata", self.Metadata())

src/viam/components/gantry/gantry.py

@@ -161,37 +161,27 @@ async def get_kinematics(
         self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
     ) -> Tuple[KinematicsFileFormat.ValueType, bytes]:
         """
-        Get the kinematics of the gantry.
+        Get the kinematics information associated with the gantry.
 
         ::
 
             my_gantry = Gantry.from_robot(robot=machine, name="my_gantry")
 
-            # Get the kinematics of the gantry.
-            format, data = await my_gantry.get_kinematics()
+            # Get the kinematics information associated with the gantry.
+            kinematics = await my_gantry.get_kinematics()
 
-        Returns:
-            Tuple[KinematicsFileFormat.ValueType, bytes]: The kinematics file format and the data.
-
-        For more information, see `Gantry component <https://docs.viam.com/dev/reference/apis/components/gantry/#getkinematics>`_.
-        """
-        ...
-
-    @abc.abstractmethod
-    async def get_geometries(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> List[Geometry]:
-        """
-        Get geometries of the gantry.
-
-        ::
+            # Get the format of the kinematics file.
+            k_file = kinematics[0]
 
-            my_gantry = Gantry.from_robot(robot=machine, name="my_gantry")
-
-            # Get the geometries of the gantry.
-            geometries = await my_gantry.get_geometries()
+            # Get the byte contents of the file.
+            k_bytes = kinematics[1]
 
         Returns:
-            List[Geometry]: A list of geometries of the gantry.
+            Tuple[KinematicsFileFormat.ValueType, bytes]: A tuple containing two values; the first [0] value represents the format of the
+            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.
 
-        For more information, see `Gantry component <https://docs.viam.com/dev/reference/apis/components/gantry/#getgeometries>`_.
+        For more information, see `Arm component <https://docs.viam.com/dev/reference/apis/components/arm/#getkinematics>`_.
         """
-        ...
+        ...

src/viam/components/gantry/service.py

@@ -1,8 +1,14 @@
 from grpclib.server import Stream
 
-from viam.proto.common import DoCommandRequest, DoCommandResponse, GetGeometriesRequest, GetGeometriesResponse, GetKinematicsRequest, GetKinematicsResponse
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+    GetGeometriesRequest,
+    GetGeometriesResponse,
+    GetKinematicsRequest,
+    GetKinematicsResponse,
+)
 from viam.proto.component.gantry import (
-    GantryServiceBase,
     GetLengthsRequest,
     GetLengthsResponse,
     GetPositionRequest,
@@ -15,14 +21,15 @@
     MoveToPositionResponse,
     StopRequest,
     StopResponse,
+    UnimplementedGantryServiceBase,
 )
 from viam.resource.rpc_service_base import ResourceRPCServiceBase
 from viam.utils import dict_to_struct, struct_to_dict
 
 from .gantry import Gantry
 
 
-class GantryRPCService(GantryServiceBase, ResourceRPCServiceBase[Gantry]):
+class GantryRPCService(UnimplementedGantryServiceBase, ResourceRPCServiceBase[Gantry]):
     """
     gRPC Service for a Gantry
     """
@@ -106,8 +113,7 @@ async def DoCommand(self, stream: Stream[DoCommandRequest, DoCommandResponse]) -
     async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematicsResponse]) -> None:
         request = await stream.recv_message()
         assert request is not None
-        name = request.name
-        gantry = self.get_resource(name)
+        gantry = self.get_resource(request.name)
         timeout = stream.deadline.time_remaining() if stream.deadline else None
         format, data = await gantry.get_kinematics(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
         response = GetKinematicsResponse(format=format, kinematics_data=data)
@@ -116,8 +122,8 @@ async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematics
     async def GetGeometries(self, stream: Stream[GetGeometriesRequest, GetGeometriesResponse]) -> None:
         request = await stream.recv_message()
         assert request is not None
-        arm = self.get_resource(request.name)
+        gantry = self.get_resource(request.name)
         timeout = stream.deadline.time_remaining() if stream.deadline else None
-        geometries = await arm.get_geometries(extra=struct_to_dict(request.extra), timeout=timeout)
+        geometries = await gantry.get_geometries(extra=struct_to_dict(request.extra), timeout=timeout)
         response = GetGeometriesResponse(geometries=geometries)
         await stream.send_message(response)

tests/test_gantry.py

@@ -2,7 +2,14 @@
 
 from viam.components.gantry import GantryClient
 from viam.components.gantry.service import GantryRPCService
-from viam.proto.common import DoCommandRequest, DoCommandResponse, GetGeometriesRequest, GetGeometriesResponse, GetKinematicsRequest, GetKinematicsResponse, KinematicsFileFormat
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+    GetGeometriesRequest,
+    GetGeometriesResponse,
+    GetKinematicsRequest,
+    GetKinematicsResponse,
+)
 from viam.proto.component.gantry import (
     GantryServiceStub,
     GetLengthsRequest,
@@ -260,4 +267,4 @@ async def test_get_geometries(self):
         async with ChannelFor([self.service]) as channel:
             client = GantryClient(self.gantry.name, channel)
             geometries = await client.get_geometries()
-            assert geometries == GEOMETRIES
+            assert geometries == GEOMETRIES