Adding a New Provider

Adding a New Provider

Developer guide for integration 1.2.x · branch HA2026_09_dev This replaces the older guide, which is preserved at Extend the integration to support a new energy provider (archived).

Adding a grid operator means writing one adapter class that translates a portal into the internal contract. The coordinator, the config flow and the sensor platform are provider-agnostic and should not need to be touched — see ADD AD-2/AD-3.

---

1. The three files you touch

File	Change
custom_components/asm/const.py	Add the provider id and its display name to PROVIDERS.
custom_components/asm/api/client_yourprovider.py	New. The adapter.
custom_components/asm/api/client.py	Register the adapter in get_client().

Optionally add a translation label for the provider name — the provider name shown in the dropdown comes from PROVIDERS, so no translation file change is needed for the value itself. If you want the label localised, add a selector translation later; the current code renders PROVIDERS directly.

---

2. Step 1 — register the provider id

# custom_components/asm/const.py

PROVIDER_WIENER_NETZE = "wiener_netze"
PROVIDER_NETZ_NOE = "netz_noe"
PROVIDER_YOURPROVIDER = "yourprovider"          # <-- add

PROVIDERS = {
    PROVIDER_WIENER_NETZE: "Wiener Netze",
    PROVIDER_NETZ_NOE: "Netz Niederösterreich (EVN)",
    PROVIDER_YOURPROVIDER: "Your Provider",     # <-- add
}

The key is what ends up in the config entry, so treat it as permanent — changing it later invalidates existing entries' unique_id.

---

3. Step 2 — write the adapter

Create custom_components/asm/api/client_yourprovider.py.

3.1 Rules

• No Home Assistant imports. Adapters are plain Python and therefore testable without the framework.
• Synchronous is fine. Every call is executed by the coordinator in the executor, so use requests (guaranteed by Core — do not add it to manifest.json).
• Raise the right error. SmartmeterLoginError for credentials problems, everything else about transport/format as SmartmeterConnectionError, "authenticated but unusable" as SmartmeterQueryError (all from api/errors.py).
• Normalise to Wh. The sensor platform declares Wh and never converts.
• Never log credentials.

3.2 Skeleton

"""Your Provider API client."""
from __future__ import annotations

import logging
from datetime import date, datetime, timedelta
from typing import Any

import requests

from .base import SmartmeterClient
from .errors import (
    SmartmeterConnectionError,
    SmartmeterLoginError,
    SmartmeterQueryError,
)

LOGGER = logging.getLogger(__name__)

BASE_URL = "https://portal.example.at/api"
REQUEST_TIMEOUT = 30
SESSION_MAX_AGE = timedelta(minutes=30)   # model the session lifetime explicitly

KWH_TO_WH = 1000
UNIT_WH = "Wh"


class YourProviderClient(SmartmeterClient):
    """Client for Your Provider."""

    def __init__(self, username, password):
        super().__init__(username, password)
        self.session = requests.Session()
        self._login_time: datetime | None = None

    # ------------------------------------------------------------- session

    def is_logged_in(self) -> bool:
        return self._login_time is not None and not self.is_login_expired()

    def is_login_expired(self) -> bool:
        if self._login_time is None:
            return True
        return datetime.now() - self._login_time >= SESSION_MAX_AGE

    def _reset(self) -> None:
        self.session.cookies.clear()
        self._login_time = None

    def _ensure_session(self) -> None:
        if not self.is_logged_in():
            self._reset()
            self.login()

    # ---------------------------------------------------------------- http

    def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
        try:
            return self.session.request(method, url, timeout=REQUEST_TIMEOUT, **kwargs)
        except requests.exceptions.RequestException as err:
            raise SmartmeterConnectionError(f"Request to {url} failed: {err}") from err

    def _get_json(self, url: str, **kwargs: Any) -> Any:
        response = self._request("GET", url, **kwargs)
        if response.status_code != 200:
            raise SmartmeterConnectionError(f"{url} returned HTTP {response.status_code}.")
        try:
            return response.json()
        except ValueError as err:
            # Portals answer with HTML when an endpoint does not exist. Report that
            # instead of the useless "Expecting value: line 1 column 1 (char 0)".
            snippet = " ".join(response.text.split())[:200]
            raise SmartmeterConnectionError(
                f"{url} returned a non-JSON response "
                f"(HTTP {response.status_code}): {snippet!r}"
            ) from err

    # --------------------------------------------------------------- login

    def login(self):
        response = self._request(
            "POST", f"{BASE_URL}/Login", json={"user": self.username, "pwd": self.password}
        )
        payload = self._safe_json(response)

        if response.status_code in (401, 403):
            raise SmartmeterLoginError("Login failed. Check username and password.")
        if response.status_code != 200:
            raise SmartmeterConnectionError(
                f"Login endpoint returned HTTP {response.status_code}."
            )
        if isinstance(payload, dict) and not payload.get("success", True):
            raise SmartmeterLoginError("Login failed. Check username and password.")

        self._login_time = datetime.now()
        return self

    # ----------------------------------------------------------- contract

    def zaehlpunkte(self) -> list[dict[str, Any]]:
        self._ensure_session()
        raw = self._get_json(f"{BASE_URL}/MeteringPoints")
        meters = [_normalise_metering_point(item) for item in raw or []]
        if not meters:
            raise SmartmeterQueryError("No metering points found in this account.")
        return [{"zaehlpunkte": meters}]

    def consumptions(self) -> list[dict[str, Any]]:
        """Return daily statistics, or [] when the portal has none."""
        return []

    def historical_data(
        self, zaehlpunktnummer: str, date_from: date = None, date_until: date = None
    ) -> list[dict[str, Any]]:
        self._ensure_session()
        if date_until is None:
            date_until = date.today()
        if date_from is None:
            date_from = date_until - timedelta(days=1)

        payload = self._get_json(
            f"{BASE_URL}/Readings",
            params={"meterId": zaehlpunktnummer, "day": date_from.isoformat()},
        )

        # --- decide the semantics of the value (see section 4) -----------------
        is_cumulative = bool(payload and payload[0].get("isMeterRead"))
        values = [float(v) for v in (payload[0].get("values") or [])]

        return [
            {
                "obisCode": "1-1:1.8.0",
                "wertetyp": "METER_READ" if is_cumulative else "DAY",
                "einheit": UNIT_WH,
                "messwerte": [
                    {
                        "zeitpunkt": f"{date_from.isoformat()}T00:00:00",
                        "messwert": round(sum(values) * KWH_TO_WH, 3),
                        "status": "VALID",
                    }
                ],
            }
        ] if values else []

---

4. Step 3 — get the semantics right

This is the part that most often goes wrong: is the number you return a meter reading or a period consumption?

Portal gives you	wertetyp	state_class Home Assistant sees	Typical OBIS
Cumulative register that only grows	METER_READ or omit	total_increasing	1-1:1.8.0, 1-1:2.8.0
Energy consumed within a day / 15 min / hour	DAY, CONSUMPTION, QUARTER_HOUR	total	1-1:1.9.0, 1-1:2.9.0

Declaring a value that resets every day as total_increasing makes Home Assistant compute huge negative deltas in the long-term statistics. If the portal only offers period values, that is fine — just mark them correctly. See ADD AD-5/AD-6.

If a day has no data, return []. The entity then reads unknown, which is honest. A 0 would be indistinguishable from a genuine zero-consumption day.

---

5. Step 4 — normalisation helpers

Keep parsing in small module-level helpers so they can be unit-tested directly:

def _normalise_metering_point(raw: Any) -> dict[str, Any]:
    if not isinstance(raw, dict):
        raise SmartmeterQueryError(f"Unexpected metering point payload: {raw!r}")

    zaehlpunktnummer = raw.get("meteringPointId") or raw.get("meterId")
    if not zaehlpunktnummer:
        raise SmartmeterQueryError("Portal returned a metering point without an id.")

    info = dict(raw)
    info["zaehlpunktnummer"] = zaehlpunktnummer
    info["zaehlpunktName"] = raw.get("name") or zaehlpunktnummer
    info["zaehlpunktAnlagentyp"] = raw.get("smartMeterType") or "CONSUMING"
    return info

The full contract your methods must satisfy is documented in SDD §3. The short version:

• zaehlpunkte() → [{"zaehlpunkte": [{"zaehlpunktnummer": ..., ...}]}]
• historical_data() → [{"obisCode": ..., "wertetyp": ..., "einheit": "Wh", "messwerte": [{"messwert": ..., "zeitpunkt": ...}]}]
• consumptions() → [{"zaehlpunktnummer": ..., "consumptionYesterday": {"value": Wh, "date": ...}}]

Everything extra that you put into the zaehlpunkte dict is automatically surfaced as diagnostic entities or as extra state attributes, so add useful portal fields — but never secrets.

---

6. Step 5 — register in the factory

# custom_components/asm/api/client.py
from .client_yourprovider import YourProviderClient

def get_client(provider: str, username, password) -> SmartmeterClient:
    if provider == PROVIDER_NETZ_NOE:
        return NetzNoeClient(username, password)
    if provider == PROVIDER_YOURPROVIDER:
        return YourProviderClient(username, password)
    return WienerNetzeClient(username, password)

Because WienerNetzeClient is the fallback, add an explicit branch rather than relying on the default.

---

7. Step 6 — test

7.1 Without Home Assistant (recommended)

The adapters do not import Home Assistant, so they can be exercised directly. Drive the session with a fake that routes URL → response and assert on the normalised output:

client = YourProviderClient("user", "pass")
client.session = FakeSession({
    ("POST", "Login"): FakeResponse(200, {"success": True}),
    ("GET", "MeteringPoints"): FakeResponse(200, [{"meteringPointId": "AT001", "name": "Home"}]),
    ("GET", "Readings"): FakeResponse(200, [{"values": [0.1, 0.2]}]),
})
assert client.zaehlpunkte()[0]["zaehlpunkte"][0]["zaehlpunktnummer"] == "AT001"
assert client.historical_data("AT001")[0]["messwerte"][0]["messwert"] == 300.0

Assert at minimum:

• successful login marks the client logged in, and the payload is what the portal expects,
• a wrong endpoint (HTML body) raises SmartmeterConnectionError, not a JSON error,
• wrong credentials raise SmartmeterLoginError,
• kWh → Wh conversion,
• the wertetyp you emit,
• an empty response yields [].

7.2 With Home Assistant

1. Restart Home Assistant to reload the code.
2. Settings → Devices & Services → Add Integration → Austria Smartmeter.
3. Your provider appears in the dropdown. Enter the credentials.
4. Enable debug logging to compare the raw portal payload against your normalised output:

logger:
  default: info
  logs:
    custom_components.asm: debug

5. Download the diagnostics file from the integration page to see the blob the coordinator produced — it contains info, readings and stats per metering point.

---

8. Checklist

• Provider constant + display name added to const.py
• Adapter class inherits SmartmeterClient and implements all six methods
• Adapter raises SmartmeterLoginError / SmartmeterConnectionError / SmartmeterQueryError appropriately
• Values normalised to Wh
• wertetyp set correctly for period values
• Empty responses return [], not a fake 0
• No credentials in logs, no Home Assistant imports in the adapter
• Registered in get_client()
• No new manifest.json requirement unless Core does not provide it
• Verified with a fake-session harness
• Provider added to the supported table in README.md and info.md
• Documentation updated: ADD §11