Active scan govee_ble only when needed and widen the window to 30s (#174557)

This commit is contained in:
J. Nick Koston
2026-06-23 21:03:38 -04:00
committed by GitHub
parent 7fd101005d
commit adf2f2854c
3 changed files with 85 additions and 7 deletions
@@ -5,7 +5,6 @@ import logging
from govee_ble import GoveeBluetoothDeviceData
from homeassistant.components.bluetooth import BluetoothScanningMode
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
@@ -29,7 +28,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: GoveeBLEConfigEntry) ->
hass,
_LOGGER,
address=address,
mode=BluetoothScanningMode.ACTIVE,
update_method=partial(process_service_info, hass, entry),
device_data=data,
entry=entry,
@@ -21,6 +21,12 @@ from .const import CONF_DEVICE_TYPE, DOMAIN
type GoveeBLEConfigEntry = ConfigEntry[GoveeBLEBluetoothProcessorCoordinator]
# Models such as the H5074 carry their measurements only in the scan response,
# so the scanner must stay active long enough to capture one; the default 10s
# window misses them most cycles. 30s is the longest active window habluetooth
# allows (AUTO_WINDOW_MAX_DURATION) and reliably spans a full broadcast cycle.
ACTIVE_SCAN_DURATION = 30.0
def process_service_info(
hass: HomeAssistant,
@@ -65,18 +71,30 @@ class GoveeBLEBluetoothProcessorCoordinator(
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.model_info: ModelInfo | None = None
# Active scanning is only needed for models that carry their payload in
# the scan response; passively broadcasting models would otherwise be
# scanned needlessly, costing fleet radio time and sensor battery.
# When the model is not yet known, scan actively so a scan-response-only
# model can still be discovered.
mode = BluetoothScanningMode.ACTIVE
scan_duration: float | None = None
if device_type := entry.data.get(CONF_DEVICE_TYPE):
self.model_info = model_info = get_model_info(device_type)
if model_info.requires_active_scan:
scan_duration = ACTIVE_SCAN_DURATION
else:
mode = BluetoothScanningMode.PASSIVE
super().__init__(
hass, logger, address, mode, update_method, scan_duration=scan_duration
)
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."""
+62
View File
@@ -0,0 +1,62 @@
"""Test the Govee BLE init."""
from unittest.mock import patch
import pytest
from homeassistant.components.bluetooth import BluetoothScanningMode
from homeassistant.components.govee_ble.const import CONF_DEVICE_TYPE, DOMAIN
from homeassistant.components.govee_ble.coordinator import ACTIVE_SCAN_DURATION
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("device_type", "expected_mode", "expected_scan_duration"),
[
pytest.param(
"H5074",
BluetoothScanningMode.ACTIVE,
ACTIVE_SCAN_DURATION,
id="h5074_scan_response",
),
pytest.param(
"H5075",
BluetoothScanningMode.ACTIVE,
ACTIVE_SCAN_DURATION,
id="h5075_scan_response",
),
pytest.param(
"H5179",
BluetoothScanningMode.PASSIVE,
None,
id="primary_advertisement",
),
],
)
async def test_active_scan_duration(
hass: HomeAssistant,
device_type: str,
expected_mode: BluetoothScanningMode,
expected_scan_duration: float | None,
) -> None:
"""Test only scan-response-only models are scanned actively."""
entry = MockConfigEntry(
domain=DOMAIN,
unique_id="61DE521B-F0BF-9F44-64D4-75BBE1738105",
data={CONF_DEVICE_TYPE: device_type},
)
entry.add_to_hass(hass)
with patch(
"homeassistant.components.bluetooth.update_coordinator.async_register_callback"
) as mock_register_callback:
assert await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert mock_register_callback.call_args.args[3] == expected_mode
assert (
mock_register_callback.call_args.kwargs["scan_duration"]
== expected_scan_duration
)