github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-18 20:22:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into jesserockz-2026-446

Browse Source
This commit is contained in:
J. Nick Koston
2026-10-12 15:07:27 -10:00
committed by GitHub
parent 86073304dc afedddafe2
commit d7bda29ada
4275 changed files with 130896 additions and 34732 deletions
Download Patch File Download Diff File Expand all files Collapse all files
script/api_protobuf/api_protobuf.py
+330 -310
View File
File diff suppressed because it is too large Load Diff
script/build_helpers.py
+25 -13
View File
@@ -22,8 +22,8 @@ from esphome.__main__ import command_compile, parse_args
from esphome.config import validate_config
from esphome.const import CONF_PLATFORM
from esphome.core import CORE
from esphome.host.toolchain import get_elf_path
from esphome.loader import get_component, get_platform
from esphome.platformio.toolchain import get_idedata
from tests.testing_helpers import ComponentManifestOverride, set_testing_manifest
# This must coincide with the version in /platformio.ini
@@ -46,6 +46,8 @@ EXIT_SKIPPED = 1
EXIT_COMPILE_ERROR = 2
EXIT_CONFIG_ERROR = 3
EXIT_NO_EXECUTABLE = 4
# A test folder with this name would be synced into src/esphome and swept away with the core tree
CORE_TREE_DIR = "esphome"
# Name of the per-component YAML config file in benchmark directories
BENCHMARK_YAML_FILENAME = "benchmark.yaml"
@@ -220,7 +222,7 @@ def create_host_config(
friendly_name: str,
libraries: str | list[str],
includes: list[str],
platformio_options: dict,
build_flags: list[str],
) -> dict:
"""Create an ESPHome host configuration for C++ builds.
@@ -229,7 +231,7 @@ def create_host_config(
friendly_name: Human-readable name
libraries: PlatformIO library specification(s)
includes: List of include folders for the build
platformio_options: Dict of platformio_options to set
build_flags: Compiler/linker flags for the build
Returns:
Configuration dict for ESPHome
@@ -239,7 +241,7 @@ def create_host_config(
"name": config_name,
"friendly_name": friendly_name,
"libraries": libraries,
"platformio_options": platformio_options,
"build_flags": build_flags,
"includes": includes,
},
HOST_KEY: {},
@@ -403,13 +405,12 @@ def compile_and_get_binary(
return EXIT_COMPILE_ERROR, None
# After a successful compilation, locate the executable:
idedata = get_idedata(config)
if idedata is None:
program_path = get_elf_path()
if not program_path.is_file():
print("Cannot find executable")
return EXIT_NO_EXECUTABLE, None
program_path: str = idedata.raw["prog_path"]
return EXIT_OK, program_path
return EXIT_OK, str(program_path)
def build_and_run(
@@ -419,7 +420,7 @@ def build_and_run(
config_prefix: str,
friendly_name: str,
libraries: str | list[str],
platformio_options: dict,
build_flags: list[str],
main_entry: str,
label: str = "build",
build_only: bool = False,
@@ -438,7 +439,7 @@ def build_and_run(
config_prefix: Prefix for the config name (e.g. "cpptests", "cppbench")
friendly_name: Human-readable name for the config
libraries: PlatformIO library specification(s)
platformio_options: PlatformIO options dict
build_flags: Compiler/linker flags for the build
main_entry: Name of the main entry file (e.g. "main.cpp")
label: Label for log messages
build_only: If True, print binary path and return without running
@@ -466,8 +467,19 @@ def build_and_run(
components = sorted(components)
# Build include list: main entry point + component folders + extra dirs
includes: list[str] = [main_entry] + components
# Build include list: main entry point + component folders + extra dirs. The core tree
# folder is listed file by file, nested files included, since a folder include would
# land in src/esphome (see CORE_TREE_DIR)
includes: list[str] = [main_entry]
for component in components:
if component != CORE_TREE_DIR:
includes.append(component)
continue
includes.extend(
str(path.relative_to(tests_dir))
for path in sorted((tests_dir / component).rglob("*"))
if path.suffix in (".cpp", ".h")
)
if extra_include_dirs:
for d in extra_include_dirs:
if d.is_dir() and (any(d.glob("*.cpp")) or any(d.glob("*.h"))):
@@ -488,7 +500,7 @@ def build_and_run(
config_name: str = f"{config_prefix}-" + hash_components(components)
config = create_host_config(
config_name, friendly_name, libraries, includes, platformio_options
config_name, friendly_name, libraries, includes, build_flags
)
exit_code, program_path = compile_and_get_binary(
script/build_language_schema.py
+4 -1
View File
@@ -511,7 +511,10 @@ def fix_lvgl_widgets():
def get_logger_tags():
pattern = re.compile(r'^static const char \*const TAG = "(\w.*)";', re.MULTILINE)
pattern = re.compile(
r'^(?:static const char \*const TAG = "|ESPHOME_LOG_TAG\(TAG, ")(\w[^"]*)"',
re.MULTILINE,
)
# tags not in components dir
tags = [
"app",
script/check_idf_py_equivalence.py
Executable
+308
View File
@@ -0,0 +1,308 @@
#!/usr/bin/env python3
"""Check that a native ESP-IDF build tree is what idf.py itself would produce.
ESPHome runs cmake and ninja directly with the arguments idf.py uses. This
runs the real ``idf.py reconfigure`` and ``idf.py build`` on a finished tree
(in place: CMake rejects a moved cache) and fails if either one changes the
cache, the generated build files or the firmware, or recompiles anything.
It catches drift when the pinned ESP-IDF version changes what idf.py does.
The color and ``size`` environment only change what is printed, so those
parts of the contract are pinned by the unit tests instead.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
WATCHED = (
"build/CMakeCache.txt",
"build/build.ninja",
"build/compile_commands.json",
"build/project_description.json",
"build/config/sdkconfig.h",
)
# Only a tree that builds the bootloader has one to watch.
BOOTLOADER_BIN = "build/bootloader/bootloader.bin"
OVERRIDE_INEFFECTIVE = (
"skip-mode tree built a bootloader; the IDF_BOOTLOADER_OVERRIDE macro in "
"esphome/build_gen/espidf.py is not taking effect"
)
# Ninja logs whose outputs mean real work when their recorded mtime changes;
# the bootloader is judged by its own sub-build log when one exists.
TOP_NINJA_LOG = "build/.ninja_log"
def _ninja_logs(skip_bootloader: bool) -> list[str]:
"""A missing sub-build log stays an error in the mode that requires one."""
logs = [TOP_NINJA_LOG]
if not skip_bootloader:
logs.append("build/bootloader/.ninja_log")
return logs
BOOTLOADER_BYPRODUCT = re.compile(r"(^|/build/)bootloader/")
MACRO_CHANGED = (
"IDF changed __build_process_project_includes; update "
"IDF_BOOTLOADER_OVERRIDE in esphome/build_gen/espidf.py"
)
VERSION_DRIFT = (
"ESPHome reads ESP-IDF version {ours!r} from {source} but idf_tools reports "
"{theirs!r}; update read_idf_version_{source} in esphome/espidf/framework.py"
)
LWIP_NOT_EMPTY = (
"lwip source {source} compiles to a non-empty object with {option} off; "
"drop it from LWIP_EMPTY_SOURCES in esphome/build_gen/espidf.py"
)
LWIP_NOTHING_MATCHED = (
"no lwip object matched {regex!r} for {option}; the lwip layout or the "
"pattern in esphome/build_gen/espidf.py changed"
)
LWIP_NM_FAILED = "nm failed on lwip object {source}: {error}"
WORK_SUFFIXES = (".obj", ".o", ".a", ".elf", ".map", ".bin", ".ld")
DEFAULT_GLOB = "tests/test_build_components/build/.esphome/build/*"
def _digest(path: Path) -> str | None:
return hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None
def watched(name: str, skip_bootloader: bool) -> list[str]:
"""The files idf.py must leave untouched for this tree's mode."""
files = [*WATCHED, f"build/{name}.elf", f"build/{name}.bin"]
if not skip_bootloader:
files.append(BOOTLOADER_BIN)
return files
def _snapshot(
build_path: Path, name: str, skip_bootloader: bool
) -> dict[str, str | None]:
return {f: _digest(build_path / f) for f in watched(name, skip_bootloader)}
def _ninja_mtimes(build_path: Path, logs: list[str]) -> dict[tuple[str, str], str]:
"""(log, output) -> recorded mtime; compaction-safe, unlike a line count."""
mtimes = {}
for name in logs:
log = build_path / name
lines = log.read_text(encoding="utf-8").splitlines() if log.is_file() else []
for fields in (line.split("\t") for line in lines if not line.startswith("#")):
if len(fields) >= 4 and (
name != TOP_NINJA_LOG or not BOOTLOADER_BYPRODUCT.search(fields[3])
):
mtimes[name, fields[3]] = fields[2]
return mtimes
def _log_problems(
build_path: Path, mtimes: dict[tuple[str, str], str], logs: list[str]
) -> list[str]:
"""A missing or unparsable ninja log would otherwise compare as unchanged."""
problems = []
for log in logs:
if not (build_path / log).is_file():
problems.append(f"missing {log}")
elif not any(k[0] == log and k[1].endswith(WORK_SUFFIXES) for k in mtimes):
problems.append(f"no build entries parsed from {log}")
return problems
def _lwip_empty_source_problems(build_path: Path) -> list[str]:
"""Compile the lwip sources the generated CMakeLists drops; any with
symbols is a problem. Leaves the tree configured with every source."""
# pylint: disable=protected-access
from esphome.build_gen.espidf import LWIP_EMPTY_SOURCES, LWIP_FULL_SOURCES_ENV
from esphome.espidf import toolchain
if (rc := toolchain.run_reconfigure(extra_env={LWIP_FULL_SOURCES_ENV: "1"})) != 0:
return [f"CMake configure with every lwip source failed with exit code {rc}"]
if rc := toolchain._run_ninja("esp-idf/lwip/liblwip.a", verbose=False, jobs=None):
return [f"building every lwip source failed with exit code {rc}"]
build = build_path / "build"
config = json.loads(
(build / "config" / "sdkconfig.json").read_text(encoding="utf-8")
)
objects = [
obj.as_posix().removesuffix(".obj")
for obj in (build / "esp-idf" / "lwip").rglob("*.obj")
]
nm = toolchain._parse_cmakecache(build / "CMakeCache.txt")["CMAKE_NM"]
problems = []
for option, regex in LWIP_EMPTY_SOURCES:
# Absent means the option is invisible here; the filter keeps those.
if config.get(option.removeprefix("CONFIG_"), True):
continue
matched = [source for source in objects if re.search(regex, source)]
if not matched:
problems.append(LWIP_NOTHING_MATCHED.format(regex=regex, option=option))
for source in matched:
name = Path(source).name
result = subprocess.run(
[nm, "--defined-only", f"{source}.obj"],
capture_output=True,
text=True,
check=False,
)
if result.returncode:
problems.append(
LWIP_NM_FAILED.format(source=name, error=result.stderr.strip())
)
elif result.stdout.strip():
problems.append(LWIP_NOT_EMPTY.format(source=name, option=option))
return problems
def _setup_core(build_path: Path, description: dict) -> tuple[str, str]:
"""Point CORE at the tree so ESPHome resolves the same IDF env as the build."""
from esphome.components.esp32.const import KEY_ESP32, KEY_IDF_VERSION, KEY_VARIANT
import esphome.config_validation as cv
from esphome.core import CORE
name = description["project_name"]
version = Path(description["idf_path"]).name
CORE.config_path = build_path.parents[2] / f"{name}.yaml"
CORE.build_path = build_path
CORE.name = name
CORE.data[KEY_ESP32] = {
KEY_IDF_VERSION: cv.Version.parse(version),
KEY_VARIANT: description["target"].upper(),
}
return name, version
def check(build_path: Path) -> list[str]:
"""Return the problems found in one build tree."""
# pylint: disable=protected-access
from esphome.build_gen.espidf import idf_macro_matches
from esphome.core import CORE
from esphome.espidf import framework, toolchain
description = json.loads(
(build_path / "build" / "project_description.json").read_text(encoding="utf-8")
)
name, version = _setup_core(build_path, description)
# Reconfiguring must not flip the tree's bootloader mode: the check
# validates the shape the build produced, not this process's flags.
skip_bootloader = toolchain.tree_skips_bootloader(build_path / "build")
CORE.skip_bootloader = skip_bootloader
# A prior tree's memoized decision must not leak into this one.
toolchain._cache().skip_bootloader = None
idf_path = toolchain._get_idf_path(version)
if not idf_macro_matches(idf_path):
return [MACRO_CHANGED]
# A managed tree always has version.txt, so the header branch is
# compared on its own or it would never be exercised here.
theirs = framework.idf_tools_version(idf_path)
for source, read in (
("txt", framework.read_idf_version_txt),
("header", framework.read_idf_version_header),
):
if (ours := read(idf_path)) != theirs:
return [VERSION_DRIFT.format(ours=ours, source=source, theirs=theirs)]
# ESP-IDF's openthread stamps the configure time into its compile flags;
# pin it before the env is cached so both configures get the same value.
os.environ["SOURCE_DATE_EPOCH"] = "0"
env = toolchain._get_idf_env(version)
python = toolchain._get_idf_tool("python")
idf_py = toolchain._get_idf_path(version) / "tools" / "idf.py"
sdkconfig = build_path / f"sdkconfig.{name}"
sdkconfig_args = ["-D", f"SDKCONFIG={sdkconfig}"] if sdkconfig.is_file() else []
# CMake writes a different build.ninja on a tree's first configure than on
# a reconfigure, so the baseline is ESPHome's own reconfigure and build.
if (rc := toolchain.run_reconfigure()) != 0:
return [f"ESPHome's CMake configure failed with exit code {rc}"]
if (rc := toolchain._run_ninja("all", verbose=False, jobs=None)) != 0:
return [f"ESPHome's ninja build failed with exit code {rc}"]
before = _snapshot(build_path, name, skip_bootloader)
logs = _ninja_logs(skip_bootloader)
mtimes_before = _ninja_mtimes(build_path, logs)
# A moved or renamed output would otherwise compare as "unchanged".
problems = [f"missing {f}" for f, digest in before.items() if digest is None]
if skip_bootloader and (build_path / BOOTLOADER_BIN).is_file():
problems.append(OVERRIDE_INEFFECTIVE)
if problems := problems + _log_problems(build_path, mtimes_before, logs):
return problems
for action in ("reconfigure", "build"):
result = subprocess.run(
[python, str(idf_py), *sdkconfig_args, action],
cwd=build_path,
env=env,
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
return [f"idf.py {action} failed:\n{result.stdout}{result.stderr}"]
after = _snapshot(build_path, name, skip_bootloader)
mtimes_after = _ninja_mtimes(build_path, logs)
problems = [f"idf.py changed {f}" for f in before if before[f] != after[f]]
problems += _log_problems(build_path, mtimes_after, logs)
for key in sorted(mtimes_before.keys() | mtimes_after.keys()):
log, out = key
if not out.endswith(WORK_SUFFIXES):
continue
if key not in mtimes_after:
problems.append(f"idf.py dropped {out} from {log}")
elif mtimes_before.get(key) != mtimes_after[key]:
problems.append(f"idf.py rebuilt {out}")
# Last: it reconfigures the tree, which would otherwise relink above.
return problems or _lwip_empty_source_problems(build_path)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0])
parser.add_argument(
"build_paths",
nargs="*",
type=Path,
help=f"ESPHome build dirs (default: the first native ESP-IDF tree in {DEFAULT_GLOB})",
)
args = parser.parse_args()
paths = args.build_paths or sorted(REPO_ROOT.glob(DEFAULT_GLOB))
# Not resolved: SDKCONFIG must be spelled as the build spelled it.
trees = [
p
for p in paths
if (p / "build" / "project_description.json").is_file()
and (p / "build" / "CMakeCache.txt").is_file()
]
if rejected := [p for p in args.build_paths if p not in trees]:
for path in rejected:
print(f"{path}: not a configured native ESP-IDF build tree")
return 1
if not trees:
print("No native ESP-IDF build tree found")
return 1
if not args.build_paths:
# The contract does not depend on the target, so one tree is enough.
trees = trees[:1]
failed = False
for tree in trees:
problems = check(tree)
print(f"{tree}: {'OK' if not problems else 'DIFFERS'}")
for problem in problems:
print(f" {problem}")
failed |= bool(problems)
if failed:
print(
"The direct cmake/ninja build no longer matches idf.py. Compare "
"_IdfPyContract and its users in esphome/espidf/toolchain.py with the "
"pinned ESP-IDF tools/idf_py_actions."
)
return 1 if failed else 0
if __name__ == "__main__":
sys.exit(main())
script/ci-custom.py
+111 -8
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import argparse
import ast
import codecs
import collections
from collections.abc import Iterator
@@ -164,7 +165,21 @@ def lint_post_check(func):
return func
def lint_re_check(regex, **kwargs):
def _nolint_in_match(content, haystack, match, mask):
"""With masking, only a trailing comment counts: the raw span still holds string contents, and
the masked text is blank exactly where comments and strings were, so NOLINT must sit after the
last real code character of the span."""
raw = content[match.start() : match.end()]
if not mask:
return "NOLINT" in raw
masked = haystack[match.start() : match.end()].rstrip()
return "NOLINT" in raw[len(masked) :]
def lint_re_check(regex, mask=False, prefilter=None, **kwargs):
"""mask=True blanks comments and string literals first so prose about the pattern is not reported;
the masked text keeps its length, so match offsets still index the original content.
prefilter is a literal every match must contain, checked before the costlier masking."""
flags = kwargs.pop("flags", re.MULTILINE)
prog = re.compile(regex, flags)
decor = lint_content_check(**kwargs)
@@ -173,8 +188,11 @@ def lint_re_check(regex, **kwargs):
@functools.wraps(func)
def new_func(fname, content):
errs = []
for match in prog.finditer(content):
if "NOLINT" in match.group(0):
if prefilter is not None and prefilter not in content:
return errs
haystack = _mask_cpp_comments_strings(content) if mask else content
for match in prog.finditer(haystack):
if _nolint_in_match(content, haystack, match, mask):
continue
lineno = content.count("\n", 0, match.start()) + 1
substr = content[: match.start()]
@@ -794,6 +812,40 @@ def lint_const_py_frozen(fname, content):
return None
@lint_content_check(include=["esphome/components/*/__init__.py"])
def lint_component_domain(fname: Path, content: str) -> str | None:
"""Require every component's __init__.py to define DOMAIN as its own name."""
if len(fname.parts) != 4:
# Platform packages such as esphome/components/<name>/sensor/__init__.py
return None
domain = fname.parts[2]
expected = f'DOMAIN = "{domain}"'
try:
tree = ast.parse(content, filename=str(fname))
except SyntaxError:
# Reported by the Python linters; nothing useful to add here.
return None
for node in tree.body:
if isinstance(node, ast.Assign):
targets = node.targets
elif isinstance(node, ast.AnnAssign):
targets = [node.target]
else:
continue
if (
any(isinstance(t, ast.Name) and t.id == "DOMAIN" for t in targets)
and isinstance(node.value, ast.Constant)
and node.value.value == domain
):
return None
return (
f"Component is missing the {highlight(expected)} constant. "
"Add it in alphabetical order with the other component metadata such as "
"CODEOWNERS and DEPENDENCIES, so other code can refer to the component by "
"name, e.g. CORE.data[DOMAIN]."
)
def relative_cpp_search_text(fname: Path, content) -> str:
parts = fname.parts
integration = parts[2]
@@ -887,6 +939,8 @@ def lint_relative_py_import(fname: Path, line, col, content):
# neither can live in a C++ namespace.
"esphome/components/esp32_hosted/esp_now_hosted.cpp",
"esphome/components/esp32_hosted/esp_now_hosted_rpc.h",
# C header shared with the vendored decoder
"esphome/components/esphome/ota/ota_esphome_inflate.h",
],
)
def lint_namespace(fname: Path, content: str) -> str | None:
@@ -1011,15 +1065,13 @@ HEAP_ALLOCATING_HELPERS = {
r"str_sprintf|"
r"str_snprintf|"
r"value_accuracy_to_string"
r")\s*\(" + CPP_RE_EOL,
# Explicit template arguments at any nesting depth, e.g. format_hex<std::array<uint8_t, 4>>(
r")\s*(?:<[^;{}()\n]*>\s*)?\(" + CPP_RE_EOL,
include=cpp_include,
exclude=[
# The definitions themselves
"esphome/core/alloc_helpers.h",
"esphome/core/alloc_helpers.cpp",
# Backward compatibility re-exports (remove before 2026.11.0)
"esphome/core/helpers.h",
"esphome/core/helpers.cpp",
# Vendored third-party library
"esphome/components/http_request/httplib.h",
],
@@ -1072,7 +1124,6 @@ def lint_no_sprintf(fname, match):
# Vendored library
"esphome/components/http_request/httplib.h",
# Deprecated helpers that return std::string
"esphome/core/helpers.cpp",
"esphome/core/alloc_helpers.cpp",
# The using declaration itself
"esphome/core/helpers.h",
@@ -1199,6 +1250,58 @@ def lint_no_std_bind(fname, match):
)
@lint_re_check(
r"[^\w]std\s*::\s*nothrow\b" + CPP_RE_EOL,
mask=True,
prefilter="nothrow",
include=cpp_include,
)
def lint_no_std_nothrow(fname, match):
return (
f"{highlight('new (std::nothrow)')} aborts on ESP-IDF when the allocation fails, exceptions are disabled "
f"there, so it never returns nullptr.\n"
f"Please use {highlight('RAMAllocator')} from esphome/core/helpers.h, which does.\n"
f" Before: {highlight('auto *buf = new (std::nothrow) uint8_t[n];')}\n"
f" After: {highlight('auto buf = RAMAllocator<uint8_t>().make_unique_array_for_overwrite(n);')}\n"
f"For one object use {highlight('RAMAllocator<T>().make_unique(args...)')}; both return empty on failure.\n"
f"Default flags prefer PSRAM; pass RAMAllocator<T>::PREFER_INTERNAL to keep it where new put it.\n"
f"(If strictly necessary, add `// NOLINT` to the end of the line)"
)
@lint_re_check(
r"^\s*(?:(?:static|constexpr|inline)\s+)*(?:const\s+)?char\s*(?:\*\s*(?:const\s+)?TAG|\s(?:const\s+)?TAG\s*\[\w*\])"
r"\s*(?:=\s*\{?|[{(])\s*\"",
prefilter="TAG",
include=["esphome/components/*.cpp", "esphome/components/**/*.cpp"],
)
def lint_log_tag_macro(fname, match):
return (
f"Declare log tags with {highlight('ESPHOME_LOG_TAG(TAG, "name");')}, which keeps the tag in flash on "
f"ESP8266.\n"
f" Before: {highlight('static const char *const TAG = "name";')}\n"
f" After: {highlight('ESPHOME_LOG_TAG(TAG, "name");')}\n"
f"Never pass such a TAG to set_timeout/set_interval names or string functions.\n"
f"(If strictly necessary, add `// NOLINT` to the end of the line)"
)
@lint_re_check(
r"(?:\b(?:set_timeout|set_interval|set_retry|cancel_timeout|cancel_interval|cancel_retry|defer)"
r"|\b(?:strcmp|strncmp|strcasecmp|strlen|strcpy|strncpy)|std::string)\s*\((?:(?:[^();]|\([^()]*\))*,)?\s*TAG\b"
r"|\bstd::string\s+\w+\s*[={(]\s*TAG\b",
mask=True,
prefilter="TAG",
include=["esphome/components/*.cpp", "esphome/components/**/*.cpp"],
)
def lint_log_tag_as_string(fname, match):
return (
f"{highlight('TAG')} is in flash on ESP8266 (ESPHOME_LOG_TAG), so it can only be used for logging.\n"
f"Use a separate name or a numeric id for scheduler calls, and do not pass it to string functions.\n"
f"(If strictly necessary, add `// NOLINT` to the end of the line)"
)
LOG_CALL_START_RE = re.compile(r"ESP_LOG\w+\s*\(")
# Comments, raw/plain string literals and single char literals are consumed whole so ; ( ) ? :
# inside them are never seen. A char literal is exactly one (escaped) char so a digit separator
script/clang-tidy
+51 -9
View File
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
import argparse
import heapq
import os
from pathlib import Path
import queue
@@ -11,6 +12,7 @@ import subprocess
import sys
import tempfile
import threading
import time
from typing import Any
import click
@@ -277,10 +279,21 @@ def clang_options(idedata, environment):
pids = set()
# Generated files the static analyzer is skipped on. The analyzer inlines the
# small dump helpers into every generated dump function and takes
# over 200 s on this file in CI, 99% of its clang-tidy time; any finding there
# would be fixed in script/api_protobuf anyway. The other checks still run.
ANALYZER_SKIPPED_FILES: frozenset[str] = frozenset(
{"esphome/components/api/api_pb2_dump.cpp"}
)
def run_tidy(executable, args, options, tmpdir, path_queue, lock, failed_files):
def run_tidy(
executable, args, options, tmpdir, path_queue, lock, failed_files, timings
):
while True:
path = path_queue.get()
started = time.monotonic()
invocation = [executable]
if tmpdir is not None:
@@ -297,16 +310,22 @@ def run_tidy(executable, args, options, tmpdir, path_queue, lock, failed_files):
if sys.stdout.isatty():
invocation.append("--use-color")
disabled_checks = []
if args.environment.startswith(("rp2", "bk72xx", "ln882h", "rtl87xx")):
# MMIO peripheral access on these bare-metal platforms is all
# fixed-address.
# bugprone-pointer-arithmetic-on-polymorphic-object (and its
# cert-ctr56-cpp alias) crashes clang-tidy 22 with infinite matcher
# recursion on lvgl_esphome.h under the RP2 defines.
invocation.append(
"--checks=-clang-analyzer-core.FixedAddressDereference,"
"-bugprone-pointer-arithmetic-on-polymorphic-object,-cert-ctr56-cpp"
)
disabled_checks += [
"-clang-analyzer-core.FixedAddressDereference",
"-bugprone-pointer-arithmetic-on-polymorphic-object",
"-cert-ctr56-cpp",
]
if os.path.normpath(path) in ANALYZER_SKIPPED_FILES:
disabled_checks.append("-clang-analyzer-*")
if disabled_checks:
invocation.append("--checks=" + ",".join(disabled_checks))
invocation.append(f"--header-filter={Path(basepath).resolve()}/.*")
invocation.append(str(Path(path).resolve()))
invocation.append("--")
@@ -319,8 +338,9 @@ def run_tidy(executable, args, options, tmpdir, path_queue, lock, failed_files):
check=False,
close_fds=False,
)
if proc.returncode != 0:
with lock:
with lock:
timings.append((time.monotonic() - started, path))
if proc.returncode != 0:
print_error_for_file(path, proc.stdout)
failed_files.append(path)
path_queue.task_done()
@@ -333,8 +353,27 @@ def progress_bar_show(value):
def split_list(a, n):
k, m = divmod(len(a), n)
return [a[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)] for i in range(n)]
"""Deal the sorted files out to n splits in turn.
Contiguous slices put whole directories on one split; expensive
neighbours such as esphome/core then land together and that split takes
several times longer than the others. Dealing spreads them out.
"""
return [a[i::n] for i in range(n)]
SLOWEST_FILES_SHOWN = 15
def print_slowest_files(timings):
"""Print the files clang-tidy spent the longest on, so the splits can be
rebalanced from CI logs."""
if not timings:
return
total = sum(seconds for seconds, _ in timings)
print(f"\nclang-tidy spent {total:.0f} s on {len(timings)} files; slowest:")
for seconds, path in heapq.nlargest(SLOWEST_FILES_SHOWN, timings):
print(f" {seconds:6.1f} s {path}")
def main():
@@ -458,6 +497,7 @@ def main():
tmpdir = tempfile.mkdtemp()
failed_files = []
timings = []
try:
executable = get_binary("clang-tidy", 22)
task_queue = queue.Queue(args.jobs)
@@ -473,6 +513,7 @@ def main():
task_queue,
lock,
failed_files,
timings,
),
)
t.daemon = True
@@ -487,6 +528,7 @@ def main():
# Wait for all threads to be done.
task_queue.join()
print_slowest_files(timings)
except FileNotFoundError:
return 1
script/cpp_benchmark.py
+11 -19
View File
@@ -25,19 +25,14 @@ CORE_BENCHMARKS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "core"
# allow benchmarks to compile on the host platform.
STUBS_DIR: Path = Path(root_path) / "tests" / "benchmarks" / "stubs"
PLATFORMIO_OPTIONS = {
"build_flags": [
"-Os", # match firmware optimization level (detects inlining regressions)
"-g", # debug symbols for profiling
"-ffunction-sections", # required for dead-code stripping with -Os
"-fdata-sections", # required for dead-code stripping with -Os
"-DUSE_BENCHMARK", # disable WarnIfComponentBlockingGuard in finish()
f"-I{STUBS_DIR}", # stub headers for ESP32-only components
],
# Use deep+ LDF mode to ensure PlatformIO detects the benchmark
# library dependency from nested includes.
"lib_ldf_mode": "deep+",
}
BUILD_FLAGS = [
"-Os", # match firmware optimization level (detects inlining regressions)
"-g", # debug symbols for profiling
"-ffunction-sections", # required for dead-code stripping with -Os
"-fdata-sections", # required for dead-code stripping with -Os
"-DUSE_BENCHMARK", # disable WarnIfComponentBlockingGuard in finish()
f"-I{STUBS_DIR}", # stub headers for ESP32-only components
]
def run_benchmarks(selected_components: list[str], build_only: bool = False) -> int:
@@ -46,7 +41,7 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
# containing {"lib_path": "/path/to/google_benchmark"}.
lib_config_json = os.environ.get("BENCHMARK_LIB_CONFIG")
pio_options = PLATFORMIO_OPTIONS
build_flags = BUILD_FLAGS
if lib_config_json:
lib_config = json.loads(lib_config_json)
benchmark_lib = f"benchmark=symlink://{lib_config['lib_path']}"
@@ -61,10 +56,7 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
"-DCODSPEED_ANALYSIS",
f'-DCODSPEED_ROOT_DIR=\\"{project_root}\\"',
]
pio_options = {
**PLATFORMIO_OPTIONS,
"build_flags": PLATFORMIO_OPTIONS["build_flags"] + codspeed_flags,
}
build_flags = BUILD_FLAGS + codspeed_flags
else:
benchmark_lib = PLATFORMIO_GOOGLE_BENCHMARK_LIB
@@ -77,7 +69,7 @@ def run_benchmarks(selected_components: list[str], build_only: bool = False) ->
config_prefix="cppbench",
friendly_name="CPP Benchmarks",
libraries=benchmark_lib,
platformio_options=pio_options,
build_flags=build_flags,
main_entry="main.cpp",
label="benchmarks",
build_only=build_only,
script/cpp_unit_test.py
+11 -19
View File
@@ -15,24 +15,16 @@ from helpers import get_all_components, root_path
# Path to /tests/components
COMPONENTS_TESTS_DIR: Path = Path(root_path) / "tests" / "components"
PLATFORMIO_OPTIONS = {
"build_type": "debug",
"build_unflags": [
"-Os", # remove size-opt flag
],
"build_flags": [
"-Og", # optimize for debug
"-DESPHOME_DEBUG", # enable debug assertions
# Enable the address and undefined behavior sanitizers
"-fsanitize=address",
"-fsanitize=undefined",
"-fno-omit-frame-pointer",
],
"debug_build_flags": [ # only for debug builds
"-g3", # max debug info
"-ggdb3",
],
}
BUILD_FLAGS = [
"-Og", # optimize for debug
"-g3", # max debug info
"-ggdb3",
"-DESPHOME_DEBUG", # enable debug assertions
# Enable the address and undefined behavior sanitizers
"-fsanitize=address",
"-fsanitize=undefined",
"-fno-omit-frame-pointer",
]
def run_tests(selected_components: list[str]) -> int:
@@ -47,7 +39,7 @@ def run_tests(selected_components: list[str]) -> int:
config_prefix="cpptests",
friendly_name="CPP Unit Tests",
libraries=PLATFORMIO_GOOGLE_TEST_LIB,
platformio_options=PLATFORMIO_OPTIONS,
build_flags=BUILD_FLAGS,
main_entry="main.cpp",
label="unit tests",
)
script/determine-jobs.py
+215 -49
View File
@@ -31,7 +31,7 @@ The CI workflow uses this information to:
- Skip or run integration tests
- Skip or run clang-tidy (and whether to do a full scan)
- Skip or run clang-format
- Skip or run Python linters (ruff, flake8, pylint, pyupgrade)
- Skip or run pylint
- Skip or run downstream esphome/device-builder tests against the PR's Python code
- Determine which components to test individually
- Decide how to split component tests (if there are many)
@@ -50,6 +50,7 @@ from __future__ import annotations
import argparse
from collections import Counter
from collections.abc import Callable
from enum import StrEnum
from functools import cache
import json
@@ -98,10 +99,6 @@ from split_components_for_ci import create_intelligent_batches
# For large PRs (>= 65 files), use split for better parallelization
CLANG_TIDY_SPLIT_THRESHOLD = 65
# Component test batch size (weighted)
# Isolated components count as 10x, groupable components count as 1x
COMPONENT_TEST_BATCH_SIZE = 40
# Above the threshold, fan out across up to this many jobs, balanced by the
# recorded per-file durations. The target is serial junit-time weight per
# bucket, not wall time (calibrated with the conftest compile cap); it
@@ -110,14 +107,25 @@ INTEGRATION_TESTS_SPLIT_THRESHOLD = 10
INTEGRATION_TESTS_SPLIT_BUCKETS = 5
INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT = 360.0
# platformio and aioesphomeapi (requirements.txt), the pytest stack
# (requirements_test.txt) and the fixture every session compiles; a change
# to any runs the full matrix
# aioesphomeapi (requirements.txt), the pytest stack (requirements_test.txt)
# and the native host build backend every test compiles with; a change to
# any runs the full matrix
INTEGRATION_TESTS_TRIGGER_FILES = frozenset(
{
"requirements.txt",
"requirements_test.txt",
"tests/integration/fixtures/cache_init.yaml",
"esphome/arduino/library.py",
"esphome/build_gen/build_tool.py",
"esphome/build_gen/host.py",
"esphome/build_helpers/ccache.py",
"esphome/build_helpers/idedata.py",
"esphome/build_helpers/native.py",
"esphome/build_helpers/ninja.py",
"esphome/build_helpers/ninja_gen.py",
"esphome/build_helpers/tools_cache.py",
"esphome/framework_helpers.py",
"esphome/host/toolchain.py",
"esphome/platformio/library.py",
}
)
@@ -239,7 +247,7 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
- conftest.py, types.py, const.py, entity_utils.py, state_utils.py, etc.
4. A file in INTEGRATION_TESTS_TRIGGER_FILES changed
- The dependency pins and the session init fixture affect every test
- The dependency pins and the host build backend affect every test
Returns (run_all=False, [test_files...]) when:
@@ -395,21 +403,33 @@ def should_run_clang_format(branch: str | None = None) -> bool:
return _any_changed_file_endswith(branch, CPP_FILE_EXTENSIONS)
# Linter pins, dependencies and linter config can change findings without a .py change.
PYTHON_LINTERS_TRIGGER_FILES = frozenset(
{
"requirements.txt",
"requirements_test.txt",
"pyproject.toml",
}
)
def should_run_python_linters(branch: str | None = None) -> bool:
"""Determine if Python linters (ruff, flake8, pylint, pyupgrade) should run based on changed files.
"""Determine if the pylint job should run based on changed files.
This function is used by the CI workflow to skip Python linting checks when no Python files
have changed, saving CI time and resources.
Python linters will run when any Python source files have changed.
The CI workflow skips pylint unless a Python file or a file in
PYTHON_LINTERS_TRIGGER_FILES has changed. ruff, flake8 and pyupgrade run in
lint-format, which this flag does not gate.
Args:
branch: Branch to compare against. If None, uses default.
Returns:
True if Python linters should run, False otherwise.
True if pylint should run, False otherwise.
"""
return _any_changed_file_endswith(branch, PYTHON_FILE_EXTENSIONS)
return any(
file.endswith(PYTHON_FILE_EXTENSIONS) or file in PYTHON_LINTERS_TRIGGER_FILES
for file in changed_files(branch)
)
# Files outside esphome/**/*.py whose changes can affect `import esphome.__main__`
@@ -520,48 +540,78 @@ ESP32_PLATFORMIO_TEST_COMPONENTS = frozenset(
}
)
# Shared by every toolchain smoke-test job: the base config and the bus
# packages each generated build includes
_SMOKE_HARNESS_TRIGGER_PATH_PREFIXES = ("tests/test_build_components/",)
# Path prefixes whose changes always trigger the PlatformIO compile test:
# anything under esphome/platformio/ (the PlatformIO runner / toolchain that
# drives every PlatformIO build). The esp32 platform component is already in
# ESP32_PLATFORMIO_TEST_COMPONENTS, so its changes are covered by the normal
# component-narrowing path.
ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES = ("esphome/platformio/",)
ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES = (
"esphome/platformio/",
*_SMOKE_HARNESS_TRIGGER_PATH_PREFIXES,
)
# Standalone files that, when changed, trigger the PlatformIO compile test:
# - esphome/build_gen/platformio.py -- the PlatformIO build generator
# - script/test_build_components.py -- the harness the job invokes
# - .github/workflows/ci.yml -- the job's own definition
ESP32_PLATFORMIO_TRIGGER_FILES = frozenset(
# Shared by every toolchain smoke-test job: the harness it invokes and the
# workflow that defines it
_SMOKE_HARNESS_TRIGGER_FILES = frozenset(
{
"esphome/build_gen/platformio.py",
"script/test_build_components.py",
".github/workflows/ci.yml",
}
)
# Standalone files that, when changed, trigger the PlatformIO compile test
ESP32_PLATFORMIO_TRIGGER_FILES = _SMOKE_HARNESS_TRIGGER_FILES | {
# The PlatformIO build generator
"esphome/build_gen/platformio.py",
# Decides which platforms take the pch script in esphome/platformio/,
# and the module that parses its switch and the ccache settings
"esphome/build_helpers/pch.py",
"esphome/build_helpers/ccache.py",
}
def _path_or_file_trigger(
files: list[str],
trigger_files: frozenset[str],
trigger_prefixes: tuple[str, ...],
) -> bool:
"""Whether any changed file matches the given infrastructure triggers."""
return any(
file in trigger_files or file.startswith(trigger_prefixes) for file in files
)
@cache
def _cached_components_closure(files: tuple[str, ...]) -> frozenset[str]:
"""Dependency closure of the changed components; cached because the
walk is expensive and every smoke-test job asks for the same list."""
component_files = [f for f in files if filter_component_and_test_files(f)]
return frozenset(get_components_with_dependencies(component_files, True))
def _esp32_platformio_path_or_file_trigger(files: list[str]) -> bool:
"""Whether any changed file is a PlatformIO infrastructure / harness trigger."""
for file in files:
if file in ESP32_PLATFORMIO_TRIGGER_FILES:
return True
if any(
file.startswith(prefix) for prefix in ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES
):
return True
return False
return _path_or_file_trigger(
files, ESP32_PLATFORMIO_TRIGGER_FILES, ESP32_PLATFORMIO_TRIGGER_PATH_PREFIXES
)
# Checks the native ESP-IDF build in CI but does not shape it, so it is kept
# out of ESP_IDF_INFRA_TRIGGER_FILES (hashed into the clang-tidy cache key).
_ESP_IDF_CHECK_TRIGGER_FILES = frozenset({"script/check_idf_py_equivalence.py"})
def _esp_idf_infra_changed(files: list[str]) -> bool:
"""Whether any changed file is ESP-IDF build/runner infrastructure."""
for file in files:
if file in ESP_IDF_INFRA_TRIGGER_FILES:
return True
if any(
file.startswith(prefix) for prefix in ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES
):
return True
return False
return _path_or_file_trigger(
files,
ESP_IDF_INFRA_TRIGGER_FILES | _ESP_IDF_CHECK_TRIGGER_FILES,
ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES,
)
def esp32_platformio_components_to_test(branch: str | None = None) -> list[str]:
@@ -599,15 +649,23 @@ def esp32_platformio_components_to_test(branch: str | None = None) -> list[str]:
Returns:
Sorted list of component names to compile.
"""
return _toolchain_components_to_test(
branch, ESP32_PLATFORMIO_TEST_COMPONENTS, _esp32_platformio_path_or_file_trigger
)
def _toolchain_components_to_test(
branch: str | None,
test_set: frozenset[str],
infra_trigger: Callable[[list[str]], bool],
) -> list[str]:
"""The shared narrowing rule for the per-toolchain smoke-test jobs."""
files = changed_files(branch)
if core_changed(files) or _esp32_platformio_path_or_file_trigger(files):
return sorted(ESP32_PLATFORMIO_TEST_COMPONENTS)
if core_changed(files) or infra_trigger(files):
return sorted(test_set)
component_files = [f for f in files if filter_component_and_test_files(f)]
changed = get_components_with_dependencies(component_files, True)
return sorted(ESP32_PLATFORMIO_TEST_COMPONENTS & set(changed))
return sorted(test_set & _cached_components_closure(tuple(files)))
def should_run_esp32_platformio(branch: str | None = None) -> bool:
@@ -628,6 +686,85 @@ def should_run_esp32_platformio(branch: str | None = None) -> bool:
return bool(esp32_platformio_components_to_test(branch))
# The `--toolchain arduino` smoke-test set: covers the core, the bundled and
# converted registry libraries, and the waveform path.
ESP8266_NATIVE_TEST_COMPONENTS = frozenset(
{
"esp8266",
"api",
"web_server",
"captive_portal",
"mqtt",
"esp8266_pwm",
"neopixelbus",
"bme280_i2c",
"uart",
}
)
# Infrastructure whose changes always trigger the native ESP8266
# compile test
ESP8266_NATIVE_TRIGGER_PATH_PREFIXES = (
"esphome/arduino8266/",
"esphome/arduino/",
"esphome/build_helpers/",
*_SMOKE_HARNESS_TRIGGER_PATH_PREFIXES,
)
# Shared library-conversion modules every native build imports; espidf-only
# infra (build_gen/espidf.py) deliberately stays out of the esp8266 set.
_NATIVE_SHARED_TRIGGER_FILES = frozenset(
{
"esphome/framework_helpers.py",
"esphome/platformio/library.py",
"esphome/platformio/extra_script.py",
}
)
# Tripwire: the shared modules must stay in the ESP-IDF trigger set too
# (now defined in clang_tidy_hash), or its smoke test silently skips them
assert _NATIVE_SHARED_TRIGGER_FILES <= ESP_IDF_INFRA_TRIGGER_FILES
ESP8266_NATIVE_TRIGGER_FILES = (
_NATIVE_SHARED_TRIGGER_FILES
| _SMOKE_HARNESS_TRIGGER_FILES
| {
"esphome/build_gen/arduino8266.py",
"esphome/build_gen/build_tool.py",
"esphome/components/esp8266/build_surgery.py",
"esphome/components/esp8266/boards.py",
"esphome/platformio/registry.py",
# esp8266/__init__.py imports copy_ccache_script from it
"esphome/platformio/toolchain.py",
".github/actions/cache-arduino8266/action.yml",
}
)
def _esp8266_native_path_or_file_trigger(files: list[str]) -> bool:
"""Whether any changed file is native-ESP8266 infrastructure / harness."""
# base_python_changed covers the top-level esphome/*.py modules the
# native backend imports directly (framework_helpers, helpers, writer,
# __main__); without it a change there would silently skip this job.
# base_python_changed is deliberately broad (any top-level esphome/*.py)
# as belt-and-braces while the backend is new; narrow it to the modules
# the backend imports once the toolchain has soaked a few releases
return base_python_changed(files) or _path_or_file_trigger(
files, ESP8266_NATIVE_TRIGGER_FILES, ESP8266_NATIVE_TRIGGER_PATH_PREFIXES
)
def esp8266_native_components_to_test(branch: str | None = None) -> list[str]:
"""The smoke set on a native-build change, nothing otherwise.
Unlike the esp32 PlatformIO job, this one does not narrow to the changed
components: the component matrix already compiles every esp8266 fixture
with this toolchain, so the only gap left is a change to the native build
itself that brings no component along.
"""
files = changed_files(branch)
if core_changed(files) or _esp8266_native_path_or_file_trigger(files):
return sorted(ESP8266_NATIVE_TEST_COMPONENTS)
return []
def determine_cpp_unit_tests(
branch: str | None = None,
) -> tuple[bool, list[str]]:
@@ -1183,6 +1320,7 @@ def detect_memory_impact_config(
"components": compatible_components,
"platform": platform,
"use_merged_config": "true",
"needs_arduino8266": platform.startswith("esp8266"),
}
@@ -1226,6 +1364,8 @@ def main() -> None:
run_device_builder = True
esp32_platformio_components = sorted(ESP32_PLATFORMIO_TEST_COMPONENTS)
run_esp32_platformio = True
esp8266_native_components = sorted(ESP8266_NATIVE_TEST_COMPONENTS)
run_esp8266_native = True
else:
integration_run_all, integration_test_files = determine_integration_tests(
args.branch
@@ -1237,6 +1377,8 @@ def main() -> None:
run_device_builder = should_run_device_builder(args.branch)
esp32_platformio_components = esp32_platformio_components_to_test(args.branch)
run_esp32_platformio = bool(esp32_platformio_components)
esp8266_native_components = esp8266_native_components_to_test(args.branch)
run_esp8266_native = bool(esp8266_native_components)
run_integration, integration_test_buckets = _compute_integration_test_buckets(
integration_run_all, integration_test_files
)
@@ -1400,21 +1542,43 @@ def main() -> None:
batches, _ = create_intelligent_batches(
components=changed_components_with_tests,
tests_dir=tests_dir,
batch_size=COMPONENT_TEST_BATCH_SIZE,
directly_changed=batch_directly_changed,
)
# Convert batches to CI matrix entries: the component list plus which
# native toolchain installs the batch's test platforms need, so the
# workflow only restores the matching multi-GB toolchain caches.
# The idf.py check does not depend on the components, so it runs once
# per workflow, in the first batch that compiles an esp32 test (a
# validate-only component is never compiled).
idf_py_check_assigned = False
skip_compile = set(validate_only_components)
for batch in batches:
platforms: set[str] = set()
compiled_platforms: set[str] = set()
for component in batch:
platforms.update(get_component_test_platforms(component))
# Variants included: the compile stage builds them, so a
# component tested only by test-<variant>.<platform>.yaml
# still needs that platform's toolchain
component_platforms = get_component_test_platforms(
component, base_only=False
)
platforms.update(component_platforms)
if component not in skip_compile:
compiled_platforms.update(component_platforms)
needs_idf = any(p.startswith("esp32") for p in platforms)
check_idf_py = not idf_py_check_assigned and any(
p.startswith("esp32") for p in compiled_platforms
)
idf_py_check_assigned |= check_idf_py
component_test_batches.append(
{
"components": " ".join(batch),
"needs_idf": any(p.startswith("esp32") for p in platforms),
"needs_idf": needs_idf,
"check_idf_py": check_idf_py,
"needs_nrf": any(p.startswith("nrf52") for p in platforms),
"needs_arduino8266": any(
p.startswith("esp8266") for p in platforms
),
}
)
@@ -1432,6 +1596,8 @@ def main() -> None:
"device_builder": run_device_builder,
"esp32_platformio": run_esp32_platformio,
"esp32_platformio_components": ",".join(esp32_platformio_components),
"esp8266_native": run_esp8266_native,
"esp8266_native_components": ",".join(esp8266_native_components),
"changed_components": changed_components,
"changed_components_with_tests": changed_components_with_tests,
"directly_changed_components_with_tests": list(directly_changed_with_tests),
script/helpers.py
+3 -5
View File
@@ -429,11 +429,9 @@ def build_all_include(header_files: list[str] | None = None) -> None:
if line
]
from esphome.writer import ENTITY_TYPES_H_TARGET
from esphome.writer import ESPHOME_H_EXCLUDE
# X-macro files are included multiple times with different macro definitions
# and must not be included bare in the all-include header
exclude = {ENTITY_TYPES_H_TARGET}
exclude = {path.as_posix() for path in ESPHOME_H_EXCLUDE}
headers = [f'#include "{h}"' for h in header_files if h not in exclude]
headers.sort()
headers.append("")
@@ -1618,7 +1616,7 @@ def lpt_partition(
Heaviest item first into the lightest group. Ties keep input order, so
pass pre-sorted items for deterministic output. script/clang-tidy's
split_list is the unweighted contiguous sibling.
split_list is the unweighted sibling that deals files out in turn.
"""
groups: list[list[str]] = [[] for _ in range(count)]
group_weights = [0.0] * count
script/split_components_for_ci.py
+261 -61
View File
@@ -13,13 +13,20 @@ from __future__ import annotations
import argparse
from collections import defaultdict
from collections.abc import Set as AbstractSet
import contextlib
from dataclasses import dataclass, field
import functools
import json
import math
from pathlib import Path
import sys
from typing import Any, NamedTuple
# Add esphome to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from esphome import yaml_util
from script.analyze_component_buses import (
ISOLATED_COMPONENTS,
ISOLATED_SIGNATURE_PREFIX,
@@ -28,13 +35,57 @@ from script.analyze_component_buses import (
create_grouping_signature,
merge_compatible_bus_groups,
)
from script.helpers import get_component_test_files, split_conflicting_groups
from script.helpers import (
_extract_components_from_yaml,
get_component_test_files,
parse_test_filename,
split_conflicting_groups,
)
# Weighting for batch creation
# Isolated components can't be grouped/merged, so they count as 10x
# Groupable components can be merged into single builds, so they count as 1x
ISOLATED_WEIGHT = 10
GROUPABLE_WEIGHT = 1
# Estimated CI seconds for one build per test platform, fitted to the batch
# job logs of a full run (every component, PR #20389). Absolute accuracy does
# not matter, only the ratios between platforms. A full run takes over an
# hour, so recording real timings is not practical.
PLATFORM_BUILD_SECONDS = {
"host": 10,
"esp8266-ard": 30,
"rtl87xx-ard": 40,
"esp32-c3-idf": 40,
"esp32-s2-idf": 40,
"rp2350-ard": 40,
"rp2040-ard": 45,
"esp32-idf": 55,
"esp32-p4-idf": 55,
"ln882x-ard": 55,
"esp32-s3-idf": 60,
"bk72xx-ard": 70,
"esp32-c6-idf": 70,
}
NRF52_BUILD_SECONDS = 50
# Arduino on any ESP32 chip builds the IDF plus the Arduino core on top
ESP32_ARDUINO_BUILD_SECONDS = 90
# The remaining ESP32 chips on ESP-IDF
DEFAULT_BUILD_SECONDS = 50
# Each extra component merged into a grouped build makes it larger
GROUPED_COMPONENT_SECONDS = 3
# Components that pull in a large library the build compiles from source
# (TensorFlow Lite Micro, FastLED, LVGL, the HTTP server), added to every
# build whose test config enables them. Fitted the same way as the platforms.
HEAVY_COMPONENT_SECONDS: dict[str, int] = {
"micro_wake_word": 200,
"fastled_clockless": 90,
"fastled_spi": 90,
"wled": 60,
"lvgl": 50,
"web_server": 50,
"online_image": 45,
"prometheus": 35,
"bluetooth_proxy": 30,
}
# Estimated build seconds per CI runner; sets how many runners are used.
# Approximate: the runner count charges each grouped build once, but a large
# group spreads over several runners, which each pay for that build.
TARGET_BATCH_SECONDS = 600
# Platform used for batching (platform-agnostic batching)
# Batches are split across CI runners and each runner tests all platforms
@@ -61,10 +112,193 @@ def has_test_files(component_name: str, tests_dir: Path) -> bool:
)
def build_seconds(platform: str) -> int:
"""Return the estimated CI seconds for one build on a test platform."""
if (seconds := PLATFORM_BUILD_SECONDS.get(platform)) is not None:
return seconds
if platform.startswith("nrf52"):
return NRF52_BUILD_SECONDS
if platform.startswith("esp32") and platform.endswith("-ard"):
return ESP32_ARDUINO_BUILD_SECONDS
return DEFAULT_BUILD_SECONDS
def _config_components(data: Any) -> set[str]:
"""Return the components a loaded test config and its ``packages:`` use."""
if not isinstance(data, dict):
return set()
found = _extract_components_from_yaml(data)
if isinstance(packages := data.get("packages"), dict):
for package in packages.values():
if isinstance(package, yaml_util.IncludeFile):
found |= _file_components(
(package.parent_file.parent / package.file).resolve()
)
else:
found |= _config_components(package)
return found
@functools.cache
def _file_components(path: Path) -> frozenset[str]:
"""Return the components a config file uses; shared packages load once."""
return frozenset(
_config_components(yaml_util.load_yaml(path, track_document_range=False))
)
def heavy_components(component: str, test_file: Path) -> frozenset[str]:
"""Return the heavy components a test compiles.
A test counts as heavy when it belongs to a heavy component or when its
config enables one, such as a display test that draws with lvgl.
"""
heavy = {component} & HEAVY_COMPONENT_SECONDS.keys()
# A config that does not load is left to the build to report
with contextlib.suppress(Exception):
heavy |= _file_components(test_file.resolve()) & HEAVY_COMPONENT_SECONDS.keys()
return frozenset(heavy)
def heavy_seconds(heavy: AbstractSet[str]) -> int:
"""Return the library seconds a set of heavy components adds to a build."""
return sum(HEAVY_COMPONENT_SECONDS[name] for name in heavy)
class _GroupedShare(NamedTuple):
"""A component's part in one (signature, platform) shared build."""
# Seconds of its files on that platform when no other member shares it
solo: int
# Heavy components its base test brings into the shared build
heavy: frozenset[str]
@dataclass
class _BatchItem:
"""A component and the builds it adds to the batch it lands in."""
component: str
# Builds that always run on their own (isolated tests, variants on a
# platform without a base test)
own_seconds: int
grouped_builds: dict[tuple[str, str], _GroupedShare] = field(default_factory=dict)
def standalone_seconds(self) -> int:
return self.own_seconds + sum(
share.solo for share in self.grouped_builds.values()
)
def _grouped_build_seconds(platform: str, members: list[_GroupedShare]) -> int:
"""Return the seconds of one (signature, platform) build in a batch.
A library two members both enable compiles once, so the union is charged.
test_build_components only groups when two or more members share the
build; a grouped member then skips its variants on that platform.
"""
if len(members) <= 1:
return sum(share.solo for share in members)
return (
build_seconds(platform)
+ GROUPED_COMPONENT_SECONDS * (len(members) - 1)
+ heavy_seconds(set().union(*(share.heavy for share in members)))
)
def _make_item(
tests_dir: Path, component: str, signature: str, is_isolated: bool
) -> _BatchItem:
# Per platform: the seconds of its test files, and the base test's heavy
# components; a platform without a base test cannot join a shared build
seconds_by_platform: dict[str, int] = defaultdict(int)
base_heavy: dict[str, frozenset[str]] = {}
for test_file in (tests_dir / component).glob("test[.-]*.yaml"):
test_name, platform = parse_test_filename(test_file)
heavy = heavy_components(component, test_file)
seconds_by_platform[platform] += build_seconds(platform) + heavy_seconds(heavy)
if test_name == "test":
base_heavy[platform] = heavy
own_seconds = 0
grouped: dict[tuple[str, str], _GroupedShare] = {}
for platform, seconds in seconds_by_platform.items():
if is_isolated or platform not in base_heavy:
own_seconds += seconds
else:
grouped[(signature, platform)] = _GroupedShare(
seconds, base_heavy[platform]
)
return _BatchItem(component, own_seconds, grouped)
class _Batch:
"""A CI runner's components and its estimated build seconds."""
def __init__(self) -> None:
self.components: list[str] = []
self.seconds = 0
self.grouped_builds: dict[tuple[str, str], list[_GroupedShare]] = defaultdict(
list
)
def added_seconds(self, item: _BatchItem) -> int:
"""Return the seconds item would add; joining an existing build is cheap."""
added = item.own_seconds
for (signature, platform), share in item.grouped_builds.items():
members = self.grouped_builds.get((signature, platform), [])
added += _grouped_build_seconds(
platform, [*members, share]
) - _grouped_build_seconds(platform, members)
return added
def add(self, item: _BatchItem) -> None:
self.seconds += self.added_seconds(item)
for build, share in item.grouped_builds.items():
self.grouped_builds[build].append(share)
self.components.append(item.component)
def balance_batches(items: list[_BatchItem], target_seconds: int) -> list[list[str]]:
"""Spread items over enough runners to stay near target_seconds each.
The runner count comes from the total estimate with every grouped build
counted once. Heaviest items go first, each to the runner it adds the
fewest seconds to among those ending near the lightest, so grouped
components follow the builds they can join.
"""
if not items:
return []
# One batch holding every item counts each grouped build once
whole = _Batch()
for item in items:
whole.add(item)
total = whole.seconds
count = min(len(items), max(1, math.ceil(total / target_seconds)))
batches = [_Batch() for _ in range(count)]
# Among the runners ending within slack of the lightest, the one the item
# adds the fewest seconds to wins, so a component lands with the bus
# group partners it shares builds with; placed away from them it builds
# alone on every platform they only share there.
slack = target_seconds // 4
for item in sorted(items, key=lambda i: (-i.standalone_seconds(), i.component)):
costs = [(batch, batch.added_seconds(item)) for batch in batches]
ceiling = min(batch.seconds + added for batch, added in costs) + slack
batch, _ = min(
(
(batch, added)
for batch, added in costs
if batch.seconds + added <= ceiling
),
key=lambda cost: (cost[1], cost[0].seconds + cost[1]),
)
batch.add(item)
return [batch.components for batch in batches if batch.components]
def create_intelligent_batches(
components: list[str],
tests_dir: Path,
batch_size: int = 40,
target_seconds: int = TARGET_BATCH_SECONDS,
directly_changed: set[str] | None = None,
) -> tuple[list[list[str]], dict[tuple[str, str], list[str]]]:
"""Create batches optimized for component grouping.
@@ -76,7 +310,7 @@ def create_intelligent_batches(
Args:
components: List of component names to batch
tests_dir: Path to tests/components directory
batch_size: Target size for each batch
target_seconds: Estimated build seconds per batch
directly_changed: Set of directly changed components (for logging only)
Returns:
@@ -157,51 +391,17 @@ def create_intelligent_batches(
# actually be split into two at build time -- throwing off CI distribution.
signature_groups = split_conflicting_groups(signature_groups)
# Create batches by keeping signature groups together
# Components with the same signature stay in the same batches
batches = []
# Sort signature groups to prioritize groupable components
# 1. Put "isolated_*" signatures last (can't be grouped with others)
# 2. Sort groupable signatures by size (largest first)
# 3. "no_buses" components CAN be grouped together
def sort_key(item):
(_platform, signature), components = item
is_isolated = signature.startswith(ISOLATED_SIGNATURE_PREFIX)
# Put "isolated_*" last (1), groupable first (0)
# Within each category, sort by size (largest first)
return (is_isolated, -len(components))
sorted_groups = sorted(signature_groups.items(), key=sort_key)
# Strategy: Create batches using weighted sizes
# - Isolated components count as 10x (since they can't be grouped/merged)
# - Groupable components count as 1x (can be merged into single builds)
# - This distributes isolated components across more runners
# - Ensures each runner has a good mix of groupable vs isolated components
current_batch = []
current_weight = 0
for (_platform, signature), group_components in sorted_groups:
is_isolated = signature.startswith(ISOLATED_SIGNATURE_PREFIX)
weight_per_component = ISOLATED_WEIGHT if is_isolated else GROUPABLE_WEIGHT
for component in group_components:
# Check if adding this component would exceed the batch size
if current_weight + weight_per_component > batch_size and current_batch:
# Start a new batch
batches.append(current_batch)
current_batch = []
current_weight = 0
# Add component to current batch
current_batch.append(component)
current_weight += weight_per_component
# Don't forget the last batch
if current_batch:
batches.append(current_batch)
items = [
_make_item(
tests_dir,
component,
signature,
signature.startswith(ISOLATED_SIGNATURE_PREFIX),
)
for (_platform, signature), group_components in sorted(signature_groups.items())
for component in group_components
]
batches = balance_batches(items, target_seconds)
return batches, signature_groups
@@ -218,11 +418,11 @@ def main() -> int:
help="JSON array of component names",
)
parser.add_argument(
"--batch-size",
"-b",
"--target-seconds",
"-t",
type=int,
default=40,
help="Target batch size (default: 40, weighted)",
default=TARGET_BATCH_SECONDS,
help=f"Estimated build seconds per batch (default: {TARGET_BATCH_SECONDS})",
)
parser.add_argument(
"--tests-dir",
@@ -268,7 +468,7 @@ def main() -> int:
batches, signature_groups = create_intelligent_batches(
components=components,
tests_dir=args.tests_dir,
batch_size=args.batch_size,
target_seconds=args.target_seconds,
directly_changed=directly_changed,
)
@@ -378,15 +578,15 @@ def main() -> int:
file=sys.stderr,
)
print(f" - Groupable (weight=1): {groupable_count}", file=sys.stderr)
print(f" - Isolated (weight=10): {isolated_count}", file=sys.stderr)
print(f" - Groupable: {groupable_count}", file=sys.stderr)
print(f" - Isolated: {isolated_count}", file=sys.stderr)
if actual_components < len(components):
print(
f"Components skipped (no test files): {len(components) - actual_components}",
file=sys.stderr,
)
print(f"Number of batches: {len(batches)}", file=sys.stderr)
print(f"Batch size target (weighted): {args.batch_size}", file=sys.stderr)
print(f"Target build seconds per batch: {args.target_seconds}", file=sys.stderr)
if len(batches) > 0:
print(
f"Average components per batch: {actual_components / len(batches):.1f}",
script/test_build_components.py
+76 -9
View File
@@ -21,6 +21,7 @@ from dataclasses import dataclass
import hashlib
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time
@@ -83,7 +84,9 @@ def show_disk_space_if_ci(esphome_command: str) -> None:
print("=" * 80)
# Use sys.stdout.flush() to ensure output appears immediately
sys.stdout.flush()
subprocess.run(["df", "-h"], check=False, stdout=sys.stdout, stderr=sys.stderr)
# Windows has no df
if df := shutil.which("df"):
subprocess.run([df, "-h"], check=False, stdout=sys.stdout, stderr=sys.stderr)
print("=" * 80 + "\n")
sys.stdout.flush()
@@ -333,6 +336,12 @@ def extract_platform_with_version(base_file: Path) -> str:
return base_file.stem.replace("build_components_base.", "")
def _wants_skip_bootloader(skip: bool, command: str, platform: str) -> bool:
"""Every esp32 target: idf variants, and esp32-ard, whose Arduino
core builds as an ESP-IDF component under the native toolchain."""
return skip and command == "compile" and platform.startswith("esp32")
def run_esphome_test(
component: str,
test_file: Path,
@@ -344,6 +353,7 @@ def run_esphome_test(
continue_on_fail: bool,
use_testing_mode: bool = False,
toolchain: str | None = None,
skip_bootloader: bool = False,
) -> TestResult:
"""Run esphome test for a single component.
@@ -408,6 +418,8 @@ def run_esphome_test(
# Add command
cmd.append(esphome_command)
if _wants_skip_bootloader(skip_bootloader, esphome_command, platform):
cmd.append("--skip-bootloader")
# Add config file
cmd.append(str(output_file))
@@ -469,6 +481,7 @@ def run_grouped_test(
esphome_command: str,
continue_on_fail: bool,
toolchain: str | None = None,
skip_bootloader: bool = False,
) -> TestResult:
"""Run esphome test for a group of components with shared bus configs.
@@ -554,6 +567,8 @@ def run_grouped_test(
# Add command
cmd.append(esphome_command)
if _wants_skip_bootloader(skip_bootloader, esphome_command, platform):
cmd.append("--skip-bootloader")
cmd.append(str(output_file))
@@ -614,6 +629,7 @@ def run_grouped_component_tests(
continue_on_fail: bool,
additional_isolated: set[str] | None = None,
toolchain: str | None = None,
skip_bootloader: bool = False,
) -> tuple[set[tuple[str, str]], list[TestResult]]:
"""Run grouped component tests.
@@ -956,6 +972,7 @@ def run_grouped_component_tests(
esphome_command=esphome_command,
continue_on_fail=continue_on_fail,
toolchain=toolchain,
skip_bootloader=skip_bootloader,
)
# Mark all components as tested
@@ -980,6 +997,7 @@ def run_individual_component_test(
tested_components: set[tuple[str, str]],
test_results: list[TestResult],
toolchain: str | None = None,
skip_bootloader: bool = False,
) -> None:
"""Run an individual component test if not already tested in a group.
@@ -1014,6 +1032,7 @@ def run_individual_component_test(
esphome_command=esphome_command,
continue_on_fail=continue_on_fail,
toolchain=toolchain,
skip_bootloader=skip_bootloader,
)
test_results.append(test_result)
@@ -1027,6 +1046,8 @@ def test_components(
isolated_components: set[str] | None = None,
base_only: bool = False,
toolchain: str | None = None,
skip_bootloader: bool = False,
fail_on_no_tests: bool = False,
) -> int:
"""Test components with optional intelligent grouping.
@@ -1061,20 +1082,32 @@ def test_components(
# toolchain build.
include_validate = esphome_command != "compile"
# Find all component tests
# A blank pattern list would slide into the reference-baseline
# fallback and exit green while building nothing
if fail_on_no_tests and not any(component_patterns):
print("No components requested (blank component list)")
return 1
# Find all component tests; remember which components each pattern
# (wildcards included) matched, for the deferred no-tests accounting
all_tests = {}
pattern_components: dict[str, set[str]] = {}
for pattern in component_patterns:
# Skip empty patterns (happens when components list is empty string)
if not pattern:
continue
all_tests.update(
find_component_tests(
tests_dir, pattern, base_only, include_validate=include_validate
)
found = find_component_tests(
tests_dir, pattern, base_only, include_validate=include_validate
)
pattern_components[pattern] = set(found)
all_tests.update(found)
if fail_on_no_tests and not all_tests:
# Nothing matched: fail before the synthetic baseline spends a
# compile reporting success on nothing
print(f"No components found matching: {component_patterns}")
return 1
# If no components found, build a reference configuration for baseline comparison
# Create a synthetic "empty" component test that will build just the base config
if not all_tests:
print(f"No components found matching: {component_patterns}")
print(
@@ -1114,6 +1147,7 @@ def test_components(
continue_on_fail=continue_on_fail,
additional_isolated=isolated_components,
toolchain=toolchain,
skip_bootloader=skip_bootloader,
)
test_results.extend(grouped_results)
@@ -1143,6 +1177,7 @@ def test_components(
tested_components=tested_components,
test_results=test_results,
toolchain=toolchain,
skip_bootloader=skip_bootloader,
)
else:
# Platform-specific test
@@ -1176,8 +1211,26 @@ def test_components(
tested_components=tested_components,
test_results=test_results,
toolchain=toolchain,
skip_bootloader=skip_bootloader,
)
silent: list[str] = []
if fail_on_no_tests:
# A green run that built nothing for a requested pattern must not
# pass CI. Per pattern so one silent pattern cannot hide behind
# the others; opt-in because some legs legitimately match nothing;
# deferred past the summary so reproduce commands still print.
built = {c for r in test_results for c in r.components}
# A pattern is silent when it matched no fixture, or when none of
# its matched components produced a build (wildcards included)
silent = [
p
for p in component_patterns
if p and not (pattern_components.get(p, set()) & built)
]
if silent:
print(f"No tests ran for requested pattern(s): {', '.join(silent)}")
# Separate results into passed and failed
passed_results = [r for r in test_results if r.success]
failed_results = [r for r in test_results if not r.success]
@@ -1209,7 +1262,7 @@ def test_components(
if os.environ.get("GITHUB_STEP_SUMMARY"):
write_github_summary(test_results, toolchain=toolchain)
if failed_results:
if failed_results or silent:
return 1
return 0
@@ -1264,6 +1317,18 @@ def main() -> int:
"--toolchain",
help="Select toolchain for compiling.",
)
parser.add_argument(
"--skip-bootloader",
action="store_true",
help="Pass --skip-bootloader to esphome compile; component builds "
"never flash, and the bootloader is covered by the toolchain jobs",
)
parser.add_argument(
"--fail-on-no-tests",
action="store_true",
help="Exit non-zero when no test matched (for CI legs whose "
"components must all have fixtures)",
)
args = parser.parse_args()
@@ -1282,8 +1347,10 @@ def main() -> int:
continue_on_fail=args.continue_on_fail,
enable_grouping=not args.no_grouping,
isolated_components=isolated_components,
fail_on_no_tests=args.fail_on_no_tests,
base_only=args.base_only,
toolchain=args.toolchain,
skip_bootloader=args.skip_bootloader,
)