Add Axion Lighting integration

CODEOWNERS

@@ -166,6 +166,8 @@ build.json @home-assistant/supervisor
 /homeassistant/components/avea/ @pattyland
 /homeassistant/components/awair/ @ahayworth @danielsjf
 /tests/components/awair/ @ahayworth @danielsjf
+/homeassistant/components/axion_dmx/ @Vrncanac
+/tests/components/axion_dmx/ @Vrncanac
 /homeassistant/components/axis/ @Kane610
 /tests/components/axis/ @Kane610
 /homeassistant/components/azure_data_explorer/ @kaareseras

homeassistant/components/axion_dmx/__init__.py

@@ -0,0 +1,54 @@
+"""The Axion Lighting integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from libaxion_dmx import AxionDmxApi
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import CONF_CHANNEL, CONF_HOST, CONF_PASSWORD
+from .coordinator import AxionDataUpdateCoordinator
+
+PLATFORMS: list[Platform] = [Platform.LIGHT]
+
+
+@dataclass
+class AxionData:
+    """Represent a config for Axion Light."""
+
+    coordinator: AxionDataUpdateCoordinator
+    api: AxionDmxApi
+
+
+type AxionConfigEntry = ConfigEntry[AxionData]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: AxionConfigEntry) -> bool:
+    """Set up Axion Lighting from a config entry."""
+
+    # Create API instance
+    api = AxionDmxApi(entry.data[CONF_HOST], entry.data[CONF_PASSWORD])
+
+    # Validate the API connection (and authentication)
+    if not await api.authenticate():
+        return False
+
+    # Create coordinator instance
+    coordinator = AxionDataUpdateCoordinator(hass, api, entry.data[CONF_CHANNEL])
+    await coordinator.async_config_entry_first_refresh()
+
+    # Store coordinator and API objects in runtime_data
+    entry.runtime_data = AxionData(coordinator=coordinator, api=api)
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: AxionConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

homeassistant/components/axion_dmx/config_flow.py

@@ -0,0 +1,72 @@
+"""Config flow for Axion Lighting integration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from libaxion_dmx import AxionDmxApi, AxionDmxAuthError, AxionDmxConnectionError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_PASSWORD
+from homeassistant.helpers import selector
+
+from .const import _LOGGER, CONF_CHANNEL, CONF_LIGHT_TYPE, DOMAIN
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_PASSWORD): str,
+        vol.Required(CONF_CHANNEL): vol.All(vol.Coerce(int), vol.Range(min=1)),
+        vol.Required(CONF_LIGHT_TYPE): selector.SelectSelector(
+            selector.SelectSelectorConfig(
+                options=[
+                    selector.SelectOptionDict(value="white", label="White"),
+                    selector.SelectOptionDict(
+                        value="tunable_white", label="Tunable White"
+                    ),
+                    selector.SelectOptionDict(value="rgb", label="RGB"),
+                    selector.SelectOptionDict(value="rgbw", label="RGBW"),
+                    selector.SelectOptionDict(value="rgbww", label="RGBWW"),
+                ],
+                translation_key="light_type",
+                multiple=False,
+                mode=selector.SelectSelectorMode.DROPDOWN,
+            )
+        ),
+    }
+)
+
+
+class AxionConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Axion Lighting."""
+
+    VERSION = 1
+
+    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:
+            api = AxionDmxApi(user_input[CONF_HOST], user_input[CONF_PASSWORD])
+            # Validate the user input and authenticate
+            try:
+                await api.authenticate()
+            except AxionDmxConnectionError:
+                errors["base"] = "cannot_connect"
+            except AxionDmxAuthError:
+                errors["base"] = "invalid_auth"
+            except Exception as e:  # noqa: BLE001
+                _LOGGER.exception("Unexpected exception: %s", e)
+                errors["base"] = "unknown"
+            else:
+                # Return the validated entry data
+                return self.async_create_entry(
+                    title=f"Axion DMX Light - Channel {user_input[CONF_CHANNEL]}",
+                    data=user_input,
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )

homeassistant/components/axion_dmx/const.py

@@ -0,0 +1,14 @@
+"""Constants for the Axion Lighting integration."""
+
+import logging
+
+_LOGGER = logging.getLogger(__name__)
+
+DOMAIN = "axion_dmx"
+CONF_CHANNEL = "channel"
+CONF_LIGHT_TYPE = "light_type"
+CONF_HOST = "host"
+CONF_PASSWORD = "password"
+
+AXION_MANUFACTURER_NAME = "Axion"
+AXION_MODEL_NAME = "Axion DMX Controller"

homeassistant/components/axion_dmx/coordinator.py

@@ -0,0 +1,35 @@
+"""Coordinator for Axion Lighting integration."""
+
+from datetime import timedelta
+from typing import Any
+
+from libaxion_dmx import AxionDmxApi
+from requests.exceptions import RequestException
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import _LOGGER
+
+
+class AxionDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
+    """Custom coordinator for Axion Lighting integration."""
+
+    def __init__(self, hass: HomeAssistant, api: AxionDmxApi, channel: int) -> None:
+        """Initialize the Axion data coordinator."""
+        self.api = api
+        self.channel = channel
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Axion Light",
+            update_interval=timedelta(seconds=5),
+        )
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Fetch data from the API endpoint."""
+        try:
+            # Return data in dictionary form
+            return await self.api.get_level(self.channel)
+        except RequestException as err:
+            raise UpdateFailed("Error communicating with API") from err