API Reference

Client

Client for the Pinergy smart-meter API.

Authentication is performed lazily — the first call to any API method that requires a token will trigger :meth:login automatically if no token has been set yet. Lazy login is thread-safe: concurrent first calls perform a single login. If the server rejects a token (HTTP 401), the stale token is dropped so the next call re-authenticates transparently.

The client owns a pooled :class:requests.Session (keep-alive connection reuse). Use it as a context manager, or call :meth:close when done, to release the pool deterministically::

with PinergyClient("user@example.com", "my-password") as client:
    balance = client.get_balance()
    print(f"Balance: €{balance.credit_balance:.2f}")

Parameters:

Name	Type	Description	Default
email	str	Registered Pinergy account email address.	required
password	str	Plaintext account password (hashed internally before sending).	required
base_url	str	Override the API base URL (useful for testing).	_BASE_URL
timeout	float	Default request timeout in seconds, applied to every network call. Accepts a float for sub-second precision.	30.0

Source code in src/pypinergy/client.py

class PinergyClient:
    """Client for the Pinergy smart-meter API.

    Authentication is performed lazily — the first call to any API method that
    requires a token will trigger :meth:`login` automatically if no token has
    been set yet. Lazy login is thread-safe: concurrent first calls perform a
    single login. If the server rejects a token (HTTP 401), the stale token is
    dropped so the next call re-authenticates transparently.

    The client owns a pooled :class:`requests.Session` (keep-alive connection
    reuse). Use it as a context manager, or call :meth:`close` when done, to
    release the pool deterministically::

        with PinergyClient("user@example.com", "my-password") as client:
            balance = client.get_balance()
            print(f"Balance: €{balance.credit_balance:.2f}")

    Args:
        email: Registered Pinergy account email address.
        password: Plaintext account password (hashed internally before sending).
        base_url: Override the API base URL (useful for testing).
        timeout: Default request timeout in seconds, applied to every network
            call. Accepts a float for sub-second precision.
    """

    def __init__(
        self,
        email: str,
        password: str,
        base_url: str = _BASE_URL,
        timeout: float = 30.0,
    ) -> None:
        self._email = email
        self._password_hash: Optional[str] = hash_password(password)
        self._base_url = base_url.rstrip("/")

        parsed = urllib.parse.urlparse(self._base_url)
        if parsed.scheme == "http" and parsed.hostname not in ("localhost", "127.0.0.1", "::1"):
            raise ValueError(
                "base_url must use https:// to prevent credential leakage "
                "(except for localhost/127.0.0.1/::1)"
            )

        self._timeout = timeout
        self._auth_token: Optional[str] = None
        # Reentrant: _ensure_auth() holds it while calling login(), which takes it again.
        self._auth_lock = threading.RLock()
        self._session = _NoRedirectSession()
        self._session.headers.update({"User-Agent": _USER_AGENT})

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    def close(self) -> None:
        """Release the underlying HTTP connection pool.

        Idempotent — safe to call more than once. The client must not be used
        for further API calls after closing.
        """
        self._session.close()

    def __enter__(self) -> "PinergyClient":
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc: Optional[BaseException],
        tb: Optional[TracebackType],
    ) -> None:
        self.close()

    # ------------------------------------------------------------------
    # Session helpers
    # ------------------------------------------------------------------

    @property
    def is_authenticated(self) -> bool:
        """True if an auth token is currently held."""
        return self._auth_token is not None

    def _ensure_auth(self) -> None:
        """Login if no token is present yet (thread-safe, double-checked)."""
        if self._auth_token is None:
            with self._auth_lock:
                if self._auth_token is None:
                    self.login()

    def _url(self, path: str) -> str:
        return f"{self._base_url}/{path.lstrip('/')}"

    def _request(
        self,
        method: str,
        path: str,
        *,
        json: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
        authenticated: bool = False,
        check_api_error: bool = True,
    ) -> Dict[str, Any]:
        """Perform an HTTP request and return the parsed JSON body.

        Every transport failure is mapped onto the ``PinergyError`` hierarchy:

        - timeout → :class:`PinergyTimeoutError`
        - HTTP 401 on an authenticated call → :class:`PinergyAuthError`
          (the stale token is also dropped so the next call re-authenticates)
        - token rejection reported as HTTP 200 ``success: false`` on an
          authenticated call → :class:`PinergyAuthError` (stale token dropped
          likewise)
        - other HTTP / network errors → :class:`PinergyHTTPError`
        - unparseable or non-object JSON body → :class:`PinergyResponseError`
        - other application-level ``success: false`` → :class:`PinergyAPIError`
          (unless *check_api_error* is False)
        """
        attempts = 2 if authenticated else 1

        for attempt in range(attempts):
            req_headers = dict(headers) if headers else {}
            if authenticated:
                self._ensure_auth()
            token_used = self._auth_token
            if authenticated:
                req_headers["auth_token"] = token_used or ""

            try:
                response = self._session.request(
                    method,
                    self._url(path),
                    json=json,
                    headers=req_headers,
                    timeout=self._timeout,
                )
                response.raise_for_status()
            except requests.exceptions.Timeout as exc:
                raise PinergyTimeoutError(str(exc)) from exc
            except requests.exceptions.HTTPError as exc:
                if authenticated and exc.response is not None and exc.response.status_code == 401:
                    with self._auth_lock:
                        if self._auth_token == token_used:
                            self._auth_token = None
                    if attempt < attempts - 1 and self._password_hash is not None:
                        continue
                    raise PinergyAuthError("Auth token rejected (401)") from exc
                raise PinergyHTTPError(str(exc)) from exc
            except requests.exceptions.RequestException as exc:
                raise PinergyHTTPError(str(exc)) from exc

            try:
                data = response.json()
            except ValueError as exc:
                raise PinergyResponseError(f"API returned malformed JSON: {exc}") from exc
            if not isinstance(data, dict):
                raise PinergyResponseError(f"API returned non-object JSON ({type(data).__name__})")

            if check_api_error:
                if authenticated and not data.get("success", True):
                    message = str(data.get("message") or "")
                    if _is_token_rejection(message):
                        with self._auth_lock:
                            if self._auth_token == token_used:
                                self._auth_token = None
                        if attempt < attempts - 1 and self._password_hash is not None:
                            continue
                        raise PinergyAuthError(message or "Auth token rejected")
                _raise_for_api_error(data)
            return data
        assert False, "unreachable"

    def _get(self, path: str) -> Dict[str, Any]:
        """Perform an authenticated GET and return the parsed JSON body."""
        return self._request("GET", path, authenticated=True)

    # ------------------------------------------------------------------
    # Authentication
    # ------------------------------------------------------------------

    def login(self) -> LoginResponse:
        """Authenticate with the Pinergy API and store the session token.

        Thread-safe — concurrent callers serialize on an internal lock.

        Raises:
            PinergyAuthError: If credentials are invalid, or this client was
                explicitly logged out (credentials are discarded on
                :meth:`logout`; create a new client to re-authenticate).
            PinergyTimeoutError: If the request times out.
            PinergyHTTPError: On network-level errors.
            PinergyResponseError: If the response body is malformed or is
                missing the ``auth_token``.

        Returns:
            :class:`~pypinergy.models.LoginResponse` with full account details.
        """
        with self._auth_lock:
            if self._password_hash is None:
                raise PinergyAuthError(
                    "Client has been logged out — credentials were discarded; "
                    "create a new PinergyClient to re-authenticate"
                )
            payload = {
                "email": self._email,
                "password": self._password_hash,
                "device_token": "",
            }
            data = self._request("POST", "/api/login/", json=payload, check_api_error=False)
            if not data.get("success"):
                raise PinergyAuthError(data.get("message", "Login failed") or "Login failed")

            token = data.get("auth_token")
            if not isinstance(token, str) or not token:
                raise PinergyResponseError(
                    "Login succeeded but the response did not include an auth_token"
                )
            self._auth_token = token
            return LoginResponse._from_dict(data)

    def logout(self) -> None:
        """Clear the stored auth token, effectively ending the session.

        Also clears the cached password hash so automatic re-login does not happen
        transparently after explicit logout. Subsequent API calls raise
        :class:`PinergyAuthError` immediately, without touching the network.
        """
        with self._auth_lock:
            self._auth_token = None
            self._password_hash = None

    def check_email(self, email: str) -> bool:
        """Check whether an email address has a registered Pinergy account.

        This endpoint does **not** require authentication.

        Args:
            email: Email address to check.

        Returns:
            True if the address is registered.
        """
        if "\r" in email or "\n" in email:
            raise ValueError("Email address cannot contain carriage return or line feed characters")
        data = self._request(
            "GET",
            "/api/checkemail",
            headers={"email_address": email},
            check_api_error=False,
        )
        return bool(data.get("success"))

    # ------------------------------------------------------------------
    # Usage
    # ------------------------------------------------------------------

    def get_usage(self) -> UsageResponse:
        """Return daily, weekly, and monthly aggregated usage.

        Returns:
            :class:`~pypinergy.models.UsageResponse` with ``day``, ``week``,
            and ``month`` lists of :class:`~pypinergy.models.UsageEntry`.

        Example::

            usage = client.get_usage()
            for entry in usage.day:
                print(f"{entry.date:%Y-%m-%d}  {entry.kwh:.2f} kWh  €{entry.amount:.2f}")
        """
        return UsageResponse._from_dict(self._get("/api/usage/"))

    def get_level_pay_usage(self) -> LevelPayUsageResponse:
        """Return half-hourly interval data for level-pay customers.

        Returns:
            :class:`~pypinergy.models.LevelPayUsageResponse`
        """
        return LevelPayUsageResponse._from_dict(self._get("/api/levelpayusage/"))

    # ------------------------------------------------------------------
    # Balance
    # ------------------------------------------------------------------

    def get_balance(self) -> BalanceResponse:
        """Return the current credit balance and meter status.

        Returns:
            :class:`~pypinergy.models.BalanceResponse`

        Example::

            bal = client.get_balance()
            print(f"€{bal.credit_balance:.2f} — {bal.top_up_in_days} days remaining")
            if bal.credit_low:
                print("Warning: credit is low!")
        """
        return BalanceResponse._from_dict(self._get("/api/balance/"))

    # ------------------------------------------------------------------
    # Top-ups
    # ------------------------------------------------------------------

    def get_active_topups(self) -> ActiveTopUpsResponse:
        """Return scheduled and automatic top-up configurations.

        Returns:
            :class:`~pypinergy.models.ActiveTopUpsResponse`
        """
        return ActiveTopUpsResponse._from_dict(self._get("/api/activetopups/"))

    # ------------------------------------------------------------------
    # Compare
    # ------------------------------------------------------------------

    def compare_usage(self) -> CompareResponse:
        """Compare this home's usage with similar homes.

        Returns:
            :class:`~pypinergy.models.CompareResponse` with ``day``, ``week``,
            and ``month`` :class:`~pypinergy.models.ComparePeriod` objects.

        Example::

            cmp = client.compare_usage()
            d = cmp.day
            print(f"Today: {d.kwh.users_home:.1f} kWh vs avg {d.kwh.average_home:.1f} kWh")
        """
        return CompareResponse._from_dict(self._get("/api/compare/"))

    # ------------------------------------------------------------------
    # Configuration
    # ------------------------------------------------------------------

    def get_config_info(self) -> ConfigInfoResponse:
        """Return valid top-up amounts and alert thresholds.

        Returns:
            :class:`~pypinergy.models.ConfigInfoResponse`
        """
        return ConfigInfoResponse._from_dict(self._get("/api/configinfo/"))

    def get_defaults_info(self) -> DefaultsInfoResponse:
        """Return reference data for house and heating types.

        Returns:
            :class:`~pypinergy.models.DefaultsInfoResponse`
        """
        return DefaultsInfoResponse._from_dict(self._get("/api/defaultsinfo/"))

    # ------------------------------------------------------------------
    # Notifications
    # ------------------------------------------------------------------

    def get_notification_preferences(self) -> NotificationPreferences:
        """Return the account's notification channel preferences.

        Returns:
            :class:`~pypinergy.models.NotificationPreferences`
        """
        return NotificationPreferences._from_dict(self._get("/api/getnotif/"))

    def update_notification_preferences(
        self,
        sms: bool,
        email: bool,
        phone: bool,
    ) -> bool:
        """Update the account's notification channel preferences.

        Args:
            sms: Enable or disable SMS notifications.
            email: Enable or disable email notifications.
            phone: Enable or disable phone notifications.

        Returns:
            True on success.
        """
        payload = {
            "sms": sms,
            "email": email,
            "phone": phone,
        }
        data = self._request("POST", "/api/setnotif/", json=payload, authenticated=True)
        return bool(data.get("success"))

    # ------------------------------------------------------------------
    # Top-Up
    # ------------------------------------------------------------------

    def top_up(self, amount: float, cc_token: str) -> TopUpResponse:
        """Initiate an instant top-up using a saved payment card.

        Args:
            amount: The amount to top up (€). Must be a positive value from the
                list of valid top-up amounts returned by :meth:`get_config_info`.
            cc_token: The tokenised credit card identifier. Use
                :attr:`~pypinergy.models.CreditCard.cc_token` from the login
                response.

        Returns:
            :class:`~pypinergy.models.TopUpResponse` with the updated balance
            and transaction details.
        """
        if amount <= 0:
            raise ValueError("Top-up amount must be positive.")
        if not cc_token:
            raise ValueError("cc_token must be a non-empty string.")

        payload = {
            "amount": amount,
            "cc_token": cc_token,
        }
        data = self._request("POST", "/api/instanttopup/", json=payload, authenticated=True)
        return TopUpResponse._from_dict(data)

    # ------------------------------------------------------------------
    # Device
    # ------------------------------------------------------------------

    def update_device_token(
        self,
        device_token: str = "",
        device_type: str = "android",
        os_version: str = "",
    ) -> bool:
        """Register or update the FCM push-notification token.

        For headless / server-side use you can pass an empty string for
        *device_token* or skip this call entirely.

        Args:
            device_token: Firebase Cloud Messaging token.
            device_type: Platform string (e.g. ``"android"``).
            os_version: OS version string.

        Returns:
            True on success.
        """
        payload = {
            "device_token": device_token,
            "device_type": device_type,
            "os_version": os_version,
        }
        data = self._request("POST", "/api/updatedevicetoken/", json=payload, authenticated=True)
        return bool(data.get("success"))

    # ------------------------------------------------------------------
    # Version (unauthenticated)
    # ------------------------------------------------------------------

    def get_version(self) -> Dict[str, Any]:
        """Return the raw version config JSON (unauthenticated).

        Returns:
            Parsed JSON dict from ``/version.json``.
        """
        return self._request("GET", "/version.json", check_api_error=False)

    def __repr__(self) -> str:
        auth_status = "authenticated" if self.is_authenticated else "unauthenticated"
        return f"PinergyClient(email={self._email!r}, status={auth_status})"

is_authenticated property

True if an auth token is currently held.

check_email(email)

Check whether an email address has a registered Pinergy account.

This endpoint does not require authentication.

Parameters:

Name	Type	Description	Default
email	str	Email address to check.	required

Returns:

Type	Description
bool	True if the address is registered.

Source code in src/pypinergy/client.py

def check_email(self, email: str) -> bool:
    """Check whether an email address has a registered Pinergy account.

    This endpoint does **not** require authentication.

    Args:
        email: Email address to check.

    Returns:
        True if the address is registered.
    """
    if "\r" in email or "\n" in email:
        raise ValueError("Email address cannot contain carriage return or line feed characters")
    data = self._request(
        "GET",
        "/api/checkemail",
        headers={"email_address": email},
        check_api_error=False,
    )
    return bool(data.get("success"))

close()

Release the underlying HTTP connection pool.

Idempotent — safe to call more than once. The client must not be used for further API calls after closing.

Source code in src/pypinergy/client.py

def close(self) -> None:
    """Release the underlying HTTP connection pool.

    Idempotent — safe to call more than once. The client must not be used
    for further API calls after closing.
    """
    self._session.close()

compare_usage()

Compare this home's usage with similar homes.

Returns:

Type	Description
CompareResponse	class:~pypinergy.models.CompareResponse with day, week,
CompareResponse	and month :class:~pypinergy.models.ComparePeriod objects.

Example::

cmp = client.compare_usage()
d = cmp.day
print(f"Today: {d.kwh.users_home:.1f} kWh vs avg {d.kwh.average_home:.1f} kWh")

Source code in src/pypinergy/client.py

def compare_usage(self) -> CompareResponse:
    """Compare this home's usage with similar homes.

    Returns:
        :class:`~pypinergy.models.CompareResponse` with ``day``, ``week``,
        and ``month`` :class:`~pypinergy.models.ComparePeriod` objects.

    Example::

        cmp = client.compare_usage()
        d = cmp.day
        print(f"Today: {d.kwh.users_home:.1f} kWh vs avg {d.kwh.average_home:.1f} kWh")
    """
    return CompareResponse._from_dict(self._get("/api/compare/"))

get_active_topups()

Return scheduled and automatic top-up configurations.

Returns:

Type	Description
ActiveTopUpsResponse	class:~pypinergy.models.ActiveTopUpsResponse

Source code in src/pypinergy/client.py

def get_active_topups(self) -> ActiveTopUpsResponse:
    """Return scheduled and automatic top-up configurations.

    Returns:
        :class:`~pypinergy.models.ActiveTopUpsResponse`
    """
    return ActiveTopUpsResponse._from_dict(self._get("/api/activetopups/"))

get_balance()

Return the current credit balance and meter status.

Returns:

Type	Description
BalanceResponse	class:~pypinergy.models.BalanceResponse

Example::

bal = client.get_balance()
print(f"€{bal.credit_balance:.2f} — {bal.top_up_in_days} days remaining")
if bal.credit_low:
    print("Warning: credit is low!")

Source code in src/pypinergy/client.py

def get_balance(self) -> BalanceResponse:
    """Return the current credit balance and meter status.

    Returns:
        :class:`~pypinergy.models.BalanceResponse`

    Example::

        bal = client.get_balance()
        print(f"€{bal.credit_balance:.2f} — {bal.top_up_in_days} days remaining")
        if bal.credit_low:
            print("Warning: credit is low!")
    """
    return BalanceResponse._from_dict(self._get("/api/balance/"))

get_config_info()

Return valid top-up amounts and alert thresholds.

Returns:

Type	Description
ConfigInfoResponse	class:~pypinergy.models.ConfigInfoResponse

Source code in src/pypinergy/client.py

def get_config_info(self) -> ConfigInfoResponse:
    """Return valid top-up amounts and alert thresholds.

    Returns:
        :class:`~pypinergy.models.ConfigInfoResponse`
    """
    return ConfigInfoResponse._from_dict(self._get("/api/configinfo/"))

get_defaults_info()

Return reference data for house and heating types.

Returns:

Type	Description
DefaultsInfoResponse	class:~pypinergy.models.DefaultsInfoResponse

Source code in src/pypinergy/client.py

def get_defaults_info(self) -> DefaultsInfoResponse:
    """Return reference data for house and heating types.

    Returns:
        :class:`~pypinergy.models.DefaultsInfoResponse`
    """
    return DefaultsInfoResponse._from_dict(self._get("/api/defaultsinfo/"))

get_level_pay_usage()

Return half-hourly interval data for level-pay customers.

Returns:

Type	Description
LevelPayUsageResponse	class:~pypinergy.models.LevelPayUsageResponse

Source code in src/pypinergy/client.py

def get_level_pay_usage(self) -> LevelPayUsageResponse:
    """Return half-hourly interval data for level-pay customers.

    Returns:
        :class:`~pypinergy.models.LevelPayUsageResponse`
    """
    return LevelPayUsageResponse._from_dict(self._get("/api/levelpayusage/"))

get_notification_preferences()

Return the account's notification channel preferences.

Returns:

Type	Description
NotificationPreferences	class:~pypinergy.models.NotificationPreferences

Source code in src/pypinergy/client.py

def get_notification_preferences(self) -> NotificationPreferences:
    """Return the account's notification channel preferences.

    Returns:
        :class:`~pypinergy.models.NotificationPreferences`
    """
    return NotificationPreferences._from_dict(self._get("/api/getnotif/"))

get_usage()

Return daily, weekly, and monthly aggregated usage.

Returns:

Type	Description
UsageResponse	class:~pypinergy.models.UsageResponse with day, week,
UsageResponse	and month lists of :class:~pypinergy.models.UsageEntry.

Example::

usage = client.get_usage()
for entry in usage.day:
    print(f"{entry.date:%Y-%m-%d}  {entry.kwh:.2f} kWh  €{entry.amount:.2f}")

Source code in src/pypinergy/client.py

def get_usage(self) -> UsageResponse:
    """Return daily, weekly, and monthly aggregated usage.

    Returns:
        :class:`~pypinergy.models.UsageResponse` with ``day``, ``week``,
        and ``month`` lists of :class:`~pypinergy.models.UsageEntry`.

    Example::

        usage = client.get_usage()
        for entry in usage.day:
            print(f"{entry.date:%Y-%m-%d}  {entry.kwh:.2f} kWh  €{entry.amount:.2f}")
    """
    return UsageResponse._from_dict(self._get("/api/usage/"))

get_version()

Return the raw version config JSON (unauthenticated).

Returns:

Type	Description
Dict[str, Any]	Parsed JSON dict from /version.json.

Source code in src/pypinergy/client.py

def get_version(self) -> Dict[str, Any]:
    """Return the raw version config JSON (unauthenticated).

    Returns:
        Parsed JSON dict from ``/version.json``.
    """
    return self._request("GET", "/version.json", check_api_error=False)

login()

Authenticate with the Pinergy API and store the session token.

Thread-safe — concurrent callers serialize on an internal lock.

Raises:

Type	Description
PinergyAuthError	If credentials are invalid, or this client was explicitly logged out (credentials are discarded on :meth:logout; create a new client to re-authenticate).
PinergyTimeoutError	If the request times out.
PinergyHTTPError	On network-level errors.
PinergyResponseError	If the response body is malformed or is missing the auth_token.

Returns:

Type	Description
LoginResponse	class:~pypinergy.models.LoginResponse with full account details.

Source code in src/pypinergy/client.py

def login(self) -> LoginResponse:
    """Authenticate with the Pinergy API and store the session token.

    Thread-safe — concurrent callers serialize on an internal lock.

    Raises:
        PinergyAuthError: If credentials are invalid, or this client was
            explicitly logged out (credentials are discarded on
            :meth:`logout`; create a new client to re-authenticate).
        PinergyTimeoutError: If the request times out.
        PinergyHTTPError: On network-level errors.
        PinergyResponseError: If the response body is malformed or is
            missing the ``auth_token``.

    Returns:
        :class:`~pypinergy.models.LoginResponse` with full account details.
    """
    with self._auth_lock:
        if self._password_hash is None:
            raise PinergyAuthError(
                "Client has been logged out — credentials were discarded; "
                "create a new PinergyClient to re-authenticate"
            )
        payload = {
            "email": self._email,
            "password": self._password_hash,
            "device_token": "",
        }
        data = self._request("POST", "/api/login/", json=payload, check_api_error=False)
        if not data.get("success"):
            raise PinergyAuthError(data.get("message", "Login failed") or "Login failed")

        token = data.get("auth_token")
        if not isinstance(token, str) or not token:
            raise PinergyResponseError(
                "Login succeeded but the response did not include an auth_token"
            )
        self._auth_token = token
        return LoginResponse._from_dict(data)

logout()

Clear the stored auth token, effectively ending the session.

Also clears the cached password hash so automatic re-login does not happen transparently after explicit logout. Subsequent API calls raise :class:PinergyAuthError immediately, without touching the network.

Source code in src/pypinergy/client.py

def logout(self) -> None:
    """Clear the stored auth token, effectively ending the session.

    Also clears the cached password hash so automatic re-login does not happen
    transparently after explicit logout. Subsequent API calls raise
    :class:`PinergyAuthError` immediately, without touching the network.
    """
    with self._auth_lock:
        self._auth_token = None
        self._password_hash = None

top_up(amount, cc_token)

Initiate an instant top-up using a saved payment card.

Parameters:

Name	Type	Description	Default
amount	float	The amount to top up (€). Must be a positive value from the list of valid top-up amounts returned by :meth:get_config_info.	required
cc_token	str	The tokenised credit card identifier. Use :attr:~pypinergy.models.CreditCard.cc_token from the login response.	required

Returns:

Type	Description
TopUpResponse	class:~pypinergy.models.TopUpResponse with the updated balance
TopUpResponse	and transaction details.

Source code in src/pypinergy/client.py

def top_up(self, amount: float, cc_token: str) -> TopUpResponse:
    """Initiate an instant top-up using a saved payment card.

    Args:
        amount: The amount to top up (€). Must be a positive value from the
            list of valid top-up amounts returned by :meth:`get_config_info`.
        cc_token: The tokenised credit card identifier. Use
            :attr:`~pypinergy.models.CreditCard.cc_token` from the login
            response.

    Returns:
        :class:`~pypinergy.models.TopUpResponse` with the updated balance
        and transaction details.
    """
    if amount <= 0:
        raise ValueError("Top-up amount must be positive.")
    if not cc_token:
        raise ValueError("cc_token must be a non-empty string.")

    payload = {
        "amount": amount,
        "cc_token": cc_token,
    }
    data = self._request("POST", "/api/instanttopup/", json=payload, authenticated=True)
    return TopUpResponse._from_dict(data)

update_device_token(device_token='', device_type='android', os_version='')

Register or update the FCM push-notification token.

For headless / server-side use you can pass an empty string for device_token or skip this call entirely.

Parameters:

Name	Type	Description	Default
device_token	str	Firebase Cloud Messaging token.	''
device_type	str	Platform string (e.g. "android").	'android'
os_version	str	OS version string.	''

Returns:

Type	Description
bool	True on success.

Source code in src/pypinergy/client.py

def update_device_token(
    self,
    device_token: str = "",
    device_type: str = "android",
    os_version: str = "",
) -> bool:
    """Register or update the FCM push-notification token.

    For headless / server-side use you can pass an empty string for
    *device_token* or skip this call entirely.

    Args:
        device_token: Firebase Cloud Messaging token.
        device_type: Platform string (e.g. ``"android"``).
        os_version: OS version string.

    Returns:
        True on success.
    """
    payload = {
        "device_token": device_token,
        "device_type": device_type,
        "os_version": os_version,
    }
    data = self._request("POST", "/api/updatedevicetoken/", json=payload, authenticated=True)
    return bool(data.get("success"))

update_notification_preferences(sms, email, phone)

Update the account's notification channel preferences.

Parameters:

Name	Type	Description	Default
sms	bool	Enable or disable SMS notifications.	required
email	bool	Enable or disable email notifications.	required
phone	bool	Enable or disable phone notifications.	required

Returns:

Type	Description
bool	True on success.

Source code in src/pypinergy/client.py

def update_notification_preferences(
    self,
    sms: bool,
    email: bool,
    phone: bool,
) -> bool:
    """Update the account's notification channel preferences.

    Args:
        sms: Enable or disable SMS notifications.
        email: Enable or disable email notifications.
        phone: Enable or disable phone notifications.

    Returns:
        True on success.
    """
    payload = {
        "sms": sms,
        "email": email,
        "phone": phone,
    }
    data = self._request("POST", "/api/setnotif/", json=payload, authenticated=True)
    return bool(data.get("success"))

---

Exceptions

Bases: Exception

Base exception for all pypinergy errors.

Source code in src/pypinergy/exceptions.py

class PinergyError(Exception):
    """Base exception for all pypinergy errors."""

Bases: PinergyError

Raised when the API returns a non-success response.

Attributes:

Name	Type	Description
error_code		The integer error code returned by the API.
message		The human-readable error message from the API.

Source code in src/pypinergy/exceptions.py

class PinergyAPIError(PinergyError):
    """Raised when the API returns a non-success response.

    Attributes:
        error_code: The integer error code returned by the API.
        message: The human-readable error message from the API.
    """

    def __init__(self, message: str, error_code: int = 0) -> None:
        super().__init__(message)
        self.error_code = error_code
        self.message = message

    def __repr__(self) -> str:
        return f"PinergyAPIError(message={self.message!r}, error_code={self.error_code})"

Bases: PinergyError

Raised when authentication fails or the session has expired.

Source code in src/pypinergy/exceptions.py

class PinergyAuthError(PinergyError):
    """Raised when authentication fails or the session has expired."""

Bases: PinergyError

Raised when an unexpected HTTP-level error occurs (e.g. 5xx, timeout).

Source code in src/pypinergy/exceptions.py

class PinergyHTTPError(PinergyError):
    """Raised when an unexpected HTTP-level error occurs (e.g. 5xx, timeout)."""

Bases: PinergyHTTPError

Raised when a request exceeds the configured timeout.

Subclasses :class:PinergyHTTPError, so existing except PinergyHTTPError handlers continue to catch timeouts exactly as before.

Source code in src/pypinergy/exceptions.py

class PinergyTimeoutError(PinergyHTTPError):
    """Raised when a request exceeds the configured timeout.

    Subclasses :class:`PinergyHTTPError`, so existing ``except PinergyHTTPError``
    handlers continue to catch timeouts exactly as before.
    """

Bases: PinergyError, ValueError

Raised when the API returns a payload that cannot be interpreted.

Covers malformed JSON bodies, non-object top-level JSON, and successful responses that are missing structurally required fields (e.g. a login response without an auth_token).

Also subclasses :class:ValueError, so callers that previously caught the raw JSON decoding error (a ValueError subclass) keep working.

Source code in src/pypinergy/exceptions.py

class PinergyResponseError(PinergyError, ValueError):
    """Raised when the API returns a payload that cannot be interpreted.

    Covers malformed JSON bodies, non-object top-level JSON, and successful
    responses that are missing structurally required fields (e.g. a login
    response without an ``auth_token``).

    Also subclasses :class:`ValueError`, so callers that previously caught the
    raw JSON decoding error (a ``ValueError`` subclass) keep working.
    """

---

Login Models

Successful login payload.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class LoginResponse:
    """Successful login payload."""

    auth_token: str = field(repr=False)
    is_legacy_meter: bool
    is_no_wan_meter: bool
    is_level_pay: bool
    is_child: bool
    is_business_connect: bool
    premises_number: str
    account_type: str
    user: User
    house: House
    credit_cards: List[CreditCard]

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "LoginResponse":
        d = _as_dict(d)
        # Performance optimization: List comprehension with locally cached
        # classmethod reference speeds up the array parsing loop by ~10% over list(map(...))
        _cc_from_dict = CreditCard._from_dict
        return cls(
            auth_token=_to_str(d.get("auth_token")),
            is_legacy_meter=bool(d.get("is_legacy_meter", False)),
            is_no_wan_meter=bool(d.get("is_no_wan_meter", False)),
            is_level_pay=bool(d.get("is_level_pay", False)),
            is_child=bool(d.get("is_child", False)),
            is_business_connect=bool(d.get("is_business_connect", False)),
            premises_number=_to_str(d.get("premises_number")),
            account_type=_to_str(d.get("account_type")),
            user=User._from_dict(_as_dict(d.get("user"))),
            house=House._from_dict(_as_dict(d.get("house"))),
            credit_cards=[_cc_from_dict(x) for x in _as_list(d.get("credit_cards"))],
        )

Authenticated user profile.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class User:
    """Authenticated user profile."""

    title: str
    name: str
    pinergy_id: str
    mobile_number: str = field(repr=False)
    sms_notifications: bool
    email_notifications: bool
    first_name: str
    last_name: str

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "User":
        d = _as_dict(d)
        return cls(
            title=_to_str(d.get("title")),
            name=_to_str(d.get("name")),
            pinergy_id=_to_str(d.get("pinergy_id")),
            mobile_number=_to_str(d.get("mobile_number")),
            sms_notifications=bool(d.get("sms_notifications", False)),
            email_notifications=bool(d.get("email_notifications", False)),
            first_name=_to_str(d.get("firstName")),
            last_name=_to_str(d.get("lastName")),
        )

Property details associated with the account.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class House:
    """Property details associated with the account."""

    type: int
    heating_type: int
    bedroom_count: int
    adult_count: int
    children_count: int

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "House":
        d = _as_dict(d)
        return cls(
            type=_to_int(d.get("type")),
            heating_type=_to_int(d.get("heating_type")),
            bedroom_count=_to_int(d.get("bedroom_count")),
            adult_count=_to_int(d.get("adult_count")),
            children_count=_to_int(d.get("children_count")),
        )

Saved payment card summary.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class CreditCard:
    """Saved payment card summary."""

    cc_token: str = field(repr=False)
    name: str
    last_4_digits: str

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "CreditCard":
        d = _as_dict(d)
        return cls(
            cc_token=_to_str(d.get("cc_token")),
            name=_to_str(d.get("name")),
            last_4_digits=_to_str(d.get("last_4_digits")),
        )

---

Usage Models

Aggregated usage across day / week / month buckets.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class UsageResponse:
    """Aggregated usage across day / week / month buckets."""

    day: List[UsageEntry]
    """Last 7 days — one entry per day."""
    week: List[UsageEntry]
    """Last 8 weeks — one entry per week."""
    month: List[UsageEntry]
    """Last 11 months — one entry per month."""

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "UsageResponse":
        d = _as_dict(d)
        # Performance optimization: List comprehension with locally cached
        # classmethod reference speeds up the array parsing loop by ~10% over list(map(...))
        _ue_from_dict = UsageEntry._from_dict
        return cls(
            day=[_ue_from_dict(x) for x in _as_list(d.get("day"))],
            week=[_ue_from_dict(x) for x in _as_list(d.get("week"))],
            month=[_ue_from_dict(x) for x in _as_list(d.get("month"))],
        )

day instance-attribute

Last 7 days — one entry per day.

month instance-attribute

Last 11 months — one entry per month.

week instance-attribute

Last 8 weeks — one entry per week.

A single aggregated usage period (day / week / month).

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class UsageEntry:
    """A single aggregated usage period (day / week / month)."""

    available: bool
    amount: float
    """Cost in euros (€)."""
    kwh: float
    """Energy consumed in kilowatt-hours."""
    co2: float
    """CO₂ in kg (typically 0.0 for renewable supply)."""
    date_ts: int
    """Raw Unix timestamp (start of period)."""
    date: datetime
    """UTC datetime for the start of the period."""

    @property
    def date_dt(self) -> datetime:
        """Alias for :attr:`date`, following the ``*_dt`` naming convention.

        :attr:`date` remains fully supported; both names refer to the same value.
        """
        return self.date

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "UsageEntry":
        d = _as_dict(d)
        ts_int, dt = _parse_ts_pair(d.get("date", 0))
        return cls(
            available=bool(d.get("available", False)),
            amount=_to_float(d.get("amount")),
            kwh=_to_float(d.get("kwh")),
            co2=_to_float(d.get("co2")),
            date_ts=ts_int or 0,
            # Re-use the constant instead of instantiating a new aware datetime per fallback
            date=dt or _EPOCH_UTC,
        )

amount instance-attribute

Cost in euros (€).

co2 instance-attribute

CO₂ in kg (typically 0.0 for renewable supply).

date instance-attribute

UTC datetime for the start of the period.

date_dt property

Alias for :attr:date, following the *_dt naming convention.

:attr:date remains fully supported; both names refer to the same value.

date_ts instance-attribute

Raw Unix timestamp (start of period).

kwh instance-attribute

Energy consumed in kilowatt-hours.

Half-hourly interval data for level pay customers.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class LevelPayUsageResponse:
    """Half-hourly interval data for level pay customers."""

    labels: List[str]
    flags: List[str]
    values: List[LevelPayDailyValue]

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "LevelPayUsageResponse":
        d = _as_dict(d)
        usage_data = _as_dict(d.get("usageData"))
        daily = _as_dict(usage_data.get("daily"))
        # Performance optimization: List comprehension with locally cached
        # classmethod reference speeds up the array parsing loop by ~10% over list(map(...))
        _lp_from_dict = LevelPayDailyValue._from_dict
        return cls(
            labels=[_to_str(x) for x in _as_list(daily.get("labels"))],
            flags=[_to_str(x) for x in _as_list(daily.get("flags"))],
            values=[_lp_from_dict(x) for x in _as_list(daily.get("values")) if x is not None],
        )

Half-hourly label and kWh per tariff band.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class LevelPayDailyValue:
    """Half-hourly label and kWh per tariff band."""

    label: str
    day_kwh: Dict[str, float]
    """Mapping of tariff band name (e.g. ``"Standard"``) to kWh consumed."""

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "LevelPayDailyValue":
        d = _as_dict(d)
        day_kwh_raw = d.get("daykWh")
        day_kwh = {}
        if isinstance(day_kwh_raw, dict):
            day_kwh = {_to_str(k): _to_float(v) for k, v in day_kwh_raw.items() if k is not None}
        return cls(
            label=_to_str(d.get("label")),
            day_kwh=day_kwh,
        )

day_kwh instance-attribute

Mapping of tariff band name (e.g. "Standard") to kWh consumed.

---

Balance Model

Current account balance and meter status.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class BalanceResponse:
    """Current account balance and meter status."""

    credit_balance: float
    """Current credit balance in euros (€)."""
    top_up_in_days: int
    """Estimated days until credit is exhausted."""
    pending_top_up: bool
    pending_top_up_by: str
    last_top_up_amount: float
    credit_low: bool
    """True when balance is below the configured alert threshold."""
    emergency_credit: bool
    """True when the meter is drawing on emergency credit."""
    power_off: bool
    """True when supply has been disconnected."""
    last_top_up_ts: Optional[int]
    last_top_up_time: Optional[datetime]
    last_reading_ts: Optional[int]
    last_reading: Optional[datetime]

    @property
    def last_top_up_dt(self) -> Optional[datetime]:
        """Alias for :attr:`last_top_up_time`, following the ``*_dt`` naming convention.

        :attr:`last_top_up_time` remains fully supported; both names refer to the same value.
        """
        return self.last_top_up_time

    @property
    def last_reading_dt(self) -> Optional[datetime]:
        """Alias for :attr:`last_reading`, following the ``*_dt`` naming convention.

        :attr:`last_reading` remains fully supported; both names refer to the same value.
        """
        return self.last_reading

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "BalanceResponse":
        d = _as_dict(d)
        ltu_ts, ltu_dt = _parse_ts_pair(d.get("last_top_up_time"))
        lr_ts, lr_dt = _parse_ts_pair(d.get("last_reading"))

        return cls(
            credit_balance=_to_float(d.get("balance")),
            top_up_in_days=_to_int(d.get("top_up_in_days")),
            pending_top_up=bool(d.get("pending_top_up", False)),
            pending_top_up_by=_to_str(d.get("pending_top_up_by")),
            last_top_up_amount=_to_float(d.get("last_top_up_amount")),
            credit_low=bool(d.get("credit_low", False)),
            emergency_credit=bool(d.get("emergency_credit", False)),
            power_off=bool(d.get("power_off", False)),
            last_top_up_ts=ltu_ts,
            last_top_up_time=ltu_dt,
            last_reading_ts=lr_ts,
            last_reading=lr_dt,
        )

credit_balance instance-attribute

Current credit balance in euros (€).

credit_low instance-attribute

True when balance is below the configured alert threshold.

emergency_credit instance-attribute

True when the meter is drawing on emergency credit.

last_reading_dt property

Alias for :attr:last_reading, following the *_dt naming convention.

:attr:last_reading remains fully supported; both names refer to the same value.

last_top_up_dt property

Alias for :attr:last_top_up_time, following the *_dt naming convention.

:attr:last_top_up_time remains fully supported; both names refer to the same value.

power_off instance-attribute

True when supply has been disconnected.

top_up_in_days instance-attribute

Estimated days until credit is exhausted.

---

Top-Up Models

Scheduled and automatic top-up configurations.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class ActiveTopUpsResponse:
    """Scheduled and automatic top-up configurations."""

    scheduled: List[ScheduledTopUp]
    auto_top_ups: List[Dict[str, Any]]

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "ActiveTopUpsResponse":
        d = _as_dict(d)
        # Performance optimization: List comprehension with locally cached
        # classmethod reference speeds up the array parsing loop by ~10% over list(map(...))
        _st_from_dict = ScheduledTopUp._from_dict
        return cls(
            scheduled=[_st_from_dict(x) for x in _as_list(d.get("scheduled"))],
            auto_top_ups=_as_list(d.get("auto_top_ups")),
        )

A top-up scheduled for a fixed calendar day.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class ScheduledTopUp:
    """A top-up scheduled for a fixed calendar day."""

    current_user: bool
    """False when this entry belongs to another resident on the same premises."""
    top_up_amount: float
    top_up_day: int
    customer: str

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "ScheduledTopUp":
        d = _as_dict(d)
        return cls(
            current_user=bool(d.get("current_user", True)),
            top_up_amount=_to_float(d.get("top_up_amount")),
            top_up_day=_to_int(d.get("top_up_day")),
            customer=_to_str(d.get("customer")),
        )

current_user instance-attribute

False when this entry belongs to another resident on the same premises.

---

Comparison Models

Comparison of this home vs. similar homes.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class CompareResponse:
    """Comparison of this home vs. similar homes."""

    day: ComparePeriod
    week: ComparePeriod
    month: ComparePeriod

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "CompareResponse":
        d = _as_dict(d)
        return cls(
            day=ComparePeriod._from_dict(_as_dict(d.get("day"))),
            week=ComparePeriod._from_dict(_as_dict(d.get("week"))),
            month=ComparePeriod._from_dict(_as_dict(d.get("month"))),
        )

Comparison data for a single period (day / week / month).

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class ComparePeriod:
    """Comparison data for a single period (day / week / month)."""

    available: bool
    euro: CompareValues
    kwh: CompareValues
    co2: CompareValues

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "ComparePeriod":
        d = _as_dict(d)
        return cls(
            available=bool(d.get("available", False)),
            euro=CompareValues._from_dict(_as_dict(d.get("euro"))),
            kwh=CompareValues._from_dict(_as_dict(d.get("kwh"))),
            co2=CompareValues._from_dict(_as_dict(d.get("co2"))),
        )

Paired user vs. average-home figures for a metric.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class CompareValues:
    """Paired user vs. average-home figures for a metric."""

    users_home: float
    average_home: float

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "CompareValues":
        d = _as_dict(d)
        return cls(
            users_home=_to_float(d.get("users_home")),
            average_home=_to_float(d.get("average_home")),
        )

---

Configuration Models

Valid top-up amounts and balance alert thresholds.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class ConfigInfoResponse:
    """Valid top-up amounts and balance alert thresholds."""

    thresholds: List[int]
    top_up_amounts: List[int]
    auto_up_amounts: List[int]
    scheduled_top_up_amounts: List[int]

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "ConfigInfoResponse":
        d = _as_dict(d)
        return cls(
            thresholds=_as_list(d.get("thresholds")),
            top_up_amounts=_as_list(d.get("top_up_amounts")),
            auto_up_amounts=_as_list(d.get("auto_up_amounts")),
            scheduled_top_up_amounts=_as_list(d.get("scheduled_top_up_amounts")),
        )

Reference data for house and heating types.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class DefaultsInfoResponse:
    """Reference data for house and heating types."""

    house_types: List[HouseType]
    heating_types: List[HeatingType]
    max_bedrooms: int
    default_bedrooms: int
    max_adults: int
    default_adults: int
    max_children: int
    default_children: int

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "DefaultsInfoResponse":
        d = _as_dict(d)
        # Performance optimization: List comprehension with locally cached
        # classmethod reference speeds up the array parsing loop by ~10% over list(map(...))
        _ht_from_dict = HouseType._from_dict
        _heat_from_dict = HeatingType._from_dict
        return cls(
            house_types=[_ht_from_dict(x) for x in _as_list(d.get("house_types"))],
            heating_types=[_heat_from_dict(x) for x in _as_list(d.get("heating_types"))],
            max_bedrooms=_to_int(d.get("max_bedrooms")),
            default_bedrooms=_to_int(d.get("default_bedrooms")),
            max_adults=_to_int(d.get("max_adults")),
            default_adults=_to_int(d.get("default_adults")),
            max_children=_to_int(d.get("max_children")),
            default_children=_to_int(d.get("default_children")),
        )

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class HouseType:
    id: int
    name: str

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "HouseType":
        d = _as_dict(d)
        return cls(id=_to_int(d.get("id")), name=_to_str(d.get("name")))

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class HeatingType:
    id: int
    name: str

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "HeatingType":
        d = _as_dict(d)
        return cls(id=_to_int(d.get("id")), name=_to_str(d.get("name")))

---

Notification Model

User notification channel preferences.

Source code in src/pypinergy/models.py

@dataclass(**_DATACLASS_KWARGS)
class NotificationPreferences:
    """User notification channel preferences."""

    sms: bool
    email: bool
    phone: bool
    should_show: int
    should_show_message: str

    @classmethod
    def _from_dict(cls, d: Mapping[str, Any]) -> "NotificationPreferences":
        d = _as_dict(d)
        return cls(
            sms=bool(d.get("sms", False)),
            email=bool(d.get("email", False)),
            phone=bool(d.get("phone", False)),
            should_show=_to_int(d.get("should_show")),
            should_show_message=_to_str(d.get("should_show_message")),
        )