Introduce the OpenRGB integration (home-assistant#153373)

Co-authored-by: Copilot <[email protected]>
Co-authored-by: Manu <[email protected]>
Co-authored-by: Joost Lekkerkerker <[email protected]>
Co-authored-by: Norbert Rittel <[email protected]>

Author: 5 people

CODEOWNERS

@@ -0,0 +1,50 @@
+"""The OpenRGB integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import CONF_NAME, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+
+from .const import DOMAIN
+from .coordinator import OpenRGBConfigEntry, OpenRGBCoordinator
+
+PLATFORMS: list[Platform] = [Platform.LIGHT]
+
+
+def _setup_server_device_registry(
+    hass: HomeAssistant, entry: OpenRGBConfigEntry, coordinator: OpenRGBCoordinator
+):
+    """Set up device registry for the OpenRGB SDK server."""
+    device_registry = dr.async_get(hass)
+
+    # Create the parent OpenRGB SDK server device
+    device_registry.async_get_or_create(
+        config_entry_id=entry.entry_id,
+        identifiers={(DOMAIN, entry.entry_id)},
+        name=entry.data[CONF_NAME],
+        model="OpenRGB SDK Server",
+        manufacturer="OpenRGB",
+        sw_version=coordinator.get_client_protocol_version(),
+        entry_type=dr.DeviceEntryType.SERVICE,
+    )
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: OpenRGBConfigEntry) -> bool:
+    """Set up OpenRGB from a config entry."""
+    coordinator = OpenRGBCoordinator(hass, entry)
+
+    await coordinator.async_config_entry_first_refresh()
+
+    _setup_server_device_registry(hass, entry, coordinator)
+
+    entry.runtime_data = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: OpenRGBConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/openrgb/__init__.py

@@ -0,0 +1,81 @@
+"""Config flow for the OpenRGB integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from openrgb import OpenRGBClient
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_validation as cv
+
+from .const import CONNECTION_ERRORS, DEFAULT_CLIENT_NAME, DEFAULT_PORT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_NAME): str,
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, host: str, port: int) -> None:
+    """Validate the user input allows us to connect."""
+
+    def _try_connect(host: str, port: int) -> None:
+        client = OpenRGBClient(host, port, DEFAULT_CLIENT_NAME)
+        client.disconnect()
+
+    await hass.async_add_executor_job(_try_connect, host, port)
+
+
+class OpenRGBConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for OpenRGB."""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            name = user_input[CONF_NAME]
+            host = user_input[CONF_HOST]
+            port = user_input[CONF_PORT]
+
+            # Prevent duplicate entries
+            self._async_abort_entries_match({CONF_HOST: host, CONF_PORT: port})
+
+            try:
+                await validate_input(self.hass, host, port)
+            except CONNECTION_ERRORS:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception(
+                    "Unknown error while connecting to OpenRGB SDK server at %s",
+                    f"{host}:{port}",
+                )
+                errors["base"] = "unknown"
+            else:
+                return self.async_create_entry(
+                    title=name,
+                    data={
+                        CONF_NAME: name,
+                        CONF_HOST: host,
+                        CONF_PORT: port,
+                    },
+                )
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=self.add_suggested_values_to_schema(
+                data_schema=STEP_USER_DATA_SCHEMA, suggested_values=user_input
+            ),
+            errors=errors,
+        )

homeassistant/components/openrgb/config_flow.py

@@ -0,0 +1,65 @@
+"""Constants for the OpenRGB integration."""
+
+from datetime import timedelta
+from enum import StrEnum
+import socket
+
+from openrgb.utils import (
+    ControllerParsingError,
+    DeviceType,
+    OpenRGBDisconnected,
+    SDKVersionError,
+)
+
+DOMAIN = "openrgb"
+
+# Defaults
+DEFAULT_PORT = 6742
+DEFAULT_CLIENT_NAME = "Home Assistant"
+
+# Update interval
+SCAN_INTERVAL = timedelta(seconds=15)
+
+DEFAULT_COLOR = (255, 255, 255)
+DEFAULT_BRIGHTNESS = 255
+OFF_COLOR = (0, 0, 0)
+
+
+class OpenRGBMode(StrEnum):
+    """OpenRGB modes."""
+
+    OFF = "Off"
+    STATIC = "Static"
+    DIRECT = "Direct"
+    CUSTOM = "Custom"
+
+
+EFFECT_OFF_OPENRGB_MODES = {OpenRGBMode.STATIC, OpenRGBMode.DIRECT, OpenRGBMode.CUSTOM}
+
+DEVICE_TYPE_ICONS: dict[DeviceType, str] = {
+    DeviceType.MOTHERBOARD: "mdi:developer-board",
+    DeviceType.DRAM: "mdi:memory",
+    DeviceType.GPU: "mdi:expansion-card",
+    DeviceType.COOLER: "mdi:fan",
+    DeviceType.LEDSTRIP: "mdi:led-variant-on",
+    DeviceType.KEYBOARD: "mdi:keyboard",
+    DeviceType.MOUSE: "mdi:mouse",
+    DeviceType.MOUSEMAT: "mdi:rug",
+    DeviceType.HEADSET: "mdi:headset",
+    DeviceType.HEADSET_STAND: "mdi:headset-dock",
+    DeviceType.GAMEPAD: "mdi:gamepad-variant",
+    DeviceType.SPEAKER: "mdi:speaker",
+    DeviceType.STORAGE: "mdi:harddisk",
+    DeviceType.CASE: "mdi:desktop-tower",
+    DeviceType.MICROPHONE: "mdi:microphone",
+    DeviceType.KEYPAD: "mdi:dialpad",
+}
+
+CONNECTION_ERRORS = (
+    ConnectionRefusedError,
+    OpenRGBDisconnected,
+    ControllerParsingError,
+    TimeoutError,
+    socket.gaierror,  # DNS errors
+    SDKVersionError,  # The OpenRGB SDK server version is incompatible with the client
+)

homeassistant/components/openrgb/const.py

@@ -0,0 +1,150 @@
+"""DataUpdateCoordinator for OpenRGB."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from openrgb import OpenRGBClient
+from openrgb.orgb import Device
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.debounce import Debouncer
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import CONNECTION_ERRORS, DEFAULT_CLIENT_NAME, DOMAIN, SCAN_INTERVAL
+
+_LOGGER = logging.getLogger(__name__)
+
+type OpenRGBConfigEntry = ConfigEntry[OpenRGBCoordinator]
+
+
+class OpenRGBCoordinator(DataUpdateCoordinator[dict[str, Device]]):
+    """Class to manage fetching OpenRGB data."""
+
+    client: OpenRGBClient
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: OpenRGBConfigEntry,
+    ) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=SCAN_INTERVAL,
+            config_entry=config_entry,
+            request_refresh_debouncer=Debouncer(
+                hass, _LOGGER, cooldown=0.5, immediate=False
+            ),
+        )
+        self.host = config_entry.data[CONF_HOST]
+        self.port = config_entry.data[CONF_PORT]
+        self.entry_id = config_entry.entry_id
+        self.server_address = f"{self.host}:{self.port}"
+        self.client_lock = asyncio.Lock()
+
+        config_entry.async_on_unload(self.async_client_disconnect)
+
+    async def _async_setup(self) -> None:
+        """Set up the coordinator by connecting to the OpenRGB SDK server."""
+        try:
+            self.client = await self.hass.async_add_executor_job(
+                OpenRGBClient,
+                self.host,
+                self.port,
+                DEFAULT_CLIENT_NAME,
+            )
+        except CONNECTION_ERRORS as err:
+            raise UpdateFailed(
+                translation_domain=DOMAIN,
+                translation_key="cannot_connect",
+                translation_placeholders={
+                    "server_address": self.server_address,
+                    "error": str(err),
+                },
+            ) from err
+
+    async def _async_update_data(self) -> dict[str, Device]:
+        """Fetch data from OpenRGB."""
+        async with self.client_lock:
+            try:
+                await self.hass.async_add_executor_job(self._client_update)
+            except CONNECTION_ERRORS as err:
+                raise UpdateFailed(
+                    translation_domain=DOMAIN,
+                    translation_key="communication_error",
+                    translation_placeholders={
+                        "server_address": self.server_address,
+                        "error": str(err),
+                    },
+                ) from err
+
+        # Return devices indexed by their key
+        return {self._get_device_key(device): device for device in self.client.devices}
+
+    def _client_update(self) -> None:
+        try:
+            self.client.update()
+        except CONNECTION_ERRORS:
+            # Try to reconnect once
+            self.client.disconnect()
+            self.client.connect()
+            self.client.update()
+
+    def _get_device_key(self, device: Device) -> str:
+        """Build a stable device key.
+
+        Note: the OpenRGB device.id is intentionally not used because it is just
+        a positional index that can change when devices are added or removed.
+        """
+        parts = (
+            self.entry_id,
+            device.type.name,
+            device.metadata.vendor or "none",
+            device.metadata.description or "none",
+            device.metadata.serial or "none",
+            device.metadata.location or "none",
+        )
+        # Double pipe is readable and is unlikely to appear in metadata
+        return "||".join(parts)
+
+    async def async_client_disconnect(self, *args) -> None:
+        """Disconnect the OpenRGB client."""
+        if not hasattr(self, "client"):
+            # If async_config_entry_first_refresh failed, client will not exist
+            return
+
+        async with self.client_lock:
+            await self.hass.async_add_executor_job(self.client.disconnect)
+
+    def get_client_protocol_version(self) -> str:
+        """Get the OpenRGB client protocol version."""
+        return f"{self.client.protocol_version} (Protocol)"
+
+    def get_device_name(self, device_key: str) -> str:
+        """Get device name with suffix if there are duplicates."""
+        device = self.data[device_key]
+        device_name = device.name
+
+        devices_with_same_name = [
+            (key, dev) for key, dev in self.data.items() if dev.name == device_name
+        ]
+
+        if len(devices_with_same_name) == 1:
+            return device_name
+
+        # Sort duplicates by device.id
+        devices_with_same_name.sort(key=lambda x: x[1].id)
+
+        # Return name with numeric suffix based on the sorted order
+        for idx, (key, _) in enumerate(devices_with_same_name, start=1):
+            if key == device_key:
+                return f"{device_name} {idx}"
+
+        # Should never reach here, but just in case
+        return device_name  # pragma: no cover

homeassistant/components/openrgb/coordinator.py

@@ -0,0 +1,31 @@
+{
+  "entity": {
+    "light": {
+      "openrgb_light": {
+        "state_attributes": {
+          "effect": {
+            "state": {
+              "breathing": "mdi:heart-pulse",
+              "chase": "mdi:run-fast",
+              "chase_fade": "mdi:run",
+              "cram": "mdi:grid",
+              "flashing": "mdi:flash",
+              "music": "mdi:music-note",
+              "neon": "mdi:lightbulb-fluorescent-tube",
+              "rainbow": "mdi:looks",
+              "random": "mdi:dice-multiple",
+              "random_flicker": "mdi:shimmer",
+              "scan": "mdi:radar",
+              "spectrum_cycle": "mdi:gradient-horizontal",
+              "spring": "mdi:flower",
+              "stack": "mdi:layers",
+              "strobe": "mdi:led-strip-variant",
+              "water": "mdi:waves",
+              "wave": "mdi:sine-wave"
+            }
+          }
+        }
+      }
+    }
+  }
+}