[ci] Add import-time regression check for esphome.__main__

Adds a CI gate that runs `python -X importtime -c "import esphome.__main__"`
via importtime-waterfall's best-of-N harness, compares the root cumulative
time against a checked-in budget (script/import_time_budget.json, seeded
at 75.2ms with a 15% margin), and fails the build when top-level imports
regress.

The CLI pays this cost on every invocation before the requested command
even runs, so silently adding a top-level dep chain (the recent zeroconf
move in #13135 being the motivating case) hurts every user. The check
gives us a signal without waiting for user reports.

- script/check_import_time.py: --check / --update / --har PATH modes.
  On regression, prints a ranked top-15 offenders table by self-time.
- script/import_time_budget.json: baseline + margin_pct.
- script/determine-jobs.py: should_run_import_time() gates the job on
  esphome/**/*.py, requirements.txt, requirements_dev.txt, pyproject.toml,
  or changes to the check itself.
- .github/workflows/ci.yml: new import-time job, runs when gated and
  uploads a waterfall HAR artifact (14-day retention) for inspection.
This commit is contained in:
J. Nick Koston
2026-04-23 14:13:01 -05:00
parent f757cd1210
commit 0dd2a57ba4
4 changed files with 340 additions and 0 deletions
+38
View File
@@ -108,6 +108,42 @@ jobs:
script/generate-esp32-boards.py --check
script/generate-rp2040-boards.py --check
import-time:
name: Check import esphome.__main__ time
runs-on: ubuntu-24.04
needs:
- common
- determine-jobs
if: needs.determine-jobs.outputs.import-time == 'true'
steps:
- name: Check out code from GitHub
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
- name: Restore Python
uses: ./.github/actions/restore-python
with:
python-version: ${{ env.DEFAULT_PYTHON }}
cache-key: ${{ needs.common.outputs.cache-key }}
- name: Install importtime-waterfall
run: |
. venv/bin/activate
pip install importtime-waterfall==1.0.0
- name: Check import time against budget
run: |
. venv/bin/activate
script/check_import_time.py --check
- name: Generate waterfall HAR
if: always()
run: |
. venv/bin/activate
script/check_import_time.py --har importtime.har
- name: Upload waterfall HAR
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: import-time-waterfall
path: importtime.har
retention-days: 14
pytest:
name: Run pytest
strategy:
@@ -176,6 +212,7 @@ jobs:
clang-tidy: ${{ steps.determine.outputs.clang-tidy }}
clang-tidy-mode: ${{ steps.determine.outputs.clang-tidy-mode }}
python-linters: ${{ steps.determine.outputs.python-linters }}
import-time: ${{ steps.determine.outputs.import-time }}
changed-components: ${{ steps.determine.outputs.changed-components }}
changed-components-with-tests: ${{ steps.determine.outputs.changed-components-with-tests }}
directly-changed-components-with-tests: ${{ steps.determine.outputs.directly-changed-components-with-tests }}
@@ -219,6 +256,7 @@ jobs:
echo "clang-tidy=$(echo "$output" | jq -r '.clang_tidy')" >> $GITHUB_OUTPUT
echo "clang-tidy-mode=$(echo "$output" | jq -r '.clang_tidy_mode')" >> $GITHUB_OUTPUT
echo "python-linters=$(echo "$output" | jq -r '.python_linters')" >> $GITHUB_OUTPUT
echo "import-time=$(echo "$output" | jq -r '.import_time')" >> $GITHUB_OUTPUT
echo "changed-components=$(echo "$output" | jq -c '.changed_components')" >> $GITHUB_OUTPUT
echo "changed-components-with-tests=$(echo "$output" | jq -c '.changed_components_with_tests')" >> $GITHUB_OUTPUT
echo "directly-changed-components-with-tests=$(echo "$output" | jq -c '.directly_changed_components_with_tests')" >> $GITHUB_OUTPUT
+260
View File
@@ -0,0 +1,260 @@
#!/usr/bin/env python3
"""Regression check for `import esphome.__main__` cost.
Runs `python -X importtime -c "import esphome.__main__"` in fresh subprocesses
and compares the root cumulative import time against a checked-in budget
(`script/import_time_budget.json`).
The CLI pays this cost on every invocation before the requested command even
runs, so a regression here hurts every user. Pair this with
`python -m importtime_waterfall --har` for human-readable waterfalls.
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import subprocess
import sys
import time
from typing import Any, TextIO
SCRIPT_DIR = Path(__file__).parent
BUDGET_PATH = SCRIPT_DIR / "import_time_budget.json"
TARGET_MODULE = "esphome.__main__"
DEFAULT_RUNS = 3
DEFAULT_MARGIN_PCT = 15
OFFENDERS_TOP_N = 15
IMPORT_TIME_PREFIX = "import time:"
def _run_importtime(module: str) -> tuple[float, str]:
"""Run `python -X importtime -c 'import <module>'` once.
Returns (wall_seconds, stderr_text).
"""
before = time.monotonic()
result = subprocess.run(
[sys.executable, "-X", "importtime", "-c", f"import {module}"],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
)
return time.monotonic() - before, result.stderr
def measure(module: str, runs: int) -> str:
"""Run `module` import `runs` times; return the fastest run's stderr."""
best_wall, best_stderr = _run_importtime(module)
for _ in range(runs - 1):
wall, stderr = _run_importtime(module)
if wall < best_wall:
best_wall, best_stderr = wall, stderr
return best_stderr
def parse_trace(stderr: str) -> list[tuple[int, int, int, str]]:
"""Parse `-X importtime` stderr into (depth, self_us, cumulative_us, name)."""
entries: list[tuple[int, int, int, str]] = []
for line in stderr.splitlines():
if not line.startswith(IMPORT_TIME_PREFIX):
continue
body = line[len(IMPORT_TIME_PREFIX) :]
parts = body.split("|")
if len(parts) != 3:
continue
try:
self_us = int(parts[0].strip())
cumulative_us = int(parts[1].strip())
except ValueError:
continue # header row
name_field = parts[2].rstrip("\n")
name = name_field.lstrip(" ")
depth = (len(name_field) - len(name)) // 2
entries.append((depth, self_us, cumulative_us, name))
return entries
def root_cumulative_us(entries: list[tuple[int, int, int, str]], module: str) -> int:
"""Return the cumulative import time of `module` (the root probe)."""
for depth, _self_us, cumulative_us, name in reversed(entries):
if depth == 0 and name == module:
return cumulative_us
raise RuntimeError(
f"Did not find a root-level trace entry for {module!r}. Is it importable?"
)
def top_offenders(
entries: list[tuple[int, int, int, str]], n: int
) -> list[tuple[str, int, int]]:
"""Return up to `n` modules with the highest self-time, deduped by name.
Returns list of (name, self_us, cumulative_us). A module imported from
multiple places is counted once, at its deepest-stacktrace occurrence
(same as what the user sees on the `--graph` output).
"""
seen: dict[str, tuple[int, int]] = {}
for _depth, self_us, cumulative_us, name in entries:
if name in seen:
continue
seen[name] = (self_us, cumulative_us)
ranked = sorted(
((name, self_us, cum_us) for name, (self_us, cum_us) in seen.items()),
key=lambda row: row[1],
reverse=True,
)
return ranked[:n]
def read_budget() -> dict[str, Any]:
if not BUDGET_PATH.exists():
return {}
with BUDGET_PATH.open() as f:
return json.load(f)
def write_budget(cumulative_us: int, margin_pct: int) -> None:
payload = {
"target_module": TARGET_MODULE,
"margin_pct": margin_pct,
"cumulative_us": cumulative_us,
}
with BUDGET_PATH.open("w") as f:
json.dump(payload, f, indent=2)
f.write("\n")
def _format_us(us: int) -> str:
if us >= 1000:
return f"{us / 1000:.1f}ms"
return f"{us}us"
def _print_offenders_table(
offenders: list[tuple[str, int, int]], stream: TextIO
) -> None:
name_w = max(len(name) for name, _, _ in offenders)
print(f"\n{'module':<{name_w}} {'self':>10} {'cumulative':>12}", file=stream)
print(f"{'-' * name_w} {'-' * 10} {'-' * 12}", file=stream)
for name, self_us, cum_us in offenders:
print(
f"{name:<{name_w}} {_format_us(self_us):>10} {_format_us(cum_us):>12}",
file=stream,
)
def cmd_check(args: argparse.Namespace) -> int:
budget = read_budget()
if not budget:
print(
f"ERROR: {BUDGET_PATH.name} missing. Run with --update first.",
file=sys.stderr,
)
return 2
stderr = measure(TARGET_MODULE, args.runs)
entries = parse_trace(stderr)
measured = root_cumulative_us(entries, TARGET_MODULE)
baseline = budget["cumulative_us"]
margin_pct = budget.get("margin_pct", DEFAULT_MARGIN_PCT)
ceiling = int(baseline * (1 + margin_pct / 100))
summary = (
f"measured {TARGET_MODULE}: {_format_us(measured)} "
f"(budget {_format_us(baseline)} + {margin_pct}% = {_format_us(ceiling)})"
)
if measured <= ceiling:
print(summary)
return 0
print(
f"REGRESSION: `import {TARGET_MODULE}` took {_format_us(measured)}, "
f"exceeding the budget of {_format_us(baseline)} + {margin_pct}% "
f"({_format_us(ceiling)}).\n"
f"Top import-time offenders (by self time):",
file=sys.stderr,
)
_print_offenders_table(top_offenders(entries, OFFENDERS_TOP_N), sys.stderr)
print(
"\nIf this regression is intentional, regenerate the budget with:\n"
" script/check_import_time.py --update\n"
"Otherwise, consider making the new import lazy "
"(import inside the function that uses it).",
file=sys.stderr,
)
return 1
def cmd_update(args: argparse.Namespace) -> int:
stderr = measure(TARGET_MODULE, args.runs)
entries = parse_trace(stderr)
measured = root_cumulative_us(entries, TARGET_MODULE)
write_budget(measured, args.margin_pct)
print(
f"Wrote {BUDGET_PATH.name}: "
f"{TARGET_MODULE}={_format_us(measured)} "
f"(margin {args.margin_pct}%)"
)
return 0
def cmd_har(args: argparse.Namespace) -> int:
out_path = Path(args.har)
result = subprocess.run(
[sys.executable, "-m", "importtime_waterfall", "--har", TARGET_MODULE],
check=True,
stdout=subprocess.PIPE,
text=True,
)
out_path.write_text(result.stdout)
print(f"Wrote waterfall HAR to {out_path}")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--runs",
type=int,
default=DEFAULT_RUNS,
help=f"Number of measurement runs (default: {DEFAULT_RUNS}, best-of-N).",
)
parser.add_argument(
"--margin-pct",
type=int,
default=DEFAULT_MARGIN_PCT,
help=(f"Margin over baseline for --update (default: {DEFAULT_MARGIN_PCT}%%)."),
)
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument(
"--check", action="store_true", help="Fail if measured time exceeds budget."
)
mode.add_argument(
"--update",
action="store_true",
help="Rewrite the budget from a fresh measurement.",
)
mode.add_argument(
"--har",
metavar="PATH",
help="Write a waterfall HAR file via `importtime_waterfall --har`.",
)
args = parser.parse_args()
if args.check:
return cmd_check(args)
if args.update:
return cmd_update(args)
if args.har:
return cmd_har(args)
return 2
if __name__ == "__main__":
sys.exit(main())
+37
View File
@@ -349,6 +349,41 @@ def should_run_python_linters(branch: str | None = None) -> bool:
return _any_changed_file_endswith(branch, PYTHON_FILE_EXTENSIONS)
# Files outside esphome/**/*.py whose changes can affect `import esphome.__main__`
# cost. requirements.txt / pyproject.toml change the dependency graph pulled in
# by top-level imports; check_import_time.py itself changes the check's behavior.
IMPORT_TIME_TRIGGER_FILES = frozenset(
{
"requirements.txt",
"requirements_dev.txt",
"pyproject.toml",
"script/check_import_time.py",
"script/import_time_budget.json",
}
)
def should_run_import_time(branch: str | None = None) -> bool:
"""Determine if the `import esphome.__main__` time regression check should run.
Runs when any Python file under `esphome/` changes (those modules are
loaded transitively from `esphome.__main__`), when dependency
declarations change, or when the check script/budget itself changes.
Args:
branch: Branch to compare against. If None, uses default.
Returns:
True if the import-time check should run, False otherwise.
"""
for file in changed_files(branch):
if file.startswith("esphome/") and file.endswith(PYTHON_FILE_EXTENSIONS):
return True
if file in IMPORT_TIME_TRIGGER_FILES:
return True
return False
def determine_cpp_unit_tests(
branch: str | None = None,
) -> tuple[bool, list[str]]:
@@ -773,6 +808,7 @@ def main() -> None:
run_clang_tidy = should_run_clang_tidy(args.branch)
run_clang_format = should_run_clang_format(args.branch)
run_python_linters = should_run_python_linters(args.branch)
run_import_time = should_run_import_time(args.branch)
changed_cpp_file_count = count_changed_cpp_files(args.branch)
# Get changed components
@@ -906,6 +942,7 @@ def main() -> None:
"clang_tidy_mode": clang_tidy_mode,
"clang_format": run_clang_format,
"python_linters": run_python_linters,
"import_time": run_import_time,
"changed_components": changed_components,
"changed_components_with_tests": changed_components_with_tests,
"directly_changed_components_with_tests": list(directly_changed_with_tests),
+5
View File
@@ -0,0 +1,5 @@
{
"target_module": "esphome.__main__",
"margin_pct": 15,
"cumulative_us": 75199
}