Event Reference¶

from aiohomematic.central.events import DataPointValueReceivedEvent

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

unsubscribe = central.event_bus.subscribe(
    event_type=DataPointValueReceivedEvent,
    event_key=None,  # All data points, or specific DataPointKey
    handler=handler,
)

from aiohomematic.central.events import DataPointStatusReceivedEvent

from aiohomematic.central.events import DataPointStateChangedEvent

from aiohomematic.central.events import RpcParameterReceivedEvent

from aiohomematic.central.events import SysvarStateChangedEvent

from aiohomematic.central.events import DeviceStateChangedEvent

from aiohomematic.central.events import DeviceRemovedEvent

from aiohomematic.central.events import FirmwareStateChangedEvent

from aiohomematic.central.events import LinkPeerChangedEvent

from aiohomematic.central.events import ConnectionStageChangedEvent
from aiohomematic.const import ConnectionStage

async def handler(*, event: ConnectionStageChangedEvent) -> None:
    print(f"{event.interface_id}: {event.previous_stage} -> {event.stage}")

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

from aiohomematic.central.events import ConnectionHealthChangedEvent

from aiohomematic.central.events import CacheInvalidatedEvent
from aiohomematic.const import CacheType, CacheInvalidationReason

from aiohomematic.central.events import CircuitBreakerStateChangedEvent
from aiohomematic.client.circuit_breaker import CircuitState

async def handler(*, event: CircuitBreakerStateChangedEvent) -> None:
    if event.new_state == CircuitState.OPEN:
        print(f"Circuit breaker OPEN for {event.interface_id}")

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

from aiohomematic.central.events import CircuitBreakerTrippedEvent

from aiohomematic.central.events import HealthRecordedEvent

async def handler(*, event: HealthRecordedEvent) -> None:
    status = "success" if event.success else "failure"
    print(f"{event.interface_id}: {status}")

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

from aiohomematic.central.events import ClientStateChangedEvent

from aiohomematic.central.events import CentralStateChangedEvent

async def handler(*, event: CentralStateChangedEvent) -> None:
    if event.new_state == "DEGRADED":
        print(f"Central {event.central_name} is degraded")

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

from aiohomematic.central.events import DataRefreshTriggeredEvent

from aiohomematic.central.events import DataRefreshCompletedEvent

async def handler(*, event: DataRefreshCompletedEvent) -> None:
    if event.success:
        print(f"Refreshed {event.items_refreshed} items in {event.duration_ms}ms")
    else:
        print(f"Refresh failed: {event.error_message}")

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

from aiohomematic.central.events import ProgramExecutedEvent

from aiohomematic.central.events import RequestCoalescedEvent

from aiohomematic.central.events import ConnectionLostEvent

from aiohomematic.central.events import RecoveryStageChangedEvent

from aiohomematic.central.events import RecoveryCompletedEvent

from aiohomematic.central.events import RecoveryFailedEvent

from aiohomematic.central.events import DataFetchCompletedEvent

from aiohomematic.central.events import SystemStatusChangedEvent
from aiohomematic.const import CentralState, FailureReason

async def handler(*, event: SystemStatusChangedEvent) -> None:
    if event.central_state == CentralState.FAILED:
        if event.failure_reason == FailureReason.AUTH:
            print("Authentication failed!")
    elif event.central_state == CentralState.DEGRADED:
        for iface_id, reason in (event.degraded_interfaces or {}).items():
            print(f"Interface {iface_id} degraded: {reason}")

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

from aiohomematic.central.events import (
    DeviceLifecycleEvent,
    DeviceLifecycleEventType,
)

async def handler(*, event: DeviceLifecycleEvent) -> None:
    if event.event_type == DeviceLifecycleEventType.CREATED:
        for address in event.device_addresses:
            print(f"New device: {address}")
    elif event.event_type == DeviceLifecycleEventType.AVAILABILITY_CHANGED:
        for address, available in event.availability_changes:
            print(f"{address}: {'available' if available else 'unavailable'}")

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

from aiohomematic.central.events import DataPointsCreatedEvent
from aiohomematic.const import DataPointCategory

from aiohomematic.central.events import DeviceTriggerEvent
from aiohomematic.const import DeviceTriggerEventType

async def handler(*, event: DeviceTriggerEvent) -> None:
    if event.trigger_type == DeviceTriggerEventType.KEYPRESS:
        print(f"Button {event.parameter} pressed on {event.device_address}")

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

# Subscribe to value updates
unsubscribe = data_point.subscribe_to_data_point_updated(
    handler=my_handler,
    custom_id="my-integration",
)

# Subscribe to device removal
unsubscribe = data_point.subscribe_to_device_removed(
    handler=my_removal_handler,
)

# Subscribe to device state updates
unsubscribe = device.subscribe_to_device_updated(
    handler=my_handler,
)

# Subscribe to firmware updates
unsubscribe = device.subscribe_to_firmware_updated(
    handler=my_firmware_handler,
)

# Subscribe to link peer changes
unsubscribe = channel.subscribe_to_link_peer_changed(
    handler=my_handler,
)

from aiohomematic.central.events import EventBus, DataPointValueReceivedEvent

# Subscribe to all events of a type
unsubscribe = event_bus.subscribe(
    event_type=DataPointValueReceivedEvent,
    event_key=None,
    handler=my_handler,
)

# Subscribe to events with a specific key
unsubscribe = event_bus.subscribe(
    event_type=DataPointValueReceivedEvent,
    event_key=my_data_point_key,
    handler=my_handler,
)

# Unsubscribe when done
unsubscribe()

from aiohomematic.central.events import EventPriority

# High priority handler (runs before normal handlers)
unsubscribe = event_bus.subscribe(
    event_type=DataPointValueReceivedEvent,
    event_key=None,
    handler=my_critical_handler,
    priority=EventPriority.HIGH,
)

from aiohomematic.central.events import EventBatch

async with EventBatch(bus=event_bus) as batch:
    batch.add(event=event1)
    batch.add(event=event2)
    batch.add(event=event3)
    # All events published when context exits

from aiohomematic_test_support.event_capture import EventCapture

capture = EventCapture()
capture.subscribe_to(event_bus, CircuitBreakerTrippedEvent)

# ... perform test actions ...

capture.assert_event_emitted(
    event_type=CircuitBreakerTrippedEvent,
    failure_count=5,
)
capture.cleanup()

# Good - type-safe, IDE autocomplete works
async def handler(*, event: DataPointValueReceivedEvent) -> None:
    print(event.dpk, event.value)

# Bad - loses type information
async def handler(*, event: Event) -> None:
    print(event.timestamp)  # Only base fields available

# Good - quick handler, offloads heavy work
async def handler(*, event: DataPointValueReceivedEvent) -> None:
    asyncio.create_task(process_update(event))

# Avoid - blocks other handlers
async def handler(*, event: DataPointValueReceivedEvent) -> None:
    await slow_database_operation(event)  # Blocks for seconds

class MyIntegration:
    def __init__(self, event_bus: EventBus) -> None:
        self._unsubscribe = event_bus.subscribe(
            event_type=DataPointValueReceivedEvent,
            event_key=None,
            handler=self._handler,
        )

    def cleanup(self) -> None:
        self._unsubscribe()

async def handler(*, event: DataPointValueReceivedEvent) -> None:
    try:
        await process(event)
    except ExpectedError:
        _LOGGER.warning("Could not process event: %s", event.dpk)