esphome/script/merge_component_configs.py

#!/usr/bin/env python3
"""Merge multiple component test configurations into a single test file.

This script combines multiple component test files that use the same common bus
configurations into a single merged test file. This allows testing multiple
compatible components together, reducing CI build time.

The merger handles:
- Component-specific substitutions (prefixing to avoid conflicts)
- Multiple instances of component configurations
- Shared common bus packages (included only once)
- Platform-specific configurations
- Uses ESPHome's built-in merge_config for proper YAML merging
"""

from __future__ import annotations

import argparse
from functools import lru_cache
from pathlib import Path
import re
import sys
from typing import Any

# Add esphome to path so we can import from it
sys.path.insert(0, str(Path(__file__).parent.parent))

from esphome import yaml_util
from esphome.config_helpers import merge_config
from script.analyze_component_buses import PACKAGE_DEPENDENCIES, get_common_bus_packages

# Prefix for dependency markers in package tracking
# Used to mark packages that are included transitively (e.g., uart via modbus)
DEPENDENCY_MARKER_PREFIX = "_dep_"


def load_yaml_file(yaml_file: Path) -> dict:
    """Load YAML file using ESPHome's YAML loader.

    Args:
        yaml_file: Path to the YAML file

    Returns:
        Parsed YAML as dictionary
    """
    if not yaml_file.exists():
        raise FileNotFoundError(f"YAML file not found: {yaml_file}")

    return yaml_util.load_yaml(yaml_file)


@lru_cache(maxsize=256)
def get_component_packages(
    component_name: str, platform: str, tests_dir_str: str
) -> dict:
    """Get packages dict from a component's test file with caching.

    This function is cached to avoid re-loading and re-parsing the same file
    multiple times when extracting packages during cross-bus merging.

    Args:
        component_name: Name of the component
        platform: Platform name (e.g., "esp32-idf")
        tests_dir_str: String path to tests/components directory (must be string for cache hashability)

    Returns:
        Dictionary with 'packages' key containing the raw packages dict from the YAML,
        or empty dict if no packages section exists
    """
    tests_dir = Path(tests_dir_str)
    test_file = tests_dir / component_name / f"test.{platform}.yaml"
    comp_data = load_yaml_file(test_file)

    if "packages" not in comp_data or not isinstance(comp_data["packages"], dict):
        return {}

    return comp_data["packages"]


def extract_packages_from_yaml(data: dict) -> dict[str, str]:
    """Extract COMMON BUS package includes from parsed YAML.

    Only extracts packages that are from test_build_components/common/,
    ignoring component-specific packages.

    Args:
        data: Parsed YAML dictionary

    Returns:
        Dictionary mapping package name to include path (as string representation)
        Only includes common bus packages (i2c, spi, uart, etc.)
    """
    if "packages" not in data:
        return {}

    packages_value = data["packages"]
    if not isinstance(packages_value, dict):
        # List format doesn't include common bus packages (those use dict format)
        return {}

    # Get common bus package names (cached)
    common_bus_packages = get_common_bus_packages()
    packages = {}

    # Dictionary format: packages: {name: value}
    for name, value in packages_value.items():
        # Only include common bus packages, ignore component-specific ones
        if name not in common_bus_packages:
            continue
        packages[name] = str(value)
        # Also track package dependencies (e.g., modbus includes uart)
        if name not in PACKAGE_DEPENDENCIES:
            continue
        for dep in PACKAGE_DEPENDENCIES[name]:
            if dep not in common_bus_packages:
                continue
            # Mark as included via dependency
            packages[f"{DEPENDENCY_MARKER_PREFIX}{dep}"] = f"(included via {name})"

    return packages


def prefix_substitutions_in_dict(
    data: Any, prefix: str, exclude: set[str] | None = None
) -> Any:
    """Recursively prefix all substitution references in a data structure.

    Args:
        data: YAML data structure (dict, list, or scalar)
        prefix: Prefix to add to substitution names
        exclude: Set of substitution names to exclude from prefixing

    Returns:
        Data structure with prefixed substitution references
    """
    if exclude is None:
        exclude = set()

    def replace_sub(text: str) -> str:
        """Replace substitution references in a string."""

        def replace_match(match):
            sub_name = match.group(1)
            if sub_name in exclude:
                return match.group(0)
            # Always use braced format in output for consistency
            return f"${{{prefix}_{sub_name}}}"

        # Match both ${substitution} and $substitution formats
        return re.sub(r"\$\{?(\w+)\}?", replace_match, text)

    if isinstance(data, dict):
        result = {}
        for key, value in data.items():
            result[key] = prefix_substitutions_in_dict(value, prefix, exclude)
        return result
    if isinstance(data, list):
        return [prefix_substitutions_in_dict(item, prefix, exclude) for item in data]
    if isinstance(data, str):
        return replace_sub(data)
    return data


# (section, id) pairs that several components intentionally share. ESPHome
# treats these as a single instance when merged, so duplicates with differing
# content are expected and must not be flagged as accidental collisions. Keyed on
# the section as well as the id so a generic name (e.g. `ldo_id`) is only exempt
# in its intended section -- an accidental collision on the same name elsewhere
# is still caught.
INTENTIONALLY_SHARED_IDS = frozenset(
    {
        # Several components each declare an `sntp_time` clock; ESPHome merges
        # them into one time source.
        ("time", "sntp_time"),
        # esp_ldo and mipi_dsi both configure the channel-3 internal LDO on the
        # ESP32-P4; only one LDO per channel may exist, so the shared id lets the
        # merge collapse them into a single LDO.
        ("esp_ldo", "ldo_id"),
    }
)


def deduplicate_by_id(data: dict) -> dict:
    """Deduplicate list items with the same ID.

    Identical items sharing an ID (e.g. a shared bus from a common package pulled
    in by several components) are collapsed to the first occurrence. Two items
    that share an ID but differ in content are a real conflict: when merged, the
    first silently wins and the others are dropped, which can make a
    cross-reference resolve to an incompatible entity. Rather than defer that to
    downstream validation (where it surfaces as a confusing, order-dependent
    failure in an unrelated build), raise immediately so the offending ID is
    named. Ids in ``INTENTIONALLY_SHARED_IDS`` are deliberately shared singletons
    and keep their collapse behaviour.

    Args:
        data: Parsed config dictionary

    Returns:
        Config with deduplicated lists

    Raises:
        ValueError: If two items share an ID but have different content.
    """
    if not isinstance(data, dict):
        return data

    result = {}
    for key, value in data.items():
        if isinstance(value, list):
            # Check for items with 'id' field
            seen_items: dict[str, Any] = {}
            deduped_list = []

            for item in value:
                if isinstance(item, dict) and "id" in item:
                    item_id = item["id"]
                    if item_id not in seen_items:
                        seen_items[item_id] = item
                        deduped_list.append(item)
                    elif (key, item_id) in INTENTIONALLY_SHARED_IDS:
                        # Deliberately shared singleton -> keep first occurrence.
                        pass
                    elif item != seen_items[item_id]:
                        raise ValueError(
                            f"Conflicting definitions for id '{item_id}' under "
                            f"'{key}' when merging test configs; give each "
                            f"component a unique id"
                        )
                    # else: identical duplicate (e.g. shared bus package) -> skip
                else:
                    # No ID, just add it
                    deduped_list.append(item)

            result[key] = deduped_list
        elif isinstance(value, dict):
            # Recursively deduplicate nested dicts
            result[key] = deduplicate_by_id(value)
        else:
            result[key] = value

    return result


def prepare_component_body(comp_data: dict, comp_name: str, comp_dir: Path) -> dict:
    """Return a component's test body as it enters the merge.

    Expands component-specific package includes inline (common bus packages are
    left for the merge to re-add once), applies ESPHome's top-level-substitutions
    -override-package-substitutions rule, then prefixes every substitution
    reference with the component name. Shared by ``merge_component_configs`` and
    the duplicate-id guard (``script/ci_check_duplicate_test_ids.py``) so the
    guard compares exactly what the build merges.
    """
    # $component_dir resolves to the component's absolute path.
    comp_abs_dir = str(comp_dir.absolute())

    # Top-level substitutions override package substitutions, so capture them
    # before expanding packages can introduce their own.
    top_level_subs = (
        comp_data["substitutions"].copy()
        if isinstance(comp_data.get("substitutions"), dict)
        else {}
    )

    packages_value = comp_data.get("packages")
    if isinstance(packages_value, dict):
        common_bus_packages = get_common_bus_packages()
        for pkg_name, pkg_value in list(packages_value.items()):
            if pkg_name in common_bus_packages:
                continue
            if isinstance(pkg_value, yaml_util.IncludeFile):
                pkg_value = pkg_value.load()
            if isinstance(pkg_value, dict):
                comp_data = merge_config(comp_data, pkg_value)
    elif isinstance(packages_value, list):
        for pkg_value in packages_value:
            if isinstance(pkg_value, yaml_util.IncludeFile):
                pkg_value = pkg_value.load()
            if isinstance(pkg_value, dict):
                comp_data = merge_config(comp_data, pkg_value)
    # Common bus packages are re-added once by the caller; drop them here.
    comp_data.pop("packages", None)

    subs = comp_data.get("substitutions") or {}
    subs.update(top_level_subs)
    prefixed_subs = {f"{comp_name}_{name}": value for name, value in subs.items()}
    prefixed_subs[f"{comp_name}_component_dir"] = comp_abs_dir
    comp_data["substitutions"] = prefixed_subs

    return prefix_substitutions_in_dict(comp_data, comp_name)


def merge_component_configs(
    component_names: list[str],
    platform: str,
    tests_dir: Path,
    output_file: Path,
) -> None:
    """Merge multiple component test configs into a single file.

    Args:
        component_names: List of component names to merge
        platform: Platform to merge for (e.g., "esp32-ard")
        tests_dir: Path to tests/components directory
        output_file: Path to output merged config file
    """
    if not component_names:
        raise ValueError("No components specified")

    # Track packages to ensure they're identical
    all_packages = None

    # Start with empty config
    merged_config_data = {}

    # Convert tests_dir to string for caching
    tests_dir_str = str(tests_dir)

    # Process each component
    for comp_name in component_names:
        comp_dir = tests_dir / comp_name
        test_file = comp_dir / f"test.{platform}.yaml"

        if not test_file.exists():
            raise FileNotFoundError(f"Test file not found: {test_file}")

        # Load the component's test file
        comp_data = load_yaml_file(test_file)

        # Merge packages from all components (cross-bus merging)
        # Components can have different packages (e.g., one with ble, another with uart)
        # as long as they don't conflict (checked by are_buses_compatible before calling this)
        comp_packages = extract_packages_from_yaml(comp_data)

        if all_packages is None:
            # First component - initialize package dict
            all_packages = comp_packages or {}
        elif comp_packages:
            # Merge packages - combine all unique package types
            # If both have the same package type, verify they're identical
            for pkg_name, pkg_config in comp_packages.items():
                if pkg_name in all_packages:
                    # Same package type - verify config matches
                    if all_packages[pkg_name] != pkg_config:
                        raise ValueError(
                            f"Component {comp_name} has conflicting config for package '{pkg_name}'. "
                            f"Expected: {all_packages[pkg_name]}, Got: {pkg_config}. "
                            f"Components with conflicting bus configs cannot be merged."
                        )
                else:
                    # New package type - add it
                    all_packages[pkg_name] = pkg_config

        # Expand component-specific packages and prefix substitutions, exactly as
        # the duplicate-id guard does, so both see the same body.
        comp_data = prepare_component_body(comp_data, comp_name, comp_dir)

        # Use ESPHome's merge_config to merge this component into the result
        # merge_config handles list merging with ID-based deduplication automatically
        merged_config_data = merge_config(merged_config_data, comp_data)

    # Add merged packages back (union of all component packages)
    # IMPORTANT: Only include common bus packages (spi, i2c, uart, etc.)
    # Do NOT re-add component-specific packages as they contain unprefixed $component_dir refs
    if all_packages:
        # Build packages dict from merged all_packages
        # all_packages is a dict mapping package_name -> str(package_value)
        # We need to reconstruct the actual package values by loading them from any component
        # Since packages with the same name must have identical configs (verified above),
        # we can load the package value from the first component that has each package
        common_bus_packages = get_common_bus_packages()
        merged_packages: dict[str, Any] = {}

        # Collect packages that are included as dependencies
        # If modbus is present, uart is included via modbus.packages.uart
        packages_to_skip: set[str] = set()
        for pkg_name in all_packages:
            if pkg_name.startswith(DEPENDENCY_MARKER_PREFIX):
                # Extract the actual package name (remove _dep_ prefix)
                dep_name = pkg_name[len(DEPENDENCY_MARKER_PREFIX) :]
                packages_to_skip.add(dep_name)

        for pkg_name in all_packages:
            # Skip dependency markers
            if pkg_name.startswith(DEPENDENCY_MARKER_PREFIX):
                continue
            # Skip non-common-bus packages
            if pkg_name not in common_bus_packages:
                continue
            # Skip packages that are included as dependencies of other packages
            # This prevents duplicate definitions (e.g., uart via modbus + uart separately)
            if pkg_name in packages_to_skip:
                continue

            # Find a component that has this package and extract its value
            # Uses cached lookup to avoid re-loading the same files
            for comp_name in component_names:
                comp_packages = get_component_packages(
                    comp_name, platform, tests_dir_str
                )
                if pkg_name in comp_packages:
                    merged_packages[pkg_name] = comp_packages[pkg_name]
                    break

        if merged_packages:
            merged_config_data["packages"] = merged_packages

    # Deduplicate items with same ID (keeps first occurrence)
    merged_config_data = deduplicate_by_id(merged_config_data)

    # Remove esphome section since it will be provided by the wrapper file
    # The wrapper file includes this merged config via packages and provides
    # the proper esphome: section with name, platform, etc.
    if "esphome" in merged_config_data:
        del merged_config_data["esphome"]

    # Write merged config
    output_file.parent.mkdir(parents=True, exist_ok=True)
    yaml_content = yaml_util.dump(merged_config_data)
    output_file.write_text(yaml_content, encoding="utf-8")

    print(f"Successfully merged {len(component_names)} components into {output_file}")


def main() -> None:
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Merge multiple component test configs into a single file"
    )
    parser.add_argument(
        "--components",
        "-c",
        required=True,
        help="Comma-separated list of component names to merge",
    )
    parser.add_argument(
        "--platform",
        "-p",
        required=True,
        help="Platform to merge for (e.g., esp32-ard)",
    )
    parser.add_argument(
        "--output",
        "-o",
        required=True,
        type=Path,
        help="Output file path for merged config",
    )
    parser.add_argument(
        "--tests-dir",
        type=Path,
        default=Path("tests/components"),
        help="Path to tests/components directory",
    )

    args = parser.parse_args()

    component_names = [c.strip() for c in args.components.split(",")]

    try:
        merge_component_configs(
            component_names=component_names,
            platform=args.platform,
            tests_dir=args.tests_dir,
            output_file=args.output,
        )
    except Exception as e:  # noqa: BLE001
        print(f"Error merging configs: {e}", file=sys.stderr)
        import traceback

        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()