Initial commit

.github/workflows/hassfest.yaml

@@ -0,0 +1,14 @@
+name: Validate with hassfest
+
+on:
+  push:
+  pull_request:
+  schedule:
+    - cron: "0 0 * * *"
+
+jobs:
+  validate:
+    runs-on: "ubuntu-latest"
+    steps:
+      - uses: "actions/checkout@v2"
+      - uses: home-assistant/actions/hassfest@master

basic_test.py

@@ -0,0 +1,22 @@
+"""Simple test script to check Tigo CCA communication."""
+
+import asyncio
+import logging
+import sys
+
+from custom_components.tigo.tigo_cca import TigoCCA
+
+logging.basicConfig(
+    format="%(asctime)-15s %(funcName)s(%(lineno)d) - %(levelname)s: %(message)s",
+    stream=sys.stderr,
+    level=getattr(logging, "DEBUG", None),
+)
+
+
+async def _test_command():
+    tigo = TigoCCA("192.168.1.125", "Tigo", "$olar")
+    status = await asyncio.wait_for(tigo.get_status(), 1)
+    print(status)
+
+
+asyncio.run(_test_command())

custom_components/tigo/__init__.py

@@ -0,0 +1,62 @@
+"""Support for Tigo CCA."""
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from .const import DOMAIN, KEY_COORDINATOR, PLATFORMS
+from .coordinator import TigoUpdateCoordinator
+from .tigo_cca import TigoCCA
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up the Tigo CCA components from a config entry."""
+    hass.data.setdefault(DOMAIN, {})
+    host = entry.data[CONF_HOST]
+    username = entry.data[CONF_USERNAME]
+    password = entry.data[CONF_PASSWORD]
+
+    try:
+        _LOGGER.debug("Connecting to Tigo CCA at %s", host)
+        tigo = TigoCCA(host, username, password)
+        status = await tigo.get_status()
+    except Exception as err:
+        raise ConfigEntryNotReady from err
+
+    # Create update coordinator
+    coordinator = TigoUpdateCoordinator(hass, entry, tigo, status)
+
+    # Fetch initial data so we have data when entities subscribe
+    await coordinator.async_config_entry_first_refresh()
+
+    hass.data[DOMAIN][entry.entry_id] = {
+        KEY_COORDINATOR: coordinator,
+    }
+
+    entry.async_on_unload(entry.add_update_listener(update_listener))
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(
+        config_entry, PLATFORMS
+    )
+
+    if unload_ok:
+        hass.data[DOMAIN].pop(config_entry.entry_id)
+
+    return unload_ok
+
+
+async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+    """Handle options update."""
+    await hass.config_entries.async_reload(config_entry.entry_id)

custom_components/tigo/config_flow.py

@@ -0,0 +1,59 @@
+"""Config flow to configure Tigo CCA integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant import config_entries
+from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DEFAULT_NAME, DOMAIN
+from .tigo_cca import TigoCCA
+
+CONFIG_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_HOST, default="tigo.local"): str,
+        vol.Required(CONF_USERNAME, default="Tigo"): str,
+        vol.Required(CONF_PASSWORD, default="$olar"): str,
+    }
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class TigoFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a Tigo CCA config flow."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle a flow initialized by the user."""
+        errors = {}
+        if user_input:
+            host = user_input[CONF_HOST]
+            username = user_input[CONF_USERNAME]
+            password = user_input[CONF_PASSWORD]
+
+            try:
+                tigo = TigoCCA(host, username, password)
+                status = await tigo.get_status()
+            except Exception:
+                errors["base"] = "connection_error"
+            else:
+                await self.async_set_unique_id(status.unit_id)
+                self._abort_if_unique_id_configured()
+
+                return self.async_create_entry(
+                    title=DEFAULT_NAME,
+                    data=user_input,
+                )
+
+        return self.async_show_form(
+            step_id="user", data_schema=CONFIG_SCHEMA, errors=errors
+        )

custom_components/tigo/const.py

@@ -0,0 +1,14 @@
+"""Constants for the Tigo CCA component."""
+
+from datetime import timedelta
+
+from homeassistant.const import Platform
+
+DOMAIN = "tigo"
+
+PLATFORMS = [Platform.SENSOR]
+
+DEFAULT_NAME = "Tigo CCA"
+SCAN_INTERVAL = timedelta(seconds=30)
+
+KEY_COORDINATOR = "coordinator"

custom_components/tigo/coordinator.py

@@ -0,0 +1,66 @@
+"""Update coordinator for Tigo CCA."""
+
+from __future__ import annotations
+
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN, SCAN_INTERVAL
+from .tigo_cca import PanelStatus, TigoCCA, TigoCcaStatus
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class TigoUpdateCoordinator(DataUpdateCoordinator[TigoCcaStatus]):
+    """Gather data for the energy device."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        entry: ConfigEntry,
+        cca: TigoCCA,
+        status: TigoCcaStatus,
+    ) -> None:
+        """Initialize update coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=entry.title,
+            update_interval=SCAN_INTERVAL,
+            update_method=self._async_update_data,
+        )
+        self.cca = cca
+        self.serial_nr = status.unit_id
+        self.device_info = _cca_device(cca, status)
+        self.panel_device_infos = {
+            p.label: _panel_device(status, p) for p in status.panels.values()
+        }
+
+    async def _async_update_data(self) -> TigoCcaStatus:
+        """Fetch data from the inverter."""
+        try:
+            return await self.cca.get_status()
+        except Exception as ex:
+            raise UpdateFailed(ex) from ex
+
+
+def _cca_device(cca: TigoCCA, status: TigoCcaStatus) -> DeviceInfo:
+    return DeviceInfo(
+        configuration_url=cca.url_root,
+        identifiers={(DOMAIN, status.unit_id)},
+        name=f"Tigo CCA {status.unit_id}",
+        model=cca.hw_platform,
+        sw_version=cca.fw_version,
+    )
+
+
+def _panel_device(cca: TigoCcaStatus, panel: PanelStatus) -> DeviceInfo:
+    return DeviceInfo(
+        identifiers={(DOMAIN, panel.mac)},
+        name=f"Panel {panel.label}",
+        via_device=(DOMAIN, cca.unit_id),
+    )

custom_components/tigo/manifest.json

@@ -0,0 +1,14 @@
+{
+  "domain": "tigo",
+  "name": "Tigo CCA",
+  "codeowners": [
+    "@mletenay"
+  ],
+  "config_flow": true,
+  "documentation": "https://github.com/mletenay/home-assistant-tigo",
+  "iot_class": "local_polling",
+  "issue_tracker": "https://github.com/mletenay/home-assistant-tigo/issues",
+  "loggers": ["tigo"],
+  "requirements": [],
+  "version": "1.0.0"
+}