alexei.dolgolyov
579553a850
HACS-compatible custom component split out from the main LedGrab repo. Creates light, switch, sensor, number, and select entities for each configured LedGrab device.
128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
"""Config flow for LED Screen Controller integration."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
from urllib.parse import urlparse, urlunparse
|
|
|
|
import aiohttp
|
|
import voluptuous as vol
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.data_entry_flow import FlowResult
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
|
|
from .const import DOMAIN, CONF_SERVER_NAME, CONF_SERVER_URL, CONF_API_KEY, DEFAULT_TIMEOUT
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
STEP_USER_DATA_SCHEMA = vol.Schema(
|
|
{
|
|
vol.Optional(CONF_SERVER_NAME, default="LED Screen Controller"): str,
|
|
vol.Required(CONF_SERVER_URL, default="http://localhost:8080"): str,
|
|
vol.Optional(CONF_API_KEY, default=""): str,
|
|
}
|
|
)
|
|
|
|
|
|
def normalize_url(url: str) -> str:
|
|
"""Normalize URL to ensure port is an integer."""
|
|
parsed = urlparse(url)
|
|
|
|
if parsed.port is not None:
|
|
netloc = parsed.hostname or "localhost"
|
|
port = int(parsed.port)
|
|
if port != (443 if parsed.scheme == "https" else 80):
|
|
netloc = f"{netloc}:{port}"
|
|
parsed = parsed._replace(netloc=netloc)
|
|
|
|
return urlunparse(parsed)
|
|
|
|
|
|
async def validate_server(
|
|
hass: HomeAssistant, server_url: str, api_key: str
|
|
) -> dict[str, Any]:
|
|
"""Validate server connectivity and API key."""
|
|
session = async_get_clientsession(hass)
|
|
timeout = aiohttp.ClientTimeout(total=DEFAULT_TIMEOUT)
|
|
|
|
# Step 1: Check connectivity via health endpoint (no auth needed)
|
|
try:
|
|
async with session.get(f"{server_url}/health", timeout=timeout) as resp:
|
|
if resp.status != 200:
|
|
raise ConnectionError(f"Server returned status {resp.status}")
|
|
data = await resp.json()
|
|
version = data.get("version", "unknown")
|
|
except aiohttp.ClientError as err:
|
|
raise ConnectionError(f"Cannot connect to server: {err}") from err
|
|
|
|
# Step 2: Validate API key via authenticated endpoint (skip if no key and auth not required)
|
|
auth_required = data.get("auth_required", True)
|
|
if api_key:
|
|
headers = {"Authorization": f"Bearer {api_key}"}
|
|
try:
|
|
async with session.get(
|
|
f"{server_url}/api/v1/output-targets",
|
|
headers=headers,
|
|
timeout=timeout,
|
|
) as resp:
|
|
if resp.status == 401:
|
|
raise PermissionError("Invalid API key")
|
|
resp.raise_for_status()
|
|
except PermissionError:
|
|
raise
|
|
except aiohttp.ClientError as err:
|
|
raise ConnectionError(f"API request failed: {err}") from err
|
|
elif auth_required:
|
|
raise PermissionError("Server requires an API key")
|
|
|
|
return {"version": version}
|
|
|
|
|
|
class LedGrabConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
|
"""Handle a config flow for LED Screen Controller."""
|
|
|
|
VERSION = 2
|
|
|
|
async def async_step_user(
|
|
self, user_input: dict[str, Any] | None = None
|
|
) -> FlowResult:
|
|
"""Handle the initial step."""
|
|
errors: dict[str, str] = {}
|
|
|
|
if user_input is not None:
|
|
server_name = user_input.get(CONF_SERVER_NAME, "LED Screen Controller")
|
|
server_url = normalize_url(user_input[CONF_SERVER_URL].rstrip("/"))
|
|
api_key = user_input[CONF_API_KEY]
|
|
|
|
try:
|
|
await validate_server(self.hass, server_url, api_key)
|
|
|
|
await self.async_set_unique_id(server_url)
|
|
self._abort_if_unique_id_configured()
|
|
|
|
return self.async_create_entry(
|
|
title=server_name,
|
|
data={
|
|
CONF_SERVER_NAME: server_name,
|
|
CONF_SERVER_URL: server_url,
|
|
CONF_API_KEY: api_key,
|
|
},
|
|
)
|
|
|
|
except ConnectionError as err:
|
|
_LOGGER.error("Connection error: %s", err)
|
|
errors["base"] = "cannot_connect"
|
|
except PermissionError:
|
|
errors["base"] = "invalid_api_key"
|
|
except Exception as err:
|
|
_LOGGER.exception("Unexpected exception: %s", err)
|
|
errors["base"] = "unknown"
|
|
|
|
return self.async_show_form(
|
|
step_id="user",
|
|
data_schema=STEP_USER_DATA_SCHEMA,
|
|
errors=errors,
|
|
)
|