RSDK-4056 Add PowerSensor (#374)

src/viam/components/power_sensor/__init__.py

@@ -0,0 +1,17 @@
+from viam.resource.registry import Registry, ResourceRegistration
+
+from .client import PowerSensorClient
+from .power_sensor import PowerSensor
+from .service import PowerSensorRPCService
+
+__all__ = [
+    "PowerSensor",
+]
+
+Registry.register_subtype(
+    ResourceRegistration(
+        PowerSensor,
+        PowerSensorRPCService,
+        lambda name, channel: PowerSensorClient(name, channel),
+    )
+)

src/viam/components/power_sensor/client.py

@@ -0,0 +1,57 @@
+from typing import Any, Dict, Mapping, Optional, Tuple
+
+from grpclib.client import Channel
+
+from viam.components.power_sensor.power_sensor import PowerSensor
+from viam.proto.common import DoCommandRequest, DoCommandResponse
+from viam.proto.component.powersensor import (
+    GetVoltageRequest,
+    GetVoltageResponse,
+    GetCurrentRequest,
+    GetCurrentResponse,
+    GetPowerRequest,
+    GetPowerResponse,
+    PowerSensorServiceStub,
+)
+from viam.resource.rpc_client_base import ReconfigurableResourceRPCClientBase
+from viam.utils import ValueTypes, dict_to_struct, struct_to_dict
+
+
+class PowerSensorClient(PowerSensor, ReconfigurableResourceRPCClientBase):
+    """gRPC client for the PowerSensor component."""
+
+    def __init__(self, name: str, channel: Channel):
+        self.channel = channel
+        self.client = PowerSensorServiceStub(channel)
+        super().__init__(name)
+
+    async def get_voltage(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> Tuple[float, bool]:
+        if extra is None:
+            extra = {}
+        request = GetVoltageRequest(name=self.name, extra=dict_to_struct(extra))
+        response: GetVoltageResponse = await self.client.GetVoltage(request, timeout=timeout)
+        return response.volts, response.is_ac
+
+    async def get_current(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> Tuple[float, bool]:
+        if extra is None:
+            extra = {}
+        request = GetCurrentRequest(name=self.name, extra=dict_to_struct(extra))
+        response: GetCurrentResponse = await self.client.GetCurrent(request, timeout=timeout)
+        return response.amperes, response.is_ac
+
+    async def get_power(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> float:
+        if extra is None:
+            extra = {}
+        request = GetPowerRequest(name=self.name, extra=dict_to_struct(extra))
+        response: GetPowerResponse = await self.client.GetPower(request, timeout=timeout)
+        return response.watts
+
+    async def get_readings(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None) -> Mapping[str, Any]:
+        if extra is None:
+            extra = {}
+        return await super().get_readings(extra=extra, timeout=timeout)
+
+    async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None) -> Mapping[str, ValueTypes]:
+        request = DoCommandRequest(name=self.name, command=dict_to_struct(command))
+        response: DoCommandResponse = await self.client.DoCommand(request, timeout=timeout)
+        return struct_to_dict(response.result)

src/viam/components/power_sensor/power_sensor.py

@@ -0,0 +1,92 @@
+import abc
+import asyncio
+from typing import Any, Dict, Final, List, Mapping, Optional, Tuple
+
+from grpclib import GRPCError
+
+from viam.errors import MethodNotImplementedError, NotSupportedError
+from viam.resource.types import RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT, Subtype
+
+from ..sensor import Sensor
+
+
+class PowerSensor(Sensor):
+    """PowerSensor reports information about voltage, current and power.
+
+    This acts as an abstract base class for any sensors that can provide data regarding voltage, current and/or power.
+    This cannot be used on its own. If the ``__init__()`` function is overridden, it must call the ``super().__init__()`` function.
+    """
+
+    SUBTYPE: Final = Subtype(RESOURCE_NAMESPACE_RDK, RESOURCE_TYPE_COMPONENT, "power_sensor")
+
+    @abc.abstractmethod
+    async def get_voltage(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Tuple[float, bool]:
+        """Get the voltage reading and bool IsAC
+
+        Returns:
+            Tuple[float, bool]: voltage (volts) and bool IsAC
+        """
+        ...
+
+    @abc.abstractmethod
+    async def get_current(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Tuple[float, bool]:
+        """Get the current reading and bool IsAC
+
+        Returns:
+            Tuple[float, bool]: current (amperes) and bool IsAC
+        """
+        ...
+
+    @abc.abstractmethod
+    async def get_power(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> float:
+        """Get the power reading in watts
+
+        Returns:
+            float: power in watts
+        """
+        ...
+
+    async def get_readings(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Mapping[str, Any]:
+        """Obtain the measurements/data specific to this sensor.
+        If a sensor is not configured to have a measurement or fails to read a piece of data, it will not appear in the readings dictionary.
+
+        Returns:
+            Mapping[str, Any]: The readings for the PowerSensor:
+            {
+               voltage: float
+               current: float
+               is_ac: bool
+               power: float
+            }
+        """
+        (vol, cur, pow) = await asyncio.gather(
+            self.get_voltage(extra=extra, timeout=timeout),
+            self.get_current(extra=extra, timeout=timeout),
+            self.get_power(extra=extra, timeout=timeout),
+            return_exceptions=True,
+        )
+
+        readings = {}
+
+        # Add returned value to the readings dictionary if value is of expected type; omit if unimplemented.
+        def add_reading(name: str, reading, returntype: List) -> None:
+            possible_error_types = (NotImplementedError, MethodNotImplementedError, NotSupportedError)
+            if type(reading) in returntype:
+                if name == "voltage":
+                    readings["voltage"] = reading[0]
+                    readings["is_ac"] = reading[1]
+                elif name == "current":
+                    readings["current"] = reading[0]
+                    readings["is_ac"] = reading[1]
+                else:
+                    readings[name] = reading
+                return
+            elif isinstance(reading, possible_error_types) or (isinstance(reading, GRPCError) and "Unimplemented" in str(reading.message)):
+                return
+            raise reading
+
+        add_reading("voltage", vol, [tuple])
+        add_reading("current", cur, [tuple])
+        add_reading("power", pow, [float])
+
+        return readings

src/viam/components/power_sensor/service.py

@@ -0,0 +1,62 @@
+from grpclib.server import Stream
+
+from viam.components.power_sensor.power_sensor import PowerSensor
+from viam.proto.common import DoCommandRequest, DoCommandResponse
+from viam.proto.component.powersensor import (
+    GetVoltageRequest,
+    GetVoltageResponse,
+    GetCurrentRequest,
+    GetCurrentResponse,
+    GetPowerRequest,
+    GetPowerResponse,
+    PowerSensorServiceBase,
+)
+from viam.resource.rpc_service_base import ResourceRPCServiceBase
+from viam.utils import dict_to_struct, struct_to_dict
+
+
+class PowerSensorRPCService(PowerSensorServiceBase, ResourceRPCServiceBase):
+    """
+    gRPC Service for a PowerSensor
+    """
+
+    RESOURCE_TYPE = PowerSensor
+
+    async def GetVoltage(self, stream: Stream[GetVoltageRequest, GetVoltageResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        sensor = self.get_resource(name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        voltage, is_ac = await sensor.get_voltage(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        response = GetVoltageResponse(volts=voltage, is_ac=is_ac)
+        await stream.send_message(response)
+
+    async def GetCurrent(self, stream: Stream[GetCurrentRequest, GetCurrentResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        sensor = self.get_resource(name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        current, is_ac = await sensor.get_current(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        response = GetCurrentResponse(amperes=current, is_ac=is_ac)
+        await stream.send_message(response)
+
+    async def GetPower(self, stream: Stream[GetPowerRequest, GetPowerResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        name = request.name
+        sensor = self.get_resource(name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        power = await sensor.get_power(extra=struct_to_dict(request.extra), timeout=timeout, metadata=stream.metadata)
+        response = GetPowerResponse(watts=power)
+        await stream.send_message(response)
+
+    async def DoCommand(self, stream: Stream[DoCommandRequest, DoCommandResponse]) -> None:
+        request = await stream.recv_message()
+        assert request is not None
+        sensor = self.get_resource(request.name)
+        timeout = stream.deadline.time_remaining() if stream.deadline else None
+        result = await sensor.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

@@ -28,6 +28,7 @@
 from viam.components.motor import Motor
 from viam.components.movement_sensor import MovementSensor
 from viam.components.pose_tracker import PoseTracker
+from viam.components.power_sensor import PowerSensor
 from viam.components.sensor import Sensor
 from viam.components.servo import Servo
 from viam.errors import ResourceNotFoundError
@@ -858,6 +859,35 @@ async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Option
         return {"command": command}
 
 
+class MockPowerSensor(PowerSensor):
+    def __init__(self, name: str, voltage: float, current: float, is_ac: bool, power: float):
+        super().__init__(name)
+        self.voltage = voltage
+        self.current = current
+        self.is_ac = is_ac
+        self.power = power
+        self.extra: Optional[Dict[str, Any]] = None
+        self.timeout: Optional[float] = None
+
+    async def get_voltage(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Tuple[float, bool]:
+        self.extra = extra
+        self.timeout = timeout
+        return (self.voltage, self.is_ac)
+
+    async def get_current(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> Tuple[float, bool]:
+        self.extra = extra
+        self.timeout = timeout
+        return (self.current, self.is_ac)
+
+    async def get_power(self, *, extra: Optional[Dict[str, Any]] = None, timeout: Optional[float] = None, **kwargs) -> float:
+        self.extra = extra
+        self.timeout = timeout
+        return self.power
+
+    async def do_command(self, command: Mapping[str, ValueTypes], *, timeout: Optional[float] = None, **kwargs) -> Mapping[str, ValueTypes]:
+        return {"command": command}
+
+
 class MockSensor(Sensor):
     def __init__(self, name: str, result: Mapping[str, Any] = {"a": 0, "b": {"foo": "bar"}, "c": [1, 8, 2], "d": "Hello world!"}):
         self.readings = result