Add MachineStatus to Robot API (#704)

Author: ale7714

src/viam/robot/client.py

@@ -27,6 +27,8 @@
     GetCloudMetadataResponse,
     GetOperationsRequest,
     GetOperationsResponse,
+    GetMachineStatusRequest,
+    GetMachineStatusResponse,
     GetStatusRequest,
     GetStatusResponse,
     GetVersionRequest,
@@ -894,3 +896,25 @@ async def get_version(self) -> GetVersionResponse:
 
         request = GetVersionRequest()
         return await self._client.GetVersion(request)
+
+    ######################
+    # Get Machine Status #
+    ######################
+
+    async def get_machine_status(self) -> GetMachineStatusResponse:
+        """
+        Get status information about the robot.
+
+        ::
+
+            machine_status = await machine.get_machine_status()
+            resource_statuses = machine_status.resources
+
+        Returns:
+            viam.proto.robot.GetMachineStatusResponse: current status of the resources (List[ResourceStatus]) of the robot.
+
+        For more information, see `Machine Management API <https://docs.viam.com/appendix/apis/robot/>`_.
+        """
+
+        request = GetMachineStatusRequest()
+        return await self._client.GetMachineStatus(request)

tests/test_robot.py

@@ -36,6 +36,9 @@
     GetOperationsResponse,
     GetStatusRequest,
     GetStatusResponse,
+    GetMachineStatusRequest,
+    GetMachineStatusResponse,
+    ResourceStatus,
     GetVersionRequest,
     GetVersionResponse,
     Operation,
@@ -161,6 +164,15 @@
     api_version="0.3.0",
 )
 
+GET_MACHINE_STATUS_RESPONSE = GetMachineStatusResponse(
+    resources=[
+        ResourceStatus(
+            name=ResourceName(namespace=RESOURCE_NAMESPACE_RDK, type=RESOURCE_TYPE_COMPONENT, subtype="arm", name="arm1"),
+            state=ResourceStatus.State.STATE_READY,
+        )
+    ]
+)
+
 
 @pytest.fixture(scope="function")
 def service() -> RobotService:
@@ -225,6 +237,11 @@ async def GetVersion(stream: Stream[GetVersionRequest, GetVersionResponse]) -> N
         assert request is not None
         await stream.send_message(GET_VERVSION_RESPONSE)
 
+    async def GetMachineStatus(stream: Stream[GetMachineStatusRequest, GetMachineStatusResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        await stream.send_message(GET_MACHINE_STATUS_RESPONSE)
+
     async def Shutdown(stream: Stream[ShutdownRequest, ShutdownResponse]) -> None:
         request = await stream.recv_message()
         assert request is not None
@@ -240,6 +257,7 @@ async def Shutdown(stream: Stream[ShutdownRequest, ShutdownResponse]) -> None:
     service.GetCloudMetadata = GetCloudMetadata
     service.Shutdown = Shutdown
     service.GetVersion = GetVersion
+    service.GetMachineStatus = GetMachineStatus
 
     return service
 
@@ -462,6 +480,14 @@ async def test_get_version(self, service: RobotService):
             assert md == GET_VERVSION_RESPONSE
             await client.close()
 
+    @pytest.mark.asyncio
+    async def test_get_machine_status(self, service: RobotService):
+        async with ChannelFor([service]) as channel:
+            client = await RobotClient.with_channel(channel, RobotClient.Options())
+            statuses = await client.get_machine_status()
+            assert statuses == GET_MACHINE_STATUS_RESPONSE
+            await client.close()
+
     @pytest.mark.asyncio
     async def test_get_operations(self, service: RobotService):
         async with ChannelFor([service]) as channel: