Add initial version of the custom component (with HACS metadata)

Adds the initial version of the custom component, with HACS metadata and structure to facilitate installation via HACS.

README.md

@@ -1,2 +1,13 @@
 # tailwind-home-assistant
 Tailwind iQ3 integration for Home Assistant
+
+This integration is not final, and currently requires your Tailwind iQ3 device to be running beta firmware.
+
+The integration will create one or more Cover entities for your garage door(s).
+The state (open/closed) is updated every 10 seconds via local polling - no cloud connection is required.
+
+# Installation
+For the best experience, install via [HACS](https://hacs.xyz/).
+At this stage, you'll need to [add it to HACS as a custom repository](https://hacs.xyz/docs/faq/custom_repositories) using the following details:
+- Repository URL: `https://github.com/pauln/tailwind-home-assistant`
+- Category: `Integration`

custom_components/tailwind_iq3/__init__.py

@@ -0,0 +1,104 @@
+"""The tailwind_iq3 integration."""
+from __future__ import annotations
+from async_timeout import timeout
+from datetime import timedelta
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_IP_ADDRESS
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.device_registry import DeviceEntry
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+    UpdateFailed,
+)
+
+from .const import DOMAIN, TAILWIND_COORDINATOR, UPDATE_INTERVAL, ATTR_RAW_STATE
+
+_LOGGER = logging.getLogger(__name__)
+PLATFORMS = ["cover"]
+
+
+async def tailwind_get_status(hass: HomeAssistant, ip_address: str) -> str:
+    websession = async_get_clientsession(hass)
+    async with websession.get(f"http://{ip_address}/status") as resp:
+        assert resp.status == 200
+        return await resp.text()
+
+
+async def tailwind_send_command(
+    hass: HomeAssistant, ip_address: str, command: str
+) -> str:
+    websession = async_get_clientsession(hass)
+    async with websession.post(f"http://{ip_address}/cmd", data=command) as resp:
+        assert resp.status == 200
+        return await resp.text()
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up tailwind_iq3 from a config entry."""
+    hass.data.setdefault(DOMAIN, {})
+
+    async def async_update_data():
+        status = -1
+        try:
+            async with timeout(10):
+                ip_address = entry.data[CONF_IP_ADDRESS]
+                status = await tailwind_get_status(hass, ip_address)
+
+        except BaseException as error:
+            raise UpdateFailed(error) from error
+
+        _LOGGER.debug("tailwind reported state: %s", status)
+        return {ATTR_RAW_STATE: int(status)}
+
+    coordinator = DataUpdateCoordinator(
+        hass,
+        _LOGGER,
+        name="tailwind devices",
+        update_method=async_update_data,
+        update_interval=timedelta(seconds=UPDATE_INTERVAL),
+    )
+
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data[DOMAIN][entry.entry_id] = {TAILWIND_COORDINATOR: coordinator}
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok
+
+
+class TailwindEntity(CoordinatorEntity):
+    """Base class for Tailwind iQ3 Entities."""
+
+    def __init__(self, coordinator: DataUpdateCoordinator, device: DeviceEntry) -> None:
+        super().__init__(coordinator)
+        self._device = device
+        self._attr_unique_id = device.device_id
+
+    @property
+    def name(self):
+        return self._device.name
+
+    @property
+    def device_info(self):
+        device_info = {
+            "identifiers": {(DOMAIN, self._device.device_id)},
+            "name": self._device.name,
+            "manufacturer": "Tailwind",
+            "model": "iQ3",
+        }
+        if self._device.parent_device_id:
+            device_info["via_device"] = (DOMAIN, self._device.parent_device_id)
+        return device_info

custom_components/tailwind_iq3/config_flow.py

@@ -0,0 +1,139 @@
+"""Config flow for tailwind_iq3."""
+from __future__ import annotations
+from typing import Any, Final
+import logging
+
+from homeassistant import config_entries
+from homeassistant.const import (
+    CONF_HOST,
+    CONF_NAME,
+    CONF_IP_ADDRESS,
+)
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.typing import DiscoveryInfoType
+
+import ipaddress
+from scapy.layers.l2 import getmacbyip
+import voluptuous as vol
+
+from .const import DOMAIN, CONF_NUM_DOORS
+
+_LOGGER: Final = logging.getLogger(__name__)
+
+
+class TailwindConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    VERSION = 1
+    CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
+
+    def __init__(self):
+        """Initialize the config flow."""
+        self._hostname = None
+        self._name = "Tailwind iQ3"
+        self._ip_address = None
+        self._num_doors = 0
+
+    async def async_step_user(self, info) -> FlowResult:
+        errors = {}
+        if info is not None:
+            if info[CONF_NAME] != "":
+                self._name = info[CONF_NAME]
+
+            mac = None
+            try:
+                ip_address = ipaddress.ip_address(info[CONF_IP_ADDRESS])
+            except ValueError:
+                # Invalid IP address specified.
+                pass
+            else:
+                # IP address is valid.  Look up MAC address for unique ID.
+                mac = getmacbyip(info[CONF_IP_ADDRESS])
+
+            if ip_address is not None and mac is not None:
+                # MAC address retrieved.
+                # Build a unique ID matching what would be autodiscovered.
+                flat_mac = mac.replace(":", "").lower()
+                self._hostname = f"tailwind-{flat_mac}"
+                await self.async_set_unique_id(self._hostname)
+                self._abort_if_unique_id_configured({CONF_HOST: self._hostname})
+
+                return self.async_create_entry(
+                    title=self._name,
+                    data=info,
+                )
+
+            errors["base"] = "invalid_ip"
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        CONF_IP_ADDRESS,
+                    ): str,
+                    vol.Optional(
+                        CONF_NAME,
+                        default=self._name,
+                    ): str,
+                    vol.Required(
+                        CONF_NUM_DOORS,
+                        default=1,
+                    ): vol.All(vol.Coerce(int), vol.Clamp(min=1, max=3)),
+                }
+            ),
+            errors=errors,
+        )
+
+    async def async_step_homekit(self, discovery_info: DiscoveryInfoType) -> FlowResult:
+        self._ip_address = discovery_info[CONF_HOST]
+        self._hostname = discovery_info["hostname"].replace(".local.", "")
+        await self.async_set_unique_id(self._hostname)
+        self._abort_if_unique_id_configured({CONF_HOST: self._hostname})
+
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_zeroconf(
+        self, discovery_info: DiscoveryInfoType
+    ) -> FlowResult:
+        self._ip_address = discovery_info[CONF_HOST]
+        self._hostname = discovery_info["hostname"].replace(".local.", "")
+        await self.async_set_unique_id(self._hostname)
+        self._abort_if_unique_id_configured({CONF_HOST: self._hostname})
+
+        return await self.async_step_discovery_confirm()
+
+    async def async_step_discovery_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle user-confirmation of discovered node."""
+        if user_input is not None:
+            if user_input[CONF_NAME] != "":
+                self._name = user_input[CONF_NAME]
+
+            self._num_doors = user_input[CONF_NUM_DOORS]
+
+            config_data = {
+                CONF_HOST: self._hostname,
+                CONF_IP_ADDRESS: self._ip_address,
+                CONF_NUM_DOORS: self._num_doors,
+            }
+            return self.async_create_entry(
+                title=self._name,
+                data=config_data,
+            )
+
+        return self.async_show_form(
+            step_id="discovery_confirm",
+            description_placeholders={"name": self._hostname},
+            data_schema=vol.Schema(
+                {
+                    vol.Optional(
+                        CONF_NAME,
+                        default=self._name,
+                    ): str,
+                    vol.Required(
+                        CONF_NUM_DOORS,
+                        default=1,
+                    ): vol.All(vol.Coerce(int), vol.Clamp(min=1, max=3)),
+                }
+            ),
+        )

custom_components/tailwind_iq3/const.py

@@ -0,0 +1,10 @@
+"""Constants for the tailwind_iq3 integration."""
+
+DOMAIN = "tailwind_iq3"
+
+TAILWIND_COORDINATOR = "coordinator"
+UPDATE_INTERVAL = 10
+
+CONF_NUM_DOORS = "num_doors"
+
+ATTR_RAW_STATE = "raw_state"

custom_components/tailwind_iq3/cover.py

@@ -0,0 +1,126 @@
+"""Support for Tailwind iQ3 Garage Door Openers."""
+import logging
+
+from homeassistant.components.cover import (
+    DEVICE_CLASS_GARAGE,
+    SUPPORT_CLOSE,
+    SUPPORT_OPEN,
+    CoverEntity,
+)
+from homeassistant.const import CONF_IP_ADDRESS
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from . import tailwind_send_command
+from .const import CONF_NUM_DOORS, DOMAIN, TAILWIND_COORDINATOR, ATTR_RAW_STATE
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    """Set up cover entities."""
+    data = hass.data[DOMAIN][config_entry.entry_id]
+    conf = config_entry.data
+    coordinator = data[TAILWIND_COORDINATOR]
+    num_doors = conf[CONF_NUM_DOORS] if CONF_NUM_DOORS in conf else 1
+
+    async_add_entities(
+        [TailwindCover(hass, coordinator, device) for device in range(num_doors)]
+    )
+
+
+class TailwindCover(CoordinatorEntity, CoverEntity):
+    """Representation of a Tailwind iQ3 cover."""
+
+    _attr_supported_features = SUPPORT_OPEN | SUPPORT_CLOSE
+    _attr_device_class = DEVICE_CLASS_GARAGE
+
+    def __init__(self, hass, coordinator, device):
+        """Initialize with API object, device id."""
+        super().__init__(coordinator)
+        self._coordinator = coordinator
+        self._device = device
+        self._hass = hass
+
+        # Default name to match Tailwind app's default names.
+        door_letter = ["A", "B", "C"][device]
+        self._attr_name = f"Garage {door_letter}"
+
+        # Set unique ID based on device unique ID and door number.
+        coordinator_id = coordinator.config_entry.unique_id.replace("tailwind-", "")
+        self._attr_unique_id = f"{coordinator_id}_door_{device}"
+
+    @property
+    def is_closed(self):
+        if self._coordinator.data is None:
+            return None
+
+        if self._coordinator.data[ATTR_RAW_STATE] == -1:
+            return None
+
+        return not self.is_open
+
+    @property
+    def is_closing(self):
+        """Return if the cover is closing or not."""
+        return False
+
+    @property
+    def is_open(self):
+        if self._coordinator.data is None:
+            return None
+
+        if self._coordinator.data[ATTR_RAW_STATE] == -1:
+            return None
+
+        raw_state = self._coordinator.data[ATTR_RAW_STATE]
+        bit_pos = 1 << self._device
+
+        """Return true if cover is open, else False."""
+        return raw_state & bit_pos
+
+    @property
+    def is_opening(self):
+        """Return if the cover is opening or not."""
+        return False
+
+    async def async_close_cover(self, **kwargs):
+        """Issue close command to cover."""
+        _LOGGER.info("Close door: %s", self._device)
+        if self.is_closing or self.is_closed:
+            return
+
+        if self._coordinator.config_entry.data is None:
+            return
+
+        ip_address = self._coordinator.config_entry.data[CONF_IP_ADDRESS]
+        command = 1 << self._device
+        command = -1 * command
+        response = await tailwind_send_command(self._hass, ip_address, str(command))
+        if int(response) != command:
+            raise HomeAssistantError(
+                f"Closing of cover {self._device.name} failed with incorrect response: {response} (expected {command})"
+            )
+
+        # Write final state to HASS
+        self.async_write_ha_state()
+
+    async def async_open_cover(self, **kwargs):
+        """Issue open command to cover."""
+        _LOGGER.info("Open door: %s", self._device)
+        if self.is_opening or self.is_open:
+            return
+
+        if self._coordinator.config_entry.data is None:
+            return
+
+        ip_address = self._coordinator.config_entry.data[CONF_IP_ADDRESS]
+        command = 1 << self._device
+        response = await tailwind_send_command(self._hass, ip_address, str(command))
+        if int(response) != command:
+            raise HomeAssistantError(
+                f"Opening of cover {self._device.name} failed with incorrect response: {response} (expected {command})"
+            )
+
+        # Write final state to HASS
+        self.async_write_ha_state()