Add event platform for OpenDisplay (#167393)

This commit is contained in:
g4bri3lDev
2026-04-09 23:37:25 +02:00
committed by GitHub
parent 2d45f9978e
commit 496c9551b3
7 changed files with 473 additions and 5 deletions
@@ -36,7 +36,7 @@ from .services import async_setup_services
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
_BASE_PLATFORMS: list[Platform] = []
_FLEX_PLATFORMS = [Platform.SENSOR]
_FLEX_PLATFORMS = [Platform.EVENT, Platform.SENSOR]
@dataclass
@@ -2,11 +2,11 @@
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import dataclass, field
import logging
from opendisplay import MANUFACTURER_ID, parse_advertisement
from opendisplay.models.advertisement import AdvertisementData
from opendisplay import MANUFACTURER_ID, AdvertisementTracker, parse_advertisement
from opendisplay.models.advertisement import AdvertisementData, ButtonChangeEvent
from homeassistant.components.bluetooth import (
BluetoothChange,
@@ -27,6 +27,7 @@ class OpenDisplayUpdate:
address: str
advertisement: AdvertisementData
button_events: list[ButtonChangeEvent] = field(default_factory=list)
class OpenDisplayCoordinator(PassiveBluetoothDataUpdateCoordinator):
@@ -42,6 +43,7 @@ class OpenDisplayCoordinator(PassiveBluetoothDataUpdateCoordinator):
connectable=True,
)
self.data: OpenDisplayUpdate | None = None
self._tracker: AdvertisementTracker = AdvertisementTracker()
@callback
def _async_handle_unavailable(
@@ -78,9 +80,11 @@ class OpenDisplayCoordinator(PassiveBluetoothDataUpdateCoordinator):
exc_info=True,
)
else:
button_events = self._tracker.update(service_info.address, advertisement)
self.data = OpenDisplayUpdate(
address=service_info.address,
advertisement=advertisement,
button_events=button_events,
)
super()._async_handle_bluetooth_event(service_info, change)
@@ -0,0 +1,93 @@
"""Event platform for OpenDisplay devices — button press/release events."""
from __future__ import annotations
from dataclasses import dataclass
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import OpenDisplayConfigEntry
from .entity import OpenDisplayEntity
PARALLEL_UPDATES = 0
@dataclass(frozen=True, kw_only=True)
class OpenDisplayEventEntityDescription(EventEntityDescription):
"""Describes an OpenDisplay button event entity."""
byte_index: int
button_id: int
async def async_setup_entry(
hass: HomeAssistant,
entry: OpenDisplayConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up OpenDisplay event entities from binary_inputs device config."""
coordinator = entry.runtime_data.coordinator
descriptions: list[OpenDisplayEventEntityDescription] = []
button_number = 0
for bi in entry.runtime_data.device_config.binary_inputs:
for button_id in range(8): # input_flags is a bitmask over 8 pin slots
if bi.input_flags & (1 << button_id):
button_number += 1
descriptions.append(
OpenDisplayEventEntityDescription(
key=f"button_{bi.instance_number}_{button_id}",
translation_key="button",
translation_placeholders={"number": str(button_number)},
device_class=EventDeviceClass.BUTTON,
event_types=["button_down", "button_up"],
byte_index=bi.button_data_byte_index,
button_id=button_id,
)
)
active_unique_ids = {f"{coordinator.address}-{d.key}" for d in descriptions}
button_unique_id_prefix = f"{coordinator.address}-button_"
entity_registry = er.async_get(hass)
for entity_entry in er.async_entries_for_config_entry(
entity_registry, entry.entry_id
):
if (
entity_entry.domain == "event"
and entity_entry.unique_id.startswith(button_unique_id_prefix)
and entity_entry.unique_id not in active_unique_ids
):
entity_registry.async_remove(entity_entry.entity_id)
async_add_entities(
OpenDisplayEventEntity(coordinator, description) for description in descriptions
)
class OpenDisplayEventEntity(OpenDisplayEntity, EventEntity):
"""A button event entity for an OpenDisplay device."""
entity_description: OpenDisplayEventEntityDescription
_last_processed_data: object | None = None
@callback
def _handle_coordinator_update(self) -> None:
"""Fire events for button transitions reported by this coordinator update."""
data = self.coordinator.data
if data is not None and data is not self._last_processed_data:
for event in data.button_events:
if (
event.byte_index == self.entity_description.byte_index
and event.button_id == self.entity_description.button_id
and event.event_type in self.event_types
):
self._trigger_event(event.event_type)
self._last_processed_data = data
self.async_write_ha_state()
@@ -51,6 +51,19 @@
}
},
"entity": {
"event": {
"button": {
"name": "Button {number}",
"state_attributes": {
"event_type": {
"state": {
"button_down": "Button down",
"button_up": "Button up"
}
}
}
}
},
"sensor": {
"battery_voltage": {
"name": "Battery voltage"
+69
View File
@@ -4,6 +4,7 @@ from time import time
from bleak.backends.scanner import AdvertisementData
from opendisplay import (
BinaryInputs,
BoardManufacturer,
ColorScheme,
DisplayConfig,
@@ -118,6 +119,74 @@ def make_service_info(
)
BINARY_INPUT = BinaryInputs(
instance_number=0,
input_type=0,
display_as=0,
reserved_pins=b"\x00" * 8,
input_flags=0x01, # bit 0 set → button_id 0 active
invert=0,
pullups=0,
pulldowns=0,
button_data_byte_index=0,
)
BUTTON_DEVICE_CONFIG = GlobalConfig(
system=DEVICE_CONFIG.system,
manufacturer=DEVICE_CONFIG.manufacturer,
power=DEVICE_CONFIG.power,
displays=DEVICE_CONFIG.displays,
binary_inputs=[BINARY_INPUT],
)
def make_v1_service_info(
dynamic_data: bytes = b"\x00" * 11,
name: str | None = "OpenDisplay 1234",
address: str = TEST_ADDRESS,
) -> BluetoothServiceInfoBleak:
"""Create a v1 advertisement service info with a custom 11-byte dynamic block."""
# temperature=25.0°C, battery≈3700 mV, loop_counter=1
return make_service_info(
name=name,
address=address,
manufacturer_data={OPENDISPLAY_MANUFACTURER_ID: dynamic_data + b"\x82\x72\x11"},
)
def make_binary_inputs(
instance_number: int = 0,
byte_index: int = 0,
input_flags: int = 0x01,
) -> BinaryInputs:
"""Create a minimal BinaryInputs config entry.
input_flags is a bitmask of active inputs: bit N set means button_id N is active.
"""
return BinaryInputs(
instance_number=instance_number,
input_type=0,
display_as=0,
reserved_pins=b"\x00" * 8,
input_flags=input_flags,
invert=0,
pullups=0,
pulldowns=0,
button_data_byte_index=byte_index,
)
def make_button_device_config(binary_inputs: list[BinaryInputs]) -> GlobalConfig:
"""Return a GlobalConfig with the given binary_inputs list."""
return GlobalConfig(
system=DEVICE_CONFIG.system,
manufacturer=DEVICE_CONFIG.manufacturer,
power=DEVICE_CONFIG.power,
displays=DEVICE_CONFIG.displays,
binary_inputs=binary_inputs,
)
VALID_SERVICE_INFO = make_service_info()
NOT_OPENDISPLAY_SERVICE_INFO = make_service_info(
+75 -1
View File
@@ -7,7 +7,16 @@ import pytest
from homeassistant.components.opendisplay.const import CONF_ENCRYPTION_KEY, DOMAIN
from . import DEVICE_CONFIG, ENCRYPTION_KEY, FIRMWARE_VERSION, TEST_ADDRESS, TEST_TITLE
from . import (
BUTTON_DEVICE_CONFIG,
DEVICE_CONFIG,
ENCRYPTION_KEY,
FIRMWARE_VERSION,
TEST_ADDRESS,
TEST_TITLE,
make_binary_inputs,
make_button_device_config,
)
from tests.common import MockConfigEntry
from tests.components.bluetooth import generate_ble_device
@@ -81,6 +90,71 @@ def mock_config_entry() -> MockConfigEntry:
)
@pytest.fixture
def mock_button_config_entry(mock_opendisplay_device: MagicMock) -> MockConfigEntry:
"""Create a mock config entry for a device with one button configured."""
mock_opendisplay_device.config = BUTTON_DEVICE_CONFIG
return MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_ADDRESS,
title=TEST_TITLE,
data={},
)
@pytest.fixture
def mock_two_button_config_entry(mock_opendisplay_device: MagicMock) -> MockConfigEntry:
"""Create a mock config entry for a device with two buttons configured."""
mock_opendisplay_device.config = make_button_device_config(
[make_binary_inputs(input_flags=0x03)]
)
return MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_ADDRESS,
title=TEST_TITLE,
data={},
)
@pytest.fixture
def mock_three_button_config_entry(
mock_opendisplay_device: MagicMock,
) -> MockConfigEntry:
"""Create a mock config entry for a device with three buttons configured."""
mock_opendisplay_device.config = make_button_device_config(
[make_binary_inputs(input_flags=0x07)]
)
return MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_ADDRESS,
title=TEST_TITLE,
data={},
)
@pytest.fixture
def mock_multi_instance_config_entry(
mock_opendisplay_device: MagicMock,
) -> MockConfigEntry:
"""Create a mock config entry with two binary_inputs instances.
Instance 0: byte_index=0, buttons 0 and 1 active → Button 1, Button 2
Instance 1: byte_index=1, button 0 active → Button 3
"""
mock_opendisplay_device.config = make_button_device_config(
[
make_binary_inputs(instance_number=0, byte_index=0, input_flags=0x03),
make_binary_inputs(instance_number=1, byte_index=1, input_flags=0x01),
]
)
return MockConfigEntry(
domain=DOMAIN,
unique_id=TEST_ADDRESS,
title=TEST_TITLE,
data={},
)
@pytest.fixture
def mock_encrypted_config_entry() -> MockConfigEntry:
"""Create a mock config entry with an encryption key."""
+215
View File
@@ -0,0 +1,215 @@
"""Test the OpenDisplay button event platform."""
from unittest.mock import MagicMock
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from . import (
TEST_ADDRESS,
make_binary_inputs,
make_button_device_config,
make_v1_service_info,
)
from tests.common import MockConfigEntry
from tests.components.bluetooth import inject_bluetooth_service_info
async def test_no_entities_without_binary_inputs(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""No event entities are created when the device has no binary inputs configured."""
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert not any(e.domain == "event" for e in entries)
async def test_entities_created_per_active_button(
hass: HomeAssistant,
mock_three_button_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""One event entity is created per active button bit in binary_inputs."""
mock_three_button_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(
mock_three_button_config_entry.entry_id
)
await hass.async_block_till_done()
entries = er.async_entries_for_config_entry(
entity_registry, mock_three_button_config_entry.entry_id
)
event_entries = [e for e in entries if e.domain == "event"]
assert len(event_entries) == 3
assert {e.unique_id for e in event_entries} == {
f"{TEST_ADDRESS}-button_0_0",
f"{TEST_ADDRESS}-button_0_1",
f"{TEST_ADDRESS}-button_0_2",
}
async def test_multiple_binary_input_instances(
hass: HomeAssistant,
mock_multi_instance_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Entities across multiple BinaryInputs instances get globally sequential button numbers."""
mock_multi_instance_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(
mock_multi_instance_config_entry.entry_id
)
await hass.async_block_till_done()
entries = er.async_entries_for_config_entry(
entity_registry, mock_multi_instance_config_entry.entry_id
)
event_entries = [e for e in entries if e.domain == "event"]
assert len(event_entries) == 3
assert {e.unique_id for e in event_entries} == {
f"{TEST_ADDRESS}-button_0_0",
f"{TEST_ADDRESS}-button_0_1",
f"{TEST_ADDRESS}-button_1_0",
}
# Button numbers must be globally sequential, not reset per instance
names = {
e.unique_id: hass.states.get(e.entity_id).attributes.get("friendly_name", "")
for e in event_entries
}
assert names[f"{TEST_ADDRESS}-button_0_0"].endswith("Button 1")
assert names[f"{TEST_ADDRESS}-button_0_1"].endswith("Button 2")
assert names[f"{TEST_ADDRESS}-button_1_0"].endswith("Button 3")
async def test_stale_entities_removed_on_config_change(
hass: HomeAssistant,
mock_two_button_config_entry: MockConfigEntry,
mock_opendisplay_device: MagicMock,
entity_registry: er.EntityRegistry,
) -> None:
"""Entities for buttons no longer in device config are removed on reload."""
mock_two_button_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_two_button_config_entry.entry_id)
await hass.async_block_till_done()
assert (
len(
[
e
for e in er.async_entries_for_config_entry(
entity_registry, mock_two_button_config_entry.entry_id
)
if e.domain == "event"
]
)
== 2
)
# Device reconfigured: now only 1 active button
mock_opendisplay_device.config = make_button_device_config(
[make_binary_inputs(input_flags=0x01)]
)
assert await hass.config_entries.async_unload(mock_two_button_config_entry.entry_id)
assert await hass.config_entries.async_setup(mock_two_button_config_entry.entry_id)
await hass.async_block_till_done()
event_entries = [
e
for e in er.async_entries_for_config_entry(
entity_registry, mock_two_button_config_entry.entry_id
)
if e.domain == "event"
]
assert len(event_entries) == 1
assert event_entries[0].unique_id == f"{TEST_ADDRESS}-button_0_0"
async def test_button_down_event_fired(
hass: HomeAssistant,
mock_button_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""A 'button_down' event fires when the button transitions to the down state."""
mock_button_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_button_config_entry.entry_id)
await hass.async_block_till_done()
# First advertisement — seeds tracker state (no events emitted)
inject_bluetooth_service_info(hass, make_v1_service_info())
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id(
"event", "opendisplay", f"{TEST_ADDRESS}-button_0_0"
)
assert entity_id is not None
state_before = hass.states.get(entity_id)
# Second advertisement — byte 0: pressed=True, button_id=0 → raw=0x80
inject_bluetooth_service_info(hass, make_v1_service_info(b"\x80" + b"\x00" * 10))
await hass.async_block_till_done()
state_after = hass.states.get(entity_id)
assert state_after is not None
assert state_after.attributes.get("event_type") == "button_down"
assert state_before != state_after
async def test_button_up_event_fired(
hass: HomeAssistant,
mock_button_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""A 'button_up' event fires when the button transitions from down to up."""
mock_button_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_button_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id(
"event", "opendisplay", f"{TEST_ADDRESS}-button_0_0"
)
assert entity_id is not None
# Seed with pressed state (pressed=True, button_id=0 → raw=0x80)
inject_bluetooth_service_info(hass, make_v1_service_info(b"\x80" + b"\x00" * 10))
await hass.async_block_till_done()
# Transition to released (pressed=False, button_id=0 → raw=0x00)
inject_bluetooth_service_info(hass, make_v1_service_info())
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state is not None
assert state.attributes.get("event_type") == "button_up"
async def test_no_event_for_wrong_button_id(
hass: HomeAssistant,
mock_two_button_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
) -> None:
"""Entity watching button_id=1 does not fire when button_id=0 changes."""
mock_two_button_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_two_button_config_entry.entry_id)
await hass.async_block_till_done()
inject_bluetooth_service_info(hass, make_v1_service_info())
await hass.async_block_till_done()
entity_id = entity_registry.async_get_entity_id(
"event", "opendisplay", f"{TEST_ADDRESS}-button_0_1"
)
assert entity_id is not None
state_before = hass.states.get(entity_id)
# Press button_id=0 only (raw=0x80: pressed=True, button_id=0)
inject_bluetooth_service_info(hass, make_v1_service_info(b"\x80" + b"\x00" * 10))
await hass.async_block_till_done()
assert hass.states.get(entity_id) == state_before