feat: send device metadata

lib/py_edge_device/carlos/edge/device/config.py

@@ -11,7 +11,11 @@
 from typing import TypeVar
 
 import yaml
-from carlos.edge.interface.device import CarlosDriver, DriverConfig, DriverFactory
+from carlos.edge.interface.device import CarlosDriver, DriverFactory
+from carlos.edge.interface.device.driver_config import (
+    DriverConfigWithDirection,
+    DriverDirection,
+)
 from loguru import logger
 from pydantic import BaseModel
 
@@ -65,9 +69,10 @@ def load_drivers(config_dir: Path | None = None) -> list[CarlosDriver]:
         driver_configs.insert(
             0,
             factory.build(
-                DriverConfig(
+                DriverConfigWithDirection(
                     identifier="__device_metrics__",
                     driver_module=DeviceMetrics.__module__,
+                    direction=DriverDirection.INPUT,
                 ).model_dump()
             ),
         )

lib/py_edge_device/carlos/edge/device/driver/_dhtxx.py

@@ -5,6 +5,8 @@
 from time import sleep
 
 from carlos.edge.interface.device import AnalogInput, DriverDirection, GpioDriverConfig
+from carlos.edge.interface.device.driver_config import DriverSignal
+from carlos.edge.interface.units import UnitOfMeasurement
 from pydantic import Field
 
 from carlos.edge.device.protocol import GPIO
@@ -124,13 +126,30 @@ def read(self) -> tuple[float, float]:
 class DHTXX(AnalogInput, ABC):
     """DHTXX Temperature and Humidity Sensor."""
 
+    TEMP_SIGNAL_ID = "temperature"
+    HUMIDITY_SIGNAL_ID = "humidity"
+
     def __init__(self, config: GpioDriverConfig):
 
         super().__init__(config=config)
 
         self._dht: DHT | None = None
         self._dht_type: DHTType | None = None
 
+    def signals(self) -> list[DriverSignal]:
+        """Returns the signals of the DHT sensor."""
+
+        return [
+            DriverSignal(
+                signal_identifier=self.TEMP_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.CELSIUS,
+            ),
+            DriverSignal(
+                signal_identifier=self.HUMIDITY_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.HUMIDITY_PERCENTAGE,
+            ),
+        ]
+
     def setup(self):
         """Sets up the DHT11 sensor."""
 
@@ -148,8 +167,8 @@ def read(self) -> dict[str, float]:
             try:
                 temperature, humidity = self._dht.read()
                 return {
-                    "temperature": temperature,
-                    "humidity": humidity,
+                    self.TEMP_SIGNAL_ID: temperature,
+                    self.HUMIDITY_SIGNAL_ID: humidity,
                 }
             except RuntimeError as ex:
                 last_error = ex

lib/py_edge_device/carlos/edge/device/driver/device_metrics.py

@@ -1,25 +1,57 @@
 import psutil
-from carlos.edge.interface.device import AnalogInput, DriverConfig, DriverFactory
+from carlos.edge.interface.device import AnalogInput, DriverFactory
+from carlos.edge.interface.device.driver_config import (
+    DriverConfigWithDirection,
+    DriverSignal,
+)
+from carlos.edge.interface.units import UnitOfMeasurement
 
 
 class DeviceMetrics(AnalogInput):
     """Provides the metrics of the device."""
 
-    def __init__(self, config: DriverConfig):
+    _CPU_LOAD_SIGNAL_ID = "cpu.load_percent"
+    _CPU_TEMP_SIGNAL_ID = "cpu.temperature"
+    _MEMORY_USAGE_SIGNAL_ID = "memory.usage_percent"
+    _DISK_USAGE_SIGNAL_ID = "disk.usage_percent"
+
+    def __init__(self, config: DriverConfigWithDirection):
 
         super().__init__(config=config)
 
+    def signals(self) -> list[DriverSignal]:
+        """Returns the signals of the DHT sensor."""
+
+        return [
+            DriverSignal(
+                signal_identifier=self._CPU_LOAD_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.PERCENTAGE,
+            ),
+            DriverSignal(
+                signal_identifier=self._CPU_TEMP_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.CELSIUS,
+            ),
+            DriverSignal(
+                signal_identifier=self._MEMORY_USAGE_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.PERCENTAGE,
+            ),
+            DriverSignal(
+                signal_identifier=self._DISK_USAGE_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.PERCENTAGE,
+            ),
+        ]
+
     def setup(self):
         pass
 
     def read(self) -> dict[str, float]:
         """Reads the device metrics."""
 
         return {
-            "cpu.load_percent": psutil.cpu_percent(interval=1.0),
-            "cpu.temperature": self._read_cpu_temp(),
-            "memory.usage_percent": psutil.virtual_memory().percent,
-            "disk.usage_percent": psutil.disk_usage("/").percent,
+            self._CPU_LOAD_SIGNAL_ID: psutil.cpu_percent(interval=1.0),
+            self._CPU_TEMP_SIGNAL_ID: self._read_cpu_temp(),
+            self._MEMORY_USAGE_SIGNAL_ID: psutil.virtual_memory().percent,
+            self._DISK_USAGE_SIGNAL_ID: psutil.disk_usage("/").percent,
         }
 
     @staticmethod
@@ -33,5 +65,5 @@ def _read_cpu_temp() -> float:
 
 
 DriverFactory().register(
-    driver_module=__name__, config=DriverConfig, factory=DeviceMetrics
+    driver_module=__name__, config=DriverConfigWithDirection, factory=DeviceMetrics
 )

lib/py_edge_device/carlos/edge/device/driver/relay.py

@@ -7,6 +7,8 @@
     DriverFactory,
     GpioDriverConfig,
 )
+from carlos.edge.interface.device.driver_config import DriverSignal
+from carlos.edge.interface.units import UnitOfMeasurement
 from pydantic import Field
 
 from carlos.edge.device.protocol import GPIO
@@ -20,11 +22,23 @@ class RelayConfig(GpioDriverConfig):
 class Relay(DigitalOutput, DigitalInput):
     """Relay."""
 
+    _STATE_SIGNAL_ID = "state"
+
     def __init__(self, config: RelayConfig):
         super().__init__(config=config)
 
         self._state = False
 
+    def signals(self) -> list[DriverSignal]:
+        """Returns the signals of the DHT sensor."""
+
+        return [
+            DriverSignal(
+                signal_identifier=self._STATE_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.UNIT_LESS,
+            ),
+        ]
+
     def setup(self):
 
         # HIGH means off, LOW means
@@ -42,27 +56,27 @@ def set(self, value: bool):
     def read(self) -> dict[str, bool]:
         """Reads the value of the relay."""
 
-        return {"state": self._state}
+        return {self._STATE_SIGNAL_ID: self._state}
 
     def test(self):
         """Tests the relay by reading the value."""
 
         self.set(False)
         time.sleep(0.01)
         state = self.read()
-        if state["state"]:
+        if state[self._STATE_SIGNAL_ID]:
             raise ValueError(f"Value of relay was not set to false. Got: {state}")
 
         self.set(True)
         time.sleep(1)
         state = self.read()
-        if not state["state"]:
+        if not state[self._STATE_SIGNAL_ID]:
             raise ValueError(f"Value of relay was not set to true. Got: {state}")
 
         self.set(False)
         time.sleep(0.01)
         state = self.read()
-        if state["state"]:
+        if state[self._STATE_SIGNAL_ID]:
             raise ValueError(f"Value of relay was not set to false. Got: {state}")
 
 

lib/py_edge_device/carlos/edge/device/driver/sht30.py

@@ -7,6 +7,8 @@
     DriverFactory,
     I2cDriverConfig,
 )
+from carlos.edge.interface.device.driver_config import DriverSignal
+from carlos.edge.interface.units import UnitOfMeasurement
 from pydantic import Field
 
 from carlos.edge.device.protocol import I2C
@@ -37,6 +39,9 @@ class SHT30(AnalogInput):
     PARAM_HIGH_REPEATABLITY = 0x06
     """Marks the measurement as high repeatability."""
 
+    _TEMPERATURE_SIGNAL_ID = "temperature"
+    _HUMIDITY_SIGNAL_ID = "humidity"
+
     def __init__(self, config: SHT30Config):
         if config.address_int not in SHT30.I2C_ADDRESSES:
             raise ValueError(
@@ -48,6 +53,20 @@ def __init__(self, config: SHT30Config):
 
         self._i2c: I2C | None = None
 
+    def signals(self) -> list[DriverSignal]:
+        """Returns the signals of the DHT sensor."""
+
+        return [
+            DriverSignal(
+                signal_identifier=self._TEMPERATURE_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.CELSIUS,
+            ),
+            DriverSignal(
+                signal_identifier=self._HUMIDITY_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.HUMIDITY_PERCENTAGE,
+            ),
+        ]
+
     def setup(self):
         self._i2c = I2C(address=self.config.address_int)
 
@@ -57,8 +76,8 @@ def read(self) -> dict[str, float]:
         read_delay_ms = 100
         humidity, temperature = self._get_measurement(read_delay_ms=read_delay_ms)
         return {
-            "temperature": float(temperature),
-            "humidity": float(humidity),
+            self._TEMPERATURE_SIGNAL_ID: float(temperature),
+            self._HUMIDITY_SIGNAL_ID: float(humidity),
         }
 
     def _get_measurement(self, read_delay_ms: int = 100) -> tuple[float, float]:

lib/py_edge_device/carlos/edge/device/driver/si1145.py

@@ -7,6 +7,8 @@
     DriverFactory,
     I2cDriverConfig,
 )
+from carlos.edge.interface.device.driver_config import DriverSignal
+from carlos.edge.interface.units import UnitOfMeasurement
 from pydantic import Field
 
 from carlos.edge.device.protocol import I2C
@@ -21,6 +23,12 @@ class Si1145Config(I2cDriverConfig):
 
 class SI1145(AnalogInput):
 
+    _VISUAL_LIGHT_SIGNAL_ID = "visual-light"
+    _VISUAL_LIGHT_RAW_SIGNAL_ID = "visual-light-raw"
+    _INFRARED_LIGHT_SIGNAL_ID = "infrared-light"
+    _INFRARED_LIGHT_RAW_SIGNAL_ID = "infrared-light-raw"
+    _UV_INDEX_SIGNAL_ID = "uv-index"
+
     def __init__(self, config: Si1145Config):
 
         if config.address_int != SDL_Pi_SI1145.ADDR:
@@ -32,6 +40,32 @@ def __init__(self, config: Si1145Config):
 
         self._si1145: SDL_Pi_SI1145 | None = None
 
+    def signals(self) -> list[DriverSignal]:
+        """Returns the signals of the DHT sensor."""
+
+        return [
+            DriverSignal(
+                signal_identifier=self._VISUAL_LIGHT_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.LUX,
+            ),
+            DriverSignal(
+                signal_identifier=self._VISUAL_LIGHT_RAW_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.UNIT_LESS,
+            ),
+            DriverSignal(
+                signal_identifier=self._INFRARED_LIGHT_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.LUX,
+            ),
+            DriverSignal(
+                signal_identifier=self._INFRARED_LIGHT_RAW_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.UNIT_LESS,
+            ),
+            DriverSignal(
+                signal_identifier=self._UV_INDEX_SIGNAL_ID,
+                unit_of_measurement=UnitOfMeasurement.UNIT_LESS,
+            ),
+        ]
+
     def setup(self):
 
         self._si1145 = SDL_Pi_SI1145()
@@ -48,11 +82,11 @@ def read(self) -> dict[str, float]:
         uv_idx = self._si1145.read_uv_index()
 
         return {
-            "visual-light-raw": float(vis_raw),
-            "visual-light": float(vis_lux),
-            "infrared-light-raw": float(ir_raw),
-            "infrared-light": float(ir_lux),
-            "uv-index": float(uv_idx),
+            self._VISUAL_LIGHT_RAW_SIGNAL_ID: float(vis_raw),
+            self._VISUAL_LIGHT_SIGNAL_ID: float(vis_lux),
+            self._INFRARED_LIGHT_RAW_SIGNAL_ID: float(ir_raw),
+            self._INFRARED_LIGHT_SIGNAL_ID: float(ir_lux),
+            self._UV_INDEX_SIGNAL_ID: float(uv_idx),
         }