KebaKeEnergyAPI

class KebaKeEnergyAPI(
    host: str,
    username: str | None = None,
    password: str | None = None,
    *,
    ssl: bool = False,
    skip_ssl_verification: bool = False,
    session: ClientSession | None = None,
)

Asynchronous client for the KEBA KeEnergy Web HMI API.

The client exposes endpoint groups and convenience methods for reading and writing controller values. A caller may provide an existing aiohttp.ClientSession to manage its own session lifecycle. Without a session, each request opens a new connection, unless the client is used with async with, which shares one session between all requests.

The installed devices are read once and reused, as they do not change while the control unit is running.

Configure the device URL, optional HTTP Basic authentication, TLS settings, and the optional shared aiohttp.ClientSession.

Parameters

• host str: The hostname or IP address, e.g. ap4400.local.
• username str | None (default: None): Username for HTTP Basic authentication.
• password str | None (default: None): Password for HTTP Basic authentication.
• ssl bool (default: False): Use HTTPS for API URLs.
• skip_ssl_verification bool (default: False): Disable TLS certificate verification, for example for self-signed certificates.
• session ClientSession | None (default: None): Optional session shared with the endpoint clients.

Example

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

Endpoints

Each endpoint group is a property of the client:

• client.system: API Endpoints to send and retrieve the system data.
• client.buffer_tank: API Endpoints to send and retrieve the buffer tank data.
• client.hot_water_tank: API Endpoints to send and retrieve the hot water tank data.
• client.heat_pump: API Endpoints to send and retrieve the heat pump data.
• client.heat_circuit: API Endpoints to send and retrieve the heat circuit data.
• client.solar_circuit: API Endpoints to send and retrieve the solar circuit data.
• client.external_heat_source: API Endpoints to retrieve the external heat source data.
• client.switch_valve: API Endpoints to retrieve the switch valve data.
• client.passive_cooling: API Endpoints to retrieve the passive cooling data.
• client.photovoltaics: API Endpoints to retrieve the photovoltaics data.

device_url

client.device_url: str

Return the base URL of the KEBA controller.

Returns

• str: e.g. https://ap4400.local

close

async def close() -> None

Close the session opened by async with.

A session passed to the client is left open, as its owner closes it.

read_data

async def read_data(
    request: Section | list[Section],
    position: int | Position | list[int] | None = None,
    *,
    human_readable: bool = True,
    extra_attributes: bool = True,
) -> dict[str, ValueResponse]

Read the requested values with one request and group them by section.

If no position is given, the installed devices are read from the control unit first (one additional request on the first call only).

Parameters

• request Section | list[Section]: Section or a list of sections e.g. [BufferTank.NAME, ...]
• position int | Position | list[int] | None (default: None): Installed device position(s), such as buffer-tank positions.
• human_readable bool (default: True): Whether values should be returned as human-readable strings.
• extra_attributes bool (default: True): Whether to append additional attributes, including heating-curve points.

Returns

• dict[str, ValueResponse]: A dictionary keyed by section prefix, containing the requested values.

Example

await client.read_data(
    request=[
        HeatCircuit.TARGET_TEMPERATURE,
        HeatCircuit.TARGET_TEMPERATURE_DAY
    ]
)

read_message_ids

async def read_message_ids(
    request: Section | list[Section],
    position: int | Position | list[int] | None = None,
) -> dict[Section, list[int]]

Read the message ids of sensors, to find out whether a sensor is in use.

A message id other than 0 means that the control unit has a message for this sensor, e.g. the sensor is not connected. Only sections with a confirmed message id are read, all of them with one request.

Parameters

• request Section | list[Section]: Section or a list of sections e.g. [HeatPump.HAS_VFD_FAILURE, ...]
• position int | Position | list[int] | None (default: None): Installed device position(s), such as heat-pump positions.

Returns

• dict[Section, list[int]]: A dictionary with the section as key and a list of message ids as value, one per device (position * quantity + sub index). 0 means the sensor is in use.

Example

await client.read_message_ids(
    request=[HeatPump.HAS_VFD_FAILURE, HeatCircuit.ROOM_TEMPERATURE]
)

write_data

async def write_data(request: dict[Section, Any]) -> None

Write multiple values with one request.

Attention

Writing values should remain within normal limits, as is the case with typical use of the Web HMI. Permanent and very frequent writing of values reduces the lifetime of the built-in flash memory.

Each key is a section, the value is a single value (position 1) or a tuple with one value per device (the first item is position 1, None skips a device). System-wide sections accept only one value. Read-only sections are ignored.

Parameters

• request dict[Section, Any]: A mapping from section to a value or a tuple of values.

Raises

• APIError: More than one value is given for a system-wide section.

Example

await client.write_data(
    request={
       HeatCircuit.TARGET_TEMPERATURE_DAY: (20, None, 5),
       HeatCircuit.TARGET_TEMPERATURE_NIGHT: (16,),
    }
)

filter_request

async def filter_request(
    request: Section | list[Section],
    position: int | Position | list[int] | None = None,
) -> list[Section]

Return the requested sections that are supported by the Web HMI software version.

All variables are read with one request. The Web HMI rejects a request with an unknown variable as a whole, so each unknown variable costs one more request. Sections of devices that are not installed are skipped. The result keeps the request order and contains each supported section once.

Parameters

• request Section | list[Section]: Section or a list of sections e.g. [BufferTank.NAME, ...]
• position int | Position | list[int] | None (default: None): Installed device position(s), such as buffer-tank positions.

Returns

• list[Section]: The supported sections e.g. [BufferTank.NAME, ...]

Raises

• APIError: The request failed for another reason than an unknown variable.

Example

sections = await client.filter_request(
    request=[
        HeatCircuit.TARGET_TEMPERATURE,
        HeatCircuit.TARGET_TEMPERATURE_DAY
    ]
)