mrambossekandClaude Opus 4.8
878f7c9c27
Control one or two xTool SafetyPro AP2 air purifiers over BLE and expose them to Home Assistant via an ESP32 (the ap2_hub external component): per-slot fan control with live gear sync, runtime-settable flash-persisted MACs, an on-device scanner, M9033 status polling (filter life, serial, firmware), buzzer control, and BLE bonding. Includes the PC debug tool (tools/ap2_ble.py) and the reverse-engineered protocol spec (SPEC.md). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
import esphome.codegen as cg
|
|
import esphome.config_validation as cv
|
|
from esphome.components import esp32_ble_client, esp32_ble_tracker
|
|
from esphome.const import CONF_ID
|
|
|
|
CODEOWNERS = ["@netplanet"]
|
|
DEPENDENCIES = ["esp32_ble_tracker"]
|
|
# esp32_ble_client gives us BLEClientBase; the hub publishes connection state to
|
|
# binary_sensor/text_sensor. The fan/text platforms load when the user adds them.
|
|
AUTO_LOAD = ["esp32_ble_client", "binary_sensor", "text_sensor", "sensor"]
|
|
|
|
# Number of AP2 slots one ESP32 manages (control connections + 1 transient
|
|
# probe in Phase 2 must stay within esp32_ble_tracker's max_connections).
|
|
NUM_SLOTS = 2
|
|
|
|
ap2_hub_ns = cg.esphome_ns.namespace("ap2_hub")
|
|
AP2Hub = ap2_hub_ns.class_(
|
|
"AP2Hub", cg.Component, esp32_ble_tracker.ESPBTDeviceListener
|
|
)
|
|
AP2Slot = ap2_hub_ns.class_("AP2Slot", esp32_ble_client.BLEClientBase)
|
|
AP2ProbeClient = ap2_hub_ns.class_("AP2ProbeClient", esp32_ble_client.BLEClientBase)
|
|
|
|
# Hidden auto-generated IDs for the slot + prober sub-objects. Declaring them here
|
|
# (rather than fabricating IDs in to_code) is what registers them in
|
|
# CORE.component_ids, so cg.register_component accepts them.
|
|
SLOT_ID_KEYS = [f"slot_{i}_id" for i in range(NUM_SLOTS)]
|
|
PROBE_ID_KEY = "probe_id"
|
|
|
|
CONFIG_SCHEMA = (
|
|
cv.Schema(
|
|
{
|
|
cv.GenerateID(): cv.declare_id(AP2Hub),
|
|
cv.GenerateID(PROBE_ID_KEY): cv.declare_id(AP2ProbeClient),
|
|
**{cv.GenerateID(k): cv.declare_id(AP2Slot) for k in SLOT_ID_KEYS},
|
|
}
|
|
)
|
|
.extend(esp32_ble_tracker.ESP_BLE_DEVICE_SCHEMA)
|
|
.extend(cv.COMPONENT_SCHEMA)
|
|
)
|
|
|
|
|
|
async def to_code(config):
|
|
hub = cg.new_Pvariable(config[CONF_ID])
|
|
await cg.register_component(hub, config)
|
|
# The hub listens to advertisements to collect scan candidates.
|
|
await esp32_ble_tracker.register_ble_device(hub, config)
|
|
|
|
for i, key in enumerate(SLOT_ID_KEYS):
|
|
slot = cg.new_Pvariable(config[key], i)
|
|
await cg.register_component(slot, config)
|
|
# Register each slot as a BLE client of the esp32_ble tracker (uses esp32_ble_id).
|
|
await esp32_ble_tracker.register_client(slot, config)
|
|
cg.add(hub.add_slot(slot))
|
|
|
|
# Transient scan prober — a third BLE client, only ever connects while the
|
|
# slots are suspended for a scan.
|
|
probe = cg.new_Pvariable(config[PROBE_ID_KEY])
|
|
await cg.register_component(probe, config)
|
|
await esp32_ble_tracker.register_client(probe, config)
|
|
cg.add(probe.set_hub(hub))
|
|
cg.add(hub.set_prober(probe))
|