Merge remote-tracking branch 'upstream/sensor/throttle-with-priority-nan-specialize' into integration

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/split_components_for_ci.py

@@ -28,7 +28,7 @@ from script.analyze_component_buses import (
     create_grouping_signature,
     merge_compatible_bus_groups,
 )
-from script.helpers import get_component_test_files
+from script.helpers import get_component_test_files, split_conflicting_groups
 
 # Weighting for batch creation
 # Isolated components can't be grouped/merged, so they count as 10x
@@ -145,6 +145,11 @@ def create_intelligent_batches(
     # improving the efficiency of test_build_components.py grouping
     signature_groups = merge_compatible_bus_groups(signature_groups)
 
+    # Split groups containing mutually-incompatible components (CONFLICTS_WITH).
+    # Without this, batch weighting assumes the group is one build when it will
+    # 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 = []

script/test_build_components.py

@@ -39,7 +39,7 @@ from script.analyze_component_buses import (
     merge_compatible_bus_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
 
 
@@ -675,6 +675,13 @@ def run_grouped_component_tests(
     # as long as they don't have conflicting configurations for the same bus type
     grouped_components = merge_compatible_bus_groups(grouped_components)
 
+    # Split groups that contain components declaring CONFLICTS_WITH each other.
+    # The bus-level merge above only considers shared bus configs; components
+    # with the same bus signature (e.g. both I2C) can still be mutually
+    # incompatible (e.g. bme680_bsec vs. bme68x_bsec2_i2c which auto-loads
+    # bme68x_bsec2). Those must end up in separate builds.
+    grouped_components = split_conflicting_groups(grouped_components)
+
     # Print detailed grouping plan
     print("\nGrouping Plan:")
     print("-" * 80)