Add event platform to govee-ble (#122031)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
J. Nick Koston 2024-07-21 09:32:58 -05:00 committed by GitHub
parent 8d01ad98eb
commit 7e82b3ecdb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 429 additions and 24 deletions

View File

@ -2,38 +2,40 @@
from __future__ import annotations
from functools import partial
import logging
from govee_ble import GoveeBluetoothDeviceData, SensorUpdate
from govee_ble import GoveeBluetoothDeviceData
from homeassistant.components.bluetooth import BluetoothScanningMode
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothProcessorCoordinator,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
PLATFORMS: list[Platform] = [Platform.SENSOR]
from .coordinator import (
GoveeBLEBluetoothProcessorCoordinator,
GoveeBLEConfigEntry,
process_service_info,
)
PLATFORMS: list[Platform] = [Platform.EVENT, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
GoveeBLEConfigEntry = ConfigEntry[PassiveBluetoothProcessorCoordinator[SensorUpdate]]
async def async_setup_entry(hass: HomeAssistant, entry: GoveeBLEConfigEntry) -> bool:
"""Set up Govee BLE device from a config entry."""
address = entry.unique_id
assert address is not None
data = GoveeBluetoothDeviceData()
coordinator = PassiveBluetoothProcessorCoordinator(
entry.runtime_data = coordinator = GoveeBLEBluetoothProcessorCoordinator(
hass,
_LOGGER,
address=address,
mode=BluetoothScanningMode.ACTIVE,
update_method=data.update,
update_method=partial(process_service_info, hass, entry),
device_data=data,
entry=entry,
)
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
# only start after all platforms have had a chance to subscribe
entry.async_on_unload(coordinator.async_start())

View File

@ -14,7 +14,7 @@ from homeassistant.components.bluetooth import (
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ADDRESS
from .const import DOMAIN
from .const import CONF_DEVICE_TYPE, DOMAIN
class GoveeConfigFlow(ConfigFlow, domain=DOMAIN):
@ -26,7 +26,9 @@ class GoveeConfigFlow(ConfigFlow, domain=DOMAIN):
"""Initialize the config flow."""
self._discovery_info: BluetoothServiceInfoBleak | None = None
self._discovered_device: DeviceData | None = None
self._discovered_devices: dict[str, str] = {}
self._discovered_devices: dict[
str, tuple[DeviceData, BluetoothServiceInfoBleak]
] = {}
async def async_step_bluetooth(
self, discovery_info: BluetoothServiceInfoBleak
@ -51,7 +53,9 @@ class GoveeConfigFlow(ConfigFlow, domain=DOMAIN):
discovery_info = self._discovery_info
title = device.title or device.get_device_name() or discovery_info.name
if user_input is not None:
return self.async_create_entry(title=title, data={})
return self.async_create_entry(
title=title, data={CONF_DEVICE_TYPE: device.device_type}
)
self._set_confirm_only()
placeholders = {"name": title}
@ -68,8 +72,10 @@ class GoveeConfigFlow(ConfigFlow, domain=DOMAIN):
address = user_input[CONF_ADDRESS]
await self.async_set_unique_id(address, raise_on_progress=False)
self._abort_if_unique_id_configured()
device, service_info = self._discovered_devices[address]
title = device.title or device.get_device_name() or service_info.name
return self.async_create_entry(
title=self._discovered_devices[address], data={}
title=title, data={CONF_DEVICE_TYPE: device.device_type}
)
current_addresses = self._async_current_ids()
@ -79,9 +85,7 @@ class GoveeConfigFlow(ConfigFlow, domain=DOMAIN):
continue
device = DeviceData()
if device.supported(discovery_info):
self._discovered_devices[address] = (
device.title or device.get_device_name() or discovery_info.name
)
self._discovered_devices[address] = (device, discovery_info)
if not self._discovered_devices:
return self.async_abort(reason="no_devices_found")
@ -89,6 +93,16 @@ class GoveeConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{vol.Required(CONF_ADDRESS): vol.In(self._discovered_devices)}
{
vol.Required(CONF_ADDRESS): vol.In(
{
address: f"{device.get_device_name(None) or discovery_info.name} ({address})"
for address, (
device,
discovery_info,
) in self._discovered_devices.items()
}
)
}
),
)

View File

@ -1,3 +1,5 @@
"""Constants for the Govee Bluetooth integration."""
DOMAIN = "govee_ble"
CONF_DEVICE_TYPE = "device_type"

View File

@ -0,0 +1,79 @@
"""The govee Bluetooth integration."""
from collections.abc import Callable
from logging import Logger
from govee_ble import GoveeBluetoothDeviceData, ModelInfo, SensorUpdate, get_model_info
from homeassistant.components.bluetooth import (
BluetoothScanningMode,
BluetoothServiceInfoBleak,
)
from homeassistant.components.bluetooth.passive_update_processor import (
PassiveBluetoothProcessorCoordinator,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CoreState, HomeAssistant
from homeassistant.helpers.dispatcher import async_dispatcher_send
from .const import CONF_DEVICE_TYPE, DOMAIN
type GoveeBLEConfigEntry = ConfigEntry[GoveeBLEBluetoothProcessorCoordinator]
def process_service_info(
hass: HomeAssistant,
entry: GoveeBLEConfigEntry,
service_info: BluetoothServiceInfoBleak,
) -> SensorUpdate:
"""Process a BluetoothServiceInfoBleak, running side effects and returning sensor data."""
coordinator = entry.runtime_data
data = coordinator.device_data
update = data.update(service_info)
if not coordinator.model_info and (device_type := data.device_type):
hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_DEVICE_TYPE: device_type}
)
coordinator.set_model_info(device_type)
if update.events and hass.state is CoreState.running:
# Do not fire events on data restore
address = service_info.device.address
for event in update.events.values():
key = event.device_key.key
signal = format_event_dispatcher_name(address, key)
async_dispatcher_send(hass, signal)
return update
def format_event_dispatcher_name(address: str, key: str) -> str:
"""Format an event dispatcher name."""
return f"{DOMAIN}_{address}_{key}"
class GoveeBLEBluetoothProcessorCoordinator(
PassiveBluetoothProcessorCoordinator[SensorUpdate]
):
"""Define a govee ble Bluetooth Passive Update Processor Coordinator."""
def __init__(
self,
hass: HomeAssistant,
logger: Logger,
address: str,
mode: BluetoothScanningMode,
update_method: Callable[[BluetoothServiceInfoBleak], SensorUpdate],
device_data: GoveeBluetoothDeviceData,
entry: ConfigEntry,
) -> None:
"""Initialize the Govee BLE Bluetooth Passive Update Processor Coordinator."""
super().__init__(hass, logger, address, mode, update_method)
self.device_data = device_data
self.entry = entry
self.model_info: ModelInfo | None = None
if device_type := entry.data.get(CONF_DEVICE_TYPE):
self.set_model_info(device_type)
def set_model_info(self, device_type: str) -> None:
"""Set the model info."""
self.model_info = get_model_info(device_type)

View File

@ -0,0 +1,107 @@
"""Support for govee_ble event entities."""
from __future__ import annotations
from govee_ble import ModelInfo, SensorType
from homeassistant.components.bluetooth import (
BluetoothServiceInfoBleak,
async_last_service_info,
)
from homeassistant.components.event import (
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .coordinator import GoveeBLEConfigEntry, format_event_dispatcher_name
BUTTON_DESCRIPTIONS = [
EventEntityDescription(
key=f"button_{i}",
translation_key=f"button_{i}",
event_types=["press"],
device_class=EventDeviceClass.BUTTON,
)
for i in range(6)
]
MOTION_DESCRIPTION = EventEntityDescription(
key="motion",
event_types=["motion"],
device_class=EventDeviceClass.MOTION,
)
class GoveeBluetoothEventEntity(EventEntity):
"""Representation of a govee ble event entity."""
_attr_should_poll = False
_attr_has_entity_name = True
def __init__(
self,
model_info: ModelInfo,
service_info: BluetoothServiceInfoBleak | None,
address: str,
description: EventEntityDescription,
) -> None:
"""Initialise a govee ble event entity."""
self.entity_description = description
# Matches logic in PassiveBluetoothProcessorEntity
name = service_info.name if service_info else model_info.model_id
self._attr_device_info = dr.DeviceInfo(
name=name,
identifiers={(DOMAIN, address)},
connections={(dr.CONNECTION_BLUETOOTH, address)},
)
self._attr_unique_id = f"{address}-{description.key}"
self._address = address
self._signal = format_event_dispatcher_name(
self._address, self.entity_description.key
)
async def async_added_to_hass(self) -> None:
"""Entity added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
async_dispatcher_connect(
self.hass,
self._signal,
self._async_handle_event,
)
)
@callback
def _async_handle_event(self) -> None:
self._trigger_event(self.event_types[0])
self.async_write_ha_state()
async def async_setup_entry(
hass: HomeAssistant,
entry: GoveeBLEConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up a govee ble event."""
coordinator = entry.runtime_data
if not (model_info := coordinator.model_info):
return
address = coordinator.address
sensor_type = model_info.sensor_type
if sensor_type is SensorType.MOTION:
descriptions = [MOTION_DESCRIPTION]
elif sensor_type is SensorType.BUTTON:
button_count = model_info.button_count
descriptions = BUTTON_DESCRIPTIONS[0:button_count]
else:
return
last_service_info = async_last_service_info(hass, address, False)
async_add_entities(
GoveeBluetoothEventEntity(model_info, last_service_info, address, description)
for description in descriptions
)

View File

@ -27,7 +27,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.sensor import sensor_device_info_to_hass_device_info
from . import GoveeBLEConfigEntry
from .coordinator import GoveeBLEConfigEntry
SENSOR_DESCRIPTIONS = {
(DeviceClass.TEMPERATURE, Units.TEMP_CELSIUS): SensorEntityDescription(

View File

@ -17,5 +17,78 @@
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
}
},
"entity": {
"event": {
"motion": {
"state_attributes": {
"event_type": {
"state": {
"motion": "[%key:component::event::entity_component::motion::name%]"
}
}
}
},
"button_0": {
"name": "Button 1",
"state_attributes": {
"event_type": {
"state": {
"press": "Press"
}
}
}
},
"button_1": {
"name": "Button 2",
"state_attributes": {
"event_type": {
"state": {
"press": "[%key:component::govee_ble::entity::event::button_0::state_attributes::event_type::state::press%]"
}
}
}
},
"button_2": {
"name": "Button 3",
"state_attributes": {
"event_type": {
"state": {
"press": "[%key:component::govee_ble::entity::event::button_0::state_attributes::event_type::state::press%]"
}
}
}
},
"button_3": {
"name": "Button 4",
"state_attributes": {
"event_type": {
"state": {
"press": "[%key:component::govee_ble::entity::event::button_0::state_attributes::event_type::state::press%]"
}
}
}
},
"button_4": {
"name": "Button 5",
"state_attributes": {
"event_type": {
"state": {
"press": "[%key:component::govee_ble::entity::event::button_0::state_attributes::event_type::state::press%]"
}
}
}
},
"button_5": {
"name": "Button 6",
"state_attributes": {
"event_type": {
"state": {
"press": "[%key:component::govee_ble::entity::event::button_0::state_attributes::event_type::state::press%]"
}
}
}
}
}
}
}

View File

@ -83,3 +83,56 @@ GVH5106_SERVICE_INFO = BluetoothServiceInfo(
service_data={},
source="local",
)
GV5125_BUTTON_0_SERVICE_INFO = BluetoothServiceInfo(
name="GV51255367",
address="C1:37:37:32:0F:45",
rssi=-36,
manufacturer_data={
60552: b"\x01\n.\xaf\xd9085Sg\x01\x01",
61320: b".\xaf\x00\x00b\\\xae\x92\x15\xb6\xa8\n\xd4\x81K\xcaK_s\xd9E40\x02",
},
service_data={},
service_uuids=[],
source="24:4C:AB:03:E6:B8",
)
GV5125_BUTTON_1_SERVICE_INFO = BluetoothServiceInfo(
name="GV51255367",
address="C1:37:37:32:0F:45",
rssi=-36,
manufacturer_data={
60552: b"\x01\n.\xaf\xd9085Sg\x01\x01",
61320: b".\xaf\x00\x00\xfb\x0e\xc9h\xd7\x05l\xaf*\xf3\x1b\xe8w\xf1\xe1\xe8\xe3\xa7\xf8\xc6",
},
service_data={},
service_uuids=[],
source="24:4C:AB:03:E6:B8",
)
GV5121_MOTION_SERVICE_INFO = BluetoothServiceInfo(
name="GV5121195A",
address="C1:37:37:32:0F:45",
rssi=-36,
manufacturer_data={
61320: b"Y\x94\x00\x00\xf0\xb9\x197\xaeP\xb67,\x86j\xc2\xf3\xd0a\xe7\x17\xc0,\xef"
},
service_data={},
service_uuids=[],
source="24:4C:AB:03:E6:B8",
)
GV5121_MOTION_SERVICE_INFO_2 = BluetoothServiceInfo(
name="GV5121195A",
address="C1:37:37:32:0F:45",
rssi=-36,
manufacturer_data={
61320: b"Y\x94\x00\x06\xa3f6e\xc8\xe6\xfdv\x04\xaf\xe7k\xbf\xab\xeb\xbf\xb3\xa3\xd5\x19"
},
service_data={},
service_uuids=[],
source="24:4C:AB:03:E6:B8",
)

View File

@ -3,7 +3,7 @@
from unittest.mock import patch
from homeassistant import config_entries
from homeassistant.components.govee_ble.const import DOMAIN
from homeassistant.components.govee_ble.const import CONF_DEVICE_TYPE, DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
@ -29,7 +29,7 @@ async def test_async_step_bluetooth_valid_device(hass: HomeAssistant) -> None:
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "H5075 2762"
assert result2["data"] == {}
assert result2["data"] == {CONF_DEVICE_TYPE: "H5075"}
assert result2["result"].unique_id == "61DE521B-F0BF-9F44-64D4-75BBE1738105"
@ -75,7 +75,7 @@ async def test_async_step_user_with_found_devices(hass: HomeAssistant) -> None:
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "H5177 2EC8"
assert result2["data"] == {}
assert result2["data"] == {CONF_DEVICE_TYPE: "H5177"}
assert result2["result"].unique_id == "4125DDBA-2774-4851-9889-6AADDD4CAC3D"
@ -198,7 +198,7 @@ async def test_async_step_user_takes_precedence_over_discovery(
)
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "H5177 2EC8"
assert result2["data"] == {}
assert result2["data"] == {CONF_DEVICE_TYPE: "H5177"}
assert result2["result"].unique_id == "4125DDBA-2774-4851-9889-6AADDD4CAC3D"
# Verify the original one was aborted

View File

@ -0,0 +1,75 @@
"""Test the Govee BLE events."""
from homeassistant.components.govee_ble.const import CONF_DEVICE_TYPE, DOMAIN
from homeassistant.const import STATE_UNKNOWN
from homeassistant.core import HomeAssistant
from . import (
GV5121_MOTION_SERVICE_INFO,
GV5121_MOTION_SERVICE_INFO_2,
GV5125_BUTTON_0_SERVICE_INFO,
GV5125_BUTTON_1_SERVICE_INFO,
)
from tests.common import MockConfigEntry
from tests.components.bluetooth import inject_bluetooth_service_info
async def test_motion_sensor(hass: HomeAssistant) -> None:
"""Test setting up creates the motion sensor."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=GV5121_MOTION_SERVICE_INFO.address,
data={CONF_DEVICE_TYPE: "H5121"},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 1
inject_bluetooth_service_info(hass, GV5121_MOTION_SERVICE_INFO)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 2
motion_sensor = hass.states.get("event.h5121_motion")
first_time = motion_sensor.state
assert motion_sensor.state != STATE_UNKNOWN
inject_bluetooth_service_info(hass, GV5121_MOTION_SERVICE_INFO_2)
await hass.async_block_till_done()
motion_sensor = hass.states.get("event.h5121_motion")
assert motion_sensor.state != first_time
assert motion_sensor.state != STATE_UNKNOWN
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
async def test_button(hass: HomeAssistant) -> None:
"""Test setting up creates the buttons."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id=GV5125_BUTTON_1_SERVICE_INFO.address,
data={CONF_DEVICE_TYPE: "H5125"},
)
entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_all()) == 6
inject_bluetooth_service_info(hass, GV5125_BUTTON_1_SERVICE_INFO)
await hass.async_block_till_done()
button_1 = hass.states.get("event.h5125_button_1")
assert button_1.state == STATE_UNKNOWN
inject_bluetooth_service_info(hass, GV5125_BUTTON_0_SERVICE_INFO)
await hass.async_block_till_done()
button_1 = hass.states.get("event.h5125_button_1")
assert button_1.state != STATE_UNKNOWN
assert len(hass.states.async_all()) == 7
assert await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()