RSDK-5028 Write analog python SDK (#460)

Author: oliviamiller

src/viam/components/board/board.py

@@ -263,3 +263,14 @@ async def set_power_mode(
             mode: the desired power mode
         """
         ...
+
+    @abc.abstractmethod
+    async def write_analog(self, pin: str, value: int, *, timeout: Optional[float] = None, **kwargs):
+        """
+        Write an analog value to a pin on the board.
+
+        Args:
+            pin (str): name of the pin.
+            value (int): value to write.
+        """
+        ...

src/viam/components/board/client.py

@@ -25,6 +25,7 @@
     SetPWMRequest,
     StatusRequest,
     StatusResponse,
+    WriteAnalogRequest,
 )
 from viam.resource.rpc_client_base import ReconfigurableResourceRPCClientBase
 from viam.utils import ValueTypes, dict_to_struct, get_geometries, struct_to_dict
@@ -178,3 +179,7 @@ async def set_power_mode(
 
     async def get_geometries(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> List[Geometry]:
         return await get_geometries(self.client, self.name, extra, timeout)
+
+    async def write_analog(self, pin: str, value: int, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None):
+        request = WriteAnalogRequest(name=self.name, pin=pin, value=value)
+        await self.client.WriteAnalog(request, timeout=timeout)

src/viam/components/board/service.py

@@ -1,6 +1,6 @@
 from grpclib.server import Stream
 
-from viam.errors import ResourceNotFoundError, MethodNotImplementedError
+from viam.errors import ResourceNotFoundError
 from viam.proto.common import DoCommandRequest, DoCommandResponse, GetGeometriesRequest, GetGeometriesResponse
 from viam.proto.component.board import (
     BoardServiceBase,
@@ -178,7 +178,14 @@ async def SetPowerMode(self, stream: Stream[SetPowerModeRequest, SetPowerModeRes
         await stream.send_message(response)
 
     async def WriteAnalog(self, stream: Stream[WriteAnalogRequest, WriteAnalogResponse]) -> None:
-        raise MethodNotImplementedError("WriteAnalog")
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        board = self.get_resource(name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        await board.write_analog(pin=request.pin, value=request.value, timeout=timeout, metadata=stream.metadata)
+        response = WriteAnalogResponse()
+        await stream.send_message(response)
 
     async def DoCommand(self, stream: Stream[DoCommandRequest, DoCommandResponse]) -> None:
         request = await stream.recv_message()

tests/mocks/components.py

@@ -388,6 +388,11 @@ async def set_power_mode(
         self.power_mode = mode
         self.power_mode_duration = duration
 
+    async def write_analog(self, pin: str, value: int, *, timeout: Optional[float] = None, **kwargs):
+        self.timeout = timeout
+        self.analog_write_pin = pin
+        self.analog_write_value = value
+
 
 class MockCamera(Camera):
     def __init__(self, name: str):

tests/test_board.py

@@ -39,6 +39,8 @@
     SetPWMRequest,
     StatusRequest,
     StatusResponse,
+    WriteAnalogRequest,
+    WriteAnalogResponse,
 )
 from viam.resource.manager import ResourceManager
 from viam.utils import dict_to_struct, struct_to_dict
@@ -144,6 +146,15 @@ async def test_get_geometries(self, board: MockBoard):
         geometries = await board.get_geometries()
         assert geometries == GEOMETRIES
 
+    @pytest.mark.asyncio
+    async def test_write_analog(self, board: MockBoard):
+        value = 10
+        pin = "pin1"
+        await board.write_analog(pin=pin, value=value, timeout=1.11)
+        assert board.timeout == loose_approx(1.11)
+        assert board.analog_write_value == value
+        assert board.analog_write_pin == pin
+
 
 class TestService:
     @pytest.mark.asyncio
@@ -320,6 +331,19 @@ async def test_set_power_mode(self, board: MockBoard, service: BoardRPCService):
             assert board.power_mode == PowerMode.POWER_MODE_OFFLINE_DEEP
             assert board.power_mode_duration == pm_duration.ToTimedelta()
 
+    @pytest.mark.asyncio
+    async def test_write_analog(self, board: MockBoard, service: BoardRPCService):
+        async with ChannelFor([service]) as channel:
+            client = BoardServiceStub(channel)
+            pin = "pin1"
+            value = 10
+            request = WriteAnalogRequest(name=board.name, pin=pin, value=value)
+            response: WriteAnalogResponse = await client.WriteAnalog(request, timeout=6.66)
+            assert response == WriteAnalogResponse()
+            assert board.timeout == loose_approx(6.66)
+            assert board.analog_write_value == value
+            assert board.analog_write_pin == pin
+
 
 class TestClient:
     @pytest.mark.asyncio
@@ -501,3 +525,14 @@ async def test_get_pwm_freq(self, board: MockBoard, service: BoardRPCService):
             mock_pin = cast(MockGPIOPin, board.gpios["pin1"])
             assert mock_pin.extra == extra
             assert mock_pin.timeout is None
+
+    @pytest.mark.asyncio
+    async def test_write_analog(self, board: MockBoard, service: BoardRPCService):
+        async with ChannelFor([service]) as channel:
+            client = BoardClient(name=board.name, channel=channel)
+            pin = "pin1"
+            value = 42
+            extra = {"foo": "bar", "baz": [1, 2, 3]}
+            await client.write_analog(pin, value, extra=extra)
+            assert board.analog_write_pin == "pin1"
+            assert board.analog_write_value == 42