Merge remote-tracking branch 'upstream/dev' into decouple_scheduler_loop_cadence

# Conflicts:
#	esphome/core/scheduler.cpp
This commit is contained in:
J. Nick Koston
2026-04-17 09:33:59 -05:00
21 changed files with 391 additions and 28 deletions
@@ -0,0 +1,185 @@
"""Tests for the _final_validate buffer size calculation in mipi_spi."""
from __future__ import annotations
from typing import Any
import pytest
from esphome.components.display import CONF_SHOW_TEST_CARD
from esphome.components.esp32 import KEY_BOARD, KEY_VARIANT, VARIANT_ESP32
from esphome.components.mipi_spi.display import CONFIG_SCHEMA, FINAL_VALIDATE_SCHEMA
from esphome.const import CONF_BUFFER_SIZE, PlatformFramework
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
def _validated(config: ConfigType) -> ConfigType:
"""Run the component config schema followed by the final validation."""
config = CONFIG_SCHEMA(config)
FINAL_VALIDATE_SCHEMA(config)
return config
def _custom_config(
width: int,
height: int,
color_depth: str | int | None = None,
**extra: Any,
) -> ConfigType:
"""Build a minimal valid custom-model config with the given dimensions."""
config: ConfigType = {
"model": "custom",
"dc_pin": 18,
"dimensions": {"width": width, "height": height},
"init_sequence": [[0xA0, 0x01]],
}
if color_depth is not None:
config["color_depth"] = color_depth
config.update(extra)
return config
# The auto buffer-size selection inside _final_validate targets ~20 kB of
# pixel buffer. For a buffer of ``depth_bytes * width * height``, it picks the
# smallest integer ``x`` in range(2, 8) such that
# ``min(20000, buffer // 4) / buffer >= 1 / x`` (falling back to ``x = 8``).
# The test cases below cover the full range of possible outcomes (1/4 .. 1/8).
@pytest.mark.parametrize(
("width", "height", "color_depth", "expected"),
[
# 16-bit color depth -- buffer = 2 * width * height
# 128*160*2 = 40960 B -> fraction = 10240/40960 = 0.25 -> x = 4
pytest.param(128, 160, "16bit", 1.0 / 4, id="16bit_tiny"),
# 200*224*2 = 89600 B -> fraction = 20000/89600 ≈ 0.2232 -> x = 5
pytest.param(200, 224, "16bit", 1.0 / 5, id="16bit_small"),
# 240*224*2 = 107520 B -> fraction ≈ 0.1860 -> x = 6
pytest.param(240, 224, "16bit", 1.0 / 6, id="16bit_medium"),
# 200*320*2 = 128000 B -> fraction = 0.15625 -> x = 7
pytest.param(200, 320, "16bit", 1.0 / 7, id="16bit_large"),
# 240*320*2 = 153600 B -> fraction ≈ 0.1302 -> default x = 8
pytest.param(240, 320, "16bit", 1.0 / 8, id="16bit_xlarge"),
# 320*480*2 = 307200 B -> fraction ≈ 0.0651 -> default x = 8
pytest.param(320, 480, "16bit", 1.0 / 8, id="16bit_huge"),
# 8-bit color depth -- buffer = width * height
# 320*240 = 76800 B -> fraction = 19200/76800 = 0.25 -> x = 4
pytest.param(320, 240, "8bit", 1.0 / 4, id="8bit_tiny"),
# 400*224 = 89600 B -> fraction ≈ 0.2232 -> x = 5
pytest.param(400, 224, "8bit", 1.0 / 5, id="8bit_small"),
# 480*224 = 107520 B -> fraction ≈ 0.1860 -> x = 6
pytest.param(480, 224, "8bit", 1.0 / 6, id="8bit_medium"),
# 400*320 = 128000 B -> fraction = 0.15625 -> x = 7
pytest.param(400, 320, "8bit", 1.0 / 7, id="8bit_large"),
# 480*320 = 153600 B -> fraction ≈ 0.1302 -> default x = 8
pytest.param(480, 320, "8bit", 1.0 / 8, id="8bit_xlarge"),
],
)
def test_buffer_size_auto_selected(
width: int,
height: int,
color_depth: str,
expected: float,
set_core_config: SetCoreConfigCallable,
) -> None:
"""Without PSRAM or an explicit buffer_size, a fraction is chosen from the display size.
Without any drawing method and without LVGL, final validation also auto-enables
``show_test_card``, which in turn makes the component require a buffer and therefore
triggers the buffer-size selection path.
"""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
config = _validated(_custom_config(width, height, color_depth))
# Sanity check: final validation should have enabled the test card for us,
# which is what causes the buffer-size calculation to actually run.
assert config.get(CONF_SHOW_TEST_CARD) is True
assert config[CONF_BUFFER_SIZE] == pytest.approx(expected)
@pytest.mark.parametrize(
"buffer_size",
[0.125, 0.25, 0.5, 1.0],
ids=["one_eighth", "one_quarter", "half", "full"],
)
def test_explicit_buffer_size_is_preserved(
buffer_size: float,
set_core_config: SetCoreConfigCallable,
) -> None:
"""An explicitly configured buffer_size is never overridden by final validation."""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
config = _validated(
_custom_config(240, 320, "16bit", buffer_size=buffer_size),
)
assert config[CONF_BUFFER_SIZE] == pytest.approx(buffer_size)
def test_buffer_size_not_set_when_psram_enabled(
set_core_config: SetCoreConfigCallable,
set_component_config,
) -> None:
"""When PSRAM is enabled the auto buffer-size selection is skipped."""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
# Presence of the psram domain in the full config is what _final_validate checks.
set_component_config("psram", True)
config = _validated(_custom_config(240, 320, "16bit"))
assert CONF_BUFFER_SIZE not in config
def test_buffer_size_not_set_when_buffer_not_required(
set_core_config: SetCoreConfigCallable,
set_component_config,
) -> None:
"""With LVGL present and no drawing methods, no buffer fraction is chosen.
LVGL suppresses the automatic show_test_card injection, which means
``requires_buffer`` is False and the early-return branch fires.
"""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
set_component_config("lvgl", [])
config = _validated(_custom_config(240, 320, "16bit"))
assert CONF_BUFFER_SIZE not in config
# And no test card should have been auto-enabled either.
assert not config.get(CONF_SHOW_TEST_CARD)
def test_buffer_size_selected_when_lvgl_with_test_card(
set_core_config: SetCoreConfigCallable,
set_component_config,
) -> None:
"""LVGL present + an explicit drawing method still triggers buffer sizing.
When LVGL is enabled, ``show_test_card`` is not injected automatically,
but users can still request it explicitly -- in that case ``requires_buffer``
is True and the buffer-size heuristic still runs.
"""
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data={KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32},
)
set_component_config("lvgl", [])
# 128x160 @ 16bit -> expected 1/4 (see test_buffer_size_auto_selected).
config = _validated(
_custom_config(128, 160, "16bit", show_test_card=True),
)
assert config[CONF_BUFFER_SIZE] == pytest.approx(1.0 / 4)
@@ -0,0 +1,19 @@
ethernet:
type: W5500
clk_pin: 6
mosi_pin: 7
miso_pin: 2
cs_pin: 10
interrupt_pin: 3
reset_pin: 4
clock_speed: 10Mhz
manual_ip:
static_ip: 192.168.178.56
gateway: 192.168.178.1
subnet: 255.255.255.0
domain: .local
mac_address: "02:AA:BB:CC:DD:01"
on_connect:
- logger.log: "Ethernet connected!"
on_disconnect:
- logger.log: "Ethernet disconnected!"
@@ -45,6 +45,11 @@ esphome:
args:
- response->status_code
- body.c_str()
- delay: 1s
- logger.log:
format: "After delay, body still: %s"
args:
- body.c_str()
http_request:
useragent: esphome/tagreader
@@ -0,0 +1,27 @@
esphome:
name: sched-interval-zero
host:
api:
logger:
level: DEBUG
globals:
- id: fire_count
type: int
initial_value: "0"
interval:
# Deliberately configure 0ms — this path goes through the C++
# Scheduler::set_timer_common_ coercion (not the Python cv.update_interval
# path, since interval: doesn't call cv.update_interval — it's an intervals
# component schema, not a PollingComponent's update_interval).
# Expected: scheduler coerces to 1ms at registration, emits ESP_LOGE,
# fires at ~1kHz instead of spinning.
- interval: 0ms
then:
- lambda: |-
id(fire_count) += 1;
if (id(fire_count) == 50) {
ESP_LOGI("test", "ZERO_INTERVAL_50_FIRES_REACHED");
}
@@ -0,0 +1,67 @@
"""Test that Scheduler::set_timer_common_ coerces interval=0 to 1ms.
Regression test for the scheduler busy-loop when interval=0 was passed
literally. Without the coercion, Scheduler::call() would spin forever
because the item's next_execution == now_64 after re-scheduling, failing
the loop's `> now_64` break condition. The device would fail to yield
back to the main loop and trigger a WDT reset.
With the coercion, interval=0 becomes interval=1 and the scheduler
fires at ~1kHz (bounded by the loop), the main loop continues to run,
and the device stays responsive to API calls.
"""
from __future__ import annotations
import asyncio
import pytest
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
async def test_scheduler_interval_zero_coerced(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""interval=0ms must be coerced to 1ms and not starve the main loop."""
loop = asyncio.get_running_loop()
reached_50: asyncio.Future[None] = loop.create_future()
coerce_warning: asyncio.Future[None] = loop.create_future()
def on_log_line(line: str) -> None:
if "ZERO_INTERVAL_50_FIRES_REACHED" in line and not reached_50.done():
reached_50.set_result(None)
if "would spin main loop" in line and not coerce_warning.done():
coerce_warning.set_result(None)
async with (
run_compiled(yaml_config, line_callback=on_log_line),
api_client_connected() as client,
):
# The API-client connection itself is evidence that the main loop
# is not starved — if set_interval(0) were spinning we could not
# get here at all.
device_info = await client.device_info()
assert device_info is not None
assert device_info.name == "sched-interval-zero"
# Coerce warning must fire at registration
try:
await asyncio.wait_for(coerce_warning, timeout=5.0)
except TimeoutError:
pytest.fail("Expected coerce warning 'would spin main loop' not seen")
# The coerced 1ms interval should fire 50 times quickly — this
# confirms the callback actually runs (not just registered) and the
# scheduler yields back to the main loop each time.
try:
await asyncio.wait_for(reached_50, timeout=5.0)
except TimeoutError:
pytest.fail(
"Coerced interval=0→1ms did not reach 50 fires within 5s, "
"which would indicate either the coercion failed or the "
"main loop is still being starved."
)
@@ -24,6 +24,7 @@ from esphome.const import (
PLATFORM_LN882X,
PLATFORM_RP2040,
PLATFORM_RTL87XX,
SCHEDULER_DONT_RUN,
)
from esphome.core import CORE, HexInt, Lambda
@@ -765,3 +766,30 @@ def test_percentage_validators__raw_number_above_one_without_percent_sign(
config_validation.unbounded_percentage(value)
with pytest.raises(Invalid, match="percent sign"):
config_validation.unbounded_possibly_negative_percentage(value)
def test_update_interval__coerces_zero_to_one_ms(
caplog: pytest.LogCaptureFixture,
) -> None:
"""update_interval: 0ms must be coerced to 1ms (not rejected) because a
literal 0ms schedule causes Scheduler::call() to spin. Coercion keeps
existing configs compiling on upgrade while emitting a user-facing
warning that directs them to set a non-zero value."""
with caplog.at_level("WARNING"):
result = config_validation.update_interval("0ms")
assert result.total_milliseconds == 1
assert "update_interval of 0ms is not supported" in caplog.text
assert "1ms" in caplog.text
def test_update_interval__preserves_nonzero_values() -> None:
"""Non-zero update_interval values must pass through unchanged."""
assert config_validation.update_interval("1ms").total_milliseconds == 1
assert config_validation.update_interval("50ms").total_milliseconds == 50
assert config_validation.update_interval("60s").total_milliseconds == 60000
def test_update_interval__never_passes_through() -> None:
"""update_interval: never must still map to SCHEDULER_DONT_RUN."""
result = config_validation.update_interval("never")
assert result.total_milliseconds == SCHEDULER_DONT_RUN