Mirrors_Framework/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-08-17 19:13:18 +08:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

[ble_device_base] Share scan parameter validation between trackers (#18003)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
J. Nick Koston
2026-08-02 19:28:22 -05:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent d6ad1560fa
commit 74fbee7d8d
5 changed files with 140 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/components/bk72xx_ble_tracker/__init__.py
+6 -75
View File
@@ -39,79 +39,10 @@ BK72xxBLETracker = bk72xx_ble_tracker_ns.class_(
)
def to_ble_units(value: cv.TimePeriod) -> int:
"""Convert a scan time to the controller's 0.625 ms units.
Used by both validation and codegen so what is validated is exactly what is
programmed — the truncation here is what makes the duty-cycle check below
meaningful.
"""
return value.total_microseconds // 625
def validate_scan_parameters(config: ConfigType) -> ConfigType:
"""Reject impossible window/interval/duration combinations at config time.
Mirrors esp32_ble_tracker: the controller cannot scan for longer than the
interval, and a too-short duration would end the scan period almost
immediately. Catching it here gives a clear error instead of a runtime
controller failure and the 1/sec retry loop.
"""
duration = config[CONF_DURATION]
interval = config[CONF_INTERVAL]
window = config[CONF_WINDOW]
if window > interval:
raise cv.Invalid(
f"Scan window ({window}) needs to be smaller than scan interval ({interval})"
)
# BLE scan interval/window are programmed in 0.625 ms units as a 16-bit value; the
# controller only accepts 2.5 ms .. 10240 ms (0x0004 .. 0x4000). Reject out-of-range
# values here instead of letting the unit conversion silently overflow.
for name, value in (("interval", interval), ("window", window)):
if value.total_microseconds < 2500 or value.total_microseconds > 10_240_000:
raise cv.Invalid(
f"Scan {name} ({value}) must be between 2.5 ms and 10240 ms"
)
# Validate what actually reaches the controller: both values are truncated to
# whole 0.625 ms units, so a window/interval pair that differs by less than one
# unit collapses to the same value — silently programming a 100 % duty cycle
# (radio permanently on) from a config that asked for less.
interval_units = to_ble_units(interval)
window_units = to_ble_units(window)
if window_units == interval_units and window < interval:
raise cv.Invalid(
f"Scan window ({window}) and interval ({interval}) both round to "
f"{interval_units} x 0.625 ms, which the controller scans at a 100 % duty "
f"cycle. Separate them by at least 0.625 ms."
)
if interval.total_microseconds * 3 > duration.total_microseconds:
raise cv.Invalid(
f"Scan duration ({duration}) must cover at least three scan intervals "
f"({interval}): the scanner listens on one of the three BLE advertising "
f"channels per interval, so a shorter duration can miss devices entirely."
)
return config
SCAN_PARAMETERS_SCHEMA = cv.All(
cv.Schema(
{
cv.Optional(CONF_DURATION, default="5min"): cv.positive_time_period_seconds,
# interval/window default to the BK reference scan rate — 100 ms / 30 ms,
# a 30 % duty cycle. Converted to the controller's 0.625 ms BLE units in
# to_code(). (LN882H's SDK recommends a different 100 / 50 ms = 50 %.)
cv.Optional(CONF_INTERVAL, default="100ms"): cv.positive_time_period,
cv.Optional(CONF_WINDOW, default="30ms"): cv.positive_time_period,
cv.Optional(CONF_CONTINUOUS, default=True): cv.boolean,
}
),
validate_scan_parameters,
)
# interval defaults to the BK reference scan rate — 100 ms with the shared 30 ms
# window, a 30 % duty cycle. Converted to the controller's 0.625 ms BLE units in
# to_code(). (LN882H's SDK recommends a different 100 / 50 ms = 50 %.)
SCAN_PARAMETERS_SCHEMA = ble_device_base.scan_parameters_schema("100ms")
CONFIG_SCHEMA = cv.Schema(
{
@@ -143,8 +74,8 @@ async def to_code(config: ConfigType) -> None:
ota.request_ota_state_listeners()
scan = config[CONF_SCAN_PARAMETERS]
cg.add(var.set_scan_interval(to_ble_units(scan[CONF_INTERVAL])))
cg.add(var.set_scan_window(to_ble_units(scan[CONF_WINDOW])))
cg.add(var.set_scan_interval(ble_device_base.to_ble_units(scan[CONF_INTERVAL])))
cg.add(var.set_scan_window(ble_device_base.to_ble_units(scan[CONF_WINDOW])))
cg.add(var.set_scan_duration(scan[CONF_DURATION].total_milliseconds))
cg.add(var.set_scan_continuous(scan[CONF_CONTINUOUS]))
esphome/components/ble_device_base/__init__.py
+88
View File
@@ -20,7 +20,9 @@ ble_aes_ccm.h.
import re
import esphome.codegen as cg
from esphome.components.const import CONF_WINDOW
import esphome.config_validation as cv
from esphome.const import CONF_ACTIVE, CONF_CONTINUOUS, CONF_DURATION, CONF_INTERVAL
from esphome.core import CORE
from esphome.types import ConfigType
@@ -76,6 +78,92 @@ async def register_ble_device(var: cg.MockObj, config: ConfigType) -> cg.MockObj
# ---- shared validation / codegen helpers (platform-neutral) ----
def to_ble_units(value: cv.TimePeriod) -> int:
"""Convert a scan time to the controller's 0.625 ms units.
Used by both validation and codegen so what is validated is exactly what is
programmed — the truncation here is what makes the duty-cycle check below
meaningful.
"""
return value.total_microseconds // 625
def validate_scan_parameters(config: ConfigType) -> ConfigType:
"""Reject impossible window/interval/duration combinations at config time.
The controller cannot scan for longer than the interval, and a too-short
duration would end the scan period almost immediately. Catching it here
gives a clear error instead of a runtime controller failure and a retry
loop.
"""
duration = config[CONF_DURATION]
interval = config[CONF_INTERVAL]
window = config[CONF_WINDOW]
if window > interval:
raise cv.Invalid(
f"Scan window ({window}) needs to be smaller than scan interval ({interval})"
)
# BLE scan interval/window are programmed in 0.625 ms units as a 16-bit value; the
# controller only accepts 2.5 ms .. 10240 ms (0x0004 .. 0x4000). Reject out-of-range
# values here instead of letting the unit conversion silently overflow.
for name, value in (("interval", interval), ("window", window)):
if value.total_microseconds < 2500 or value.total_microseconds > 10_240_000:
raise cv.Invalid(
f"Scan {name} ({value}) must be between 2.5 ms and 10240 ms"
)
# Validate what actually reaches the controller: both values are truncated to
# whole 0.625 ms units, so a window/interval pair that differs by less than one
# unit collapses to the same value — silently programming a 100 % duty cycle
# (radio permanently on) from a config that asked for less.
interval_units = to_ble_units(interval)
window_units = to_ble_units(window)
if window_units == interval_units and window < interval:
raise cv.Invalid(
f"Scan window ({window}) and interval ({interval}) both truncate to "
f"{interval_units} x 0.625 ms, which the controller scans at a 100 % duty "
f"cycle. Separate them by at least 0.625 ms."
)
if interval.total_microseconds * 3 > duration.total_microseconds:
raise cv.Invalid(
f"Scan duration ({duration}) must cover at least three scan intervals "
f"({interval}): the scanner listens on one of the three BLE advertising "
f"channels per interval, so a shorter duration can miss devices entirely."
)
return config
def scan_parameters_schema(
interval_default: str,
*,
window_default: str = "30ms",
supports_active: bool = False,
) -> cv.All:
"""Build the scan_parameters value schema shared by all BLE trackers.
interval_default and window_default are per chip (e.g. esp32 320/30 ms,
bk72xx/rp2 100/30 ms — the reference scan rates of the respective stacks;
LN882H's SDK recommends 100/50 ms). Pass supports_active=True only when
the tracker supports active scanning; it exposes the `active` option
(whose own default is on, esp32_ble_tracker behavior).
"""
schema = {
cv.Optional(CONF_DURATION, default="5min"): cv.positive_time_period_seconds,
cv.Optional(CONF_INTERVAL, default=interval_default): cv.positive_time_period,
cv.Optional(CONF_WINDOW, default=window_default): cv.positive_time_period,
cv.Optional(CONF_CONTINUOUS, default=True): cv.boolean,
}
if supports_active:
schema[cv.Optional(CONF_ACTIVE, default=True)] = cv.boolean
return cv.All(cv.Schema(schema), validate_scan_parameters)
BT_UUID16_FORMAT = "XXXX"
BT_UUID32_FORMAT = "XXXXXXXX"
BT_UUID128_FORMAT = "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX"
esphome/components/esp32_ble_tracker/__init__.py
+10 -39
View File
@@ -125,25 +125,6 @@ ESP32BLEStopScanAction = esp32_ble_tracker_ns.class_(
)
def validate_scan_parameters(config):
duration = config[CONF_DURATION]
interval = config[CONF_INTERVAL]
window = config[CONF_WINDOW]
if window > interval:
raise cv.Invalid(
f"Scan window ({window}) needs to be smaller than scan interval ({interval})"
)
if interval.total_milliseconds * 3 > duration.total_milliseconds:
raise cv.Invalid(
"Scan duration needs to be at least three times the scan interval to"
"cover all BLE channels."
)
return config
def validate_max_connections_deprecated(config: ConfigType) -> ConfigType:
if CONF_MAX_CONNECTIONS in config:
_LOGGER.warning(
@@ -153,6 +134,13 @@ def validate_max_connections_deprecated(config: ConfigType) -> ConfigType:
return config
# 320 ms is the ESP-IDF reference scan interval; the shared schema also
# tightens validation to the controller's 2.5 ms .. 10240 ms range and rejects
# window/interval pairs that collapse to the same 0.625 ms unit count.
SCAN_PARAMETERS_SCHEMA = ble_device_base.scan_parameters_schema(
"320ms", supports_active=True
)
# Codegen helpers are owned by ble_device_base; kept under the historical names
# here for the components that import them from this module.
as_hex = ble_device_base.as_hex
@@ -168,24 +156,7 @@ CONFIG_SCHEMA = cv.All(
cv.Optional(CONF_MAX_CONNECTIONS): cv.All(
cv.positive_int, cv.Range(min=0, max=IDF_MAX_CONNECTIONS)
),
cv.Optional(CONF_SCAN_PARAMETERS, default={}): cv.All(
cv.Schema(
{
cv.Optional(
CONF_DURATION, default="5min"
): cv.positive_time_period_seconds,
cv.Optional(
CONF_INTERVAL, default="320ms"
): cv.positive_time_period_milliseconds,
cv.Optional(
CONF_WINDOW, default="30ms"
): cv.positive_time_period_milliseconds,
cv.Optional(CONF_ACTIVE, default=True): cv.boolean,
cv.Optional(CONF_CONTINUOUS, default=True): cv.boolean,
}
),
validate_scan_parameters,
),
cv.Optional(CONF_SCAN_PARAMETERS, default={}): SCAN_PARAMETERS_SCHEMA,
cv.Optional(CONF_ON_BLE_ADVERTISE): automation.validate_automation(
{
cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(
@@ -255,8 +226,8 @@ async def to_code(config):
params = config[CONF_SCAN_PARAMETERS]
cg.add(var.set_scan_duration(params[CONF_DURATION]))
cg.add(var.set_scan_interval(int(params[CONF_INTERVAL].total_milliseconds / 0.625)))
cg.add(var.set_scan_window(int(params[CONF_WINDOW].total_milliseconds / 0.625)))
cg.add(var.set_scan_interval(ble_device_base.to_ble_units(params[CONF_INTERVAL])))
cg.add(var.set_scan_window(ble_device_base.to_ble_units(params[CONF_WINDOW])))
cg.add(var.set_scan_active(params[CONF_ACTIVE]))
cg.add(var.set_scan_continuous(params[CONF_CONTINUOUS]))
tests/component_tests/ble_device_base/__init__.py
View File
tests/component_tests/bk72xx_ble_tracker/test_scan_parameter_validation.py → tests/component_tests/ble_device_base/test_scan_parameter_validation.py
+36 -13
View File
@@ -1,16 +1,20 @@
"""Tests for bk72xx_ble_tracker scan parameter validation."""
"""Tests for the shared BLE tracker scan parameter validation."""
from __future__ import annotations
import pytest
from esphome import config_validation as cv
from esphome.components.bk72xx_ble_tracker import SCAN_PARAMETERS_SCHEMA, to_ble_units
from esphome.components.bk72xx_ble_tracker import (
SCAN_PARAMETERS_SCHEMA as BK72XX_SCHEMA,
)
from esphome.components.ble_device_base import to_ble_units
from esphome.components.esp32_ble_tracker import SCAN_PARAMETERS_SCHEMA as ESP32_SCHEMA
def _validate(**kwargs: str) -> dict:
"""Run a scan_parameters config through the schema, applying defaults."""
return SCAN_PARAMETERS_SCHEMA(dict(kwargs))
"""Run a scan_parameters config through a passive tracker's real schema."""
return BK72XX_SCHEMA(kwargs)
# --- to_ble_units ---
@@ -36,14 +40,37 @@ def test_to_ble_units_truncates() -> None:
assert to_ble_units(cv.positive_time_period("2500us")) == 4
# --- accepted configurations ---
# --- the real per-chip schemas ---
def test_defaults_are_valid() -> None:
"""The documented default 100 ms / 30 ms pair validates."""
def test_bk72xx_defaults_are_valid() -> None:
"""bk72xx pins the BK reference rate: 100 ms interval, shared 30 ms window."""
config = _validate()
assert to_ble_units(config["interval"]) == 160
assert to_ble_units(config["window"]) == 48
assert "active" not in config
def test_esp32_defaults_are_valid() -> None:
"""esp32 pins the ESP-IDF reference rate and exposes active (default on)."""
config = ESP32_SCHEMA({})
assert to_ble_units(config["interval"]) == 512
assert to_ble_units(config["window"]) == 48
assert config["active"] is True
def test_esp32_active_can_disable() -> None:
config = ESP32_SCHEMA({"active": False})
assert config["active"] is False
def test_passive_schema_rejects_active_key() -> None:
"""Trackers without active scan support must not silently accept the option."""
with pytest.raises(cv.Invalid):
_validate(active="true")
# --- accepted configurations ---
def test_minimum_separation_accepted() -> None:
@@ -100,12 +127,8 @@ def test_out_of_range_rejected(interval: str, window: str, offender: str) -> Non
def test_unit_collapse_rejected() -> None:
"""Regression: 3000us/2500us both floor to 4 units — a hidden 100 % duty cycle.
This is the configuration that previously validated and programmed the radio
permanently on despite asking for roughly 83 %.
"""
with pytest.raises(cv.Invalid, match="both round to 4 x 0.625 ms"):
"""3000us/2500us both floor to 4 units — a hidden 100 % duty cycle."""
with pytest.raises(cv.Invalid, match="both truncate to 4 x 0.625 ms"):
_validate(interval="3000us", window="2500us")