Add gridX integration

.strict-typing

@@ -243,6 +243,7 @@ homeassistant.components.google_weather.*
 homeassistant.components.govee_ble.*
 homeassistant.components.gpsd.*
 homeassistant.components.greeneye_monitor.*
+homeassistant.components.gridx.*
 homeassistant.components.group.*
 homeassistant.components.guardian.*
 homeassistant.components.habitica.*

homeassistant/components/gridx/__init__.py

@@ -0,0 +1,96 @@
+"""The GridX integration."""
+
+from __future__ import annotations
+
+import httpx
+
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers.httpx_client import create_async_httpx_client
+
+from .client import async_create_connector, load_oem_config
+from .const import CONF_OEM, DOMAIN, LOGGER
+from .coordinator import GridxHistoricalCoordinator, GridxLiveCoordinator
+from .types import GridxConfigEntry, GridxData
+
+PLATFORMS = [Platform.SENSOR]
+API_BASE_URL = "https://api.gridx.de"
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: GridxConfigEntry) -> bool:
+    """Set up GridX from a config entry."""
+    username: str = entry.data[CONF_USERNAME]
+    password: str = entry.data[CONF_PASSWORD]
+    oem: str = entry.data[CONF_OEM]
+
+    config = load_oem_config(oem, username, password)
+    httpx_client = create_async_httpx_client(
+        hass,
+        auto_cleanup=False,
+        base_url=API_BASE_URL,
+    )
+
+    try:
+        connector = await async_create_connector(config, httpx_client)
+    except PermissionError as err:
+        await httpx_client.aclose()
+        LOGGER.error("GridX authentication failed: %s", err)
+        raise ConfigEntryAuthFailed(
+            translation_domain=DOMAIN,
+            translation_key="invalid_auth",
+        ) from err
+    except httpx.HTTPStatusError as err:
+        await httpx_client.aclose()
+        status = err.response.status_code if err.response else None
+        LOGGER.error("Error connecting to GridX: %s", err)
+        if status in (401, 403):
+            raise ConfigEntryAuthFailed(
+                translation_domain=DOMAIN,
+                translation_key="invalid_auth",
+            ) from err
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="cannot_connect",
+        ) from err
+    except httpx.HTTPError as err:
+        await httpx_client.aclose()
+        LOGGER.error("Error connecting to GridX: %s", err)
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="cannot_connect",
+        ) from err
+    except (RuntimeError, TypeError, ValueError) as err:
+        await httpx_client.aclose()
+        LOGGER.error("Error connecting to GridX: %s", err)
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="cannot_connect",
+        ) from err
+
+    live_coordinator = GridxLiveCoordinator(hass, entry, connector)
+    hist_coordinator = GridxHistoricalCoordinator(hass, entry, connector)
+
+    try:
+        await live_coordinator.async_config_entry_first_refresh()
+        await hist_coordinator.async_config_entry_first_refresh()
+    except Exception:
+        await connector.close()
+        raise
+
+    entry.runtime_data = GridxData(
+        connector=connector,
+        live_coordinator=live_coordinator,
+        hist_coordinator=hist_coordinator,
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: GridxConfigEntry) -> bool:
+    """Unload a GridX config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        await entry.runtime_data.connector.close()
+    return unload_ok

homeassistant/components/gridx/client.py

@@ -0,0 +1,54 @@
+"""Client helpers for the GridX integration."""
+
+from __future__ import annotations
+
+from importlib import import_module
+from importlib.resources import files
+import json
+from typing import Any, Protocol
+
+import httpx
+
+
+class GridxConnector(Protocol):
+    """Protocol for the GridX connector used by the integration."""
+
+    async def retrieve_live_data(self) -> list[dict[str, Any]]:
+        """Retrieve live data for all systems."""
+
+    async def retrieve_historical_data(
+        self,
+        *,
+        start: str,
+        end: str,
+        resolution: str,
+    ) -> list[dict[str, Any]]:
+        """Retrieve historical data for all systems."""
+
+    async def close(self) -> None:
+        """Close the connector and any owned clients."""
+
+
+def load_oem_config(oem: str, username: str, password: str) -> dict[str, Any]:
+    """Load OEM connector config and inject credentials."""
+    config_path = files("gridx_connector").joinpath("config", f"{oem}.config.json")
+    config: dict[str, Any] = json.loads(config_path.read_text())
+    config["login"]["username"] = username
+    config["login"]["password"] = password
+    return config
+
+
+async def async_create_connector(
+    config: dict[str, Any],
+    httpx_client: httpx.AsyncClient,
+) -> GridxConnector:
+    """Create a GridX connector without importing the dependency at module import time."""
+    connector_module = import_module("gridx_connector.async_connector")
+    async_gridbox_connector = connector_module.AsyncGridboxConnector
+
+    connector: GridxConnector = await async_gridbox_connector.create(
+        config,
+        httpx_client=httpx_client,
+        owns_httpx_client=True,
+    )
+    return connector