[ci] Move conflict-aware grouping helpers into script/helpers.py with tests

script/analyze_component_buses.py

@@ -24,8 +24,7 @@ Example output:
 from __future__ import annotations
 
 import argparse
-import ast
-from functools import cache, lru_cache
+from functools import lru_cache
 import json
 from pathlib import Path
 import re
@@ -43,9 +42,6 @@ from esphome.config_helpers import Extend, Remove
 # Path to common bus configs
 COMMON_BUS_PATH = Path("tests/test_build_components/common")
 
-# Path to esphome component sources (used to parse AUTO_LOAD / CONFLICTS_WITH metadata)
-COMPONENTS_SRC_PATH = Path("esphome/components")
-
 # Package dependencies - maps packages to the packages they include
 # When a component uses a package on the left, it automatically gets
 # the packages on the right as well
@@ -515,86 +511,6 @@ def merge_compatible_bus_groups(
     return merged_groups
 
 
-@cache
-def _parse_component_init(name: str) -> tuple[frozenset[str], frozenset[str]]:
-    """Return (AUTO_LOAD, CONFLICTS_WITH) list-literal values for a single component."""
-    init_file = COMPONENTS_SRC_PATH / name / "__init__.py"
-    result: dict[str, frozenset[str]] = {
-        "AUTO_LOAD": frozenset(),
-        "CONFLICTS_WITH": frozenset(),
-    }
-    if not init_file.exists():
-        return result["AUTO_LOAD"], result["CONFLICTS_WITH"]
-    try:
-        tree = ast.parse(init_file.read_text())
-    except (OSError, SyntaxError):
-        return result["AUTO_LOAD"], result["CONFLICTS_WITH"]
-    for node in tree.body:
-        if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.List):
-            continue
-        for target in node.targets:
-            if not isinstance(target, ast.Name) or target.id not in result:
-                continue
-            result[target.id] = frozenset(
-                e.value
-                for e in node.value.elts
-                if isinstance(e, ast.Constant) and isinstance(e.value, str)
-            )
-    return result["AUTO_LOAD"], result["CONFLICTS_WITH"]
-
-
-def split_conflicting_groups(
-    grouped_components: dict[tuple[str, str], list[str]],
-) -> dict[tuple[str, str], list[str]]:
-    """Split groups so that components declaring mutual CONFLICTS_WITH end up in separate builds.
-
-    A conflict propagates through AUTO_LOAD: if X declares CONFLICTS_WITH=[Y]
-    and Z auto-loads Y, then X and Z conflict (e.g. bme680_bsec vs.
-    bme68x_bsec2_i2c which auto-loads bme68x_bsec2). Only components that
-    appear in the batch (and their AUTO_LOAD closures) are parsed.
-    """
-    batch = {c for comps in grouped_components.values() for c in comps}
-
-    # Walk each batch component's AUTO_LOAD chain once, collecting the full
-    # loaded-alongside set and the names it rejects via CONFLICTS_WITH. Two
-    # components conflict when either one's rejects intersect the other's
-    # loaded set (relation must be symmetric -- e.g. ethernet rejects wifi
-    # but wifi does not reject ethernet).
-    walks: dict[str, tuple[set[str], set[str]]] = {}
-    for comp in batch:
-        loaded, rejects, stack = {comp}, set(), [comp]
-        while stack:
-            auto_load, conflicts_with = _parse_component_init(stack.pop())
-            rejects |= conflicts_with
-            new = auto_load - loaded
-            loaded |= new
-            stack.extend(new)
-        walks[comp] = (loaded, rejects)
-
-    def conflicts(a: str, b: str) -> bool:
-        loaded_a, rejects_a = walks[a]
-        loaded_b, rejects_b = walks[b]
-        return not rejects_a.isdisjoint(loaded_b) or not rejects_b.isdisjoint(loaded_a)
-
-    result: dict[tuple[str, str], list[str]] = {}
-    for (platform, signature), components in grouped_components.items():
-        buckets: list[list[str]] = []
-        for comp in components:
-            for bucket in buckets:
-                if not any(conflicts(comp, other) for other in bucket):
-                    bucket.append(comp)
-                    break
-            else:
-                buckets.append([comp])
-        if len(buckets) == 1:
-            result[(platform, signature)] = buckets[0]
-            continue
-        for index, bucket in enumerate(buckets):
-            key = signature if index == 0 else f"{signature}__conflict{index}"
-            result[(platform, key)] = bucket
-    return result
-
-
 def create_grouping_signature(
     platform_buses: dict[str, list[str]], platform: str
 ) -> str:

script/helpers.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import ast
 from collections.abc import Callable
+from dataclasses import dataclass, field
 from functools import cache
 import hashlib
 import json
@@ -139,6 +141,109 @@ def get_component_test_files(
     return list(tests_dir.glob("test.*.yaml"))
 
 
+@dataclass(frozen=True)
+class ComponentMetadata:
+    """Statically-parsed AUTO_LOAD and CONFLICTS_WITH declarations."""
+
+    auto_load: frozenset[str] = field(default_factory=frozenset)
+    conflicts_with: frozenset[str] = field(default_factory=frozenset)
+
+
+@cache
+def parse_component_metadata(name: str) -> ComponentMetadata:
+    """Return the AUTO_LOAD / CONFLICTS_WITH declarations for a component.
+
+    Parses the component's ``esphome/components/<name>/__init__.py`` statically.
+    Callable forms (``def AUTO_LOAD():``) require runtime imports and are
+    reported as empty -- safe for conflict detection since they cannot be
+    evaluated without executing the module.
+    """
+    init_file = Path(root_path) / ESPHOME_COMPONENTS_PATH / name / "__init__.py"
+    if not init_file.exists():
+        return ComponentMetadata()
+    try:
+        tree = ast.parse(init_file.read_text())
+    except (OSError, SyntaxError):
+        return ComponentMetadata()
+    fields: dict[str, frozenset[str]] = {
+        "AUTO_LOAD": frozenset(),
+        "CONFLICTS_WITH": frozenset(),
+    }
+    for node in tree.body:
+        if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.List):
+            continue
+        for target in node.targets:
+            if not isinstance(target, ast.Name) or target.id not in fields:
+                continue
+            fields[target.id] = frozenset(
+                e.value
+                for e in node.value.elts
+                if isinstance(e, ast.Constant) and isinstance(e.value, str)
+            )
+    return ComponentMetadata(
+        auto_load=fields["AUTO_LOAD"],
+        conflicts_with=fields["CONFLICTS_WITH"],
+    )
+
+
+@dataclass
+class _ConflictWalk:
+    loaded: set[str]
+    rejects: set[str]
+
+
+def split_conflicting_groups(
+    grouped_components: dict[tuple[str, str], list[str]],
+) -> dict[tuple[str, str], list[str]]:
+    """Split groups so components declaring mutual CONFLICTS_WITH end up in separate builds.
+
+    A conflict propagates through AUTO_LOAD: if X declares CONFLICTS_WITH=[Y]
+    and Z auto-loads Y, then X and Z conflict (e.g. bme680_bsec vs.
+    bme68x_bsec2_i2c which auto-loads bme68x_bsec2). Only components that
+    appear in the batch (and their AUTO_LOAD closures) are parsed. The
+    conflict relation is treated as symmetric even when only one side
+    declares it (e.g. ethernet rejects wifi but wifi does not declare the
+    reverse).
+    """
+    batch = {c for comps in grouped_components.values() for c in comps}
+
+    walks: dict[str, _ConflictWalk] = {}
+    for comp in batch:
+        walk = _ConflictWalk(loaded={comp}, rejects=set())
+        stack = [comp]
+        while stack:
+            metadata = parse_component_metadata(stack.pop())
+            walk.rejects |= metadata.conflicts_with
+            new = metadata.auto_load - walk.loaded
+            walk.loaded |= new
+            stack.extend(new)
+        walks[comp] = walk
+
+    def conflicts(a: str, b: str) -> bool:
+        wa, wb = walks[a], walks[b]
+        return not wa.rejects.isdisjoint(wb.loaded) or not wb.rejects.isdisjoint(
+            wa.loaded
+        )
+
+    result: dict[tuple[str, str], list[str]] = {}
+    for (platform, signature), components in grouped_components.items():
+        buckets: list[list[str]] = []
+        for comp in components:
+            for bucket in buckets:
+                if not any(conflicts(comp, other) for other in bucket):
+                    bucket.append(comp)
+                    break
+            else:
+                buckets.append([comp])
+        if len(buckets) == 1:
+            result[(platform, signature)] = buckets[0]
+            continue
+        for index, bucket in enumerate(buckets):
+            key = signature if index == 0 else f"{signature}__conflict{index}"
+            result[(platform, key)] = bucket
+    return result
+
+
 def styled(color: str | tuple[str, ...], msg: str, reset: bool = True) -> str:
     prefix = "".join(color) if isinstance(color, tuple) else color
     suffix = colorama.Style.RESET_ALL if reset else ""

script/test_build_components.py

@@ -37,10 +37,9 @@ from script.analyze_component_buses import (
     create_grouping_signature,
     is_platform_component,
     merge_compatible_bus_groups,
-    split_conflicting_groups,
     uses_local_file_references,
 )
-from script.helpers import get_component_test_files
+from script.helpers import get_component_test_files, split_conflicting_groups
 from script.merge_component_configs import merge_component_configs