Device API Reference¶

Device interfaces provide base classes for different connection patterns.

BaseDeviceInterface¶

ucapi_framework.device.BaseDeviceInterface ¶

BaseDeviceInterface(device_config: Any, loop: AbstractEventLoop | None = None, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: ABC

Base class for all device interfaces.

Provides common functionality: - Event emitter for device state changes - Connection lifecycle management - Property accessors for device information - Logging helpers

Create device interface instance.

:param device_config: Device configuration :param loop: Event loop :param config_manager: Optional config manager for persisting configuration updates :param driver: Optional reference to the integration driver. Allows devices to dynamically register entities at runtime (e.g., when a hub discovers new sub-devices).

Attributes¶

device_config `property` ¶

device_config: Any

Return the device configuration.

driver `property` ¶

driver: Any | None

Get the integration driver reference.

Allows devices to interact with the driver, such as dynamically registering entities when new sub-devices are discovered.

:return: The driver instance or None if not set

identifier `abstractmethod` `property` ¶

identifier: str

Return the device identifier.

name `abstractmethod` `property` ¶

name: str

Return the device name.

address `abstractmethod` `property` ¶

address: str | None

Return the device address.

log_id `abstractmethod` `property` ¶

log_id: str

Return a log identifier for the device.

is_connected `abstractmethod` `property` ¶

is_connected: bool

Return True if device is currently connected, False otherwise.

state `property` ¶

state: Any

Return the current device state.

Functions¶

push_update ¶

push_update() -> None

Notify the framework that this device's state has changed.

Emits DeviceEvents.UPDATE with no payload. Any entity that has called subscribe_to_device() will have its sync_state() method invoked automatically in response.

This is the recommended way to trigger state propagation when using the coordinator pattern::

async def _poll(self):
    self.volume = await self._fetch_volume()
    self.push_update()

For the legacy attribute-routing pattern, emit DeviceEvents.UPDATE directly with entity_id and an update dict instead.

update_config ¶

update_config(**kwargs) -> bool

Update device configuration attributes and persist changes.

This method allows devices to update their configuration when runtime changes occur, such as: - New authentication tokens received - IP address changes detected - Device firmware updates changing capabilities - Dynamic configuration from device responses

The configuration is updated both in memory and persisted to storage if a config_manager is available.

Example usage

Update token after authentication¶

self.update_config(token="new_token_value")

Update multiple fields¶

self.update_config( address="192.168.1.100", token="new_token", firmware_version="2.0.1" )

:param kwargs: Configuration attributes to update :return: True if config was persisted successfully, False if no config_manager or update failed :raises AttributeError: If trying to update non-existent configuration attribute

get_device_attributes ¶

get_device_attributes(entity_id: str) -> dict[str, Any] | EntityAttributes

Provide entity-specific attributes beyond STATE.

This method is called by the framework when refreshing entity state to allow devices to supply additional attributes (e.g., SOURCE_LIST, SOUND_MODE_LIST, VOLUME, BRIGHTNESS, etc.) that should be included in the entity's attributes.

Can return either a dict or an EntityAttributes dataclass (e.g., MediaPlayerAttributes). When returning a dataclass, None values are automatically filtered out.

The default implementation returns an empty dict, meaning only STATE will be updated. Override this method to provide device-specific attributes.

Example with dict

def get_device_attributes(self, entity_id: str) -> dict[str, Any]: return { media_player.Attributes.SOURCE_LIST: list(self.source_list), media_player.Attributes.SOUND_MODE_LIST: list(self.sound_mode_list), media_player.Attributes.VOLUME: self.volume, }

Example with dataclass (recommended): def get_device_attributes(self, entity_id: str) -> MediaPlayerAttributes: return self.attributes # MediaPlayerAttributes instance

Example for multi-entity devices

def init(self, ...): self.zone_attrs = { "zone1": MediaPlayerAttributes(), "zone2": MediaPlayerAttributes(), }

def get_device_attributes(self, entity_id: str) -> EntityAttributes: if "zone1" in entity_id: return self.zone_attrs["zone1"] elif "zone2" in entity_id: return self.zone_attrs["zone2"]

Example for sensors (using entity registry): def init(self, ...): self.sensor_attrs = {}

def get_device_attributes(self, entity_id: str) -> SensorAttributes:
    sensor_id = entity_id.split('.')[-1]
    return self.sensor_attrs.get(sensor_id, SensorAttributes())

:param entity_id: Entity identifier to get attributes for :return: Dictionary or EntityAttributes dataclass of entity attributes

connect `abstractmethod` `async` ¶

connect() -> bool

Establish connection to the device.

:return: True if connection successful, False otherwise

disconnect `abstractmethod` `async` ¶

disconnect() -> None

Disconnect from the device.

StatelessHTTPDevice¶

ucapi_framework.device.StatelessHTTPDevice ¶

StatelessHTTPDevice(device_config: Any, loop: AbstractEventLoop | None = None, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: BaseDeviceInterface

Base class for devices with stateless HTTP API.

No persistent connection is maintained. Each command creates a new HTTP session for the request.

Good for: REST APIs, simple HTTP devices without a persistent connection (e.g., websockets)

Initialize stateless HTTP device.

Attributes¶

is_connected `property` ¶

is_connected: bool

Return True if device is currently connected.

Functions¶

connect `async` ¶

connect() -> bool

Establish connection (verify device is reachable).

For stateless devices, this typically means verifying the device responds to a basic request.

:return: True if connection successful, False otherwise

disconnect `async` ¶

disconnect() -> None

Disconnect from device (mark as disconnected).

verify_connection `abstractmethod` `async` ¶

verify_connection() -> None

Verify the device connection.

Should make a simple request to verify device is reachable. Raises exception if connection fails.

PollingDevice¶

ucapi_framework.device.PollingDevice ¶

PollingDevice(device_config: Any, loop: AbstractEventLoop | None = None, poll_interval: int = 30, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: BaseDeviceInterface

Base class for devices requiring periodic status polling.

Maintains a polling task that periodically queries the device for status updates.

Good for: Devices without push notifications, devices with changing state

Initialize polling device.

:param device_config: Device configuration :param loop: Event loop :param poll_interval: Polling interval in seconds (default: 30) :param config_manager: Optional config manager for persisting configuration updates :param driver: Optional reference to the integration driver

Attributes¶

is_connected `property` ¶

is_connected: bool

Return True if device is currently connected and polling.

Functions¶

connect `async` ¶

connect() -> bool

Establish connection and start polling.

:return: True if connection successful, False otherwise

disconnect `async` ¶

disconnect() -> None

Stop polling and disconnect.

establish_connection `abstractmethod` `async` ¶

establish_connection() -> None

Establish initial connection to device.

Called once when connect() is invoked.

poll_device `abstractmethod` `async` ¶

poll_device() -> None

Poll the device for status updates.

Called periodically based on poll_interval. Should emit UPDATE events with changed state.

WebSocketDevice¶

ucapi_framework.device.WebSocketDevice ¶

WebSocketDevice(device_config: Any, loop: AbstractEventLoop | None = None, reconnect: bool = True, reconnect_interval: int = BACKOFF_SEC, reconnect_max: int = BACKOFF_MAX, ping_interval: int = 30, ping_timeout: int = 10, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: BaseDeviceInterface

Base class for devices with WebSocket connections.

Maintains a persistent WebSocket connection with automatic reconnection, exponential backoff, and optional ping/keepalive support.

Features: - Automatic reconnection on connection loss - Configurable exponential backoff (default: 2s initial, 30s max) - Optional ping/pong keepalive (default: 30s interval) - Graceful error handling and recovery

Good for: Devices with WebSocket APIs, real-time updates

Initialize WebSocket device.

:param device_config: Device configuration :param loop: Event loop :param reconnect: Enable automatic reconnection (default: True) :param reconnect_interval: Initial reconnection interval in seconds (default: 2) :param reconnect_max: Maximum reconnection interval in seconds (default: 30) :param ping_interval: Ping/keepalive interval in seconds, 0 to disable (default: 30) :param ping_timeout: Ping timeout in seconds (default: 10) :param config_manager: Optional config manager for persisting configuration updates :param driver: Optional reference to the integration driver

Attributes¶

is_connected `property` ¶

is_connected: bool

Check if WebSocket is currently connected.

:return: True if WebSocket is connected, False otherwise

Functions¶

connect `async` ¶

connect() -> bool

Establish WebSocket connection with automatic reconnection.

If reconnection is enabled, this will continuously attempt to maintain a connection until disconnect() is called.

:return: True if connection task started successfully, False otherwise

disconnect `async` ¶

disconnect() -> None

Close WebSocket connection and stop reconnection attempts.

send_ping `async` ¶

send_ping() -> None

Send ping to WebSocket connection.

Default implementation does nothing. Override this if your WebSocket implementation requires explicit ping messages.

For websockets library, pings are handled automatically. For custom implementations, send your protocol-specific keepalive.

Example for custom protocol

async def send_ping(self): await self._ws.send(json.dumps({"type": "ping"}))

create_websocket `abstractmethod` `async` ¶

create_websocket() -> Any

Create and return WebSocket connection.

Called automatically by the connection loop. Raise an exception if connection cannot be established.

Example using websockets library

async def create_websocket(self): import websockets return await websockets.connect( f"ws://{self.address}/socket", ping_interval=None, # We handle pings ourselves )

:return: WebSocket connection object

close_websocket `abstractmethod` `async` ¶

close_websocket() -> None

Close the WebSocket connection.

Example

async def close_websocket(self): if self._ws: await self._ws.close()

receive_message `abstractmethod` `async` ¶

receive_message() -> Any

Receive a message from WebSocket.

Should block until a message is available or connection is closed.

Example

async def receive_message(self): try: message = await self._ws.recv() return json.loads(message) except websockets.ConnectionClosed: return None

:return: Message data or None if connection closed

handle_message `abstractmethod` `async` ¶

handle_message(message: Any) -> None

Handle incoming WebSocket message.

Called for each message received from the WebSocket connection.

:param message: Message data

WebSocketPollingDevice¶

ucapi_framework.device.WebSocketPollingDevice ¶

WebSocketPollingDevice(device_config: Any, loop: AbstractEventLoop | None = None, poll_interval: int = 30, ping_interval: int = 30, ping_timeout: int = 10, keep_polling_on_disconnect: bool = True, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: WebSocketDevice, PollingDevice

Base class for devices with WebSocket + Polling hybrid pattern.

Combines WebSocket for real-time updates with periodic polling for health checks and state verification. This is a common pattern for smart TVs and IoT devices where: - WebSocket provides instant notifications when device is active - Polling provides fallback health checks and state verification - REST API provides additional control endpoints

The polling and WebSocket run concurrently and independently. By default, polling continues even when WebSocket is disconnected (via disconnect()), providing resilience and allowing fallback to polling-only mode. Use disconnect_all() or set keep_polling_on_disconnect=False to stop both.

Implementation uses multiple inheritance to compose WebSocketDevice and PollingDevice functionality without code duplication.

WebSocket reconnection is automatically disabled for this class since polling provides the resilience. WebSocket will reconnect naturally through the hybrid connect() implementation.

Good for: Smart TVs, media players, IoT devices with multiple communication methods

Initialize WebSocket + Polling device.

:param device_config: Device configuration :param loop: Event loop :param poll_interval: Polling interval in seconds (default: 30) :param ping_interval: WebSocket ping interval in seconds, 0 to disable (default: 30) :param ping_timeout: WebSocket ping timeout in seconds (default: 10) :param keep_polling_on_disconnect: Continue polling when WebSocket disconnects (default: True) :param config_manager: Optional config manager for persisting configuration updates

Attributes¶

is_websocket_connected `property` ¶

is_websocket_connected: bool

Check if WebSocket is currently connected.

:return: True if WebSocket is connected, False otherwise

Functions¶

connect `async` ¶

connect() -> bool

Establish WebSocket connection and start polling.

Both WebSocket and polling tasks run concurrently. If WebSocket connection fails, polling continues to provide state updates.

:return: True if at least polling started successfully, False otherwise

disconnect `async` ¶

disconnect(stop_polling: bool | None = None) -> None

Stop WebSocket and optionally polling.

If keep_polling_on_disconnect is True (default) and stop_polling is not explicitly set, only the WebSocket connection is stopped and polling continues. This allows the device to fall back to polling-only mode.

If keep_polling_on_disconnect is False or stop_polling=True, both WebSocket and polling are stopped, fully disconnecting the device.

:param stop_polling: Override to force stop polling (True) or keep it running (False). If None, uses keep_polling_on_disconnect setting.

disconnect_all `async` ¶

disconnect_all() -> None

Stop both WebSocket and polling, fully disconnecting the device.

This method always stops both WebSocket and polling regardless of the keep_polling_on_disconnect setting.

ExternalClientDevice¶

ucapi_framework.device.ExternalClientDevice ¶

ExternalClientDevice(device_config: Any, loop: AbstractEventLoop | None = None, enable_watchdog: bool = True, watchdog_interval: int = 30, reconnect_delay: int = 5, max_reconnect_attempts: int | None = 3, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: BaseDeviceInterface

Base class for devices using external client libraries.

Use this when wrapping a third-party library that: - Manages its own WebSocket/TCP connection internally - Provides event callbacks for state changes - Exposes a connection state property - May disconnect without proper notification

Features: - Watchdog polling to verify external client connection state - Automatic reconnection when watchdog detects disconnect - Configurable watchdog interval and reconnection attempts - Early exit in connect() if client is already connected

Z-Wave JS, Home Assistant WebSocket, MQTT clients,

or any library that manages its own connection.

Initialize external client device.

:param device_config: Device configuration :param loop: Event loop :param enable_watchdog: Enable watchdog to monitor connection state (default: True) :param watchdog_interval: Interval to check connection state (seconds) :param reconnect_delay: Delay between reconnection attempts (seconds) :param max_reconnect_attempts: Max reconnection attempts before giving up. None = disable reconnection, 0 = infinite, positive int = limit :param config_manager: Optional config manager :param driver: Optional reference to the integration driver

Attributes¶

is_connected `property` ¶

is_connected: bool

Return True if device is connected.

Checks both internal state and external client state for accuracy.

Functions¶

connect `async` ¶

connect() -> bool

Connect to device via external client and start watchdog.

:return: True if connection successful, False otherwise

disconnect `async` ¶

disconnect() -> None

Disconnect from device and stop watchdog.

create_client `abstractmethod` `async` ¶

create_client() -> Any

Create the external client instance.

Example

async def create_client(self): return ZWaveClient(self.address)

:return: External client instance

connect_client `abstractmethod` `async` ¶

connect_client() -> None

Connect the external client.

This is a good place to set up event handlers on the client.

Example

async def connect_client(self): await self._client.connect() self._client.on("value_updated", self._on_value_updated)

disconnect_client `abstractmethod` `async` ¶

disconnect_client() -> None

Disconnect the external client.

This is a good place to remove event handlers from the client.

Example

async def disconnect_client(self): self._client.off("value_updated", self._on_value_updated) await self._client.disconnect()

check_client_connected `abstractmethod` ¶

check_client_connected() -> bool

Check if the external client is connected.

This should query the external client's actual connection state, not rely on internal tracking.

Example

def check_client_connected(self) -> bool: return self._client is not None and self._client.connected

:return: True if external client is connected

PersistentConnectionDevice¶

ucapi_framework.device.PersistentConnectionDevice ¶

PersistentConnectionDevice(device_config: Any, loop: AbstractEventLoop | None = None, backoff_max: int = BACKOFF_MAX, config_manager: BaseConfigManager | None = None, driver: BaseIntegrationDriver | None = None)

Bases: BaseDeviceInterface

Base class for devices with persistent TCP/protocol connections.

Maintains a persistent connection with reconnection logic and backoff.

Good for: Proprietary protocols, TCP connections, devices requiring persistent sessions

Initialize persistent connection device.

:param device_config: Device configuration :param loop: Event loop :param backoff_max: Maximum backoff time in seconds :param config_manager: Optional config manager for persisting configuration updates

Attributes¶

is_connected `property` ¶

is_connected: bool

Return True if device has an active connection.

Functions¶

connect `async` ¶

connect() -> bool

Establish persistent connection with reconnection logic.

:return: True if connection task started successfully, False otherwise

disconnect `async` ¶

disconnect() -> None

Close persistent connection.

establish_connection `abstractmethod` `async` ¶

establish_connection() -> Any

Establish connection to device.

:return: Connection object

close_connection `abstractmethod` `async` ¶

close_connection() -> None

Close the connection.

maintain_connection `abstractmethod` `async` ¶

maintain_connection() -> None

Maintain the connection.

This method should block while the connection is active. Return when connection is lost or should be closed.

DeviceEvents¶

ucapi_framework.device.DeviceEvents ¶

Bases: StrEnum

Common device events.

UPDATE Event

Emitted when device state changes. Can optionally include entity_id to target a specific entity, which is useful for multi-entity devices and sensors.

Usage: # Update all entities for this device self.events.emit(DeviceEvents.UPDATE, update={...})

# Update specific entity (recommended for sensors and multi-entity devices)
self.events.emit(
    DeviceEvents.UPDATE,
    entity_id="sensor.device_id.temperature",
    update=SensorAttributes(STATE=sensor.States.ON, VALUE=23.5, UNIT="°C")
)

The framework automatically calls on_device_update() which will refresh the specified entity or all entities for the device.

Device API Reference¶

BaseDeviceInterface¶

ucapi_framework.device.BaseDeviceInterface ¶

Attributes¶

device_config property ¶

driver property ¶

identifier abstractmethod property ¶

name abstractmethod property ¶

address abstractmethod property ¶

log_id abstractmethod property ¶

is_connected abstractmethod property ¶

state property ¶

Functions¶

push_update ¶

update_config ¶

Update token after authentication¶

Update multiple fields¶

get_device_attributes ¶

connect abstractmethod async ¶

disconnect abstractmethod async ¶

StatelessHTTPDevice¶

ucapi_framework.device.StatelessHTTPDevice ¶

Attributes¶

is_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

verify_connection abstractmethod async ¶

PollingDevice¶

ucapi_framework.device.PollingDevice ¶

Attributes¶

is_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

establish_connection abstractmethod async ¶

poll_device abstractmethod async ¶

WebSocketDevice¶

ucapi_framework.device.WebSocketDevice ¶

Attributes¶

is_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

send_ping async ¶

create_websocket abstractmethod async ¶

close_websocket abstractmethod async ¶

receive_message abstractmethod async ¶

handle_message abstractmethod async ¶

WebSocketPollingDevice¶

ucapi_framework.device.WebSocketPollingDevice ¶

Attributes¶

is_websocket_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

disconnect_all async ¶

ExternalClientDevice¶

ucapi_framework.device.ExternalClientDevice ¶

Attributes¶

is_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

create_client abstractmethod async ¶

connect_client abstractmethod async ¶

disconnect_client abstractmethod async ¶

check_client_connected abstractmethod ¶

PersistentConnectionDevice¶

ucapi_framework.device.PersistentConnectionDevice ¶

Attributes¶

is_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

establish_connection abstractmethod async ¶

close_connection abstractmethod async ¶

maintain_connection abstractmethod async ¶

DeviceEvents¶

ucapi_framework.device.DeviceEvents ¶

device_config `property` ¶

driver `property` ¶

identifier `abstractmethod` `property` ¶

name `abstractmethod` `property` ¶

address `abstractmethod` `property` ¶

log_id `abstractmethod` `property` ¶

is_connected `abstractmethod` `property` ¶

state `property` ¶

connect `abstractmethod` `async` ¶

disconnect `abstractmethod` `async` ¶

is_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

verify_connection `abstractmethod` `async` ¶

is_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

establish_connection `abstractmethod` `async` ¶

poll_device `abstractmethod` `async` ¶

is_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

send_ping `async` ¶

create_websocket `abstractmethod` `async` ¶

close_websocket `abstractmethod` `async` ¶

receive_message `abstractmethod` `async` ¶

handle_message `abstractmethod` `async` ¶

is_websocket_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

disconnect_all `async` ¶

is_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

create_client `abstractmethod` `async` ¶

connect_client `abstractmethod` `async` ¶

disconnect_client `abstractmethod` `async` ¶

check_client_connected `abstractmethod` ¶

is_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

establish_connection `abstractmethod` `async` ¶

close_connection `abstractmethod` `async` ¶

maintain_connection `abstractmethod` `async` ¶