RSDK-10279: Button component (#877)

Author: lia-viam

src/viam/components/button/__init__.py

@@ -0,0 +1,10 @@
+import viam.gen.component.button.v1.button_pb2
+from viam.resource.registry import Registry, ResourceRegistration
+
+from .client import ButtonClient
+from .service import ButtonRPCService
+from .button import Button
+
+__all__ = ["Button"]
+
+Registry.register_api(ResourceRegistration(Button, ButtonRPCService, lambda name, channel: ButtonClient(name, channel)))

src/viam/components/button/button.py

@@ -0,0 +1,41 @@
+import abc
+from typing import Any, Final, Mapping, Optional
+
+from viam.components.component_base import ComponentBase
+from viam.resource.types import API, RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT
+
+
+class Button(ComponentBase):
+    """
+    Button represents a device that can be pushed.
+
+    This acts as an abstract base class for any drivers representing specific
+    button implementations. This cannot be used on its own. If the ``__init__()`` function is
+    overridden, it must call the ``super().__init__()`` function.
+
+    ::
+
+        from viam.components.button import Button
+
+    For more information, see `Button component <https://docs.viam.com/dev/reference/apis/components/button/>` _.
+    """
+
+    API: Final = API(  # pyright: ignore [reportIncompatibleVariableOverride]
+        RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT, "button"
+    )
+
+    @abc.abstractmethod
+    async def push(self, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> None:
+        """
+        Push the button.
+
+        ::
+
+            my_button = Button.from_robot(robot=machine, name="my_button")
+
+            # Push the button
+            await my_button.push()
+
+        For more information, see `Button component <https://docs.viam.com/dev/reference/apis/components/button/>` _.
+        """
+        ...

src/viam/components/button/client.py

@@ -0,0 +1,54 @@
+from typing import Any, Mapping, Optional
+
+from grpclib.client import Channel
+
+from viam.gen.component.button.v1.button_pb2 import (
+    PushRequest
+)
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+)
+from viam.proto.component.button import ButtonServiceStub
+from viam.resource.rpc_client_base import ReconfigurableResourceRPCClientBase
+from viam.utils import (
+    ValueTypes,
+    dict_to_struct,
+    struct_to_dict,
+)
+
+from .button import Button
+
+
+class ButtonClient(Button, ReconfigurableResourceRPCClientBase):
+    """
+    gRPC client for Button component
+    """
+
+    def __init__(self, name: str, channel: Channel):
+        self.channel = channel
+        self.client = ButtonServiceStub(channel)
+        super().__init__(name)
+
+    async def push(
+        self,
+        *,
+        extra: Optional[Mapping[str, Any]] = None,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ) -> None:
+        md = kwargs.get("metadata", self.Metadata()).proto
+        request = PushRequest(name=self.name, extra=dict_to_struct(extra))
+        await self.client.Push(request, timeout=timeout, metadata=md)
+
+    async def do_command(
+        self,
+        command: Mapping[str, ValueTypes],
+        *,
+        timeout: Optional[float] = None,
+        **kwargs,
+    ) -> Mapping[str, ValueTypes]:
+        md = kwargs.get("metadata", self.Metadata()).proto
+        request = DoCommandRequest(name=self.name, command=dict_to_struct(command))
+        response: DoCommandResponse = await self.client.DoCommand(request, timeout=timeout, metadata=md)
+        return struct_to_dict(response.result)

src/viam/components/button/service.py

@@ -0,0 +1,46 @@
+from grpclib.server import Stream
+
+from viam.gen.component.button.v1.button_pb2 import (
+    PushRequest,
+    PushResponse,
+)
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+)
+from viam.proto.component.button import ButtonServiceBase
+from viam.resource.rpc_service_base import ResourceRPCServiceBase
+from viam.utils import dict_to_struct, struct_to_dict
+
+from .button import Button
+
+
+class ButtonRPCService(ButtonServiceBase, ResourceRPCServiceBase[Button]):
+    """
+    gRPC Service for a generic Button
+    """
+
+    RESOURCE_TYPE = Button
+
+    async def Push(self, stream: Stream[PushRequest, PushResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        button = self.get_resource(name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        await button.push(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        await stream.send_message(PushResponse())
+
+    async def DoCommand(self, stream: Stream[DoCommandRequest, DoCommandResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        button = self.get_resource(name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        result = await button.do_command(
+            command=struct_to_dict(request.command),
+            timeout=timeout,
+            metadata=stream.metadata,
+        )
+        response = DoCommandResponse(result=dict_to_struct(result))
+        await stream.send_message(response)

tests/mocks/components.py

@@ -15,6 +15,7 @@
 from viam.components.arm import Arm, JointPositions, KinematicsFileFormat
 from viam.components.audio_input import AudioInput
 from viam.components.base import Base
+from viam.components.button import Button
 from viam.components.board import Board, Tick
 from viam.components.camera import Camera, DistortionParameters, IntrinsicParameters
 from viam.components.encoder import Encoder
@@ -1043,3 +1044,18 @@ async def set_position(
 
     async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None, **kwargs) -> Mapping[str, ValueTypes]:
         return {"command": command}
+
+class MockButton(Button):
+    def __init__(self, name: str):
+        self.pushed = False
+        self.timeout: Optional[float] = None
+        self.extra: Optional[Mapping[str, Any]] = None
+        super().__init__(name)
+
+    async def push(self, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> None:
+        self.extra = extra
+        self.timeout = timeout
+        self.pushed = True
+
+    async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None, **kwargs) -> Mapping[str, ValueTypes]:
+        return {"command": command}

tests/test_button.py

@@ -0,0 +1,87 @@
+import pytest
+from grpclib.testing import ChannelFor
+
+from viam.components.button import ButtonClient
+from viam.components.button.service import ButtonRPCService
+from viam.gen.component.button.v1.button_pb2 import (
+    PushRequest
+)
+from viam.proto.common import (
+    DoCommandRequest,
+    DoCommandResponse,
+)
+from viam.proto.component.button import ButtonServiceStub
+from viam.resource.manager import ResourceManager
+from viam.utils import dict_to_struct, struct_to_dict
+
+from . import loose_approx
+from .mocks.components import MockButton
+
+EXTRA_PARAMS = {"foo": "bar", "baz": [1, 2, 3]}
+
+
+@pytest.fixture(scope="function")
+def button() -> MockButton:
+    return MockButton(name="button")
+
+
+class TestButton:
+    async def test_push(self, button):
+        await button.push(timeout=1.23, extra=EXTRA_PARAMS)
+        assert button.pushed is True
+        assert button.timeout == loose_approx(1.23)
+        assert button.extra == EXTRA_PARAMS
+
+    async def test_do(self, button):
+        command = {"command": "args"}
+        resp = await button.do_command(command)
+        assert resp == {"command": command}
+
+
+@pytest.fixture(scope="function")
+def manager(button) -> ResourceManager:
+    return ResourceManager([button])
+
+
+@pytest.fixture(scope="function")
+def service(manager) -> ButtonRPCService:
+    return ButtonRPCService(manager)
+
+
+class TestService:
+    async def test_push(self, button, service):
+        async with ChannelFor([service]) as channel:
+            client = ButtonServiceStub(channel)
+            request = PushRequest(name=button.name, extra=dict_to_struct(EXTRA_PARAMS))
+            assert button.extra is None
+            await client.Push(request, timeout=1.23)
+            assert button.pushed is True
+            assert button.extra == EXTRA_PARAMS
+            assert button.timeout == loose_approx(1.23)
+
+    async def test_do(self, button: MockButton, service: ButtonRPCService):
+        async with ChannelFor([service]) as channel:
+            client = ButtonServiceStub(channel)
+            command = {"command": "args"}
+            request = DoCommandRequest(name=button.name, command=dict_to_struct(command))
+            response: DoCommandResponse = await client.DoCommand(request)
+            result = struct_to_dict(response.result)
+            assert result == {"command": command}
+
+
+class TestClient:
+    async def test_push(self, button, service):
+        async with ChannelFor([service]) as channel:
+            client = ButtonClient(button.name, channel)
+            assert button.extra is None
+            await client.push(timeout=3.45, extra=EXTRA_PARAMS)
+            assert button.pushed is True
+            assert button.extra == EXTRA_PARAMS
+            assert button.timeout == loose_approx(3.45)
+
+    async def test_do(self, button, manager, service):
+        async with ChannelFor([service]) as channel:
+            client = ButtonClient(button.name, channel)
+            command = {"command": "args"}
+            resp = await client.do_command(command)
+            assert resp == {"command": command}