warp-charger

Async Python client for the local API of Tinkerforge WARP wallboxes and the WARP Energy Manager.

It keeps one WebSocket open for push updates (no polling) and sends commands over HTTP. It powers the Tinkerforge WARP Home Assistant integration, but has no Home Assistant dependency.

• Full state mirror: every API topic (evse/state, meters/0/values, …) in one dict, kept current by the device's push stream
• Topic and connection callbacks
• Automatic reconnect with backoff; survives device reboots
• HTTP Digest authentication (via aiohttp's DigestAuthMiddleware), HTTP and HTTPS
• Helpers: enum tables, meter value IDs, UID encoding, NFC scan detection

Requires Python 3.13+ and aiohttp 3.12+.

Install

pip install warp-charger

Quick start

import asyncio

import aiohttp

from warp_charger import WarpClient
from warp_charger.const import CHARGER_STATE


async def main() -> None:
    async with aiohttp.ClientSession() as session:
        client = WarpClient(session, "http://warp3-AbCd.local")
        task = asyncio.create_task(client.run())  # reconnect loop
        await client.wait_ready(timeout=30)  # first full state dump received

        state = client.states["evse/state"]
        print("Charger:", CHARGER_STATE[state["charger_state"]])

        client.subscribe("evse/state", lambda payload: print("now", payload["charger_state"]))

        await client.put("evse/external_current", {"current": 16000})  # 16 A
        await asyncio.sleep(60)

        await client.stop()
        await task


asyncio.run(main())

Authentication

If the web interface has a login, put the credentials into the session. The client does not handle them itself:

session = aiohttp.ClientSession(middlewares=(aiohttp.DigestAuthMiddleware("user", "password"),))
client = WarpClient(session, "https://warp3-AbCd.local")

The device does not validate digest nonces, so a captured login can be replayed on plain HTTP. Prefer HTTPS when a password is set. For the self-signed certificate, pass a connector with ssl=False.

API

WarpClient(session, base_url, **tunables)

Optional keyword tunables are reconnect_delay (5 s), max_reconnect_delay (60 s), jitter, stable_after, going_away_delay, receive_timeout (15 s), request_timeout (10 s), kick_threshold, kick_window and clock.

Callbacks run in the event loop. Exceptions in callbacks are logged, never propagated.

After a reconnect, the new dump replaces states in one step. Then every subscribed topic is called once, then the connection callbacks.

Exceptions

WarpError is the base class of:

• WarpAuthError: HTTP 401.
• WarpConnectionError: network errors, timeouts, unexpected status.
• WarpCommandError: HTTP 400; str(err) is the device's error message.

Helpers

• warp_charger.const
  • CHARGER_STATE, IEC61851_STATE, ERROR_STATE, DC_FAULT_STATE: int → key.
  • CHARGE_MODE, plus CHARGE_MODE_DEFAULT (4 is a state only, never selectable).
  • DURATION_LIMITS for charge-limit durations.
  • METER_VALUES: meter value id → MeterValue(key, unit, digits), for decoding meters/N/values with meters/N/value_ids.
  • INT32_MAX.
• warp_charger.uid: encode(int) -> str and decode(str) -> int. These convert between the numeric UID from mDNS TXT records and the device UID string shown in info/name (base58, or zbase32 for large UIDs).
• warp_charger.NfcScanDetector: feed(nfc/seen_tags payload) returns the newly scanned (tag_type, tag_id) pairs. It uses the firmware web UI's 2 s threshold.

Device API in a nutshell

The topics and payloads are the device's MQTT/HTTP API, documented in the WARP API reference.

• Read: everything arrives over /ws. On connect the device sends a full dump of all topics, then only changes, at most once per second per topic.
• Write: put("<topic>_update", {...}) for configuration and state topics, e.g. evse/auto_start_charging_update. put("<command>") for commands, e.g. evse/start_charging or reboot.
  • The device answers 200 even for some silent no-ops; for example, start/stop without a vehicle is ignored.
• Limits:
  • At most 5 WebSocket clients (web UI tabs count). When full, the device drops the client that answered pings least recently.
  • The ESP32 serves one request at a time, so avoid parallel or high-rate requests.

Never GET a command path. The firmware executes payload-less commands on GET, so GET /reboot reboots the device. get() is meant for state topics only.

Recording a fixture

scripts/record_dump.py writes one state dump as JSONL (one {"topic", "payload"} per line). It anonymises MACs, IPs, SSIDs, tag ids, user names and coordinates. It is read-only: it only opens the WebSocket.

uv run python scripts/record_dump.py warp3-AbCd.local --replace-uid -o warp3.jsonl
uv run python scripts/record_dump.py 192.168.1.20 -u admin -p secret --https --insecure

Development

uv sync
uv run pytest
uv run ruff check . && uv run ruff format --check .

The tests run against a fake device on an aiohttp test server and never touch real hardware.

Release: bump version in pyproject.toml, then tag vX.Y.Z. CI publishes to PyPI via trusted publishing.

License

MIT