Add Elgato Key Light integration

CODEOWNERS

@@ -85,6 +85,7 @@ homeassistant/components/ecobee/* @marthoc
 homeassistant/components/ecovacs/* @OverloadUT
 homeassistant/components/egardia/* @jeroenterheerdt
 homeassistant/components/eight_sleep/* @mezz64
+homeassistant/components/elgato/* @frenck
 homeassistant/components/elv/* @majuss
 homeassistant/components/emby/* @mezz64
 homeassistant/components/emulated_hue/* @NobleKangaroo

homeassistant/components/elgato/.translations/en.json

@@ -0,0 +1,27 @@
+{
+  "config": {
+    "title": "Elgato Key Light",
+    "flow_title": "Elgato Key Light: {serial_number}",
+    "step": {
+      "user": {
+        "title": "Link your Elgato Key Light",
+        "description": "Set up your Elgato Key Light to integrate with Home Assistant.",
+        "data": {
+          "host": "Host or IP address",
+          "port": "Port number"
+        }
+      },
+      "zeroconf_confirm": {
+        "description": "Do you want to add the Elgato Key Light with serial number `{serial_number}` to Home Assistant?",
+        "title": "Discovered Elgato Key Light device"
+      }
+    },
+    "error": {
+      "connection_error": "Failed to connect to Elgato Key Light device."
+    },
+    "abort": {
+      "already_configured": "This Elgato Key Light device is already configured.",
+      "connection_error": "Failed to connect to Elgato Key Light device."
+    }
+  }
+}

homeassistant/components/elgato/__init__.py

@@ -0,0 +1,55 @@
+"""Support for Elgato Key Lights."""
+import logging
+
+from elgato import Elgato, ElgatoConnectionError
+
+from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DATA_ELGATO_CLIENT, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the Elgato Key Light components."""
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Elgato Key Light from a config entry."""
+    session = async_get_clientsession(hass)
+    elgato = Elgato(entry.data[CONF_HOST], port=entry.data[CONF_PORT], session=session,)
+
+    # Ensure we can connect to it
+    try:
+        await elgato.info()
+    except ElgatoConnectionError as exception:
+        raise ConfigEntryNotReady from exception
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = {DATA_ELGATO_CLIENT: elgato}
+
+    hass.async_create_task(
+        hass.config_entries.async_forward_entry_setup(entry, LIGHT_DOMAIN)
+    )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload Elgato Key Light config entry."""
+    # Unload entities for this entry/device.
+    await hass.config_entries.async_forward_entry_unload(entry, LIGHT_DOMAIN)
+
+    # Cleanup
+    del hass.data[DOMAIN][entry.entry_id]
+    if not hass.data[DOMAIN]:
+        del hass.data[DOMAIN]
+
+    return True

homeassistant/components/elgato/config_flow.py

@@ -0,0 +1,146 @@
+"""Config flow to configure the Elgato Key Light integration."""
+import logging
+from typing import Any, Dict, Optional
+
+from elgato import Elgato, ElgatoError, Info
+import voluptuous as vol
+
+from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.helpers import ConfigType
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import CONF_SERIAL_NUMBER, DOMAIN  # pylint: disable=unused-import
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a Elgato Key Light config flow."""
+
+    VERSION = 1
+    CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
+
+    async def async_step_user(
+        self, user_input: Optional[ConfigType] = None
+    ) -> Dict[str, Any]:
+        """Handle a flow initiated by the user."""
+        if user_input is None:
+            return self._show_setup_form()
+
+        try:
+            info = await self._get_elgato_info(
+                user_input[CONF_HOST], user_input[CONF_PORT]
+            )
+        except ElgatoError:
+            return self._show_setup_form({"base": "connection_error"})
+
+        # Check if already configured
+        if await self._device_already_configured(info):
+            # This serial number is already configured
+            return self.async_abort(reason="already_configured")
+
+        return self.async_create_entry(
+            title=info.serial_number,
+            data={
+                CONF_HOST: user_input[CONF_HOST],
+                CONF_PORT: user_input[CONF_PORT],
+                CONF_SERIAL_NUMBER: info.serial_number,
+            },
+        )
+
+    async def async_step_zeroconf(
+        self, user_input: Optional[ConfigType] = None
+    ) -> Dict[str, Any]:
+        """Handle zeroconf discovery."""
+        if user_input is None:
+            return self.async_abort(reason="connection_error")
+
+        # Hostname is format: my-ke.local.
+        host = user_input["hostname"].rstrip(".")
+        try:
+            info = await self._get_elgato_info(host, user_input[CONF_PORT])
+        except ElgatoError:
+            return self.async_abort(reason="connection_error")
+
+        # Check if already configured
+        if await self._device_already_configured(info):
+            # This serial number is already configured
+            return self.async_abort(reason="already_configured")
+
+        # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+        self.context.update(
+            {
+                CONF_HOST: host,
+                CONF_PORT: user_input[CONF_PORT],
+                CONF_SERIAL_NUMBER: info.serial_number,
+                "title_placeholders": {"serial_number": info.serial_number},
+            }
+        )
+
+        # Prepare configuration flow
+        return self._show_confirm_dialog()
+
+    # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+    async def async_step_zeroconf_confirm(
+        self, user_input: ConfigType = None
+    ) -> Dict[str, Any]:
+        """Handle a flow initiated by zeroconf."""
+        if user_input is None:
+            return self._show_confirm_dialog()
+
+        try:
+            info = await self._get_elgato_info(
+                self.context.get(CONF_HOST), self.context.get(CONF_PORT)
+            )
+        except ElgatoError:
+            return self.async_abort(reason="connection_error")
+
+        # Check if already configured
+        if await self._device_already_configured(info):
+            # This serial number is already configured
+            return self.async_abort(reason="already_configured")
+
+        return self.async_create_entry(
+            title=self.context.get(CONF_SERIAL_NUMBER),
+            data={
+                CONF_HOST: self.context.get(CONF_HOST),
+                CONF_PORT: self.context.get(CONF_PORT),
+                CONF_SERIAL_NUMBER: self.context.get(CONF_SERIAL_NUMBER),
+            },
+        )
+
+    def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
+        """Show the setup form to the user."""
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_HOST): str,
+                    vol.Optional(CONF_PORT, default=9123): int,
+                }
+            ),
+            errors=errors or {},
+        )
+
+    def _show_confirm_dialog(self) -> Dict[str, Any]:
+        """Show the confirm dialog to the user."""
+        # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+        serial_number = self.context.get(CONF_SERIAL_NUMBER)
+        return self.async_show_form(
+            step_id="zeroconf_confirm",
+            description_placeholders={"serial_number": serial_number},
+        )
+
+    async def _get_elgato_info(self, host: str, port: int) -> Info:
+        """Get device information from an Elgato Key Light device."""
+        session = async_get_clientsession(self.hass)
+        elgato = Elgato(host, port=port, session=session,)
+        return await elgato.info()
+
+    async def _device_already_configured(self, info: Info) -> bool:
+        """Return if a Elgato Key Light is already configured."""
+        for entry in self._async_current_entries():
+            if entry.data[CONF_SERIAL_NUMBER] == info.serial_number:
+                return True
+        return False

homeassistant/components/elgato/const.py

@@ -0,0 +1,17 @@
+"""Constants for the Elgato Key Light integration."""
+
+# Integration domain
+DOMAIN = "elgato"
+
+# Hass data keys
+DATA_ELGATO_CLIENT = "elgato_client"
+
+# Attributes
+ATTR_IDENTIFIERS = "identifiers"
+ATTR_MANUFACTURER = "manufacturer"
+ATTR_MODEL = "model"
+ATTR_ON = "on"
+ATTR_SOFTWARE_VERSION = "sw_version"
+ATTR_TEMPERATURE = "temperature"
+
+CONF_SERIAL_NUMBER = "serial_number"