Implement config flow

custom_components/flickerstrip/__init__.py

@@ -1,7 +1,29 @@
-from homeassistant import core
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import HomeAssistant
+
+from .common import Flickerstrip
+from .const import (
+    DOMAIN,
+)
 
 
-async def async_setup(hass: core.HomeAssistant, config: dict) -> bool:
-    """Set up the Flickerstrip component."""
-    # @TODO: Add setup code.
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Setup flickerstrip from config entry."""
+
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = Flickerstrip(
+        hass=hass,
+        host=entry.data[CONF_HOST],
+        port=entry.data[CONF_PORT],
+    )
+
+    entry.async_on_unload(entry.add_update_listener(update_listener))
+
     return True
+
+
+async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Update when config_entry options update."""
+    if entry.options:
+        await hass.config_entries.async_reload(entry.entry_id)

custom_components/flickerstrip/common.py

@@ -0,0 +1,58 @@
+from typing import Any
+import flickerstrip_py as flstrp
+
+from homeassistant.core import HomeAssistant
+from homeassistant.components.light import (
+    ATTR_BRIGHTNESS,
+    LightEntity,
+)
+
+from .const import (
+    DEFAULT_PORT,
+)
+
+
+class Flickerstrip(LightEntity):
+    """Flickerstrip class."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        host: str,
+        port: int = DEFAULT_PORT,
+    ):
+        self.hass = hass
+        self.host = host
+        self.port = port
+        self._flickerstrip = flstrp.Flickerstrip(host)
+
+    @property
+    def name(self) -> str:
+        """Return the display name of this light."""
+        return self._name
+
+    @property
+    def brightness(self):
+        """Return the brightness of the light."""
+        return self._flickerstrip.status.brightness
+
+    @property
+    def is_on(self) -> bool | None:
+        """Return true if light is on."""
+        return self._flickerstrip.status.power
+
+    def turn_on(self, **kwargs: Any) -> None:
+        """Instruct the light to turn on."""
+        brightness = kwargs.get(ATTR_BRIGHTNESS, 100)
+        self._flickerstrip.set_brightness(brightness)
+        self._flickerstrip.power_on()
+
+    def turn_off(self, **kwargs: Any) -> None:
+        """Instruct the light to turn off."""
+        self._flickerstrip.power_off()
+
+    def update(self) -> None:
+        """Fetch new state data for this light.
+        This is the only method that should fetch new data for Home Assistant.
+        """
+        self._flickerstrip.force_update()

custom_components/flickerstrip/config_flow.py

@@ -0,0 +1,124 @@
+import ipaddress
+import socket
+from typing import Any
+from urllib.parse import ParseResult, urlparse
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.components import ssdp
+from homeassistant.components.device_tracker import (
+    CONF_CONSIDER_HOME,
+    DEFAULT_CONSIDER_HOME,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PORT
+from homeassistant.core import callback
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import (
+    DOMAIN,
+    DEFAULT_PORT,
+)
+
+
+class FlickerstripConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Flickerstrip config flow."""
+    VERSION = 1
+
+    def __init__(self) -> None:
+        """Initialize Flickerstrip conflig flow."""
+        self._host: str | None = None
+        self._port: int | None = None
+        self._entry: ConfigEntry | None = None
+
+    async def async_check_configured_entry(self) -> ConfigEntry | None:
+        """Check if entry is configured."""
+        assert self._host
+        current_host = await self.hass.async_add_executor_job(
+            socket.gethostbyname, self._host
+        )
+
+        for entry in self._async_current_entries(include_ignore=False):
+            entry_host = await self.hass.async_add_executor_job(
+                socket.gethostbyname, entry.data[CONF_HOST]
+            )
+            if entry_host == current_host:
+                return entry
+        return None
+
+    @callback
+    def _async_create_entry(self) -> FlowResult:
+        """Async create flow handler entry."""
+        return self.async_create_entry(
+            title=self._name,
+            data={
+                CONF_HOST: self._host,
+                CONF_PORT: self._port,
+            },
+            options={
+                CONF_CONSIDER_HOME: DEFAULT_CONSIDER_HOME.total_seconds(),
+            },
+        )
+
+    async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo) -> FlowResult :
+        ssdp_location: ParseResult = urlparse(discovery_info.ssdp_location or "")
+        self._host = ssdp_location.hostname
+        self._port = ssdp_location.port
+
+        self.context[CONF_HOST] = self._host
+        self.context[CONF_PORT] = self._port
+
+        if not self._host or ipaddress.ip_address(self._host).is_link_local:
+            return self.async_abort(reason="ignore_ip6_link_local")
+
+        if uuid := discovery_info.upnp.get(ssdp.ATTR_UPNP_UDN):
+            if uuid.startswith("uuid:"):
+                uuid = uuid[5:]
+            await self.async_set_unique_id(uuid)
+            self._abort_if_unique_id_configured({CONF_HOST: self._host})
+
+        for progress in self._async_in_progress():
+            if progress.get("context", {}).get(CONF_HOST) == self._host:
+                return self.async_abort(reason="already_in_progress")
+
+        if entry := await self.async_check_configured_entry():
+            if uuid and not entry.unique_id:
+                self.hass.config_entries.async_update_entry(entry, unique_id=uuid)
+            return self.async_abort(reason="already_configured")
+
+        self.context.update(
+            {
+                "title_placeholders": {"name": "Flickerstrip"},
+                "configuration_url": f"http://{self._host}",
+            }
+        )
+
+        return self._async_create_entry()
+
+    def _show_setup_form_init(self, errors: dict[str, str] | None = None) -> FlowResult:
+        """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=DEFAULT_PORT): vol.Coerce(int),
+                }
+            ),
+            errors=errors or {},
+        )
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle a flow initiated by the user."""
+        if user_input is None:
+            return self._show_setup_form_init()
+
+        self._host = user_input[CONF_HOST]
+        self._port = user_input[CONF_PORT]
+
+        if await self.async_check_configured_entry():
+            return self._show_setup_form_init({"base": "already_configured"})
+
+        return self._async_create_entry()

custom_components/flickerstrip/const.py

@@ -1 +1,3 @@
 DOMAIN = "flickerstrip"
+
+DEFAULT_PORT = 80

custom_components/flickerstrip/manifest.json

@@ -6,6 +6,15 @@
   "domain": "flickerstrip",
   "iot_class": "local_polling",
   "name": "Flickerstrip",
-  "requirements": [],
+  "ssdp": [
+    {
+      "modelName": "Flickerstrip LED Strip",
+      "manufacturer": "HomeAutomaton",
+      "deviceType": "urn:schemas-upnp-org:device:Basic:1"
+    }
+  ],
+  "requirements": [
+    "git+https://git.fifitido.net/lib/flickerstrip-py@d808ca8ac8#egg=flickerstrip-py"
+  ],
   "version": "1.0.0"
 }