Use template forwarder structs for callback deduplication

Replace per-site lambda generation with TriggerForwarder<Ts...>,
TriggerOnTrueForwarder, and TriggerOnFalseForwarder structs. The
compiler generates one operator() per forwarder type shared across
all call sites, avoiding flash duplication from unique lambdas.

Also cleans up API: replaces condition/callback_args with
bool_filter using TRIGGER_ON_TRUE/TRIGGER_ON_FALSE constants.
This commit is contained in:
J. Nick Koston
2026-03-25 14:16:30 -10:00
parent 6c905bd036
commit c127cacb9a
5 changed files with 109 additions and 99 deletions
+29 -21
View File
@@ -137,6 +137,9 @@ UpdateComponentAction = cg.esphome_ns.class_("UpdateComponentAction", Action)
SuspendComponentAction = cg.esphome_ns.class_("SuspendComponentAction", Action)
ResumeComponentAction = cg.esphome_ns.class_("ResumeComponentAction", Action)
Automation = cg.esphome_ns.class_("Automation")
TriggerForwarder = cg.esphome_ns.class_("TriggerForwarder")
TriggerOnTrueForwarder = cg.esphome_ns.class_("TriggerOnTrueForwarder")
TriggerOnFalseForwarder = cg.esphome_ns.class_("TriggerOnFalseForwarder")
LambdaCondition = cg.esphome_ns.class_("LambdaCondition", Condition)
StatelessLambdaCondition = cg.esphome_ns.class_("StatelessLambdaCondition", Condition)
@@ -663,46 +666,51 @@ async def build_automation(
return obj
TRIGGER_ON_TRUE = "on_true"
TRIGGER_ON_FALSE = "on_false"
async def build_callback_automation(
parent: MockObj,
callback_method: str,
args: TemplateArgsType,
config: ConfigType,
callback_args: TemplateArgsType | None = None,
condition: str | None = None,
bool_filter: str | None = None,
) -> None:
"""Build an Automation and register it as a callback on the parent.
Eliminates the need for a Trigger wrapper object by registering the
automation's trigger() directly as a callback on the parent component.
Uses template forwarder structs (TriggerForwarder, TriggerOnTrueForwarder,
TriggerOnFalseForwarder) so the compiler deduplicates the operator() body
across all call sites with the same signature.
:param parent: The component object (e.g., button, sensor).
:param callback_method: Name of the callback method (e.g., "add_on_press_callback").
:param args: Automation template args as list of (type, name) tuples.
:param config: The automation config dict.
:param callback_args: Lambda parameter types if different from args (e.g., for
conditional triggers where the callback receives (bool state) but the
automation is Automation<> with no args). Defaults to args.
:param condition: Optional C++ condition. Use callback arg names directly
(e.g., "state", "!state").
:param bool_filter: Optional bool filter. Use TRIGGER_ON_TRUE to trigger only
when the bool callback arg is true, TRIGGER_ON_FALSE for false.
The automation will be Automation<> (no args) while the callback receives bool.
"""
arg_types = [arg[0] for arg in args]
templ = cg.TemplateArguments(*arg_types)
obj = cg.new_Pvariable(config[CONF_AUTOMATION_ID], templ)
actions = await build_action_list(config[CONF_THEN], templ, args)
cg.add(obj.add_actions(actions))
# Build trigger call expression: automation->trigger(arg1, arg2, ...)
trigger_args = [MockObj(arg[1], "") for arg in args]
trigger_expr = obj.trigger(*trigger_args)
if condition is not None:
body = [f"if ({condition}) {{ ", trigger_expr, "; }"]
# Use template forwarder structs for deduplication. The compiler generates
# one operator() per forwarder type; different automation pointers are just
# data in the struct.
if bool_filter == TRIGGER_ON_TRUE:
forwarder = cg.RawExpression(f"{TriggerOnTrueForwarder}{{{obj}}}")
elif bool_filter == TRIGGER_ON_FALSE:
forwarder = cg.RawExpression(f"{TriggerOnFalseForwarder}{{{obj}}}")
else:
body = [trigger_expr, ";"]
# Use callback_args for the lambda parameters if provided (e.g., when the
# callback signature differs from the automation args due to filtering).
lambda_params = callback_args if callback_args is not None else args
# ESPHome codegen allocates all variables as static pointers, so they
# are accessible without explicit lambda capture. Using "" avoids
# -Wcapture-of-non-automatic-storage-duration warnings.
lambda_expr = LambdaExpression(body, lambda_params, capture="")
cg.add(getattr(parent, callback_method)(lambda_expr))
forwarder_type = (
TriggerForwarder.template(templ)
if arg_types
else TriggerForwarder.template()
)
forwarder = cg.RawExpression(f"{forwarder_type}{{{obj}}}")
cg.add(getattr(parent, callback_method)(forwarder))
+2 -4
View File
@@ -562,8 +562,7 @@ async def _build_binary_sensor_automations(var, config):
"add_on_state_callback",
[],
conf,
callback_args=[(bool, "state")],
condition="state",
bool_filter=automation.TRIGGER_ON_TRUE,
)
for conf in config.get(CONF_ON_RELEASE, []):
@@ -572,8 +571,7 @@ async def _build_binary_sensor_automations(var, config):
"add_on_state_callback",
[],
conf,
callback_args=[(bool, "state")],
condition="!state",
bool_filter=automation.TRIGGER_ON_FALSE,
)
for conf in config.get(CONF_ON_CLICK, []):
+2 -4
View File
@@ -157,8 +157,7 @@ async def _build_switch_automations(var, config):
"add_on_state_callback",
[],
conf,
callback_args=[(bool, "state")],
condition="state",
bool_filter=automation.TRIGGER_ON_TRUE,
)
for conf in config.get(CONF_ON_TURN_OFF, []):
await automation.build_callback_automation(
@@ -166,8 +165,7 @@ async def _build_switch_automations(var, config):
"add_on_state_callback",
[],
conf,
callback_args=[(bool, "state")],
condition="!state",
bool_filter=automation.TRIGGER_ON_FALSE,
)
+25
View File
@@ -492,4 +492,29 @@ template<typename... Ts> class Automation {
ActionList<Ts...> actions_;
};
/// Callback forwarder that triggers an Automation directly.
/// One operator() instantiation per Automation<Ts...> signature, shared across all call sites.
template<typename... Ts> struct TriggerForwarder {
Automation<Ts...> *automation;
void operator()(Ts... args) const { this->automation->trigger(args...); }
};
/// Callback forwarder that triggers an Automation<> only when the bool arg is true.
struct TriggerOnTrueForwarder {
Automation<> *automation;
void operator()(bool state) const {
if (state)
this->automation->trigger();
}
};
/// Callback forwarder that triggers an Automation<> only when the bool arg is false.
struct TriggerOnFalseForwarder {
Automation<> *automation;
void operator()(bool state) const {
if (!state)
this->automation->trigger();
}
};
} // namespace esphome
+51 -70
View File
@@ -5,8 +5,15 @@ from unittest.mock import patch
import pytest
from esphome.automation import has_non_synchronous_actions
from esphome.cpp_generator import LambdaExpression, MockObj, RawExpression
from esphome.automation import (
TRIGGER_ON_FALSE,
TRIGGER_ON_TRUE,
TriggerForwarder,
TriggerOnFalseForwarder,
TriggerOnTrueForwarder,
has_non_synchronous_actions,
)
from esphome.cpp_generator import MockObj, RawExpression
from esphome.util import RegistryEntry
@@ -178,94 +185,68 @@ def test_has_non_synchronous_actions_dict_input(
assert has_non_synchronous_actions({"logger.log": "hello"}) is False
def _build_trigger_lambda(
def _build_forwarder(
automation_name: str,
args: list[tuple[str, str]],
callback_args: list[tuple[str, str]] | None = None,
condition: str | None = None,
bool_filter: str | None = None,
) -> str:
"""Build a trigger callback lambda the same way build_callback_automation does.
"""Build a trigger forwarder expression the same way build_callback_automation does.
Mirrors the logic in automation.build_callback_automation lines 694-708.
Mirrors the forwarder selection logic in automation.build_callback_automation.
"""
import esphome.codegen as cg
obj = MockObj(automation_name, "->")
# Convert string type names to RawExpression (matching real codegen where
# types are MockObj/MockObjClass objects, not plain strings)
typed_args = [(RawExpression(t), n) for t, n in args]
trigger_args = [MockObj(arg[1], "") for arg in args]
trigger_expr = obj.trigger(*trigger_args)
if condition is not None:
body = [f"if ({condition}) {{ ", trigger_expr, "; }"]
else:
body = [trigger_expr, ";"]
lambda_params = (
[(RawExpression(t), n) for t, n in callback_args]
if callback_args is not None
else typed_args
)
lambda_expr = LambdaExpression(body, lambda_params, capture="")
return str(lambda_expr)
if bool_filter == TRIGGER_ON_TRUE:
return f"{TriggerOnTrueForwarder}{{{obj}}}"
if bool_filter == TRIGGER_ON_FALSE:
return f"{TriggerOnFalseForwarder}{{{obj}}}"
arg_types = [RawExpression(t) for t, _ in args]
templ = cg.TemplateArguments(*arg_types) if arg_types else cg.TemplateArguments()
forwarder_type = TriggerForwarder.template(templ)
return f"{forwarder_type}{{{obj}}}"
def test_trigger_callback_lambda_no_args() -> None:
"""Button on_press: no args, no condition."""
result = _build_trigger_lambda("auto_1", [])
assert result == "[]() {\n auto_1->trigger();\n}"
def test_trigger_forwarder_no_args() -> None:
"""Button on_press: TriggerForwarder<> with no args."""
result = _build_forwarder("auto_1", [])
assert result == "TriggerForwarder<>{auto_1}"
def test_trigger_callback_lambda_single_float_arg() -> None:
"""Sensor on_value: single float arg."""
result = _build_trigger_lambda("auto_1", [("float", "x")])
assert result == "[](float x) {\n auto_1->trigger(x);\n}"
def test_trigger_forwarder_single_float_arg() -> None:
"""Sensor on_value: TriggerForwarder<float>."""
result = _build_forwarder("auto_1", [("float", "x")])
assert result == "TriggerForwarder<float>{auto_1}"
def test_trigger_callback_lambda_single_bool_arg() -> None:
"""Switch on_state / binary_sensor on_state: single bool arg."""
result = _build_trigger_lambda("auto_1", [("bool", "x")])
assert result == "[](bool x) {\n auto_1->trigger(x);\n}"
def test_trigger_forwarder_single_bool_arg() -> None:
"""Switch on_state: TriggerForwarder<bool>."""
result = _build_forwarder("auto_1", [("bool", "x")])
assert result == "TriggerForwarder<bool>{auto_1}"
def test_trigger_callback_lambda_condition_true() -> None:
"""Binary_sensor on_press: condition filters on state=true."""
result = _build_trigger_lambda(
"auto_1",
[],
callback_args=[("bool", "state")],
condition="state",
)
assert result == ("[](bool state) {\n if (state) { auto_1->trigger(); }\n}")
def test_trigger_forwarder_on_true() -> None:
"""Binary_sensor on_press / switch on_turn_on: TriggerOnTrueForwarder."""
result = _build_forwarder("auto_1", [], bool_filter=TRIGGER_ON_TRUE)
assert result == "TriggerOnTrueForwarder{auto_1}"
def test_trigger_callback_lambda_condition_false() -> None:
"""Binary_sensor on_release: condition filters on state=false."""
result = _build_trigger_lambda(
"auto_1",
[],
callback_args=[("bool", "state")],
condition="!state",
)
assert result == ("[](bool state) {\n if (!state) { auto_1->trigger(); }\n}")
def test_trigger_forwarder_on_false() -> None:
"""Binary_sensor on_release / switch on_turn_off: TriggerOnFalseForwarder."""
result = _build_forwarder("auto_1", [], bool_filter=TRIGGER_ON_FALSE)
assert result == "TriggerOnFalseForwarder{auto_1}"
def test_trigger_callback_lambda_multiple_args() -> None:
"""Binary_sensor on_state_change: two optional<bool> args."""
result = _build_trigger_lambda(
def test_trigger_forwarder_multiple_args() -> None:
"""Binary_sensor on_state_change: TriggerForwarder with two args."""
result = _build_forwarder(
"auto_1",
[("optional<bool>", "x_previous"), ("optional<bool>", "x")],
)
assert result == (
"[](optional<bool> x_previous, optional<bool> x) {\n"
" auto_1->trigger(x_previous, x);\n}"
)
assert result == "TriggerForwarder<optional<bool>, optional<bool>>{auto_1}"
def test_trigger_callback_lambda_string_arg() -> None:
"""Text_sensor on_value: std::string arg."""
result = _build_trigger_lambda("auto_1", [("std::string", "x")])
assert result == "[](std::string x) {\n auto_1->trigger(x);\n}"
def test_trigger_callback_lambda_empty_capture() -> None:
"""All generated lambdas use empty capture to avoid static storage warnings."""
result = _build_trigger_lambda("auto_1", [("float", "x")])
assert result.startswith("[](")
def test_trigger_forwarder_string_arg() -> None:
"""Text_sensor on_value: TriggerForwarder<std::string>."""
result = _build_forwarder("auto_1", [("std::string", "x")])
assert result == "TriggerForwarder<std::string>{auto_1}"