Add Supervisor's OS APIs

aiohasupervisor/models/__init__.py

@@ -32,6 +32,20 @@
     HomeAssistantStopOptions,
     HomeAssistantUpdateOptions,
 )
+from aiohasupervisor.models.os import (
+    BootSlot,
+    BootSlotName,
+    DataDisk,
+    GreenInfo,
+    GreenOptions,
+    MigrateDataOptions,
+    OSInfo,
+    OSUpdate,
+    RaucState,
+    SetBootSlotOptions,
+    YellowInfo,
+    YellowOptions,
+)
 from aiohasupervisor.models.resolution import (
     Check,
     CheckOptions,
@@ -112,4 +126,16 @@
     "HomeAssistantStats",
     "HomeAssistantStopOptions",
     "HomeAssistantUpdateOptions",
+    "RaucState",
+    "BootSlotName",
+    "BootSlot",
+    "OSInfo",
+    "OSUpdate",
+    "MigrateDataOptions",
+    "DataDisk",
+    "SetBootSlotOptions",
+    "GreenInfo",
+    "GreenOptions",
+    "YellowInfo",
+    "YellowOptions",
 ]

aiohasupervisor/models/os.py

@@ -0,0 +1,126 @@
+"""Models for OS APIs."""
+
+from dataclasses import dataclass
+from enum import StrEnum
+from pathlib import PurePath
+
+from .base import Options, Request, ResponseData
+
+# --- ENUMS ----
+
+
+class RaucState(StrEnum):
+    """RaucState type."""
+
+    GOOD = "good"
+    BAD = "bad"
+    ACTIVE = "active"
+
+
+class BootSlotName(StrEnum):
+    """BootSlotName type."""
+
+    A = "A"
+    B = "B"
+
+
+# --- OBJECTS ----
+
+
+@dataclass(frozen=True, slots=True)
+class BootSlot(ResponseData):
+    """BootSlot model."""
+
+    state: str
+    status: RaucState | None
+    version: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class OSInfo(ResponseData):
+    """OSInfo model."""
+
+    version: str | None
+    version_latest: str | None
+    update_available: bool
+    board: str | None
+    boot: str | None
+    data_disk: str | None
+    boot_slots: dict[str, BootSlot]
+
+
+@dataclass(frozen=True, slots=True)
+class OSUpdate(Request):
+    """OSUpdate model."""
+
+    version: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class MigrateDataOptions(Request):
+    """MigrateDataOptions model."""
+
+    device: str
+
+
+@dataclass(frozen=True, slots=True)
+class DataDisk(ResponseData):
+    """DataDisk model."""
+
+    name: str
+    vendor: str
+    model: str
+    serial: str
+    size: int
+    id: str
+    dev_path: PurePath
+
+
+@dataclass(frozen=True, slots=True)
+class DataDiskList(ResponseData):
+    """DataDiskList model."""
+
+    disks: list[DataDisk]
+
+
+@dataclass(frozen=True, slots=True)
+class SetBootSlotOptions(Request):
+    """SetBootSlotOptions model."""
+
+    boot_slot: BootSlotName
+
+
+@dataclass(frozen=True, slots=True)
+class GreenInfo(ResponseData):
+    """GreenInfo model."""
+
+    activity_led: bool
+    power_led: bool
+    system_health_led: bool
+
+
+@dataclass(frozen=True, slots=True)
+class GreenOptions(Options):
+    """GreenOptions model."""
+
+    activity_led: bool | None = None
+    power_led: bool | None = None
+    system_health_led: bool | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class YellowInfo(ResponseData):
+    """YellowInfo model."""
+
+    disk_led: bool
+    heartbeat_led: bool
+    power_led: bool
+
+
+@dataclass(frozen=True, slots=True)
+class YellowOptions(Options):
+    """YellowOptions model."""
+
+    disk_led: bool | None = None
+    heartbeat_led: bool | None = None
+    power_led: bool | None = None

aiohasupervisor/os.py

@@ -0,0 +1,69 @@
+"""OS client for supervisor."""
+
+from .client import _SupervisorComponentClient
+from .models.os import (
+    DataDisk,
+    DataDiskList,
+    GreenInfo,
+    GreenOptions,
+    MigrateDataOptions,
+    OSInfo,
+    OSUpdate,
+    SetBootSlotOptions,
+    YellowInfo,
+    YellowOptions,
+)
+
+
+class OSClient(_SupervisorComponentClient):
+    """Handles OS access in supervisor."""
+
+    async def info(self) -> OSInfo:
+        """Get OS info."""
+        result = await self._client.get("os/info")
+        return OSInfo.from_dict(result.data)
+
+    async def update(self, options: OSUpdate | None = None) -> None:
+        """Update OS."""
+        await self._client.post(
+            "os/update", json=options.to_dict() if options else None
+        )
+
+    async def config_sync(self) -> None:
+        """Trigger config reload on OS."""
+        await self._client.post("os/config/sync")
+
+    async def migrate_data(self, options: MigrateDataOptions) -> None:
+        """Migrate data to new data disk and reboot."""
+        await self._client.post("os/datadisk/move", json=options.to_dict())
+
+    async def list_data_disks(self) -> list[DataDisk]:
+        """Get all data disks."""
+        result = await self._client.get("os/datadisk/list")
+        return DataDiskList.from_dict(result.data).disks
+
+    async def wipe_data(self) -> None:
+        """Trigger data disk wipe on host and reboot."""
+        await self._client.post("os/datadisk/wipe")
+
+    async def set_boot_slot(self, options: SetBootSlotOptions) -> None:
+        """Change active boot slot on host and reboot."""
+        await self._client.post("os/boot-slot", json=options.to_dict())
+
+    async def green_info(self) -> GreenInfo:
+        """Get info for green board (if in use)."""
+        result = await self._client.get("os/boards/green")
+        return GreenInfo.from_dict(result.data)
+
+    async def green_options(self, options: GreenOptions) -> None:
+        """Set options for green board (if in use)."""
+        await self._client.post("os/boards/green", json=options.to_dict())
+
+    async def yellow_info(self) -> YellowInfo:
+        """Get info for yellow board (if in use)."""
+        result = await self._client.get("os/boards/yellow")
+        return YellowInfo.from_dict(result.data)
+
+    async def yellow_options(self, options: YellowOptions) -> None:
+        """Set options for yellow board (if in use)."""
+        await self._client.post("os/boards/yellow", json=options.to_dict())

aiohasupervisor/root.py

@@ -8,6 +8,7 @@
 from .client import _SupervisorClient
 from .homeassistant import HomeAssistantClient
 from .models.root import AvailableUpdate, AvailableUpdates, RootInfo
+from .os import OSClient
 from .resolution import ResolutionClient
 from .store import StoreClient
 from .supervisor import SupervisorManagementClient
@@ -26,6 +27,7 @@ def __init__(
         """Initialize client."""
         self._client = _SupervisorClient(api_host, token, request_timeout, session)
         self._addons = AddonsClient(self._client)
+        self._os = OSClient(self._client)
         self._resolution = ResolutionClient(self._client)
         self._store = StoreClient(self._client)
         self._supervisor = SupervisorManagementClient(self._client)
@@ -41,6 +43,11 @@ def homeassistant(self) -> HomeAssistantClient:
         """Get Home Assistant component client."""
         return self._homeassistant
 
+    @property
+    def os(self) -> OSClient:
+        """Get OS component client."""
+        return self._os
+
     @property
     def resolution(self) -> ResolutionClient:
         """Get resolution center component client."""

tests/fixtures/os_datadisk_list.json

@@ -0,0 +1,17 @@
+{
+  "result": "ok",
+  "data": {
+    "devices": ["SSK-SSK-Storage-DF123"],
+    "disks": [
+      {
+        "name": "SSK SSK Storage (DF123)",
+        "vendor": "SSK",
+        "model": "SSK Storage",
+        "serial": "DF123",
+        "size": 250059350016,
+        "id": "SSK-SSK-Storage-DF123",
+        "dev_path": "/dev/sda"
+      }
+    ]
+  }
+}

tests/fixtures/os_green_info.json

@@ -0,0 +1,4 @@
+{
+  "result": "ok",
+  "data": { "activity_led": true, "power_led": true, "system_health_led": true }
+}

tests/fixtures/os_info.json

@@ -0,0 +1,19 @@
+{
+  "result": "ok",
+  "data": {
+    "version": "13.0",
+    "version_latest": "13.1",
+    "update_available": true,
+    "board": "odroid-n2",
+    "boot": "B",
+    "data_disk": "BJTD4R-0xaabbccdd",
+    "boot_slots": {
+      "A": { "state": "inactive", "status": "good", "version": null },
+      "B": {
+        "state": "booted",
+        "status": "good",
+        "version": "13.0"
+      }
+    }
+  }
+}

tests/fixtures/os_yellow_info.json

@@ -0,0 +1,4 @@
+{
+  "result": "ok",
+  "data": { "disk_led": true, "heartbeat_led": true, "power_led": true }
+}