RSDK-6375 Add generic service wrappers (#526)

Author: benjirewis

src/viam/services/generic/__init__.py

@@ -0,0 +1,18 @@
+import viam.gen.service.generic.v1.generic_pb2  # Need this import for Generic service descriptors to resolve
+from viam.resource.registry import Registry, ResourceRegistration
+
+from .client import GenericClient
+from .generic import Generic
+from .service import GenericRPCService
+
+__all__ = [
+    "Generic",
+]
+
+Registry.register_subtype(
+    ResourceRegistration(
+        Generic,
+        GenericRPCService,
+        lambda name, channel: GenericClient(name, channel),
+    )
+)

src/viam/services/generic/client.py

@@ -0,0 +1,55 @@
+from typing import Any, Mapping, Optional
+
+from grpclib import GRPCError, Status
+from grpclib.client import Channel
+
+from viam.proto.common import DoCommandRequest, DoCommandResponse
+from viam.proto.service.generic import GenericServiceStub
+from viam.resource.rpc_client_base import ReconfigurableResourceRPCClientBase
+from viam.utils import ValueTypes, dict_to_struct, struct_to_dict
+
+from .generic import Generic
+
+
+class GenericClient(Generic, ReconfigurableResourceRPCClientBase):
+    """
+    gRPC client for the Generic service.
+    """
+
+    def __init__(self, name: str, channel: Channel):
+        self.client = GenericServiceStub(channel)
+        super().__init__(name)
+
+    async def do_command(
+        self,
+        command: Mapping[str, Any],
+        *,
+        timeout: Optional[float] = None,
+        **__,
+    ) -> Mapping[str, Any]:
+        request = DoCommandRequest(name=self.name, command=dict_to_struct(command))
+        try:
+            response: DoCommandResponse = await self.client.DoCommand(request, timeout=timeout)
+        except GRPCError as e:
+            if e.status == Status.UNIMPLEMENTED:
+                raise NotImplementedError()
+            raise e
+
+        return struct_to_dict(response.result)
+
+
+async def do_command(
+    channel: Channel, name: str, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None
+) -> Mapping[str, ValueTypes]:
+    """Convenience method to allow service clients to execute ``do_command`` functions
+
+    Args:
+        channel (Channel): A gRPC channel
+        name (str): The name of the component
+        command (Dict[str, Any]): The command to execute
+
+    Returns:
+        Dict[str, Any]: The result of the executed command
+    """
+    client = GenericClient(name, channel)
+    return await client.do_command(command, timeout=timeout)

src/viam/services/generic/generic.py

@@ -0,0 +1,58 @@
+from typing import Final
+
+from viam.resource.types import RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_SERVICE, Subtype
+
+from ..service_base import ServiceBase
+
+
+class Generic(ServiceBase):
+    """
+    Generic service, which represents any type of service that can execute arbitrary commands
+
+    This acts as an abstract base class for any drivers representing generic services.
+    This cannot be used on its own. If the ``__init__()`` function is overridden, it must call the ``super().__init__()`` function.
+
+    To create a Generic service (an arbitrary service that can process commands), this ``Generic`` service should be subclassed
+    and the ``do_command`` function implemented.
+
+    Example::
+
+        class ComplexService(Generic):
+
+            async def do_command(
+                self,
+                command: Mapping[str, ValueTypes],
+                *,
+                timeout: Optional[float] = None,
+                **kwargs
+            ) -> Mapping[str, ValueTypes]:
+                result = {key: False for key in command.keys()}
+                for (name, args) in command.items():
+                    if name == 'set_val':
+                        self.set_val(*args)
+                        result[name] = True
+                    if name == 'get_val':
+                        result[name] = self.val
+                    if name == 'complex_command':
+                        self.complex_command(*args)
+                        result[name] = True
+                return result
+
+            def set_val(self, val: int):
+                self.val = val
+
+            def complex_command(self, arg1, arg2, arg3):
+                ...
+
+    To execute commands, simply call the ``do_command`` function with the appropriate parameters.
+    ::
+
+        await service.do_command({'set_val': 10})
+        service.val  # 10
+        await service.do_command({'set_val': 5})
+        service.val  # 5
+    """
+
+    SUBTYPE: Final = Subtype(  # pyright: ignore [reportIncompatibleVariableOverride]
+        RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_SERVICE, "generic"
+    )

src/viam/services/generic/service.py

@@ -0,0 +1,29 @@
+from grpclib import GRPCError, Status
+from grpclib.server import Stream
+
+from viam.services.service_base import ServiceBase
+from viam.proto.common import DoCommandRequest, DoCommandResponse
+from viam.proto.service.generic import GenericServiceBase
+from viam.resource.rpc_service_base import ResourceRPCServiceBase
+from viam.utils import dict_to_struct, struct_to_dict
+
+
+class GenericRPCService(GenericServiceBase, ResourceRPCServiceBase):
+    """
+    gRPC Service for a Generic service
+    """
+
+    RESOURCE_TYPE = ServiceBase
+
+    async def DoCommand(self, stream: Stream[DoCommandRequest, DoCommandResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        service = self.get_resource(name)
+        try:
+            timeout = stream.deadline.time_remaining() if stream.deadline else None
+            result = await service.do_command(struct_to_dict(request.command), timeout=timeout, metadata=stream.metadata)
+        except NotImplementedError:
+            raise GRPCError(Status.UNIMPLEMENTED, f"``DO`` command is unimplemented for service named: {name}")
+        response = DoCommandResponse(result=dict_to_struct(result))
+        await stream.send_message(response)

tests/mocks/components.py

@@ -549,7 +549,7 @@ async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Option
         return {"command": command}
 
 
-class MockGeneric(GenericComponent):
+class MockGenericComponent(GenericComponent):
     timeout: Optional[float] = None
     geometries = GEOMETRIES
 

tests/mocks/services.py

@@ -287,6 +287,7 @@
 )
 from viam.proto.service.slam import MappingMode
 from viam.proto.service.vision import Classification, Detection
+from viam.services.generic import Generic as GenericService
 from viam.services.mlmodel import File, LabelType, Metadata, MLModel, TensorInfo
 from viam.services.mlmodel.utils import flat_tensors_to_ndarrays, ndarrays_to_flat_tensors
 from viam.services.navigation import Navigation
@@ -1390,3 +1391,11 @@ async def DeleteRegistryItem(self, stream: Stream[DeleteRegistryItemRequest, Del
 
     async def GetRegistryItem(self, stream: Stream[GetRegistryItemRequest, GetRegistryItemResponse]) -> None:
         raise NotImplementedError()
+
+
+class MockGenericService(GenericService):
+    timeout: Optional[float] = None
+
+    async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None, **kwargs) -> Mapping[str, ValueTypes]:
+        self.timeout = timeout
+        return {key: True for key in command.keys()}

tests/test_generic.py renamed to tests/test_generic_component.py

@@ -8,11 +8,11 @@
 from viam.utils import dict_to_struct, struct_to_dict
 
 from . import loose_approx
-from .mocks.components import GEOMETRIES, MockGeneric
+from .mocks.components import GEOMETRIES, MockGenericComponent
 
 
-class TestGeneric:
-    generic = MockGeneric(name="generic")
+class TestGenericComponent:
+    generic = MockGenericComponent(name="generic")
 
     @pytest.mark.asyncio
     async def test_do(self):
@@ -30,7 +30,7 @@ class TestService:
     @classmethod
     def setup_class(cls):
         cls.name = "generic"
-        cls.generic = MockGeneric(name=cls.name)
+        cls.generic = MockGenericComponent(name=cls.name)
         cls.manager = ResourceManager([cls.generic])
         cls.service = GenericRPCService(cls.manager)
 
@@ -57,7 +57,7 @@ class TestClient:
     @classmethod
     def setup_class(cls):
         cls.name = "generic"
-        cls.generic = MockGeneric(name=cls.name)
+        cls.generic = MockGenericComponent(name=cls.name)
         cls.manager = ResourceManager([cls.generic])
         cls.service = GenericRPCService(cls.manager)
 

tests/test_generic_service.py

@@ -0,0 +1,57 @@
+import pytest
+from grpclib.testing import ChannelFor
+
+from viam.components.generic import GenericClient, GenericRPCService
+from viam.proto.common import DoCommandRequest, DoCommandResponse
+from viam.proto.component.generic import GenericServiceStub
+from viam.resource.manager import ResourceManager
+from viam.utils import dict_to_struct, struct_to_dict
+
+from . import loose_approx
+from .mocks.services import MockGenericService
+
+
+class TestGenericService:
+    generic = MockGenericService(name="generic")
+
+    @pytest.mark.asyncio
+    async def test_do(self):
+        result = await self.generic.do_command({"command": "args"}, timeout=1.82)
+        assert result == {"command": True}
+        assert self.generic.timeout == loose_approx(1.82)
+
+
+class TestService:
+    @classmethod
+    def setup_class(cls):
+        cls.name = "generic"
+        cls.generic = MockGenericService(name=cls.name)
+        cls.manager = ResourceManager([cls.generic])
+        cls.service = GenericRPCService(cls.manager)
+
+    @pytest.mark.asyncio
+    async def test_do(self):
+        async with ChannelFor([self.service]) as channel:
+            client = GenericServiceStub(channel)
+            request = DoCommandRequest(name=self.name, command=dict_to_struct({"command": "args"}))
+            response: DoCommandResponse = await client.DoCommand(request, timeout=4.4)
+            result = struct_to_dict(response.result)
+            assert result == {"command": True}
+            assert self.generic.timeout == loose_approx(4.4)
+
+
+class TestClient:
+    @classmethod
+    def setup_class(cls):
+        cls.name = "generic"
+        cls.generic = MockGenericService(name=cls.name)
+        cls.manager = ResourceManager([cls.generic])
+        cls.service = GenericRPCService(cls.manager)
+
+    @pytest.mark.asyncio
+    async def test_do(self):
+        async with ChannelFor([self.service]) as channel:
+            client = GenericClient(self.name, channel)
+            result = await client.do_command({"command": "args"}, timeout=7.86)
+            assert result == {"command": True}
+            assert self.generic.timeout == loose_approx(7.86)

tests/test_motor.py

@@ -156,9 +156,9 @@ async def test_set_power(self, motor: MockMotor, service: MotorRPCService):
         async with ChannelFor([service]) as channel:
             client = MotorServiceStub(channel)
             request = SetPowerRequest(name=motor.name, power_pct=13)
-            await client.SetPower(request, timeout=1.23)
+            await client.SetPower(request, timeout=2.34)
             assert motor.power == 13
-            assert motor.timeout == loose_approx(1.23)
+            assert motor.timeout == loose_approx(2.34)
 
     @pytest.mark.asyncio
     async def test_get_position(self, motor: MockMotor, service: MotorRPCService):