Add 1-wire and modbus support

Author: adamberkes

robopipe_api/models/modbus.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel, Field
+
+from typing import Annotated
+
+from .device import Device
+
+
+class Modbus(Device):
+    value: Annotated[int, Field(ge=0, le=(2**16 - 1))]
+    save: bool = True
+
+
+class ModbusUpdate(BaseModel):
+    value: float

robopipe_api/models/ow.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel
+
+from typing import Literal
+
+from .device import Device
+
+
+class OWPower(Device):
+    value: Literal[0, 1]
+
+
+class OWPowerUpdate(BaseModel):
+    value: Literal[0, 1]
+
+
+class OWBus(Device):
+    bus: str
+    interval: int = 10
+    do_scan: bool
+    do_reset: bool
+    scan_interval: int = 60
+
+
+class OWBusUpdate(BaseModel):
+    interval: int = 10
+    do_scan: bool = False
+    do_reset: bool = False
+    scan_interval: int = 60

robopipe_api/routers/controller.py

@@ -1,14 +1,17 @@
 from fastapi import APIRouter
+from pydantic import BaseModel
 
 from dataclasses import asdict
 from typing import Any, TypeVar
 
-from ..controller.devices import AI, AO, DI, DO, LED
+from ..controller.devices import AI, AO, DI, DO, LED, OWPOWER, OWBUS, REGISTER
 from ..models.analog_input import AnalogInput, AnalogInputUpdate
 from ..models.analog_output import AnalogOutput, AnalogOutputUpdate
 from ..models.digital_input import DigitalInput, DigitalInputUpdate
 from ..models.digital_output import DigitalOutput, DigitalOutputUpdate
 from ..models.led import Led, LedUpdate
+from ..models.modbus import Modbus, ModbusUpdate
+from ..models.ow import OWPower, OWPowerUpdate, OWBus, OWBusUpdate
 from .common import DevicesDep, DeviceType, DEVICE_TYPES, DeviceDep
 
 router = APIRouter(prefix="/controller", tags=["controller"])
@@ -32,42 +35,46 @@ def list_all_circuits_of_device_type(devices: DevicesDep, device_type: DeviceTyp
 S = TypeVar("S")
 
 
-def create_endpoints(
-    device_type: str,
-    output_type: S,
-    input_type: T,
-):
+def create_endpoints(device_type: str, output_type: S, input_type: T):
     async def get_all(devices: DevicesDep) -> list[output_type]:
         return [device.full() for device in devices.by_name(device_type)]
 
     async def set_all(devices: DevicesDep, body: input_type) -> list[output_type]:
         return [
-            await device.set(**asdict(body)) for device in devices.by_name(device_type)
+            await device.set(**body.model_dump())
+            for device in devices.by_name(device_type)
         ]
 
     async def get(device: DeviceDep) -> output_type:
         return device.full()
 
     async def set(device: DeviceDep, body: input_type) -> output_type:
-        return await device.set(**asdict(body))
+        return await device.set(**body.model_dump())
 
     return [get_all, set_all, get, set]
 
 
 DEVICE_ENDPOINTS = [
-    [AI, AnalogInput, AnalogInputUpdate, "analog input"],
-    [AO, AnalogOutput, AnalogOutputUpdate, "analog output"],
-    [DI, DigitalInput, DigitalInputUpdate, "digital input"],
-    [DO, DigitalOutput, DigitalOutputUpdate, "digital output"],
-    [LED, Led, LedUpdate, "led"],
+    [AI, AnalogInput, AnalogInputUpdate, "analog input", None],
+    [AO, AnalogOutput, AnalogOutputUpdate, "analog output", None],
+    [DI, DigitalInput, DigitalInputUpdate, "digital input", None],
+    [DO, DigitalOutput, DigitalOutputUpdate, "digital output", None],
+    [LED, Led, LedUpdate, "led", None],
+    [OWPOWER, OWPower, OWPowerUpdate, "one-wire power supply", "supplies"],
+    [OWBUS, OWBus, OWBusUpdate, "one-wire bus", "buses"],
+    [REGISTER, Modbus, ModbusUpdate, "modbus register", None],
 ]
 
 
-def register_device_endpoints(endpoints: list[tuple[str, Any, Any, str]]):
+def register_device_endpoints(endpoints: list[tuple[str, Any, Any, str, str | None]]):
     for endpoint in endpoints:
-        path, out, inp, name = endpoint
+        path, out, inp, name, last_plural = endpoint
         get_all, set_all, get, set = create_endpoints(path, out, inp)
-        plural_name = name + "s"
+        plural_name = (
+            name + "s"
+            if last_plural is None
+            else " ".join([*name.split(" ")[:-1], last_plural])
+        )
 
         router.add_api_route(
             rf"/{path}", get_all, name=f"List all {plural_name}", methods=["GET"]