Add Kiosker integration

homeassistant/components/kiosker/__init__.py

@@ -0,0 +1,29 @@
+"""The Kiosker integration."""
+
+from __future__ import annotations
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .coordinator import KioskerConfigEntry, KioskerDataUpdateCoordinator
+
+_PLATFORMS: list[Platform] = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: KioskerConfigEntry) -> bool:
+    """Set up Kiosker from a config entry."""
+
+    coordinator = KioskerDataUpdateCoordinator(hass, entry)
+
+    await coordinator.async_config_entry_first_refresh()
+
+    entry.runtime_data = coordinator
+
+    await hass.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: KioskerConfigEntry) -> bool:
+    """Unload a config entry."""
+    return await hass.config_entries.async_unload_platforms(entry, _PLATFORMS)

homeassistant/components/kiosker/config_flow.py

@@ -0,0 +1,198 @@
+"""Config flow for the Kiosker integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from kiosker import (
+    AuthenticationError,
+    BadRequestError,
+    ConnectionError,
+    IPAuthenticationError,
+    KioskerAPI,
+    PingError,
+    TLSVerificationError,
+)
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_HOST, CONF_SSL, CONF_VERIFY_SSL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
+
+from .const import CONF_API_TOKEN, DEFAULT_SSL, DEFAULT_SSL_VERIFY, DOMAIN, PORT
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST): str,
+        vol.Required(CONF_API_TOKEN): str,
+        vol.Optional(CONF_SSL, default=DEFAULT_SSL): bool,
+        vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_SSL_VERIFY): bool,
+    }
+)
+STEP_ZEROCONF_CONFIRM_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_API_TOKEN): str,
+        vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_SSL_VERIFY): bool,
+    }
+)
+
+
+async def validate_input(
+    hass: HomeAssistant, data: dict[str, Any]
+) -> tuple[dict[str, str], str | None]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    Returns a tuple of (errors dict, device_id). If validation succeeds, errors will be empty.
+    """
+    api = KioskerAPI(
+        host=data[CONF_HOST],
+        port=PORT,
+        token=data[CONF_API_TOKEN],
+        ssl=data[CONF_SSL],
+        verify=data[CONF_VERIFY_SSL],
+    )
+
+    try:
+        # Test connection by getting status
+        status = await hass.async_add_executor_job(api.status)
+    except ConnectionError:
+        return ({"base": "cannot_connect"}, None)
+    except AuthenticationError:
+        return ({"base": "invalid_auth"}, None)
+    except IPAuthenticationError:
+        return ({"base": "invalid_ip_auth"}, None)
+    except TLSVerificationError:
+        return ({"base": "tls_error"}, None)
+    except BadRequestError:
+        return ({"base": "bad_request"}, None)
+    except PingError:
+        return ({"base": "cannot_connect"}, None)
+    except Exception:
+        _LOGGER.exception("Unexpected exception while connecting to Kiosker")
+        return ({"base": "unknown"}, None)
+
+    # Ensure we have a device_id from the status response
+    if not status.device_id:
+        _LOGGER.error("Device did not return a valid device_id")
+        return ({"base": "cannot_connect"}, None)
+
+    return ({}, status.device_id)
+
+
+class KioskerConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Kiosker."""
+
+    VERSION = 1
+    MINOR_VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize the config flow."""
+
+        self._discovered_host: str | None = None
+        self._discovered_device_id: str | None = None
+        self._discovered_version: str | None = None
+        self._discovered_ssl: bool | None = None
+
+    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:
+            validation_errors, device_id = await validate_input(self.hass, user_input)
+            if validation_errors:
+                errors.update(validation_errors)
+            elif device_id:
+                # Use device ID as unique identifier
+                await self.async_set_unique_id(device_id, raise_on_progress=False)
+                self._abort_if_unique_id_configured()
+
+                # Use first 8 characters of device_id for consistency with entity naming
+                display_id = device_id[:8] if len(device_id) > 8 else device_id
+                title = f"Kiosker {display_id}"
+                return self.async_create_entry(title=title, data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+    async def async_step_zeroconf(
+        self, discovery_info: ZeroconfServiceInfo
+    ) -> ConfigFlowResult:
+        """Handle zeroconf discovery."""
+        host = discovery_info.host
+
+        # Extract device information from zeroconf properties
+        properties = discovery_info.properties
+        device_id = properties.get("uuid")
+        app_name = properties.get("app", "Kiosker")
+        version = properties.get("version", "")
+        ssl = properties.get("ssl", "false").lower() == "true"
+
+        # Use device_id from zeroconf
+        if device_id:
+            device_name = f"{app_name} ({device_id[:8].upper()})"
+            unique_id = device_id
+        else:
+            _LOGGER.debug("Zeroconf properties did not include a valid device_id")
+            return self.async_abort(reason="cannot_connect")
+
+        # Set unique ID and check for duplicates
+        await self.async_set_unique_id(unique_id)
+        self._abort_if_unique_id_configured()
+
+        # Store discovery info for confirmation step
+        self.context["title_placeholders"] = {
+            "name": device_name,
+            "host": host,
+        }
+
+        # Store discovered information for later use
+        self._discovered_host = host
+        self._discovered_device_id = device_id
+        self._discovered_version = version
+        self._discovered_ssl = ssl
+
+        # Show confirmation dialog
+        return await self.async_step_zeroconf_confirm()
+
+    async def async_step_zeroconf_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle zeroconf confirmation."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            # Use stored discovery info and user-provided token
+            host = self._discovered_host
+            ssl = self._discovered_ssl
+
+            # Create config with discovered host and user-provided token
+            config_data = {
+                CONF_HOST: host,
+                CONF_API_TOKEN: user_input[CONF_API_TOKEN],
+                CONF_SSL: ssl,
+                CONF_VERIFY_SSL: user_input.get(CONF_VERIFY_SSL, DEFAULT_SSL_VERIFY),
+            }
+
+            validation_errors, device_id = await validate_input(self.hass, config_data)
+            if validation_errors:
+                errors.update(validation_errors)
+            elif device_id:
+                # Use first 8 characters of device_id for consistency with entity naming
+                display_id = device_id[:8] if len(device_id) > 8 else device_id
+                title = f"Kiosker {display_id}"
+                return self.async_create_entry(title=title, data=config_data)
+
+        # Show form to get API token for discovered device
+        return self.async_show_form(
+            step_id="zeroconf_confirm",
+            data_schema=STEP_ZEROCONF_CONFIRM_DATA_SCHEMA,
+            description_placeholders=self.context["title_placeholders"],
+            errors=errors,
+        )

homeassistant/components/kiosker/const.py

@@ -0,0 +1,12 @@
+"""Constants for the Kiosker integration."""
+
+DOMAIN = "kiosker"
+
+# Configuration keys
+CONF_API_TOKEN = "api_token"
+
+# Default values
+PORT = 8081
+POLL_INTERVAL = 15
+DEFAULT_SSL = False
+DEFAULT_SSL_VERIFY = False

homeassistant/components/kiosker/coordinator.py

@@ -0,0 +1,105 @@
+"""DataUpdateCoordinator for Kiosker."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import timedelta
+import logging
+
+from kiosker import (
+    AuthenticationError,
+    BadRequestError,
+    Blackout,
+    ConnectionError,
+    IPAuthenticationError,
+    KioskerAPI,
+    PingError,
+    ScreensaverState,
+    Status,
+    TLSVerificationError,
+)
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_SSL, CONF_VERIFY_SSL
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import CONF_API_TOKEN, DOMAIN, POLL_INTERVAL, PORT
+
+_LOGGER = logging.getLogger(__name__)
+
+type KioskerConfigEntry = ConfigEntry[KioskerDataUpdateCoordinator]
+
+
+@dataclass
+class KioskerData:
+    """Data structure for Kiosker integration."""
+
+    status: Status
+    blackout: Blackout | None
+    screensaver: ScreensaverState | None
+
+
+class KioskerDataUpdateCoordinator(DataUpdateCoordinator[KioskerData]):
+    """Class to manage fetching data from the Kiosker API."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        config_entry: KioskerConfigEntry,
+    ) -> None:
+        """Initialize."""
+        self.api = KioskerAPI(
+            host=config_entry.data[CONF_HOST],
+            port=PORT,
+            token=config_entry.data[CONF_API_TOKEN],
+            ssl=config_entry.data.get(CONF_SSL, False),
+            verify=config_entry.data.get(CONF_VERIFY_SSL, False),
+        )
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=POLL_INTERVAL),
+            config_entry=config_entry,
+        )
+
+    def _fetch_all_data(self) -> tuple[Status, Blackout, ScreensaverState]:
+        """Fetch all data from the API in a single executor job."""
+        status = self.api.status()
+        blackout = self.api.blackout_get()
+        screensaver = self.api.screensaver_get_state()
+        return status, blackout, screensaver
+
+    async def _async_update_data(self) -> KioskerData:
+        """Update data via library."""
+        try:
+            status, blackout, screensaver = await self.hass.async_add_executor_job(
+                self._fetch_all_data
+            )
+        except AuthenticationError as exc:
+            raise ConfigEntryAuthFailed(
+                "Authentication failed. Check your API token."
+            ) from exc
+        except IPAuthenticationError as exc:
+            raise ConfigEntryAuthFailed(
+                "IP authentication failed. Check your IP whitelist."
+            ) from exc
+        except (ConnectionError, PingError) as exc:
+            raise UpdateFailed(f"Connection failed: {exc}") from exc
+        except TLSVerificationError as exc:
+            raise UpdateFailed(f"TLS verification failed: {exc}") from exc
+        except BadRequestError as exc:
+            raise UpdateFailed(f"Bad request: {exc}") from exc
+        except (OSError, TimeoutError) as exc:
+            raise UpdateFailed(f"Connection timeout: {exc}") from exc
+        except Exception as exc:
+            _LOGGER.exception("Unexpected error updating Kiosker data")
+            raise UpdateFailed(f"Unexpected error: {exc}") from exc
+
+        return KioskerData(
+            status=status,
+            blackout=blackout,
+            screensaver=screensaver,
+        )