diff --git a/esphome/components/bk72xx_ble_tracker/__init__.py b/esphome/components/bk72xx_ble_tracker/__init__.py index 39c9ada731..e53e8e13d7 100644 --- a/esphome/components/bk72xx_ble_tracker/__init__.py +++ b/esphome/components/bk72xx_ble_tracker/__init__.py @@ -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])) diff --git a/esphome/components/ble_device_base/__init__.py b/esphome/components/ble_device_base/__init__.py index d9789b0e9f..8100b41d99 100644 --- a/esphome/components/ble_device_base/__init__.py +++ b/esphome/components/ble_device_base/__init__.py @@ -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" diff --git a/esphome/components/esp32_ble_tracker/__init__.py b/esphome/components/esp32_ble_tracker/__init__.py index e462da1a49..7ffde76429 100644 --- a/esphome/components/esp32_ble_tracker/__init__.py +++ b/esphome/components/esp32_ble_tracker/__init__.py @@ -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])) diff --git a/tests/component_tests/ble_device_base/__init__.py b/tests/component_tests/ble_device_base/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/component_tests/bk72xx_ble_tracker/test_scan_parameter_validation.py b/tests/component_tests/ble_device_base/test_scan_parameter_validation.py similarity index 70% rename from tests/component_tests/bk72xx_ble_tracker/test_scan_parameter_validation.py rename to tests/component_tests/ble_device_base/test_scan_parameter_validation.py index 968a24dad5..363c129f4b 100644 --- a/tests/component_tests/bk72xx_ble_tracker/test_scan_parameter_validation.py +++ b/tests/component_tests/ble_device_base/test_scan_parameter_validation.py @@ -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")