Add Enphase Envoy (IQ Gateway) powermeter support

CHANGELOG.md

@@ -15,6 +15,7 @@
 - **Added** MQTT Insights: optional `[MQTT_INSIGHTS]` section publishes internal state (grid power, targets, saturation, consumer topology, EMA poll interval) to MQTT with Home Assistant Device Discovery; per-consumer active/pause + manual target control; Shelly battery offline availability; auto-configured in the HA app when Mosquitto is installed ([#292](https://github.com/tomquist/astrameter/pull/292), [#294](https://github.com/tomquist/astrameter/pull/294), [#297](https://github.com/tomquist/astrameter/pull/297), [#300](https://github.com/tomquist/astrameter/pull/300), [#306](https://github.com/tomquist/astrameter/pull/306)).
 - **Added** opt-in web-based configuration editor (`WEB_CONFIG_ENABLED = True` in `[GENERAL]`) accessible at `http://<host>:52500/config`; supports editing all config sections and keys with type-appropriate inputs, comment preservation, and a Save & Restart button ([#319](https://github.com/tomquist/astrameter/pull/319)).
 - **Added** HomeWizard P1 powermeter via the device WebSocket API, with optional `VERIFY_SSL` ([#231](https://github.com/tomquist/astrameter/pull/231), [#254](https://github.com/tomquist/astrameter/pull/254)).
+- **Added** Enphase IQ Gateway (Envoy) powermeter via the local HTTPS `production.json` API, with optional Enlighten-cloud token acquisition and automatic refresh on 401, and auto-detection of single- vs three-phase readings ([#245](https://github.com/tomquist/astrameter/pull/245)).
 - **Added** SMA Energy Meter / Sunny Home Manager support via Speedwire multicast with device auto-detection and per-phase readings ([#252](https://github.com/tomquist/astrameter/pull/252)).
 - **Added** SML powermeter for smart meters over a local serial port (IR head), with optional per-phase OBIS overrides ([#229](https://github.com/tomquist/astrameter/pull/229)).
 - **Added** multi-phase support to the MQTT powermeter via `TOPICS` / `JSON_PATHS` ([#280](https://github.com/tomquist/astrameter/pull/280), [issue #136](https://github.com/tomquist/b2500-meter/issues/136)).

README.md

@@ -740,6 +740,31 @@ SERIAL = your_device_serial
 # THROTTLE_INTERVAL = 0
 ```
 
+### Enphase Envoy (IQ Gateway)
+
+Reads grid power from an [Enphase IQ Gateway / Envoy](https://enphase.com/installers/microinverters/iq-gateway) over the local HTTPS API (`/production.json?details=1`). The reading comes from the `net-consumption` measurement (positive = grid import, negative = export). Per-phase readings are reported automatically when the gateway exposes them; otherwise the aggregate single-phase value is used. Requires consumption CTs installed on the Envoy.
+
+```ini
+[ENVOY]
+HOST = 192.168.1.120
+# Option A: pre-obtained long-lived JWT (recommended)
+TOKEN = eyJ...
+# Option B: let AstraMeter fetch and refresh tokens via the Enphase Enlighten cloud
+# USERNAME = you@example.com
+# PASSWORD = your-enphase-password
+# SERIAL = 123456789012
+# Envoy ships a self-signed certificate; verification is disabled by default.
+# VERIFY_SSL = False
+```
+
+**Token acquisition.** Generate a long-lived (~1 year) static token at <https://entrez.enphaseenergy.com/>. Alternatively, configure `USERNAME`/`PASSWORD`/`SERIAL` and AstraMeter will fetch a token on first use and refresh it automatically when the Envoy returns 401.
+
+**TLS.** `VERIFY_SSL` defaults to `False` because Enphase does not publish a CA bundle for the IQ Gateway's self-signed certificate. This option **only affects the local Envoy connection** — Enphase Enlighten cloud requests (login and token endpoints) always verify TLS using the system trust store, regardless of this setting.
+
+**MFA.** The auto-fetch flow does not support Enlighten accounts with multi-factor authentication enabled. Those users must supply a static `TOKEN`.
+
+**CT direction.** If your readings have the wrong sign (export shows as import or vice versa), one or more CTs are mounted backwards. Flip them in software with the global `POWER_MULTIPLIER = -1` (or per-phase, e.g. `POWER_MULTIPLIER = 1, -1, 1`).
+
 ### SMA Energy Meter
 
 Reads an [SMA Energy Meter](https://www.sma.de/) (EM 1.0/2.0) or Sunny Home Manager via the **Speedwire** multicast protocol (UDP). The listener joins the default multicast group and reports per-phase active power (L1, L2, L3). Use `SERIAL_NUMBER = 0` to auto-detect the first meter seen on the network, or set the device serial to pin a specific unit. Like other UDP-based features, this requires the host to receive multicast traffic (use Docker host networking or equivalent).

config.ini.example

@@ -303,6 +303,22 @@ THROTTLE_INTERVAL = 0
 ## Per-powermeter throttling override (optional)
 #THROTTLE_INTERVAL = 0
 
+#[ENVOY]
+## Enphase IQ Gateway (Envoy) via local HTTPS API
+## Reads grid power from /production.json?details=1 (net-consumption).
+## Auto-detects single- vs three-phase from the response.
+#HOST = 192.168.1.120
+## Option A: long-lived JWT token from https://entrez.enphaseenergy.com/
+#TOKEN = eyJ...
+## Option B: let AstraMeter obtain a token via the Enphase Enlighten cloud
+##   (auto-refreshes on 401; not compatible with MFA-enabled Enlighten accounts)
+#USERNAME = you@example.com
+#PASSWORD = your-enphase-password
+#SERIAL = 123456789012
+## Envoy uses a self-signed certificate; verification is disabled by default.
+## Affects only the local Envoy connection — cloud requests always verify TLS.
+#VERIFY_SSL = False
+
 #[SMA_ENERGY_METER]
 ## SMA Energy Meter / Sunny Home Manager via Speedwire multicast
 ## Listens for UDP multicast broadcasts and reports per-phase power (L1, L2, L3)

src/astrameter/config/config_loader.py

@@ -13,6 +13,7 @@
 from astrameter.powermeter import (
     AmisReader,
     Emlog,
+    Envoy,
     ESPHome,
     HomeAssistant,
     HomeWizardPowermeter,
@@ -58,6 +59,7 @@
 JSON_HTTP_SECTION = "JSON_HTTP"
 TQ_EM_SECTION = "TQ_EM"
 HOMEWIZARD_SECTION = "HOMEWIZARD"
+ENVOY_SECTION = "ENVOY"
 SMA_ENERGY_METER_SECTION = "SMA_ENERGY_METER"
 MQTT_INSIGHTS_SECTION = "MQTT_INSIGHTS"
 
@@ -379,6 +381,8 @@ def create_powermeter(
         return create_json_http_powermeter(section, config)
     elif section.startswith(HOMEWIZARD_SECTION):
         return create_homewizard_powermeter(section, config)
+    elif section.startswith(ENVOY_SECTION):
+        return create_envoy_powermeter(section, config)
     elif section.startswith(SMA_ENERGY_METER_SECTION):
         return create_sma_energy_meter_powermeter(section, config)
     elif section.startswith("MQTT") and not section.startswith(MQTT_INSIGHTS_SECTION):
@@ -661,6 +665,19 @@ def create_homewizard_powermeter(
     )
 
 
+def create_envoy_powermeter(
+    section: str, config: configparser.ConfigParser
+) -> Powermeter:
+    return Envoy(
+        host=config.get(section, "HOST", fallback=""),
+        token=config.get(section, "TOKEN", fallback=""),
+        username=config.get(section, "USERNAME", fallback=""),
+        password=config.get(section, "PASSWORD", fallback=""),
+        serial=config.get(section, "SERIAL", fallback=""),
+        verify_ssl=config.getboolean(section, "VERIFY_SSL", fallback=False),
+    )
+
+
 def create_sma_energy_meter_powermeter(
     section: str, config: configparser.ConfigParser
 ) -> Powermeter:

src/astrameter/powermeter/__init__.py

@@ -1,6 +1,7 @@
 from .amisreader import AmisReader
 from .base import Powermeter
 from .emlog import Emlog
+from .envoy import Envoy
 from .esphome import ESPHome
 from .homeassistant import HomeAssistant
 from .homewizard import HomeWizardPowermeter
@@ -31,6 +32,7 @@
     "DeadbandPowermeter",
     "ESPHome",
     "Emlog",
+    "Envoy",
     "HampelPowermeter",
     "HomeAssistant",
     "HomeWizardPowermeter",

src/astrameter/powermeter/envoy.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import ssl
+from typing import Any
+
+import aiohttp
+from aiohttp import ClientResponseError, ClientTimeout, TCPConnector
+
+from .base import Powermeter
+
+logger = logging.getLogger("astrameter")
+
+ENLIGHTEN_LOGIN_URL = "https://enlighten.enphaseenergy.com/login/login.json"
+ENTREZ_TOKEN_URL = "https://entrez.enphaseenergy.com/tokens"
+DEFAULT_TIMEOUT_SECONDS = 10.0
+
+
+def _build_ssl_context(verify_ssl: bool) -> ssl.SSLContext:
+    ctx = ssl.create_default_context()
+    if not verify_ssl:
+        # Order matters: verify_mode=CERT_NONE requires check_hostname=False first.
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+    return ctx
+
+
+async def _obtain_token(
+    cloud_session: aiohttp.ClientSession,
+    username: str,
+    password: str,
+    serial: str,
+) -> str:
+    async with cloud_session.post(
+        ENLIGHTEN_LOGIN_URL,
+        data={"user[email]": username, "user[password]": password},
+    ) as resp:
+        resp.raise_for_status()
+        login_payload = await resp.json(content_type=None)
+    session_id = (
+        login_payload.get("session_id") if isinstance(login_payload, dict) else None
+    )
+    if not session_id:
+        message = (
+            login_payload.get("message", "unknown")
+            if isinstance(login_payload, dict)
+            else "unknown"
+        )
+        raise ValueError(
+            f"Envoy: Enlighten login response missing session_id (message: {message})"
+        )
+
+    async with cloud_session.post(
+        ENTREZ_TOKEN_URL,
+        json={
+            "session_id": session_id,
+            "serial_num": serial,
+            "username": username,
+        },
+    ) as resp:
+        resp.raise_for_status()
+        token = (await resp.text()).strip()
+
+    if not token.startswith("eyJ") or token.count(".") != 2:
+        raise ValueError(
+            f"Envoy: entrez token endpoint did not return a JWT (body: {token[:200]!r})"
+        )
+
+    logger.info("Envoy: obtained new JWT token from Enlighten cloud")
+    return token
+
+
+class Envoy(Powermeter):
+    def __init__(
+        self,
+        host: str,
+        token: str = "",
+        username: str = "",
+        password: str = "",
+        serial: str = "",
+        verify_ssl: bool = False,
+    ) -> None:
+        if not host:
+            raise ValueError("Envoy: HOST is required")
+        has_credentials = bool(username and password and serial)
+        if not token and not has_credentials:
+            raise ValueError("Envoy: provide either TOKEN or USERNAME/PASSWORD/SERIAL")
+
+        self.host = host
+        self._username = username
+        self._password = password
+        self._serial = serial
+        self._has_credentials = has_credentials
+        self._verify_ssl = verify_ssl
+        self._ssl_context = _build_ssl_context(verify_ssl)
+        self._token = token
+        self._token_lock = asyncio.Lock()
+        self._session: aiohttp.ClientSession | None = None
+        self._cloud_session: aiohttp.ClientSession | None = None
+
+        if not verify_ssl:
+            logger.warning(
+                "Envoy: TLS certificate verification is disabled for the local "
+                "Envoy (VERIFY_SSL=False); use only on a trusted LAN. Enphase "
+                "Enlighten cloud requests are unaffected and always use system TLS."
+            )
+
+    async def start(self) -> None:
+        if self._session is not None:
+            return
+        timeout = ClientTimeout(total=DEFAULT_TIMEOUT_SECONDS)
+        self._session = aiohttp.ClientSession(
+            connector=TCPConnector(ssl=self._ssl_context),
+            timeout=timeout,
+        )
+        # Separate session for the Enphase cloud: always uses default system TLS,
+        # never weakened by VERIFY_SSL=False on the local Envoy.
+        self._cloud_session = aiohttp.ClientSession(timeout=timeout)
+
+    async def stop(self) -> None:
+        if self._session is not None:
+            await self._session.close()
+            self._session = None
+        if self._cloud_session is not None:
+            await self._cloud_session.close()
+            self._cloud_session = None
+
+    async def _ensure_token(self) -> None:
+        if self._token:
+            return
+        async with self._token_lock:
+            if self._token:
+                return
+            if self._cloud_session is None:
+                raise RuntimeError("Cloud session not started; call start() first")
+            self._token = await _obtain_token(
+                self._cloud_session, self._username, self._password, self._serial
+            )
+
+    async def _refresh_token(self) -> None:
+        async with self._token_lock:
+            if self._cloud_session is None:
+                raise RuntimeError("Cloud session not started; call start() first")
+            self._token = await _obtain_token(
+                self._cloud_session, self._username, self._password, self._serial
+            )
+
+    async def _get_production(self) -> dict[str, Any]:
+        if self._session is None:
+            raise RuntimeError("Session not started; call start() first")
+        url = f"https://{self.host}/production.json?details=1"
+        headers = {"Authorization": f"Bearer {self._token}"}
+        async with self._session.get(url, headers=headers) as resp:
+            resp.raise_for_status()
+            data = await resp.json(content_type=None)
+        return data if isinstance(data, dict) else {}
+
+    async def _fetch_production(self) -> dict[str, Any]:
+        await self._ensure_token()
+        old_token = self._token
+        try:
+            return await self._get_production()
+        except ClientResponseError as e:
+            if e.status != 401 or not self._has_credentials:
+                raise
+            # If another coroutine already refreshed while we were awaiting,
+            # skip our own refresh and retry with the fresh token.
+            if self._token == old_token:
+                logger.info("Envoy: token rejected (401), refreshing")
+                await self._refresh_token()
+            return await self._get_production()
+
+    async def get_powermeter_watts(self) -> list[float]:
+        data = await self._fetch_production()
+        consumption = data.get("consumption")
+        if not isinstance(consumption, list):
+            raise ValueError(
+                "Envoy: production.json missing 'consumption' array; "
+                "consumption CTs are required"
+            )
+
+        entry = next(
+            (
+                c
+                for c in consumption
+                if isinstance(c, dict) and c.get("measurementType") == "net-consumption"
+            ),
+            None,
+        )
+        if entry is None:
+            raise ValueError(
+                "Envoy: response does not expose 'net-consumption'; "
+                "consumption CTs are required"
+            )
+
+        lines = entry.get("lines")
+        if isinstance(lines, list) and lines:
+            values: list[float] = []
+            for i, line in enumerate(lines[:3]):
+                if not isinstance(line, dict) or "wNow" not in line:
+                    raise ValueError(
+                        f"Envoy: malformed net-consumption line entry at index {i}"
+                    )
+                try:
+                    values.append(float(line["wNow"]))
+                except (TypeError, ValueError) as err:
+                    raise ValueError(
+                        f"Envoy: non-numeric 'wNow' in net-consumption line at index {i}"
+                    ) from err
+            return values
+
+        if "wNow" not in entry:
+            raise ValueError("Envoy: net-consumption entry missing 'wNow'")
+        try:
+            return [float(entry["wNow"])]
+        except (TypeError, ValueError) as err:
+            raise ValueError("Envoy: non-numeric 'wNow' in net-consumption") from err