KEBA KeEnergy API for Python

A Python wrapper for the KEBA KeEnergy API used by the Web HMI.

Getting started

pip install keba-keenergy-api

Usage

import asyncio
from typing import Any

from keba_keenergy_api import KebaKeEnergyAPI
from keba_keenergy_api.constants import HeatCircuit
from keba_keenergy_api.constants import HeatCircuitOperatingMode

async def main():
    client = KebaKeEnergyAPI(
        host="ap4400.local",
        username="test",
        password="test",
        ssl=True,
        skip_ssl_verification=True
    )

    # Get current outdoor temperature
    outdoor_temperature = await client.system.get_outdoor_temperature()

    # Get heat circuit temperature from heat circuit 2
    heat_circuit_temperature = await client.heat_circuit.get_target_temperature(
        position=2
    )

    # Read multiple values
    data = await client.read_data(
        request=[
            HeatCircuit.TARGET_TEMPERATURE,
            HeatCircuit.TARGET_TEMPERATURE_DAY
        ],
    )

    # Enable "day" mode for heat circuit 2
    await client.heat_circuit.set_operating_mode(
        mode=HeatCircuitOperatingMode.DAY.value,
        position=2
    )

    # Write multiple values
    await client.write_data(
        request={
            # Write heat circuit on position 1 and 3
            HeatCircuit.TARGET_TEMPERATURE_DAY: (20, None, 5),
            # Write night temperature on position 1
            HeatCircuit.TARGET_TEMPERATURE_NIGHT: (16,),
        },
    )

asyncio.run(main())

By default, the library creates a new connection to KEBA KeEnergy API with each coroutine. If you are calling a large number of coroutines, an aiohttp ClientSession() can be used for connection pooling:

import asyncio

from keba_keenergy_api import KebaKeEnergyAPI

from aiohttp import ClientSession

async def main():
    async with ClientSession() as session:
        client = KebaKeEnergyAPI(
            host="ap4400.local",
            username="test",
            password="test",
            ssl=True,
            skip_ssl_verification=True,
            session=session
        )
        ...

asyncio.run(main())