1
0
mirror of https://github.com/esphome/esphome.git synced 2025-11-14 22:05:54 +00:00

Compare commits

..

1 Commits

Author SHA1 Message Date
J. Nick Koston
1f408ce41c [template.alarm_control_panel] Use FixedVector for iteration-only sensor storage 2025-11-13 12:35:43 -06:00
10 changed files with 303 additions and 236 deletions

View File

@@ -52,10 +52,8 @@ static void log_invalid_parameter(const char *name, const LogString *message) {
}
static const LogString *color_mode_to_human(ColorMode color_mode) {
if (color_mode == ColorMode::ON_OFF)
return LOG_STR("On/Off");
if (color_mode == ColorMode::BRIGHTNESS)
return LOG_STR("Brightness");
if (color_mode == ColorMode::UNKNOWN)
return LOG_STR("Unknown");
if (color_mode == ColorMode::WHITE)
return LOG_STR("White");
if (color_mode == ColorMode::COLOR_TEMPERATURE)
@@ -70,7 +68,7 @@ static const LogString *color_mode_to_human(ColorMode color_mode) {
return LOG_STR("RGB + cold/warm white");
if (color_mode == ColorMode::RGB_COLOR_TEMPERATURE)
return LOG_STR("RGB + color temperature");
return LOG_STR("Unknown");
return LOG_STR("");
}
// Helper to log percentage values

View File

@@ -137,7 +137,11 @@ async def to_code(config):
cg.add(var.set_arming_night_time(config[CONF_ARMING_NIGHT_TIME]))
supports_arm_night = True
for sensor in config.get(CONF_BINARY_SENSORS, []):
if sensors := config.get(CONF_BINARY_SENSORS, []):
# Initialize FixedVector with the exact number of sensors
cg.add(var.init_sensors(len(sensors)))
for sensor in sensors:
bs = await cg.get_variable(sensor[CONF_INPUT])
flags = BinarySensorFlags[FLAG_NORMAL]

View File

@@ -20,10 +20,13 @@ void TemplateAlarmControlPanel::add_sensor(binary_sensor::BinarySensor *sensor,
// Save the flags and type. Assign a store index for the per sensor data type.
SensorDataStore sd;
sd.last_chime_state = false;
this->sensor_map_[sensor].flags = flags;
this->sensor_map_[sensor].type = type;
AlarmSensor alarm_sensor;
alarm_sensor.sensor = sensor;
alarm_sensor.info.flags = flags;
alarm_sensor.info.type = type;
alarm_sensor.info.store_index = this->next_store_index_++;
this->sensors_.push_back(alarm_sensor);
this->sensor_data_.push_back(sd);
this->sensor_map_[sensor].store_index = this->next_store_index_++;
};
static const LogString *sensor_type_to_string(AlarmSensorType type) {
@@ -45,7 +48,7 @@ void TemplateAlarmControlPanel::dump_config() {
ESP_LOGCONFIG(TAG,
"TemplateAlarmControlPanel:\n"
" Current State: %s\n"
" Number of Codes: %u\n"
" Number of Codes: %zu\n"
" Requires Code To Arm: %s\n"
" Arming Away Time: %" PRIu32 "s\n"
" Arming Home Time: %" PRIu32 "s\n"
@@ -58,7 +61,8 @@ void TemplateAlarmControlPanel::dump_config() {
(this->arming_home_time_ / 1000), (this->arming_night_time_ / 1000), (this->pending_time_ / 1000),
(this->trigger_time_ / 1000), this->get_supported_features());
#ifdef USE_BINARY_SENSOR
for (auto const &[sensor, info] : this->sensor_map_) {
for (const auto &alarm_sensor : this->sensors_) {
const uint16_t flags = alarm_sensor.info.flags;
ESP_LOGCONFIG(TAG,
" Binary Sensor:\n"
" Name: %s\n"
@@ -67,11 +71,10 @@ void TemplateAlarmControlPanel::dump_config() {
" Armed night bypass: %s\n"
" Auto bypass: %s\n"
" Chime mode: %s",
sensor->get_name().c_str(), LOG_STR_ARG(sensor_type_to_string(info.type)),
TRUEFALSE(info.flags & BINARY_SENSOR_MODE_BYPASS_ARMED_HOME),
TRUEFALSE(info.flags & BINARY_SENSOR_MODE_BYPASS_ARMED_NIGHT),
TRUEFALSE(info.flags & BINARY_SENSOR_MODE_BYPASS_AUTO),
TRUEFALSE(info.flags & BINARY_SENSOR_MODE_CHIME));
alarm_sensor.sensor->get_name().c_str(), LOG_STR_ARG(sensor_type_to_string(alarm_sensor.info.type)),
TRUEFALSE(flags & BINARY_SENSOR_MODE_BYPASS_ARMED_HOME),
TRUEFALSE(flags & BINARY_SENSOR_MODE_BYPASS_ARMED_NIGHT),
TRUEFALSE(flags & BINARY_SENSOR_MODE_BYPASS_AUTO), TRUEFALSE(flags & BINARY_SENSOR_MODE_CHIME));
}
#endif
}
@@ -121,7 +124,9 @@ void TemplateAlarmControlPanel::loop() {
#ifdef USE_BINARY_SENSOR
// Test all of the sensors regardless of the alarm panel state
for (auto const &[sensor, info] : this->sensor_map_) {
for (const auto &alarm_sensor : this->sensors_) {
const auto &info = alarm_sensor.info;
auto *sensor = alarm_sensor.sensor;
// Check for chime zones
if (info.flags & BINARY_SENSOR_MODE_CHIME) {
// Look for the transition from closed to open
@@ -242,11 +247,11 @@ void TemplateAlarmControlPanel::arm_(optional<std::string> code, alarm_control_p
void TemplateAlarmControlPanel::bypass_before_arming() {
#ifdef USE_BINARY_SENSOR
for (auto const &[sensor, info] : this->sensor_map_) {
for (const auto &alarm_sensor : this->sensors_) {
// Check for faulted bypass_auto sensors and remove them from monitoring
if ((info.flags & BINARY_SENSOR_MODE_BYPASS_AUTO) && (sensor->state)) {
ESP_LOGW(TAG, "'%s' is faulted and will be automatically bypassed", sensor->get_name().c_str());
this->bypassed_sensor_indicies_.push_back(info.store_index);
if ((alarm_sensor.info.flags & BINARY_SENSOR_MODE_BYPASS_AUTO) && (alarm_sensor.sensor->state)) {
ESP_LOGW(TAG, "'%s' is faulted and will be automatically bypassed", alarm_sensor.sensor->get_name().c_str());
this->bypassed_sensor_indicies_.push_back(alarm_sensor.info.store_index);
}
}
#endif

View File

@@ -1,11 +1,12 @@
#pragma once
#include <cinttypes>
#include <map>
#include <vector>
#include "esphome/core/automation.h"
#include "esphome/core/component.h"
#include "esphome/core/defines.h"
#include "esphome/core/helpers.h"
#include "esphome/components/alarm_control_panel/alarm_control_panel.h"
@@ -49,6 +50,13 @@ struct SensorInfo {
uint8_t store_index;
};
#ifdef USE_BINARY_SENSOR
struct AlarmSensor {
binary_sensor::BinarySensor *sensor;
SensorInfo info;
};
#endif
class TemplateAlarmControlPanel final : public alarm_control_panel::AlarmControlPanel, public Component {
public:
TemplateAlarmControlPanel();
@@ -63,6 +71,12 @@ class TemplateAlarmControlPanel final : public alarm_control_panel::AlarmControl
void bypass_before_arming();
#ifdef USE_BINARY_SENSOR
/** Initialize the sensors vector with the specified capacity.
*
* @param capacity The number of sensors to allocate space for.
*/
void init_sensors(size_t capacity) { this->sensors_.init(capacity); }
/** Add a binary_sensor to the alarm_panel.
*
* @param sensor The BinarySensor instance.
@@ -122,8 +136,8 @@ class TemplateAlarmControlPanel final : public alarm_control_panel::AlarmControl
protected:
void control(const alarm_control_panel::AlarmControlPanelCall &call) override;
#ifdef USE_BINARY_SENSOR
// This maps a binary sensor to its alarm specific info
std::map<binary_sensor::BinarySensor *, SensorInfo> sensor_map_;
// List of binary sensors with their alarm-specific info
FixedVector<AlarmSensor> sensors_;
// a list of automatically bypassed sensors
std::vector<uint8_t> bypassed_sensor_indicies_;
#endif

View File

@@ -1,17 +1,10 @@
import logging
import esphome.codegen as cg
from esphome.components.esp32 import add_idf_component
from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
from esphome.config_helpers import merge_config
import esphome.config_validation as cv
from esphome.const import CONF_ID, CONF_OTA, CONF_PLATFORM, CONF_WEB_SERVER
from esphome.const import CONF_ID
from esphome.core import CORE, coroutine_with_priority
from esphome.coroutine import CoroPriority
import esphome.final_validate as fv
from esphome.types import ConfigType
_LOGGER = logging.getLogger(__name__)
CODEOWNERS = ["@esphome/core"]
DEPENDENCIES = ["network", "web_server_base"]
@@ -19,53 +12,6 @@ DEPENDENCIES = ["network", "web_server_base"]
web_server_ns = cg.esphome_ns.namespace("web_server")
WebServerOTAComponent = web_server_ns.class_("WebServerOTAComponent", OTAComponent)
def _web_server_ota_final_validate(config: ConfigType) -> None:
"""Merge multiple web_server OTA instances into one.
Multiple web_server OTA instances register duplicate HTTP handlers for /update,
causing undefined behavior. Merge them into a single instance.
"""
full_conf = fv.full_config.get()
ota_confs = full_conf.get(CONF_OTA, [])
web_server_ota_configs: list[ConfigType] = []
other_ota_configs: list[ConfigType] = []
for ota_conf in ota_confs:
if ota_conf.get(CONF_PLATFORM) == CONF_WEB_SERVER:
web_server_ota_configs.append(ota_conf)
else:
other_ota_configs.append(ota_conf)
if len(web_server_ota_configs) <= 1:
return
# Merge all web_server OTA configs into the first one
merged = web_server_ota_configs[0]
for ota_conf in web_server_ota_configs[1:]:
# Validate that IDs are consistent if manually specified
if (
merged[CONF_ID].is_manual
and ota_conf[CONF_ID].is_manual
and merged[CONF_ID] != ota_conf[CONF_ID]
):
raise cv.Invalid(
f"Found multiple web_server OTA configurations but {CONF_ID} is inconsistent"
)
merged = merge_config(merged, ota_conf)
_LOGGER.warning(
"Found and merged %d web_server OTA configurations into one instance",
len(web_server_ota_configs),
)
# Replace OTA configs with merged web_server + other OTA platforms
other_ota_configs.append(merged)
full_conf[CONF_OTA] = other_ota_configs
fv.full_config.set(full_conf)
CONFIG_SCHEMA = (
cv.Schema(
{
@@ -76,8 +22,6 @@ CONFIG_SCHEMA = (
.extend(cv.COMPONENT_SCHEMA)
)
FINAL_VALIDATE_SCHEMA = _web_server_ota_final_validate
@coroutine_with_priority(CoroPriority.WEB_SERVER_OTA)
async def to_code(config):

View File

@@ -12,6 +12,7 @@ from esphome.components.network import (
from esphome.components.psram import is_guaranteed as psram_is_guaranteed
from esphome.config_helpers import filter_source_files_from_platform
import esphome.config_validation as cv
from esphome.config_validation import only_with_esp_idf
from esphome.const import (
CONF_AP,
CONF_BSSID,
@@ -351,7 +352,7 @@ CONFIG_SCHEMA = cv.All(
single=True
),
cv.Optional(CONF_USE_PSRAM): cv.All(
cv.only_on_esp32, cv.requires_component("psram"), cv.boolean
only_with_esp_idf, cv.requires_component("psram"), cv.boolean
),
}
),

View File

@@ -1,4 +1,4 @@
pylint==4.0.3
pylint==4.0.2
flake8==7.3.0 # also change in .pre-commit-config.yaml when updating
ruff==0.14.4 # also change in .pre-commit-config.yaml when updating
pyupgrade==3.21.1 # also change in .pre-commit-config.yaml when updating

View File

@@ -1,18 +1,6 @@
"""Tests for the web_server OTA platform."""
from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any
import pytest
from esphome import config_validation as cv
from esphome.components.web_server.ota import _web_server_ota_final_validate
from esphome.const import CONF_ID, CONF_OTA, CONF_PLATFORM, CONF_WEB_SERVER
from esphome.core import ID
import esphome.final_validate as fv
def test_web_server_ota_generated(generate_main: Callable[[str], str]) -> None:
@@ -112,144 +100,3 @@ def test_web_server_ota_esp8266(generate_main: Callable[[str], str]) -> None:
# Check web server OTA component is present
assert "WebServerOTAComponent" in main_cpp
assert "web_server::WebServerOTAComponent" in main_cpp
@pytest.mark.parametrize(
("ota_configs", "expected_count", "warning_expected"),
[
pytest.param(
[
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web", is_manual=False),
}
],
1,
False,
id="single_instance_no_merge",
),
pytest.param(
[
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web_1", is_manual=False),
},
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web_2", is_manual=False),
},
],
1,
True,
id="two_instances_merged",
),
pytest.param(
[
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web_1", is_manual=False),
},
{
CONF_PLATFORM: "esphome",
CONF_ID: ID("ota_esphome", is_manual=False),
},
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web_2", is_manual=False),
},
],
2,
True,
id="mixed_platforms_web_server_merged",
),
],
)
def test_web_server_ota_instance_merging(
ota_configs: list[dict[str, Any]],
expected_count: int,
warning_expected: bool,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test web_server OTA instance merging behavior."""
full_conf = {CONF_OTA: ota_configs.copy()}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
_web_server_ota_final_validate({})
updated_conf = fv.full_config.get()
# Verify total number of OTA platforms
assert len(updated_conf[CONF_OTA]) == expected_count
# Verify warning
if warning_expected:
assert any(
"Found and merged" in record.message
and "web_server OTA" in record.message
for record in caplog.records
), "Expected merge warning not found in log"
else:
assert len(caplog.records) == 0, "Unexpected warnings logged"
finally:
fv.full_config.reset(token)
def test_web_server_ota_consistent_manual_ids(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that consistent manual IDs can be merged successfully."""
ota_configs = [
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web", is_manual=True),
},
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web", is_manual=True),
},
]
full_conf = {CONF_OTA: ota_configs}
token = fv.full_config.set(full_conf)
try:
with caplog.at_level(logging.WARNING):
_web_server_ota_final_validate({})
updated_conf = fv.full_config.get()
assert len(updated_conf[CONF_OTA]) == 1
assert updated_conf[CONF_OTA][0][CONF_ID].id == "ota_web"
assert any(
"Found and merged" in record.message and "web_server OTA" in record.message
for record in caplog.records
)
finally:
fv.full_config.reset(token)
def test_web_server_ota_inconsistent_manual_ids() -> None:
"""Test that inconsistent manual IDs raise an error."""
ota_configs = [
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web_1", is_manual=True),
},
{
CONF_PLATFORM: CONF_WEB_SERVER,
CONF_ID: ID("ota_web_2", is_manual=True),
},
]
full_conf = {CONF_OTA: ota_configs}
token = fv.full_config.set(full_conf)
try:
with pytest.raises(
cv.Invalid,
match="Found multiple web_server OTA configurations but id is inconsistent",
):
_web_server_ota_final_validate({})
finally:
fv.full_config.reset(token)

View File

@@ -0,0 +1,136 @@
esphome:
name: template-alarm-many-sensors
friendly_name: "Template Alarm Control Panel with Many Sensors"
logger:
host:
api:
binary_sensor:
- platform: template
id: sensor1
name: "Door 1"
- platform: template
id: sensor2
name: "Door 2"
- platform: template
id: sensor3
name: "Window 1"
- platform: template
id: sensor4
name: "Window 2"
- platform: template
id: sensor5
name: "Motion 1"
- platform: template
id: sensor6
name: "Motion 2"
- platform: template
id: sensor7
name: "Glass Break 1"
- platform: template
id: sensor8
name: "Glass Break 2"
- platform: template
id: sensor9
name: "Smoke Detector"
- platform: template
id: sensor10
name: "CO Detector"
alarm_control_panel:
- platform: template
id: test_alarm
name: "Test Alarm"
codes:
- "1234"
requires_code_to_arm: true
arming_away_time: 5s
arming_home_time: 3s
arming_night_time: 3s
pending_time: 10s
trigger_time: 300s
restore_mode: ALWAYS_DISARMED
binary_sensors:
- input: sensor1
bypass_armed_home: false
bypass_armed_night: false
bypass_auto: true
chime: true
trigger_mode: DELAYED
- input: sensor2
bypass_armed_home: false
bypass_armed_night: false
bypass_auto: true
chime: true
trigger_mode: DELAYED
- input: sensor3
bypass_armed_home: true
bypass_armed_night: false
bypass_auto: false
chime: false
trigger_mode: DELAYED
- input: sensor4
bypass_armed_home: true
bypass_armed_night: false
bypass_auto: false
chime: false
trigger_mode: DELAYED
- input: sensor5
bypass_armed_home: false
bypass_armed_night: true
bypass_auto: false
chime: false
trigger_mode: INSTANT
- input: sensor6
bypass_armed_home: false
bypass_armed_night: true
bypass_auto: false
chime: false
trigger_mode: INSTANT
- input: sensor7
bypass_armed_home: false
bypass_armed_night: false
bypass_auto: false
chime: false
trigger_mode: INSTANT
- input: sensor8
bypass_armed_home: false
bypass_armed_night: false
bypass_auto: false
chime: false
trigger_mode: INSTANT
- input: sensor9
bypass_armed_home: false
bypass_armed_night: false
bypass_auto: false
chime: false
trigger_mode: INSTANT_ALWAYS
- input: sensor10
bypass_armed_home: false
bypass_armed_night: false
bypass_auto: false
chime: false
trigger_mode: INSTANT_ALWAYS
on_disarmed:
- logger.log: "Alarm disarmed"
on_arming:
- logger.log: "Alarm arming"
on_armed_away:
- logger.log: "Alarm armed away"
on_armed_home:
- logger.log: "Alarm armed home"
on_armed_night:
- logger.log: "Alarm armed night"
on_pending:
- logger.log: "Alarm pending"
on_triggered:
- logger.log: "Alarm triggered"
on_cleared:
- logger.log: "Alarm cleared"
on_chime:
- logger.log: "Chime activated"
on_ready:
- logger.log: "Sensors ready state changed"

View File

@@ -0,0 +1,118 @@
"""Integration test for template alarm control panel with many sensors."""
from __future__ import annotations
import aioesphomeapi
from aioesphomeapi.model import APIIntEnum
import pytest
from .state_utils import InitialStateHelper
from .types import APIClientConnectedFactory, RunCompiledFunction
class EspHomeACPFeatures(APIIntEnum):
"""ESPHome AlarmControlPanel feature numbers."""
ARM_HOME = 1
ARM_AWAY = 2
ARM_NIGHT = 4
TRIGGER = 8
ARM_CUSTOM_BYPASS = 16
ARM_VACATION = 32
@pytest.mark.asyncio
async def test_template_alarm_control_panel_many_sensors(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""Test template alarm control panel with 10 binary sensors using FixedVector."""
async with run_compiled(yaml_config), api_client_connected() as client:
# Get entity info first
entities, _ = await client.list_entities_services()
# Find the alarm control panel and binary sensors
alarm_info: aioesphomeapi.AlarmControlPanelInfo | None = None
binary_sensors: list[aioesphomeapi.BinarySensorInfo] = []
for entity in entities:
if isinstance(entity, aioesphomeapi.AlarmControlPanelInfo):
alarm_info = entity
elif isinstance(entity, aioesphomeapi.BinarySensorInfo):
binary_sensors.append(entity)
assert alarm_info is not None, "Alarm control panel entity info not found"
assert alarm_info.name == "Test Alarm"
assert alarm_info.requires_code is True
assert alarm_info.requires_code_to_arm is True
# Verify we have 10 binary sensors
assert len(binary_sensors) == 10, (
f"Expected 10 binary sensors, got {len(binary_sensors)}"
)
# Verify sensor names
expected_sensor_names = {
"Door 1",
"Door 2",
"Window 1",
"Window 2",
"Motion 1",
"Motion 2",
"Glass Break 1",
"Glass Break 2",
"Smoke Detector",
"CO Detector",
}
actual_sensor_names = {sensor.name for sensor in binary_sensors}
assert actual_sensor_names == expected_sensor_names, (
f"Sensor names mismatch. Expected: {expected_sensor_names}, "
f"Got: {actual_sensor_names}"
)
# Use InitialStateHelper to wait for all initial states
state_helper = InitialStateHelper(entities)
def on_state(state: aioesphomeapi.EntityState) -> None:
# We'll receive subsequent states here after initial states
pass
client.subscribe_states(state_helper.on_state_wrapper(on_state))
# Wait for all initial states
await state_helper.wait_for_initial_states(timeout=5.0)
# Verify the alarm state is disarmed initially
alarm_state = state_helper.initial_states.get(alarm_info.key)
assert alarm_state is not None, "Alarm control panel initial state not received"
assert isinstance(alarm_state, aioesphomeapi.AlarmControlPanelEntityState)
assert alarm_state.state == aioesphomeapi.AlarmControlPanelState.DISARMED, (
f"Expected initial state DISARMED, got {alarm_state.state}"
)
# Verify all 10 binary sensors have initial states
binary_sensor_states = [
state_helper.initial_states.get(sensor.key) for sensor in binary_sensors
]
assert all(state is not None for state in binary_sensor_states), (
"Not all binary sensors have initial states"
)
# Verify all binary sensor states are BinarySensorState type
for i, state in enumerate(binary_sensor_states):
assert isinstance(state, aioesphomeapi.BinarySensorState), (
f"Binary sensor {i} state is not BinarySensorState: {type(state)}"
)
# Verify supported features
expected_features = (
EspHomeACPFeatures.ARM_HOME
| EspHomeACPFeatures.ARM_AWAY
| EspHomeACPFeatures.ARM_NIGHT
| EspHomeACPFeatures.TRIGGER
)
assert alarm_info.supported_features == expected_features, (
f"Expected supported_features={expected_features} (ARM_HOME|ARM_AWAY|ARM_NIGHT|TRIGGER), "
f"got {alarm_info.supported_features}"
)