Getting Started with aiohomematic¶

This is the full walkthrough for using aiohomematic as a standalone Python library. It covers CentralConfig and CentralUnit usage, manual lifecycle management, and explicit interface configuration.

Looking for the fastest path? The Quick Start shows the minimum working example (≈ 5 minutes) with CentralConfig and CentralUnit.

Tip: For definitions of terms like Backend, Interface, Device, Channel, and Parameter, see the Glossary.

Verified code: a known-good, end-to-end working script lives at example.py in the repository root. When the snippets below conflict with example.py, treat example.py as the source of truth.

Installation¶

pip install aiohomematic

Quick Start¶

Using the Async Context Manager (Recommended)¶

CentralUnit supports the async context manager protocol directly -- async with central: calls start() on entry and stop() on exit, guaranteeing cleanup even when exceptions occur:

import asyncio
from aiohomematic.central import CentralConfig
from aiohomematic.const import ParamsetKey

async def main():
    config = CentralConfig.for_ccu(
        host="192.168.1.100",
        username="Admin",
        password="your-password",
    )
    central = await config.create_central()

    async with central:
        # List all devices
        for device in central.devices:
            print(f"{device.address}: {device.name} ({device.model})")

        # Read a value
        device = central.device_coordinator.get_device(address="VCU0000001")
        client = central.client_coordinator.get_client(interface_id=device.interface_id)
        state = await client.get_value(
            channel_address="VCU0000001:1",
            paramset_key=ParamsetKey.VALUES,
            parameter="STATE",
        )
        print(f"Current state: {state}")

        # Write a value
        await client.set_value(
            channel_address="VCU0000001:1",
            paramset_key=ParamsetKey.VALUES,
            parameter="STATE",
            value=True,
        )

    # Connection is automatically closed when exiting the context

asyncio.run(main())

Connection Options¶

CentralConfig.for_ccu() and CentralConfig.for_homegear() support several options:

# CCU with TLS
config = CentralConfig.for_ccu(
    host="192.168.1.100",
    username="Admin",
    password="secret",
    tls=True,
    verify_tls=False,  # Set to True in production
)

# Homegear backend
config = CentralConfig.for_homegear(
    host="192.168.1.100",
    username="Admin",
    password="secret",
)

# Custom central ID
config = CentralConfig.for_ccu(
    host="192.168.1.100",
    username="Admin",
    password="secret",
    central_id="my-living-room-ccu",
)

Manual Lifecycle Management¶

For more control over the lifecycle, you can manage start/stop manually instead of using the context manager:

import asyncio
from aiohomematic.central import CentralConfig

async def main():
    config = CentralConfig.for_ccu(
        name="my-ccu",
        host="192.168.1.100",
        username="Admin",
        password="your-password",
        central_id="my-ccu",
    )

    central = await config.create_central()
    await central.start()

    try:
        for device in central.devices:
            print(f"{device.address}: {device.name}")
    finally:
        await central.stop()

asyncio.run(main())

Using CentralUnit with Explicit Interface Configuration¶

For full control over which interfaces are enabled and their ports, configure interface_configs explicitly instead of using the for_ccu()/for_homegear() presets:

import asyncio
from aiohomematic.central import CentralConfig
from aiohomematic.client import InterfaceConfig
from aiohomematic.const import Interface

async def main():
    # Define interfaces manually
    interface_configs = {
        InterfaceConfig(
            central_name="my-ccu",
            interface=Interface.HMIP_RF,
            port=2010,
        ),
        InterfaceConfig(
            central_name="my-ccu",
            interface=Interface.BIDCOS_RF,
            port=2001,
        ),
    }

    # Create configuration
    config = CentralConfig(
        name="my-ccu",
        host="192.168.1.100",
        username="Admin",
        password="your-password",
        central_id="unique-id",
        interface_configs=interface_configs,
    )

    # Create and start central unit
    central = await config.create_central()
    await central.start()

    try:
        # Access devices
        for device in central.devices:
            print(f"{device.address}: {device.name}")

    finally:
        await central.stop()

asyncio.run(main())

Configuration Presets¶

aiohomematic provides convenient factory methods for common backend types:

CCU3/CCU2¶

from aiohomematic.central import CentralConfig

# Basic setup with HmIP-RF and BidCos-RF
config = CentralConfig.for_ccu(
    host="192.168.1.100",
    username="Admin",
    password="secret",
)

# With TLS and additional interfaces
config = CentralConfig.for_ccu(
    host="192.168.1.100",
    username="Admin",
    password="secret",
    tls=True,
    enable_bidcos_wired=True,
    enable_virtual_devices=True,
)

Homegear¶

from aiohomematic.central import CentralConfig

config = CentralConfig.for_homegear(
    host="192.168.1.50",
    username="homegear",
    password="secret",
)

Common Patterns¶

The examples below assume a running central: CentralUnit (see Quick Start above). Reading and writing raw parameter values requires resolving the client responsible for the device's interface first:

from aiohomematic.const import ParamsetKey

def get_client_for(*, address: str):
    """Resolve the client responsible for a device address."""
    device = central.device_coordinator.get_device(address=address)
    client = central.client_coordinator.get_client(interface_id=device.interface_id)
    return client

Device Discovery¶

# List all devices
for device in central.devices:
    print(f"Device: {device.address}")
    print(f"  Name: {device.name}")
    print(f"  Model: {device.model}")
    print(f"  Channels: {len(device.channels)}")

    # Iterate channels (device.channels is Mapping[str, ChannelProtocol], keyed by channel address)
    for channel in device.channels.values():
        print(f"  Channel {channel.no}: {channel.address}")
        # Generic data points expose per-parameter values
        for dp in channel.generic_data_points:
            print(f"    - {dp.parameter}: {dp.value}")

Reading Values¶

# Read from a specific channel and parameter
client = get_client_for(address="VCU0000001")
value = await client.get_value(
    channel_address="VCU0000001:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="STATE",
)

# Read directly from a device's channel via address
device = central.device_coordinator.get_device(address="VCU0000001")
if device:
    # device.channels is keyed by channel address (e.g. "VCU0000001:1"), not integer no
    channel = device.channels.get("VCU0000001:1")
    if channel:
        # Look up a generic data point by parameter name
        state_dp = next(
            (dp for dp in channel.generic_data_points if dp.parameter == "STATE"),
            None,
        )
        if state_dp:
            print(f"State: {state_dp.value}")

Writing Values¶

# Turn on a switch
client = get_client_for(address="VCU0000001")
await client.set_value(
    channel_address="VCU0000001:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="STATE",
    value=True,
)

# Set dimmer level (0.0 to 1.0)
client = get_client_for(address="VCU0000002")
await client.set_value(
    channel_address="VCU0000002:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="LEVEL",
    value=0.5,
)

# Set thermostat temperature
client = get_client_for(address="VCU0000003")
await client.set_value(
    channel_address="VCU0000003:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="SET_POINT_TEMPERATURE",
    value=21.5,
)

Subscribing to Events¶

from aiohomematic.central.events import DataPointValueReceivedEvent

async def on_value_changed(*, event: DataPointValueReceivedEvent) -> None:
    print(f"Update: {event.dpk.channel_address}.{event.dpk.parameter} = {event.value}")

# Subscribe to all data point updates
unsubscribe = central.event_bus.subscribe(
    event_type=DataPointValueReceivedEvent,
    event_key=None,
    handler=on_value_changed,
)

# ... your application logic ...

# Unsubscribe when done
unsubscribe()

Using the EventBus for Device-Level Events¶

The EventBus is the single mechanism for all event delivery -- data point updates and device lifecycle changes are just different event types on the same bus:

from aiohomematic.central.events import DataPointValueReceivedEvent, DeviceStateChangedEvent

async def on_datapoint_update(*, event: DataPointValueReceivedEvent) -> None:
    print(f"DataPoint {event.dpk} = {event.value}")

async def on_device_update(*, event: DeviceStateChangedEvent) -> None:
    print(f"Device updated: {event.device_address}")

# Subscribe to specific events
central.event_bus.subscribe(
    event_type=DataPointValueReceivedEvent,
    event_key=None,
    handler=on_datapoint_update,
)

central.event_bus.subscribe(
    event_type=DeviceStateChangedEvent,
    event_key=None,
    handler=on_device_update,
)

Error Handling¶

Common Exceptions¶

from aiohomematic.exceptions import (
    AioHomematicException,      # Base exception
    ClientException,            # Client/connection errors
    NoConnectionException,      # No connection to backend
    AuthFailure,                # Authentication failed
    ValidationException,        # Value validation failed
)

try:
    client = get_client_for(address="VCU0000001")
    await client.set_value(
        channel_address="VCU0000001:1",
        paramset_key=ParamsetKey.VALUES,
        parameter="LEVEL",
        value=1.5,  # Invalid: must be 0.0-1.0
    )
except ValidationException as e:
    print(f"Validation error: {e}")
except NoConnectionException as e:
    print(f"Connection lost: {e}")
except AioHomematicException as e:
    print(f"General error: {e}")

Connection Recovery¶

The library automatically handles connection recovery. You can monitor connection state:

# Check connection status
if central.client_coordinator.has_clients and not central.connection_state.is_any_issue:
    print("Connected to backend")
else:
    print("Not connected")

# Subscribe to device availability changes
from aiohomematic.central.events import DeviceStateChangedEvent

async def on_device_updated(*, event: DeviceStateChangedEvent) -> None:
    print(f"Device {event.device_address} was updated")

unsubscribe = central.event_bus.subscribe(
    event_type=DeviceStateChangedEvent,
    event_key=None,
    handler=on_device_updated,
)

Working with Specific Device Types¶

Switches¶

# Get switch state
client = get_client_for(address="VCU0000001")
state = await client.get_value(
    channel_address="VCU0000001:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="STATE",
)

# Toggle switch
await client.set_value(
    channel_address="VCU0000001:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="STATE",
    value=not state,
)

Dimmers¶

# Get current level (0.0-1.0)
client = get_client_for(address="VCU0000002")
level = await client.get_value(
    channel_address="VCU0000002:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="LEVEL",
)

# Set to 75%
await client.set_value(
    channel_address="VCU0000002:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="LEVEL",
    value=0.75,
)

Thermostats¶

client = get_client_for(address="VCU0000003")

# Read current temperature
current_temp = await client.get_value(
    channel_address="VCU0000003:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="ACTUAL_TEMPERATURE",
)

# Read set point
set_point = await client.get_value(
    channel_address="VCU0000003:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="SET_POINT_TEMPERATURE",
)

# Set new temperature
await client.set_value(
    channel_address="VCU0000003:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="SET_POINT_TEMPERATURE",
    value=22.0,
)

Blinds/Covers¶

client = get_client_for(address="VCU0000004")

# Get current position (0.0=closed, 1.0=open)
position = await client.get_value(
    channel_address="VCU0000004:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="LEVEL",
)

# Open blinds fully
await client.set_value(
    channel_address="VCU0000004:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="LEVEL",
    value=1.0,
)

# Stop movement
await client.set_value(
    channel_address="VCU0000004:1",
    paramset_key=ParamsetKey.VALUES,
    parameter="STOP",
    value=True,
)

Programs and System Variables¶

CCU programs and system variables are surfaced as hub data points created by the HubCoordinator. They behave like regular data points (value, subscribe, set) rather than living on a Hub.programs / Hub.sysvars collection directly.

See the Consumer API guide for the supported access patterns: developer/consumer_api.md, section "Hub data points". For the design rationale and lifecycle, see architecture.md and ADR 0022 — Unified Schedule Access.

Best Practices¶

Always use async context: All network operations are asynchronous.
Clean up properly: Always call stop() to clean up resources (or use async with central:).
Handle disconnections: The library auto-reconnects, but your code should handle temporary disconnections gracefully.
Use keyword arguments: All API methods use keyword-only parameters for clarity.
Validate before writing: Check parameter constraints before writing values to avoid validation errors.
Subscribe to events: Use event subscriptions instead of polling for real-time updates.

Next Steps¶

See Common Operations for more detailed examples
Check the Architecture documentation for advanced usage
Review the Consumer API for integration patterns