RSDK-6664 Get metadata client (#545)

Author: maximpertsov

src/viam/robot/client.py

@@ -23,6 +23,8 @@
     FrameSystemConfig,
     FrameSystemConfigRequest,
     FrameSystemConfigResponse,
+    GetCloudMetadataRequest,
+    GetCloudMetadataResponse,
     GetOperationsRequest,
     GetOperationsResponse,
     GetStatusRequest,
@@ -623,3 +625,15 @@ async def stop_all(self, extra: Dict[ResourceName, Dict[str, Any]] = {}):
             ep.append(StopExtraParameters(name=name, params=dict_to_struct(params)))
         request = StopAllRequest(extra=ep)
         await self._client.StopAll(request)
+
+    ######################
+    # Get Cloud Metadata #
+    ######################
+
+    async def get_cloud_metadata(self) -> GetCloudMetadataResponse:
+        """
+        Returns app-related information about the robot.
+        """
+
+        request = GetCloudMetadataRequest()
+        return await self._client.GetCloudMetadata(request)

src/viam/services/motion/client.py

@@ -1,4 +1,4 @@
-from typing import Any, Final, List, Mapping, Optional, Sequence, Iterable
+from typing import Any, Final, Iterable, List, Mapping, Optional, Sequence
 
 from grpclib.client import Channel
 

tests/test_robot.py

@@ -29,6 +29,8 @@
     FrameSystemConfig,
     FrameSystemConfigRequest,
     FrameSystemConfigResponse,
+    GetCloudMetadataRequest,
+    GetCloudMetadataResponse,
     GetOperationsRequest,
     GetOperationsResponse,
     GetStatusRequest,
@@ -140,6 +142,12 @@
 
 OPERATIONS_RESPONSE = [Operation(id=OPERATION_ID)]
 
+GET_CLOUD_METADATA_RESPONSE = GetCloudMetadataResponse(
+    robot_part_id="the-robot-part",
+    primary_org_id="the-primary-org",
+    location_id="the-location",
+)
+
 
 @pytest.fixture(scope="function")
 def service() -> RobotService:
@@ -194,12 +202,18 @@ async def GetOperations(stream: Stream[GetOperationsRequest, GetOperationsRespon
         response = GetOperationsResponse(operations=OPERATIONS_RESPONSE)
         await stream.send_message(response)
 
+    async def GetCloudMetadata(stream: Stream[GetCloudMetadataRequest, GetCloudMetadataResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        await stream.send_message(GET_CLOUD_METADATA_RESPONSE)
+
     manager = ResourceManager(resources)
     service = RobotService(manager)
     service.FrameSystemConfig = Config
     service.TransformPose = TransformPose
     service.DiscoverComponents = DiscoverComponents
     service.GetOperations = GetOperations
+    service.GetCloudMetadata = GetCloudMetadata
 
     return service
 
@@ -407,6 +421,14 @@ async def test_discover_components(self, service: RobotService):
             assert discoveries == DISCOVERY_RESPONSE
             await client.close()
 
+    @pytest.mark.asyncio
+    async def test_get_cloud_metadata(self, service: RobotService):
+        async with ChannelFor([service]) as channel:
+            client = await RobotClient.with_channel(channel, RobotClient.Options())
+            md = await client.get_cloud_metadata()
+            assert md == GET_CLOUD_METADATA_RESPONSE
+            await client.close()
+
     @pytest.mark.asyncio
     async def test_get_operations(self, service: RobotService):
         async with ChannelFor([service]) as channel: