Automated Protos Update (viamrobotics#1041)

Co-authored-by: Naveed Jooma <[email protected]>

Author: viambot

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

@@ -1,8 +1,15 @@
-from typing import Any, Dict, List, Mapping, Optional
+from typing import Any, Dict, List, Mapping, Optional, Tuple
 
 from grpclib.client import Channel
 
-from viam.proto.common import DoCommandRequest, DoCommandResponse, Geometry
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+    Geometry,
+    GetKinematicsRequest,
+    GetKinematicsResponse,
+    KinematicsFileFormat,
+)
 from viam.proto.component.gantry import (
     GantryServiceStub,
     GetLengthsRequest,
@@ -110,6 +117,14 @@ async def do_command(
         response: DoCommandResponse = await self.client.DoCommand(request, timeout=timeout, metadata=md)
         return struct_to_dict(response.result)
 
+    async def get_kinematics(
+        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)
+
     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)

src/viam/components/gantry/gantry.py

@@ -1,6 +1,7 @@
 import abc
-from typing import Any, Dict, Final, List, Optional
+from typing import Any, Dict, Final, List, Optional, Tuple
 
+from viam.components.arm import KinematicsFileFormat
 from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
 
 from ..component_base import ComponentBase
@@ -154,3 +155,33 @@ async def is_moving(self) -> bool:
         For more information, see `Gantry component <https://docs.viam.com/dev/reference/apis/components/gantry/#ismoving>`_.
         """
         ...
+
+    @abc.abstractmethod
+    async def get_kinematics(
+        self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs
+    ) -> Tuple[KinematicsFileFormat.ValueType, bytes]:
+        """
+        Get the kinematics information associated with the gantry.
+
+        ::
+
+            my_gantry = Gantry.from_robot(robot=machine, name="my_gantry")
+
+            # Get the kinematics information associated with the gantry.
+            kinematics = await my_gantry.get_kinematics()
+
+            # Get the format of the kinematics file.
+            k_file = kinematics[0]
+
+            # Get the byte contents of the file.
+            k_bytes = kinematics[1]
+
+        Returns:
+            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 `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
+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
     """
@@ -103,11 +110,20 @@ async def DoCommand(self, stream: Stream[DoCommandRequest, DoCommandResponse]) -
         response = DoCommandResponse(result=dict_to_struct(result))
         await stream.send_message(response)
 
+    async def GetKinematics(self, stream: Stream[GetKinematicsRequest, GetKinematicsResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        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)
+        await stream.send_message(response)
+
     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)

src/viam/gen/app/v1/robot_pb2.py

@@ -179,6 +179,7 @@ class JobConfig(google.protobuf.message.Message):
     RESOURCE_FIELD_NUMBER: builtins.int
     METHOD_FIELD_NUMBER: builtins.int
     COMMAND_FIELD_NUMBER: builtins.int
+    LOG_CONFIGURATION_FIELD_NUMBER: builtins.int
     name: builtins.str
     'unique name of the job.'
     schedule: builtins.str
@@ -194,13 +195,17 @@ class JobConfig(google.protobuf.message.Message):
         command argument of the gRPC request.
         """
 
-    def __init__(self, *, name: builtins.str=..., schedule: builtins.str=..., resource: builtins.str=..., method: builtins.str=..., command: google.protobuf.struct_pb2.Struct | None=...) -> None:
+    @property
+    def log_configuration(self) -> global___LogConfiguration:
+        """configuration for this job's logger."""
+
+    def __init__(self, *, name: builtins.str=..., schedule: builtins.str=..., resource: builtins.str=..., method: builtins.str=..., command: google.protobuf.struct_pb2.Struct | None=..., log_configuration: global___LogConfiguration | None=...) -> None:
         ...
 
-    def HasField(self, field_name: typing.Literal['command', b'command']) -> builtins.bool:
+    def HasField(self, field_name: typing.Literal['command', b'command', 'log_configuration', b'log_configuration']) -> builtins.bool:
         ...
 
-    def ClearField(self, field_name: typing.Literal['command', b'command', 'method', b'method', 'name', b'name', 'resource', b'resource', 'schedule', b'schedule']) -> None:
+    def ClearField(self, field_name: typing.Literal['command', b'command', 'log_configuration', b'log_configuration', 'method', b'method', 'name', b'name', 'resource', b'resource', 'schedule', b'schedule']) -> None:
         ...
 global___JobConfig = JobConfig
 

src/viam/gen/app/v1/robot_pb2.pyi

@@ -40,12 +40,16 @@ async def IsMoving(self, stream: 'grpclib.server.Stream[component.gantry.v1.gant
     async def DoCommand(self, stream: 'grpclib.server.Stream[common.v1.common_pb2.DoCommandRequest, common.v1.common_pb2.DoCommandResponse]') -> None:
         pass
 
+    @abc.abstractmethod
+    async def GetKinematics(self, stream: 'grpclib.server.Stream[common.v1.common_pb2.GetKinematicsRequest, common.v1.common_pb2.GetKinematicsResponse]') -> None:
+        pass
+
     @abc.abstractmethod
     async def GetGeometries(self, stream: 'grpclib.server.Stream[common.v1.common_pb2.GetGeometriesRequest, common.v1.common_pb2.GetGeometriesResponse]') -> None:
         pass
 
     def __mapping__(self) -> typing.Dict[str, grpclib.const.Handler]:
-        return {'/viam.component.gantry.v1.GantryService/GetPosition': grpclib.const.Handler(self.GetPosition, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.GetPositionRequest, component.gantry.v1.gantry_pb2.GetPositionResponse), '/viam.component.gantry.v1.GantryService/MoveToPosition': grpclib.const.Handler(self.MoveToPosition, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.MoveToPositionRequest, component.gantry.v1.gantry_pb2.MoveToPositionResponse), '/viam.component.gantry.v1.GantryService/Home': grpclib.const.Handler(self.Home, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.HomeRequest, component.gantry.v1.gantry_pb2.HomeResponse), '/viam.component.gantry.v1.GantryService/GetLengths': grpclib.const.Handler(self.GetLengths, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.GetLengthsRequest, component.gantry.v1.gantry_pb2.GetLengthsResponse), '/viam.component.gantry.v1.GantryService/Stop': grpclib.const.Handler(self.Stop, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.StopRequest, component.gantry.v1.gantry_pb2.StopResponse), '/viam.component.gantry.v1.GantryService/IsMoving': grpclib.const.Handler(self.IsMoving, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.IsMovingRequest, component.gantry.v1.gantry_pb2.IsMovingResponse), '/viam.component.gantry.v1.GantryService/DoCommand': grpclib.const.Handler(self.DoCommand, grpclib.const.Cardinality.UNARY_UNARY, common.v1.common_pb2.DoCommandRequest, common.v1.common_pb2.DoCommandResponse), '/viam.component.gantry.v1.GantryService/GetGeometries': grpclib.const.Handler(self.GetGeometries, grpclib.const.Cardinality.UNARY_UNARY, common.v1.common_pb2.GetGeometriesRequest, common.v1.common_pb2.GetGeometriesResponse)}
+        return {'/viam.component.gantry.v1.GantryService/GetPosition': grpclib.const.Handler(self.GetPosition, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.GetPositionRequest, component.gantry.v1.gantry_pb2.GetPositionResponse), '/viam.component.gantry.v1.GantryService/MoveToPosition': grpclib.const.Handler(self.MoveToPosition, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.MoveToPositionRequest, component.gantry.v1.gantry_pb2.MoveToPositionResponse), '/viam.component.gantry.v1.GantryService/Home': grpclib.const.Handler(self.Home, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.HomeRequest, component.gantry.v1.gantry_pb2.HomeResponse), '/viam.component.gantry.v1.GantryService/GetLengths': grpclib.const.Handler(self.GetLengths, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.GetLengthsRequest, component.gantry.v1.gantry_pb2.GetLengthsResponse), '/viam.component.gantry.v1.GantryService/Stop': grpclib.const.Handler(self.Stop, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.StopRequest, component.gantry.v1.gantry_pb2.StopResponse), '/viam.component.gantry.v1.GantryService/IsMoving': grpclib.const.Handler(self.IsMoving, grpclib.const.Cardinality.UNARY_UNARY, component.gantry.v1.gantry_pb2.IsMovingRequest, component.gantry.v1.gantry_pb2.IsMovingResponse), '/viam.component.gantry.v1.GantryService/DoCommand': grpclib.const.Handler(self.DoCommand, grpclib.const.Cardinality.UNARY_UNARY, common.v1.common_pb2.DoCommandRequest, common.v1.common_pb2.DoCommandResponse), '/viam.component.gantry.v1.GantryService/GetKinematics': grpclib.const.Handler(self.GetKinematics, grpclib.const.Cardinality.UNARY_UNARY, common.v1.common_pb2.GetKinematicsRequest, common.v1.common_pb2.GetKinematicsResponse), '/viam.component.gantry.v1.GantryService/GetGeometries': grpclib.const.Handler(self.GetGeometries, grpclib.const.Cardinality.UNARY_UNARY, common.v1.common_pb2.GetGeometriesRequest, common.v1.common_pb2.GetGeometriesResponse)}
 
 class UnimplementedGantryServiceBase(GantryServiceBase):
 
@@ -70,6 +74,9 @@ async def IsMoving(self, stream: 'grpclib.server.Stream[component.gantry.v1.gant
     async def DoCommand(self, stream: 'grpclib.server.Stream[common.v1.common_pb2.DoCommandRequest, common.v1.common_pb2.DoCommandResponse]') -> None:
         raise grpclib.exceptions.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
 
+    async def GetKinematics(self, stream: 'grpclib.server.Stream[common.v1.common_pb2.GetKinematicsRequest, common.v1.common_pb2.GetKinematicsResponse]') -> None:
+        raise grpclib.exceptions.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
+
     async def GetGeometries(self, stream: 'grpclib.server.Stream[common.v1.common_pb2.GetGeometriesRequest, common.v1.common_pb2.GetGeometriesResponse]') -> None:
         raise grpclib.exceptions.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
 
@@ -83,4 +90,5 @@ def __init__(self, channel: grpclib.client.Channel) -> None:
         self.Stop = grpclib.client.UnaryUnaryMethod(channel, '/viam.component.gantry.v1.GantryService/Stop', component.gantry.v1.gantry_pb2.StopRequest, component.gantry.v1.gantry_pb2.StopResponse)
         self.IsMoving = grpclib.client.UnaryUnaryMethod(channel, '/viam.component.gantry.v1.GantryService/IsMoving', component.gantry.v1.gantry_pb2.IsMovingRequest, component.gantry.v1.gantry_pb2.IsMovingResponse)
         self.DoCommand = grpclib.client.UnaryUnaryMethod(channel, '/viam.component.gantry.v1.GantryService/DoCommand', common.v1.common_pb2.DoCommandRequest, common.v1.common_pb2.DoCommandResponse)
+        self.GetKinematics = grpclib.client.UnaryUnaryMethod(channel, '/viam.component.gantry.v1.GantryService/GetKinematics', common.v1.common_pb2.GetKinematicsRequest, common.v1.common_pb2.GetKinematicsResponse)
         self.GetGeometries = grpclib.client.UnaryUnaryMethod(channel, '/viam.component.gantry.v1.GantryService/GetGeometries', common.v1.common_pb2.GetGeometriesRequest, common.v1.common_pb2.GetGeometriesResponse)