From d3892b8399c7f015bbbced0b50e943dcbacd2b17 Mon Sep 17 00:00:00 2001
From: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com>
Date: Sun, 28 Jun 2026 07:03:09 -0400
Subject: [PATCH] [platformio] Extract toolchain-agnostic PlatformIO library
converter (#17243)
---
esphome/espidf/component.py | 724 ++------------------
esphome/platformio/library.py | 717 +++++++++++++++++++
tests/unit_tests/test_espidf_component.py | 83 +--
tests/unit_tests/test_platformio_library.py | 229 +++++++
4 files changed, 1023 insertions(+), 730 deletions(-)
create mode 100644 esphome/platformio/library.py
create mode 100644 tests/unit_tests/test_platformio_library.py
diff --git a/esphome/espidf/component.py b/esphome/espidf/component.py
index cfd42916b2..5029e014a4 100644
--- a/esphome/espidf/component.py
+++ b/esphome/espidf/component.py
@@ -1,163 +1,42 @@
-from collections import deque
-from collections.abc import Callable
-from dataclasses import dataclass, field
-import glob
-import hashlib
-import itertools
-import json
+"""ESP-IDF backend for the shared PlatformIO library converter.
+
+The toolchain-agnostic resolution/download/caching pipeline lives in
+``esphome.platformio.library``; this module only adds the ESP-IDF specifics:
+emitting an ``idf_component_register`` ``CMakeLists.txt`` + ``idf_component.yml``
+for each resolved library, running any PlatformIO ``extraScript``, and the
+ESP-IDF platform/framework compatibility defaults.
+"""
+
import logging
import os
from pathlib import Path
-import re
-import tempfile
-from typing import Any, TypeVar
-from urllib.parse import urlparse, urlsplit, urlunsplit
-from esphome import git, yaml_util
from esphome.core import CORE, Library
-from esphome.espidf.framework import archive_extract_all, download_from_mirrors, rmdir
from esphome.helpers import write_file_if_changed
+from esphome.platformio.library import (
+ DEFAULT_BUILD_FLAGS,
+ DEFAULT_BUILD_INCLUDE_DIR,
+ DEFAULT_BUILD_SRC_FILTER,
+ ESPHOME_DATA_EXTRA_CMAKE_KEY,
+ ESPHOME_DATA_KEY,
+ SRC_FILE_EXTENSIONS,
+ ConvertedLibrary as IDFComponent,
+ LibraryBackend,
+ PathType,
+ collect_filtered_files,
+ convert_libraries,
+ ensure_list,
+ split_list_by_condition,
+)
_LOGGER = logging.getLogger(__name__)
-PathType = str | os.PathLike
-
-#
-# Constants from platformio
-#
-
-FILTER_REGEX = re.compile(r"([+-])<([^>]+)>")
-DEFAULT_BUILD_SRC_FILTER = (
- "+<*> -<.git/> -<.svn/> - - - -"
-)
-DEFAULT_BUILD_SRC_DIRS = "src"
-DEFAULT_BUILD_INCLUDE_DIR = "include"
-DEFAULT_BUILD_FLAGS = []
-SRC_FILE_EXTENSIONS = [
- ".c",
- ".cpp",
- ".cc",
- ".cxx",
- ".c++",
- ".S",
- ".spp",
- ".SPP",
- ".sx",
- ".s",
- ".asm",
- ".ASM",
-]
-
ESP32_PLATFORM = "espressif32"
-DOMAIN = "pio_components"
-
-ESPHOME_DATA_KEY = "ESPHOME"
-ESPHOME_DATA_EXTRA_CMAKE_KEY = "EXTRA_CMAKE"
-class Source:
- def download(self, dir_suffix: str, force: bool = False, salt: str = "") -> Path:
- raise NotImplementedError
-
-
-class URLSource(Source):
- def __init__(self, url: str):
- self.url = url
-
- def download(self, dir_suffix: str, force: bool = False, salt: str = "") -> Path:
- base_dir = Path(CORE.data_dir) / DOMAIN
- h = hashlib.new("sha256")
- h.update(self.url.encode())
- if salt:
- h.update(salt.encode())
- path = base_dir / h.hexdigest()[:8] / dir_suffix
- # Marker file written last to signal a complete extraction. Using a
- # marker (instead of just `path.is_dir()`) means an interrupted
- # extraction is correctly detected and re-run on the next invocation,
- # and lets us extract directly into ``path`` — avoiding a
- # post-extraction rename that races with antivirus on Windows.
- extracted_marker = path / ".esphome_extracted"
- if not extracted_marker.is_file() or force:
- rmdir(path, msg=f"Clean up library directory {path}")
-
- # Download in temporary file
- with tempfile.NamedTemporaryFile() as tmp:
- _LOGGER.info("Downloading %s ...", self.url)
- _LOGGER.debug("Location: %s", path)
-
- download_from_mirrors([self.url], {}, tmp.file)
-
- _LOGGER.debug("Extracting archive to %s ...", path)
- archive_extract_all(tmp.file, path)
- extracted_marker.touch()
- return path
-
- def __str__(self):
- return self.url
-
-
-class GitSource(Source):
- def __init__(self, url: str, ref: str | None):
- self.url = url
- self.ref = ref
-
- def download(self, dir_suffix: str, force: bool = False, salt: str = "") -> Path:
- path, _ = git.clone_or_update(
- url=self.url,
- ref=self.ref,
- refresh=git.NEVER_REFRESH if not force else None,
- domain=f"{DOMAIN}/{salt}" if salt else DOMAIN,
- submodules=[],
- subpath=Path(dir_suffix),
- )
- return path
-
- def __str__(self):
- return f"{self.url}#{self.ref}" if self.ref else self.url
-
-
-class InvalidIDFComponent(Exception):
- pass
-
-
-class IDFComponent:
- def __init__(self, name: str, version: str, source: Source | None):
- self.name = name
- self.version = version
- self.source = source
- self.data = {}
- self.dependencies: list[IDFComponent] = []
- self._path: Path | None = None
-
- def __str__(self):
- return f"{self.name}@{self.version}={self.source}"
-
- @property
- def path(self) -> Path:
- if self._path is None:
- raise RuntimeError(f"path not set for component {self}")
- return self._path
-
- @path.setter
- def path(self, value: Path) -> None:
- self._path = value
-
- def get_sanitized_name(self):
- return re.sub(r"[^a-zA-Z0-9_.\-/]", "_", self.name)
-
- def get_require_name(self):
- return self.get_sanitized_name().replace("/", "__")
-
- def download(self, force: bool = False, salt: str = ""):
- """
- The dependency name should match the directory name at the end of the override path.
- The ESP-IDF build system uses the directory name as the component name, so the directory of the override_path should match the component name.
- If you want to specify the full name of the component with the namespace, replace / in the component name with __.
- @see https://docs.espressif.com/projects/idf-component-manager/en/latest/reference/manifest_file.html
- """
- self.path = self.source.download(
- self.get_sanitized_name(), force=force, salt=salt
- )
+def _idf_framework() -> str:
+ """The framework token an ESP-IDF library manifest is expected to declare."""
+ return "arduino" if CORE.using_arduino else "espidf"
def _apply_extra_script(component: IDFComponent) -> None:
@@ -190,119 +69,6 @@ def _apply_extra_script(component: IDFComponent) -> None:
component.data["build"]["flags"] = flags
-T = TypeVar("T")
-
-
-def _ensure_list(obj: T | list[T]) -> list[T]:
- """
- Convert an object to a list if it isn't already a list.
-
- Args:
- obj: Object that may or may not already be a list.
-
- Returns:
- list[T]: The original list if ``obj`` is a list, otherwise a single-item
- list containing ``obj``.
- """
- return [obj] if not isinstance(obj, list) else obj
-
-
-def _owner_pkgname_to_name(owner: str | None, pkgname: str) -> str:
- """
- Convert owner and package name to a standardized component name.
-
- This function combines owner and package name with a forward slash when
- both are provided, otherwise returns just the package name.
-
- Args:
- owner: The owner/username of the package (can be None)
- pkgname: The name of the package
-
- Returns:
- str: The standardized component name in "owner/pkgname" format or just "pkgname"
- """
- return f"{owner}/{pkgname}" if owner else pkgname
-
-
-def _collect_filtered_files(src_dir: PathType, src_filters: list[str]) -> list[str]:
- """
- Recursively match files in a directory according to include/exclude patterns.
-
- This function processes a list of filter strings that indicate which files
- to include or exclude. Each filter is parsed into patterns with a sign:
- '+' for inclusion and '-' for exclusion. Directory patterns ending with '/'
- are normalized to include all their contents recursively.
-
- Args:
- src_dir (PathType): Root directory to search within.
- src_filters (list[str]): List of filter strings, which may contain multiple
- patterns. Each pattern can start with '+' or '-' to indicate inclusion
- or exclusion.
-
- Returns:
- list[str]: List of matched file paths as strings. Only files (not directories)
- are returned, even if a directory matches a pattern.
- """
- matches = list(
- itertools.chain.from_iterable(
- FILTER_REGEX.findall(src_filter) for src_filter in src_filters
- )
- )
-
- selected = set()
-
- for sign, pattern in matches:
- pattern = pattern.strip()
-
- if pattern.endswith("/"):
- pattern = pattern.rstrip("/") + "/**"
-
- # glob.escape has no pathlib equivalent and the matcher works on raw
- # path strings, so PTH118/PTH207 don't apply here.
- full_pattern = os.path.join(glob.escape(str(src_dir)), pattern) # noqa: PTH118
-
- matched = []
- for item in glob.glob(full_pattern, recursive=True): # noqa: PTH207
- if not Path(item).is_dir():
- matched.append(item)
- else:
- # PlatformIO quirk: a directory matched with "*" should include all its
- # nested files and subdirectories, not just the directory itself.
- for root, _, files in os.walk(item):
- matched.extend([str(Path(root) / f) for f in files])
-
- if sign == "+":
- selected.update(matched)
- elif sign == "-":
- selected.difference_update(matched)
-
- return [r for r in selected if Path(r).is_file()]
-
-
-def _split_list_by_condition(
- items: list[str], match_fn: Callable[[str], str | None]
-) -> tuple[list[str], list[str]]:
- """
- Splits a list into two lists based on a matching function.
-
- Args:
- items: List of items to split.
- match_fn: Function that returns a value for items that should go into the "matched" list.
-
- Returns:
- A tuple (matched, non_matched)
- """
- matched = []
- non_matched = []
- for item in items:
- result = match_fn(item)
- if result:
- matched.append(result)
- else:
- non_matched.append(item)
- return matched, non_matched
-
-
def generate_cmakelists_txt(component: IDFComponent) -> str:
"""
Generate a CMakeLists.txt file for an ESP-IDF component.
@@ -333,15 +99,15 @@ def generate_cmakelists_txt(component: IDFComponent) -> str:
build_include_dir = component.data.get("build", {}).get(
"includeDir", DEFAULT_BUILD_INCLUDE_DIR
)
- build_src_filter = _ensure_list(
+ build_src_filter = ensure_list(
component.data.get("build", {}).get("srcFilter", DEFAULT_BUILD_SRC_FILTER)
)
- build_flags = _ensure_list(
+ build_flags = ensure_list(
component.data.get("build", {}).get("flags", DEFAULT_BUILD_FLAGS)
)
# List all sources files
- build_src_files = _collect_filtered_files(
+ build_src_files = collect_filtered_files(
component.path / Path(build_src_dir), build_src_filter
)
@@ -361,13 +127,13 @@ def generate_cmakelists_txt(component: IDFComponent) -> str:
]
# Handle build flags
- include_dir_flags, build_flags = _split_list_by_condition(
+ include_dir_flags, build_flags = split_list_by_condition(
build_flags, lambda a: a[2:].strip() if a.startswith("-I") else None
)
- link_directories, build_flags = _split_list_by_condition(
+ link_directories, build_flags = split_list_by_condition(
build_flags, lambda a: a[2:].strip() if a.startswith("-L") else None
)
- link_libraries, build_flags = _split_list_by_condition(
+ link_libraries, build_flags = split_list_by_condition(
build_flags, lambda a: a[2:].strip() if a.startswith("-l") else None
)
@@ -379,7 +145,7 @@ def generate_cmakelists_txt(component: IDFComponent) -> str:
]
# Split build_flags list into private and public lists
- private_build_flags, public_build_flags = _split_list_by_condition(
+ private_build_flags, public_build_flags = split_list_by_condition(
build_flags, lambda a: a if a.startswith("-W") else None
)
@@ -453,6 +219,8 @@ def generate_idf_component_yml(component: IDFComponent) -> str:
Returns:
YAML string representation of ESP-IDF component configuration
"""
+ from esphome import yaml_util
+
data = {}
description = component.data.get("description")
@@ -477,410 +245,24 @@ def generate_idf_component_yml(component: IDFComponent) -> str:
return yaml_util.dump(data)
-def _check_library_data(data: dict):
- """
- Check if a library data is compatible with the ESP-IDF framework.
-
- A platform mismatch (e.g. an AVR-only library on ESP32) raises
- ``InvalidIDFComponent`` so the caller skips the library. A framework
- mismatch only logs a warning — PIO manifests often understate the
- frameworks they actually compile under, and IDF (unlike PIO's
- ``lib_compat_mode``) has no opt-out, so we include the library anyway.
-
- Args:
- data: PIO library manifest dict being processed.
-
- Raises:
- InvalidIDFComponent: If the library does not support the ESP32 platform.
- """
- platforms = data.get("platforms", "*")
- if isinstance(platforms, str):
- platforms = [a.strip() for a in platforms.split(",")]
- platforms = _ensure_list(platforms)
-
- # Check if library supports ESP-IDF platform
- valid_platforms = "*" in platforms or ESP32_PLATFORM in platforms
-
- if not valid_platforms:
- raise InvalidIDFComponent(f"Unsupported library platforms: {platforms}")
-
- frameworks = data.get("frameworks", "*")
- if isinstance(frameworks, str):
- frameworks = [a.strip() for a in frameworks.split(",")]
- frameworks = _ensure_list(frameworks)
-
- # Check if library declares the active framework. PIO library manifests
- # often list only "arduino" even when the library actually compiles fine
- # under ESP-IDF, and IDF (unlike PIO with `lib_compat_mode`) has no way to
- # opt out of the check. Warn instead of failing so the user isn't forced to
- # fork the library to fix the manifest.
- framework = "arduino" if CORE.using_arduino else "espidf"
- valid_framework = "*" in frameworks or framework in frameworks
-
- if not valid_framework:
- _LOGGER.warning(
- "Library %s declares frameworks %s that do not include '%s'; including anyway",
- data.get("name", ""),
- frameworks,
- framework,
- )
-
-
-def _parse_library_json(library_json_path: PathType):
- """
- Load and parse a JSON file describing a library.
-
- Args:
- library_json_path (PathType): Path to the JSON file.
-
- Returns:
- dict: Parsed JSON content as a Python dictionary.
- """
- with Path(library_json_path).open(encoding="utf8") as fp:
- return json.load(fp)
-
-
-def _parse_library_properties(library_properties_path: PathType):
- """
- Parse a key-value platformio .properties style file into a dictionary.
-
- Args:
- library_properties_path (PathType): Path to the properties file.
-
- Returns:
- dict[str, str]: Mapping of parsed property keys to values.
- """
- with Path(library_properties_path).open(encoding="utf8") as fp:
- data = {}
- for line in fp.read().splitlines():
- line = line.strip()
- if not line or "=" not in line:
- continue
- # skip comments
- if line.startswith("#"):
- continue
- key, value = line.split("=", 1)
- if not value.strip():
- continue
- data[key.strip()] = value.strip()
- return data
-
-
-def _make_registry_client() -> Any:
- """Create a minimal PlatformIO registry client with no system filtering.
-
- ``is_system_compatible`` is forced True so version selection is driven purely
- by the requested version requirements -- ESP-IDF/target compatibility is
- handled elsewhere, not by the PlatformIO registry.
- """
- from platformio.package.manager._registry import PackageManagerRegistryMixin
-
- class _Registry(PackageManagerRegistryMixin):
- def __init__(self) -> None:
- self._registry_client = None
- self.pkg_type = "library"
-
- @staticmethod
- def is_system_compatible(value: Any, custom_system: Any = None) -> bool:
- return True
-
- return _Registry()
-
-
-def _resolve_registry_version(
- owner: str | None, pkgname: str, requirements: set[str]
-) -> tuple[str, str, str, str]:
- """Resolve a registry package to the single highest version satisfying ALL
- the given requirements; return ``(owner, name, version, download_url)``.
-
- Intersecting every requirement (rather than resolving each consumer in
- isolation) makes the result independent of processing order and guarantees
- no stated constraint is violated -- e.g. ``esphome/libsodium`` requested as
- both ``==1.10021.0`` and ``^1.10018.1`` resolves to ``1.10021.0``.
- """
- from platformio.package.meta import PackageSpec
-
- registry = _make_registry_client()
- package = registry.fetch_registry_package(PackageSpec(owner=owner, name=pkgname))
- owner = package["owner"]["username"]
- name = package["name"]
-
- # Chaining the per-requirement filter intersects all constraints.
- versions = package.get("versions") or []
- for requirement in sorted(requirements):
- versions = registry.get_compatible_registry_versions(
- versions, PackageSpec(owner=owner, name=name, requirements=requirement)
- )
- if not versions:
- raise RuntimeError(
- f"No version of {owner}/{name} satisfies all requirements "
- f"{sorted(requirements)} requested across the library tree"
- )
-
- best = registry.pick_best_registry_version(versions)
- pkgfile = registry.pick_compatible_pkg_file(best["files"])
- if not pkgfile:
- raise RuntimeError(f"No package file for {owner}/{name}@{best['name']}")
- return owner, name, best["name"], pkgfile["download_url"]
-
-
-def _normalize_dependencies(dependencies: Any) -> list[dict]:
- """Normalize a library manifest's ``dependencies`` to a list of dicts.
-
- PIO's library.json accepts both the list-of-dicts form and the shorthand
- dict form (``{"owner/Name": "version_spec"}``); normalize the latter so
- callers see a uniform list.
- """
- if not dependencies:
- return []
- if isinstance(dependencies, dict):
- normalized = []
- for raw_name, spec in dependencies.items():
- if "/" in raw_name:
- owner, pkgname = raw_name.split("/", 1)
- else:
- owner, pkgname = None, raw_name
- entry = {"name": pkgname, "owner": owner}
- if isinstance(spec, dict):
- entry.update(spec)
- else:
- entry["version"] = spec
- normalized.append(entry)
- return normalized
- return [d for d in dependencies if isinstance(d, dict)]
-
-
-@dataclass
-class _LibNode:
- """A node in the library dependency graph being resolved as a batch."""
-
- key: str
- is_git: bool
- owner: str | None = None
- pkgname: str | None = None
- requirements: set[str] = field(default_factory=set)
- url: str | None = None
- ref: str | None = None
- edges: set[str] = field(default_factory=set)
-
-
-def _node_key(
- name: str | None, version: str | None, repository: str | None
-) -> tuple[str, bool, tuple[str | None, str | None]]:
- """Return ``(key, is_git, locator)`` for a library or dependency spec.
-
- The key is derived from the *input* spec (the registry name as written, or
- the git URL path), not the resolved canonical name. So a package referenced
- inconsistently -- bare ``name`` vs ``owner/name``, or git vs registry -- maps
- to distinct keys and isn't deduplicated; ``generate_idf_components`` warns
- about that after resolution rather than merging the nodes.
- """
- if repository:
- split_result = urlsplit(repository)
- key = str(split_result.path).strip("/").removesuffix(".git")
- ref = split_result.fragment.strip() or None
- url = urlunsplit(split_result._replace(fragment=""))
- return key, True, (url, ref)
- if name and "/" in name:
- owner, pkgname = name.split("/", 1)
- else:
- owner, pkgname = None, name
- return name, False, (owner, pkgname)
+def _emit_idf_component(component: IDFComponent) -> None:
+ """Write the ESP-IDF build files for a resolved library into its cache dir."""
+ _apply_extra_script(component)
+ write_file_if_changed(
+ component.path / "CMakeLists.txt",
+ generate_cmakelists_txt(component),
+ )
+ write_file_if_changed(
+ component.path / "idf_component.yml",
+ generate_idf_component_yml(component),
+ )
def generate_idf_components(libraries: list[Library]) -> list[IDFComponent]:
- """Resolve and convert a batch of PlatformIO libraries to IDF components.
-
- Resolves the whole set together rather than each library independently: it
- walks the dependency graph collecting every version *requirement* per
- component name, then resolves each name once to a single version satisfying
- all of them. So a transitive dependency shared under
- different specs (e.g. ``esphome/libsodium``, pulled by both ``noise-c`` and
- ``esp_wireguard``) becomes one component instead of two clashing
- ``override_path`` entries -- order-independently, and without ever violating
- a stated constraint.
-
- The returned list holds the top-level components (those directly requested);
- transitive dependencies are converted too and wired into each component's
- generated manifest.
-
- ``lib_ignore`` from ``esphome->platformio_options`` excludes libraries by
- short name (part after the ``/``), matched against both the top-level
- libraries and every dependency discovered during the graph walk.
- """
- nodes: dict[str, _LibNode] = {}
-
- lib_ignore = {
- name.split("/")[-1].lower()
- for name in CORE.platformio_options.get("lib_ignore", [])
- }
-
- # The generated CMakeLists.txt/idf_component.yml inside the shared cache
- # bake in the dependency wiring, which lib_ignore changes; salt the cache
- # path so configs with different lib_ignore values don't fight over (and
- # constantly rewrite) the same converted component files.
- salt = (
- hashlib.sha256(",".join(sorted(lib_ignore)).encode()).hexdigest()[:8]
- if lib_ignore
- else ""
+ """Resolve and convert a batch of PlatformIO libraries to IDF components."""
+ backend = LibraryBackend(
+ platform=ESP32_PLATFORM,
+ framework=_idf_framework(),
+ emit=_emit_idf_component,
)
-
- def is_ignored(name: str | None) -> bool:
- if not lib_ignore or name is None:
- return False
- return name.split("/")[-1].lower() in lib_ignore
-
- def add_spec(name: str | None, version: str | None, repository: str | None) -> str:
- key, is_git, locator = _node_key(name, version, repository)
- node = nodes.get(key) or _LibNode(key=key, is_git=is_git)
- nodes[key] = node
- if is_git:
- node.is_git = True
- node.url, node.ref = locator
- else:
- node.owner, node.pkgname = locator
- if version:
- node.requirements.add(version)
- return key
-
- top_level = [
- add_spec(library.name, library.version, library.repository)
- for library in libraries
- if not is_ignored(library.name)
- ]
-
- # Collect + resolve to a fixpoint: a node is (re)resolved whenever its
- # requirement set has grown since the last time, so every requirement in the
- # graph is accounted for before conversion.
- components: dict[str, IDFComponent] = {}
- resolved_requirements: dict[str, frozenset[str]] = {}
- top_level_keys = set(top_level)
- worklist = deque(dict.fromkeys(top_level))
- while worklist:
- key = worklist.popleft()
- node = nodes[key]
-
- # A node is queued once per referring edge; skip the (uncached) registry
- # lookup + download + dependency walk unless its requirement set grew
- # since the last resolve. Requirements only ever grow, so this still
- # converges the fixpoint and terminates dependency cycles.
- requirements = frozenset(node.requirements)
- if resolved_requirements.get(key) == requirements:
- continue
- resolved_requirements[key] = requirements
-
- if node.is_git:
- component = IDFComponent(key, "*", GitSource(node.url, node.ref))
- else:
- owner, name, version, url = _resolve_registry_version(
- node.owner, node.pkgname, node.requirements
- )
- component = IDFComponent(
- _owner_pkgname_to_name(owner, name), version, URLSource(url)
- )
- component.download(salt=salt)
-
- library_json_path = component.path / "library.json"
- library_properties_path = component.path / "library.properties"
- if library_json_path.is_file():
- component.data = _parse_library_json(library_json_path)
- elif library_properties_path.is_file():
- component.data = _parse_library_properties(library_properties_path)
- else:
- raise RuntimeError(
- f"Invalid PIO library {key}: missing library.json and "
- "library.properties"
- )
-
- try:
- _check_library_data(component.data)
- except InvalidIDFComponent as e:
- # Skip an incompatible transitive dependency, but fail fast if a
- # top-level library the build explicitly requested is incompatible.
- if key in top_level_keys:
- raise RuntimeError(
- f"Requested library {key} is not compatible with ESP-IDF: {e}"
- ) from e
- _LOGGER.debug("Skip incompatible dependency %s: %s", key, str(e))
- continue
- components[key] = component
-
- # Requirements changed (we got past the short-circuit above), so
- # (re)walk this component's dependencies.
- node.edges = set()
- for dependency in _normalize_dependencies(component.data.get("dependencies")):
- if "name" not in dependency or "version" not in dependency:
- continue
- try:
- _check_library_data(dependency)
- except InvalidIDFComponent as e:
- _LOGGER.debug("Skip dependency %s: %s", dependency.get("name"), str(e))
- continue
- dep_name = _owner_pkgname_to_name(
- dependency.get("owner"), dependency.get("name")
- )
- if is_ignored(dep_name):
- _LOGGER.debug("Skip ignored dependency %s", dep_name)
- continue
- # The version field may actually be a URL (git/archive dependency).
- dep_version = dependency["version"]
- dep_url = None
- try:
- parsed = urlparse(dep_version)
- if all([parsed.scheme, parsed.netloc]):
- dep_url, dep_version = dep_version, None
- except (TypeError, ValueError):
- pass
- dep_key = add_spec(dep_name, dep_version, dep_url)
- node.edges.add(dep_key)
- worklist.append(dep_key)
-
- # A git source wins over any registry version requested for the same
- # component. That's intentional, but warn so a dropped registry pin isn't a
- # silent surprise.
- for node in nodes.values():
- if node.is_git and node.requirements:
- _LOGGER.warning(
- "Library %s is requested both from a git source (%s) and as "
- "registry version(s) %s; using the git source.",
- node.key,
- node.url,
- sorted(node.requirements),
- )
-
- # Two graph nodes that resolve to the same component name (e.g. a package
- # referenced both bare and as ``owner/name``) are not deduplicated and can
- # produce conflicting component definitions. Warn so it's not silent.
- canonical_keys: dict[str, str] = {}
- for node_key, component in components.items():
- canonical = component.get_sanitized_name()
- if canonical_keys.setdefault(canonical, node_key) != node_key:
- _LOGGER.warning(
- "Library %s is referenced under multiple names (%s and %s); these "
- "are not deduplicated. Reference it consistently as %s.",
- canonical,
- canonical_keys[canonical],
- node_key,
- canonical,
- )
-
- # Wire each component's dependencies to the single resolved instances, then
- # regenerate build files.
- for key, component in components.items():
- component.dependencies = [
- components[dep_key]
- for dep_key in sorted(nodes[key].edges)
- if dep_key in components
- ]
- for component in components.values():
- _apply_extra_script(component)
- write_file_if_changed(
- component.path / "CMakeLists.txt",
- generate_cmakelists_txt(component),
- )
- write_file_if_changed(
- component.path / "idf_component.yml",
- generate_idf_component_yml(component),
- )
-
- return [components[key] for key in top_level if key in components]
+ return convert_libraries(libraries, backend)
diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py
new file mode 100644
index 0000000000..43282c7aa0
--- /dev/null
+++ b/esphome/platformio/library.py
@@ -0,0 +1,717 @@
+"""Toolchain-agnostic PlatformIO library converter.
+
+Resolves a batch of PlatformIO/Arduino library specs (added via
+``cg.add_library(...)``) into local, build-ready directories: it fetches each
+library (registry/git/url), parses its ``library.json`` / ``library.properties``
+manifest, resolves the whole dependency graph to a single version per name, and
+caches the result under ``/pio_components``.
+
+The toolchain-specific part — turning a resolved library into build files
+(ESP-IDF ``idf_component_register`` CMakeLists, or a Zephyr module) — is supplied
+by a :class:`LibraryBackend`. This module owns everything that is the same
+regardless of which toolchain consumes the result.
+"""
+
+from collections import deque
+from collections.abc import Callable
+from dataclasses import dataclass, field
+import glob
+import hashlib
+import itertools
+import json
+import logging
+import os
+from pathlib import Path
+import re
+import tempfile
+from typing import Any, TypeVar
+from urllib.parse import urlparse, urlsplit, urlunsplit
+
+from esphome import git
+from esphome.core import CORE, Library
+from esphome.framework_helpers import archive_extract_all, download_from_mirrors, rmdir
+
+_LOGGER = logging.getLogger(__name__)
+
+PathType = str | os.PathLike
+
+#
+# Constants from platformio
+#
+
+FILTER_REGEX = re.compile(r"([+-])<([^>]+)>")
+DEFAULT_BUILD_SRC_FILTER = (
+ "+<*> -<.git/> -<.svn/> - - - -"
+)
+DEFAULT_BUILD_SRC_DIRS = "src"
+DEFAULT_BUILD_INCLUDE_DIR = "include"
+DEFAULT_BUILD_FLAGS = []
+SRC_FILE_EXTENSIONS = [
+ ".c",
+ ".cpp",
+ ".cc",
+ ".cxx",
+ ".c++",
+ ".S",
+ ".spp",
+ ".SPP",
+ ".sx",
+ ".s",
+ ".asm",
+ ".ASM",
+]
+
+DOMAIN = "pio_components"
+
+ESPHOME_DATA_KEY = "ESPHOME"
+ESPHOME_DATA_EXTRA_CMAKE_KEY = "EXTRA_CMAKE"
+
+
+class Source:
+ def download(self, dir_suffix: str, force: bool = False, salt: str = "") -> Path:
+ raise NotImplementedError
+
+
+class URLSource(Source):
+ def __init__(self, url: str):
+ self.url = url
+
+ def download(self, dir_suffix: str, force: bool = False, salt: str = "") -> Path:
+ base_dir = Path(CORE.data_dir) / DOMAIN
+ h = hashlib.new("sha256")
+ h.update(self.url.encode())
+ if salt:
+ h.update(salt.encode())
+ path = base_dir / h.hexdigest()[:8] / dir_suffix
+ # Marker file written last to signal a complete extraction. Using a
+ # marker (instead of just `path.is_dir()`) means an interrupted
+ # extraction is correctly detected and re-run on the next invocation,
+ # and lets us extract directly into ``path`` — avoiding a
+ # post-extraction rename that races with antivirus on Windows.
+ extracted_marker = path / ".esphome_extracted"
+ if not extracted_marker.is_file() or force:
+ rmdir(path, msg=f"Clean up library directory {path}")
+
+ # Download in temporary file
+ with tempfile.NamedTemporaryFile() as tmp:
+ _LOGGER.info("Downloading %s ...", self.url)
+ _LOGGER.debug("Location: %s", path)
+
+ download_from_mirrors([self.url], {}, tmp.file)
+
+ _LOGGER.debug("Extracting archive to %s ...", path)
+ archive_extract_all(tmp.file, path)
+ extracted_marker.touch()
+ return path
+
+ def __str__(self):
+ return self.url
+
+
+class GitSource(Source):
+ def __init__(self, url: str, ref: str | None):
+ self.url = url
+ self.ref = ref
+
+ def download(self, dir_suffix: str, force: bool = False, salt: str = "") -> Path:
+ path, _ = git.clone_or_update(
+ url=self.url,
+ ref=self.ref,
+ refresh=git.NEVER_REFRESH if not force else None,
+ domain=f"{DOMAIN}/{salt}" if salt else DOMAIN,
+ submodules=[],
+ subpath=Path(dir_suffix),
+ )
+ return path
+
+ def __str__(self):
+ return f"{self.url}#{self.ref}" if self.ref else self.url
+
+
+class InvalidLibrary(Exception):
+ pass
+
+
+class ConvertedLibrary:
+ """A resolved PlatformIO library plus its parsed manifest and on-disk path.
+
+ Toolchain-neutral: ESP-IDF treats it as a component, Zephyr as a module. The
+ backend reads ``name``/``version``/``data``/``dependencies``/``path`` to emit
+ its build files.
+ """
+
+ def __init__(self, name: str, version: str, source: Source | None):
+ self.name = name
+ self.version = version
+ self.source = source
+ self.data = {}
+ self.dependencies: list[ConvertedLibrary] = []
+ self._path: Path | None = None
+
+ def __str__(self):
+ return f"{self.name}@{self.version}={self.source}"
+
+ @property
+ def path(self) -> Path:
+ if self._path is None:
+ raise RuntimeError(f"path not set for library {self}")
+ return self._path
+
+ @path.setter
+ def path(self, value: Path) -> None:
+ self._path = value
+
+ def get_sanitized_name(self):
+ return re.sub(r"[^a-zA-Z0-9_.\-/]", "_", self.name)
+
+ def get_require_name(self):
+ return self.get_sanitized_name().replace("/", "__")
+
+ def download(self, force: bool = False, salt: str = ""):
+ """Fetch the library into the shared cache and record its ``path``.
+
+ The cache directory is named after the sanitized library name; backends
+ rely on that name to identify the unit they build (e.g. ESP-IDF uses the
+ directory name as the component name, replacing ``/`` with ``__`` via
+ ``get_require_name``).
+ """
+ self.path = self.source.download(
+ self.get_sanitized_name(), force=force, salt=salt
+ )
+
+
+@dataclass
+class LibraryBackend:
+ """Toolchain hooks for :func:`convert_libraries`.
+
+ ``platform``/``framework`` drive the manifest compatibility check.
+ ``emit`` writes the toolchain-specific build files into a resolved library's
+ ``path`` (e.g. the ESP-IDF ``CMakeLists.txt`` + ``idf_component.yml``, or a
+ Zephyr ``module.yml`` + ``CMakeLists.txt``).
+ """
+
+ platform: str
+ framework: str
+ emit: Callable[["ConvertedLibrary"], None]
+
+
+T = TypeVar("T")
+
+
+def ensure_list(obj: T | list[T]) -> list[T]:
+ """
+ Convert an object to a list if it isn't already a list.
+
+ Args:
+ obj: Object that may or may not already be a list.
+
+ Returns:
+ list[T]: The original list if ``obj`` is a list, otherwise a single-item
+ list containing ``obj``.
+ """
+ return [obj] if not isinstance(obj, list) else obj
+
+
+def _owner_pkgname_to_name(owner: str | None, pkgname: str) -> str:
+ """
+ Convert owner and package name to a standardized component name.
+
+ This function combines owner and package name with a forward slash when
+ both are provided, otherwise returns just the package name.
+
+ Args:
+ owner: The owner/username of the package (can be None)
+ pkgname: The name of the package
+
+ Returns:
+ str: The standardized component name in "owner/pkgname" format or just "pkgname"
+ """
+ return f"{owner}/{pkgname}" if owner else pkgname
+
+
+def collect_filtered_files(src_dir: PathType, src_filters: list[str]) -> list[str]:
+ """
+ Recursively match files in a directory according to include/exclude patterns.
+
+ This function processes a list of filter strings that indicate which files
+ to include or exclude. Each filter is parsed into patterns with a sign:
+ '+' for inclusion and '-' for exclusion. Directory patterns ending with '/'
+ are normalized to include all their contents recursively.
+
+ Args:
+ src_dir (PathType): Root directory to search within.
+ src_filters (list[str]): List of filter strings, which may contain multiple
+ patterns. Each pattern can start with '+' or '-' to indicate inclusion
+ or exclusion.
+
+ Returns:
+ list[str]: List of matched file paths as strings. Only files (not directories)
+ are returned, even if a directory matches a pattern.
+ """
+ matches = list(
+ itertools.chain.from_iterable(
+ FILTER_REGEX.findall(src_filter) for src_filter in src_filters
+ )
+ )
+
+ selected = set()
+
+ for sign, pattern in matches:
+ pattern = pattern.strip()
+
+ if pattern.endswith("/"):
+ pattern = pattern.rstrip("/") + "/**"
+
+ # glob.escape has no pathlib equivalent and the matcher works on raw
+ # path strings, so PTH118/PTH207 don't apply here.
+ full_pattern = os.path.join(glob.escape(str(src_dir)), pattern) # noqa: PTH118
+
+ matched = []
+ for item in glob.glob(full_pattern, recursive=True): # noqa: PTH207
+ if not Path(item).is_dir():
+ matched.append(item)
+ else:
+ # PlatformIO quirk: a directory matched with "*" should include all its
+ # nested files and subdirectories, not just the directory itself.
+ for root, _, files in os.walk(item):
+ matched.extend([str(Path(root) / f) for f in files])
+
+ # FILTER_REGEX only ever captures "+" or "-", so the else is the "-" case.
+ if sign == "+":
+ selected.update(matched)
+ else:
+ selected.difference_update(matched)
+
+ return [r for r in selected if Path(r).is_file()]
+
+
+def split_list_by_condition(
+ items: list[str], match_fn: Callable[[str], str | None]
+) -> tuple[list[str], list[str]]:
+ """
+ Splits a list into two lists based on a matching function.
+
+ Args:
+ items: List of items to split.
+ match_fn: Function that returns a value for items that should go into the "matched" list.
+
+ Returns:
+ A tuple (matched, non_matched)
+ """
+ matched = []
+ non_matched = []
+ for item in items:
+ result = match_fn(item)
+ if result:
+ matched.append(result)
+ else:
+ non_matched.append(item)
+ return matched, non_matched
+
+
+def check_library_data(data: dict, platform: str, framework: str):
+ """
+ Check whether a library manifest is compatible with the target toolchain.
+
+ A platform mismatch (e.g. an AVR-only library on ESP32) raises
+ ``InvalidLibrary`` so the caller skips the library. A framework mismatch only
+ logs a warning — PIO manifests often understate the frameworks they actually
+ compile under, and there's no opt-out at this layer, so we include the library
+ anyway.
+
+ Args:
+ data: PIO library manifest dict being processed.
+ platform: The PlatformIO platform token the build targets (e.g.
+ ``espressif32``).
+ framework: The active framework name (e.g. ``espidf``, ``arduino``,
+ ``zephyr``) the manifest is expected to declare.
+
+ Raises:
+ InvalidLibrary: If the library does not support the target platform.
+ """
+ platforms = data.get("platforms", "*")
+ if isinstance(platforms, str):
+ platforms = [a.strip() for a in platforms.split(",")]
+ platforms = ensure_list(platforms)
+
+ # Check if library supports the target platform
+ valid_platforms = "*" in platforms or platform in platforms
+
+ if not valid_platforms:
+ raise InvalidLibrary(f"Unsupported library platforms: {platforms}")
+
+ frameworks = data.get("frameworks", "*")
+ if isinstance(frameworks, str):
+ frameworks = [a.strip() for a in frameworks.split(",")]
+ frameworks = ensure_list(frameworks)
+
+ # Check if library declares the active framework. PIO library manifests
+ # often list only "arduino" even when the library actually compiles fine
+ # under the target framework, and there's no way to opt out of the check at
+ # this layer. Warn instead of failing so the user isn't forced to fork the
+ # library to fix the manifest.
+ valid_framework = "*" in frameworks or framework in frameworks
+
+ if not valid_framework:
+ _LOGGER.warning(
+ "Library %s declares frameworks %s that do not include '%s'; including anyway",
+ data.get("name", ""),
+ frameworks,
+ framework,
+ )
+
+
+def _parse_library_json(library_json_path: PathType):
+ """
+ Load and parse a JSON file describing a library.
+
+ Args:
+ library_json_path (PathType): Path to the JSON file.
+
+ Returns:
+ dict: Parsed JSON content as a Python dictionary.
+ """
+ with Path(library_json_path).open(encoding="utf8") as fp:
+ return json.load(fp)
+
+
+def _parse_library_properties(library_properties_path: PathType):
+ """
+ Parse a key-value platformio .properties style file into a dictionary.
+
+ Args:
+ library_properties_path (PathType): Path to the properties file.
+
+ Returns:
+ dict[str, str]: Mapping of parsed property keys to values.
+ """
+ with Path(library_properties_path).open(encoding="utf8") as fp:
+ data = {}
+ for line in fp.read().splitlines():
+ line = line.strip()
+ if not line or "=" not in line:
+ continue
+ # skip comments
+ if line.startswith("#"):
+ continue
+ key, value = line.split("=", 1)
+ if not value.strip():
+ continue
+ data[key.strip()] = value.strip()
+ return data
+
+
+def _make_registry_client() -> Any:
+ """Create a minimal PlatformIO registry client with no system filtering.
+
+ ``is_system_compatible`` is forced True so version selection is driven purely
+ by the requested version requirements -- target compatibility is handled
+ elsewhere, not by the PlatformIO registry.
+ """
+ from platformio.package.manager._registry import PackageManagerRegistryMixin
+
+ class _Registry(PackageManagerRegistryMixin):
+ def __init__(self) -> None:
+ self._registry_client = None
+ self.pkg_type = "library"
+
+ @staticmethod
+ def is_system_compatible(value: Any, custom_system: Any = None) -> bool:
+ return True
+
+ return _Registry()
+
+
+def _resolve_registry_version(
+ owner: str | None, pkgname: str, requirements: set[str]
+) -> tuple[str, str, str, str]:
+ """Resolve a registry package to the single highest version satisfying ALL
+ the given requirements; return ``(owner, name, version, download_url)``.
+
+ Intersecting every requirement (rather than resolving each consumer in
+ isolation) makes the result independent of processing order and guarantees
+ no stated constraint is violated -- e.g. ``esphome/libsodium`` requested as
+ both ``==1.10021.0`` and ``^1.10018.1`` resolves to ``1.10021.0``.
+ """
+ from platformio.package.meta import PackageSpec
+
+ registry = _make_registry_client()
+ package = registry.fetch_registry_package(PackageSpec(owner=owner, name=pkgname))
+ owner = package["owner"]["username"]
+ name = package["name"]
+
+ # Chaining the per-requirement filter intersects all constraints.
+ versions = package.get("versions") or []
+ for requirement in sorted(requirements):
+ versions = registry.get_compatible_registry_versions(
+ versions, PackageSpec(owner=owner, name=name, requirements=requirement)
+ )
+ if not versions:
+ raise RuntimeError(
+ f"No version of {owner}/{name} satisfies all requirements "
+ f"{sorted(requirements)} requested across the library tree"
+ )
+
+ best = registry.pick_best_registry_version(versions)
+ pkgfile = registry.pick_compatible_pkg_file(best["files"])
+ if not pkgfile:
+ raise RuntimeError(f"No package file for {owner}/{name}@{best['name']}")
+ return owner, name, best["name"], pkgfile["download_url"]
+
+
+def _normalize_dependencies(dependencies: Any) -> list[dict]:
+ """Normalize a library manifest's ``dependencies`` to a list of dicts.
+
+ PIO's library.json accepts both the list-of-dicts form and the shorthand
+ dict form (``{"owner/Name": "version_spec"}``); normalize the latter so
+ callers see a uniform list.
+ """
+ if not dependencies:
+ return []
+ if isinstance(dependencies, dict):
+ normalized = []
+ for raw_name, spec in dependencies.items():
+ if "/" in raw_name:
+ owner, pkgname = raw_name.split("/", 1)
+ else:
+ owner, pkgname = None, raw_name
+ entry = {"name": pkgname, "owner": owner}
+ if isinstance(spec, dict):
+ entry.update(spec)
+ else:
+ entry["version"] = spec
+ normalized.append(entry)
+ return normalized
+ return [d for d in dependencies if isinstance(d, dict)]
+
+
+@dataclass
+class _LibNode:
+ """A node in the library dependency graph being resolved as a batch."""
+
+ key: str
+ is_git: bool
+ owner: str | None = None
+ pkgname: str | None = None
+ requirements: set[str] = field(default_factory=set)
+ url: str | None = None
+ ref: str | None = None
+ edges: set[str] = field(default_factory=set)
+
+
+def _node_key(
+ name: str | None, version: str | None, repository: str | None
+) -> tuple[str, bool, tuple[str | None, str | None]]:
+ """Return ``(key, is_git, locator)`` for a library or dependency spec.
+
+ The key is derived from the *input* spec (the registry name as written, or
+ the git URL path), not the resolved canonical name. So a package referenced
+ inconsistently -- bare ``name`` vs ``owner/name``, or git vs registry -- maps
+ to distinct keys and isn't deduplicated; ``convert_libraries`` warns about
+ that after resolution rather than merging the nodes.
+ """
+ if repository:
+ split_result = urlsplit(repository)
+ key = str(split_result.path).strip("/").removesuffix(".git")
+ ref = split_result.fragment.strip() or None
+ url = urlunsplit(split_result._replace(fragment=""))
+ return key, True, (url, ref)
+ if name and "/" in name:
+ owner, pkgname = name.split("/", 1)
+ else:
+ owner, pkgname = None, name
+ return name, False, (owner, pkgname)
+
+
+def convert_libraries(
+ libraries: list[Library], backend: LibraryBackend
+) -> list[ConvertedLibrary]:
+ """Resolve and convert a batch of PlatformIO libraries for ``backend``.
+
+ Resolves the whole set together rather than each library independently: it
+ walks the dependency graph collecting every version *requirement* per
+ component name, then resolves each name once to a single version satisfying
+ all of them. So a transitive dependency shared under
+ different specs (e.g. ``esphome/libsodium``, pulled by both ``noise-c`` and
+ ``esp_wireguard``) becomes one component instead of two clashing
+ ``override_path`` entries -- order-independently, and without ever violating
+ a stated constraint.
+
+ The returned list holds the top-level components (those directly requested);
+ transitive dependencies are converted too and wired into each component's
+ generated manifest. ``backend.emit`` is called once per converted library to
+ write its toolchain-specific build files.
+
+ ``lib_ignore`` from ``esphome->platformio_options`` excludes libraries by
+ short name (part after the ``/``), matched against both the top-level
+ libraries and every dependency discovered during the graph walk.
+ """
+ nodes: dict[str, _LibNode] = {}
+
+ lib_ignore = {
+ name.split("/")[-1].lower()
+ for name in CORE.platformio_options.get("lib_ignore", [])
+ }
+
+ # The generated build files inside the shared cache bake in the dependency
+ # wiring, which lib_ignore changes; salt the cache path so configs with
+ # different lib_ignore values don't fight over (and constantly rewrite) the
+ # same converted component files.
+ salt = (
+ hashlib.sha256(",".join(sorted(lib_ignore)).encode()).hexdigest()[:8]
+ if lib_ignore
+ else ""
+ )
+
+ def is_ignored(name: str | None) -> bool:
+ if not lib_ignore or name is None:
+ return False
+ return name.split("/")[-1].lower() in lib_ignore
+
+ def add_spec(name: str | None, version: str | None, repository: str | None) -> str:
+ key, is_git, locator = _node_key(name, version, repository)
+ node = nodes.get(key) or _LibNode(key=key, is_git=is_git)
+ nodes[key] = node
+ if is_git:
+ node.is_git = True
+ node.url, node.ref = locator
+ else:
+ node.owner, node.pkgname = locator
+ if version:
+ node.requirements.add(version)
+ return key
+
+ top_level = [
+ add_spec(library.name, library.version, library.repository)
+ for library in libraries
+ if not is_ignored(library.name)
+ ]
+
+ # Collect + resolve to a fixpoint: a node is (re)resolved whenever its
+ # requirement set has grown since the last time, so every requirement in the
+ # graph is accounted for before conversion.
+ components: dict[str, ConvertedLibrary] = {}
+ resolved_requirements: dict[str, frozenset[str]] = {}
+ top_level_keys = set(top_level)
+ worklist = deque(dict.fromkeys(top_level))
+ while worklist:
+ key = worklist.popleft()
+ node = nodes[key]
+
+ # A node is queued once per referring edge; skip the (uncached) registry
+ # lookup + download + dependency walk unless its requirement set grew
+ # since the last resolve. Requirements only ever grow, so this still
+ # converges the fixpoint and terminates dependency cycles.
+ requirements = frozenset(node.requirements)
+ if resolved_requirements.get(key) == requirements:
+ continue
+ resolved_requirements[key] = requirements
+
+ if node.is_git:
+ component = ConvertedLibrary(key, "*", GitSource(node.url, node.ref))
+ else:
+ owner, name, version, url = _resolve_registry_version(
+ node.owner, node.pkgname, node.requirements
+ )
+ component = ConvertedLibrary(
+ _owner_pkgname_to_name(owner, name), version, URLSource(url)
+ )
+ component.download(salt=salt)
+
+ library_json_path = component.path / "library.json"
+ library_properties_path = component.path / "library.properties"
+ if library_json_path.is_file():
+ component.data = _parse_library_json(library_json_path)
+ elif library_properties_path.is_file():
+ component.data = _parse_library_properties(library_properties_path)
+ else:
+ raise RuntimeError(
+ f"Invalid PIO library {key}: missing library.json and "
+ "library.properties"
+ )
+
+ try:
+ check_library_data(component.data, backend.platform, backend.framework)
+ except InvalidLibrary as e:
+ # Skip an incompatible transitive dependency, but fail fast if a
+ # top-level library the build explicitly requested is incompatible.
+ if key in top_level_keys:
+ raise RuntimeError(
+ f"Requested library {key} is not compatible with "
+ f"{backend.framework}: {e}"
+ ) from e
+ _LOGGER.debug("Skip incompatible dependency %s: %s", key, str(e))
+ continue
+ components[key] = component
+
+ # Requirements changed (we got past the short-circuit above), so
+ # (re)walk this component's dependencies.
+ node.edges = set()
+ for dependency in _normalize_dependencies(component.data.get("dependencies")):
+ if "name" not in dependency or "version" not in dependency:
+ continue
+ try:
+ check_library_data(dependency, backend.platform, backend.framework)
+ except InvalidLibrary as e:
+ _LOGGER.debug("Skip dependency %s: %s", dependency.get("name"), str(e))
+ continue
+ dep_name = _owner_pkgname_to_name(
+ dependency.get("owner"), dependency.get("name")
+ )
+ if is_ignored(dep_name):
+ _LOGGER.debug("Skip ignored dependency %s", dep_name)
+ continue
+ # The version field may actually be a URL (git/archive dependency).
+ dep_version = dependency["version"]
+ dep_url = None
+ try:
+ parsed = urlparse(dep_version)
+ if all([parsed.scheme, parsed.netloc]):
+ dep_url, dep_version = dep_version, None
+ except (TypeError, ValueError):
+ pass
+ dep_key = add_spec(dep_name, dep_version, dep_url)
+ node.edges.add(dep_key)
+ worklist.append(dep_key)
+
+ # A git source wins over any registry version requested for the same
+ # component. That's intentional, but warn so a dropped registry pin isn't a
+ # silent surprise.
+ for node in nodes.values():
+ if node.is_git and node.requirements:
+ _LOGGER.warning(
+ "Library %s is requested both from a git source (%s) and as "
+ "registry version(s) %s; using the git source.",
+ node.key,
+ node.url,
+ sorted(node.requirements),
+ )
+
+ # Two graph nodes that resolve to the same component name (e.g. a package
+ # referenced both bare and as ``owner/name``) are not deduplicated and can
+ # produce conflicting component definitions. Warn so it's not silent.
+ canonical_keys: dict[str, str] = {}
+ for node_key, component in components.items():
+ canonical = component.get_sanitized_name()
+ if canonical_keys.setdefault(canonical, node_key) != node_key:
+ _LOGGER.warning(
+ "Library %s is referenced under multiple names (%s and %s); these "
+ "are not deduplicated. Reference it consistently as %s.",
+ canonical,
+ canonical_keys[canonical],
+ node_key,
+ canonical,
+ )
+
+ # Wire each component's dependencies to the single resolved instances, then
+ # emit build files.
+ for key, component in components.items():
+ component.dependencies = [
+ components[dep_key]
+ for dep_key in sorted(nodes[key].edges)
+ if dep_key in components
+ ]
+ for component in components.values():
+ backend.emit(component)
+
+ return [components[key] for key in top_level if key in components]
diff --git a/tests/unit_tests/test_espidf_component.py b/tests/unit_tests/test_espidf_component.py
index 87e168dc94..d43a1d5276 100644
--- a/tests/unit_tests/test_espidf_component.py
+++ b/tests/unit_tests/test_espidf_component.py
@@ -14,23 +14,23 @@ from esphome.const import (
Platform,
)
from esphome.core import CORE, Library
-import esphome.espidf.component
from esphome.espidf.component import (
+ generate_cmakelists_txt,
+ generate_idf_component_yml,
+ generate_idf_components,
+)
+import esphome.platformio.library
+from esphome.platformio.library import (
+ ConvertedLibrary as IDFComponent,
GitSource,
- IDFComponent,
- InvalidIDFComponent,
URLSource,
- _check_library_data,
- _collect_filtered_files,
_node_key,
_normalize_dependencies,
_parse_library_json,
_parse_library_properties,
_resolve_registry_version,
- _split_list_by_condition,
- generate_cmakelists_txt,
- generate_idf_component_yml,
- generate_idf_components,
+ collect_filtered_files,
+ split_list_by_condition,
)
@@ -70,7 +70,7 @@ def test_collect_filtered_files_basic(tmp_path):
f2.parent.mkdir(parents=True)
f2.write_text("int b;")
- result = _collect_filtered_files(tmp_path, ["+<*>"])
+ result = collect_filtered_files(tmp_path, ["+<*>"])
assert str(f1) in result
assert str(f2) in result
@@ -81,7 +81,7 @@ def test_collect_filtered_files_exclude(tmp_path):
f1.write_text("int a;")
f2.write_text("int b;")
- result = _collect_filtered_files(tmp_path, ["+<*> -<*.cpp>"])
+ result = collect_filtered_files(tmp_path, ["+<*> -<*.cpp>"])
assert str(f1) in result
assert str(f2) not in result
@@ -89,7 +89,7 @@ def test_collect_filtered_files_exclude(tmp_path):
def test_split_list_by_condition():
items = ["-Iinclude", "-Llib", "-Wall"]
- matched, rest = _split_list_by_condition(
+ matched, rest = split_list_by_condition(
items, lambda x: x[2:] if x.startswith("-I") else None
)
@@ -202,41 +202,6 @@ def test_generate_idf_component_yml_missing_path_raises(tmp_component):
generate_idf_component_yml(tmp_component)
-def test_check_library_data_valid(esp32_idf_core):
- _check_library_data({"platforms": "*", "frameworks": "*"})
-
-
-def test_check_library_data_valid2(esp32_idf_core):
- _check_library_data({"platforms": "*"})
-
-
-def test_check_library_data_valid3(esp32_idf_core):
- _check_library_data({})
-
-
-def test_check_library_data_valid4(esp32_idf_core):
- _check_library_data({"platforms": "espressif32", "frameworks": "*"})
-
-
-def test_check_library_data_valid5(esp32_idf_core):
- _check_library_data({"platforms": "*", "frameworks": "espidf"})
-
-
-def test_check_library_data_invalid_platform(esp32_idf_core):
- with pytest.raises(InvalidIDFComponent):
- _check_library_data({"platforms": ["other"], "frameworks": "*"})
-
-
-def test_check_library_data_invalid_framework(
- esp32_idf_core: None, caplog: pytest.LogCaptureFixture
-) -> None:
- # Framework mismatch is a warning, not a hard skip: the library is still
- # included so that PIO manifests that only list "arduino" (but actually
- # compile under IDF) can be used without forking them.
- _check_library_data({"name": "lib", "platforms": "*", "frameworks": ["other"]})
- assert "do not include 'espidf'" in caplog.text
-
-
def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
from esphome.espidf.extra_script import captured_as_build_flags, run_extra_script
@@ -453,7 +418,7 @@ def _patch_registry(monkeypatch, versions):
``get_compatible_registry_versions`` / ``pick_best_registry_version`` run on
the canned data so the intersection logic is exercised for real.
"""
- registry = esphome.espidf.component._make_registry_client()
+ registry = esphome.platformio.library._make_registry_client()
monkeypatch.setattr(
registry,
"fetch_registry_package",
@@ -467,7 +432,7 @@ def _patch_registry(monkeypatch, versions):
},
)
monkeypatch.setattr(
- esphome.espidf.component, "_make_registry_client", lambda: registry
+ esphome.platformio.library, "_make_registry_client", lambda: registry
)
@@ -535,7 +500,7 @@ def test_generate_idf_components_dedupes_shared_dependency(
return owner, pkgname, version, f"http://x/{pkgname}.tar.gz"
monkeypatch.setattr(
- esphome.espidf.component, "_resolve_registry_version", fake_resolve
+ esphome.platformio.library, "_resolve_registry_version", fake_resolve
)
top = generate_idf_components(
@@ -594,7 +559,7 @@ def test_generate_idf_components_lib_ignore_filters_top_level_and_dependencies(
return owner, pkgname, "1.0.0", f"http://x/{pkgname}.tar.gz"
monkeypatch.setattr(
- esphome.espidf.component, "_resolve_registry_version", fake_resolve
+ esphome.platformio.library, "_resolve_registry_version", fake_resolve
)
# lib_ignore is read from CORE.platformio_options (stored there by
# _add_platformio_options); matched by lowercase short name.
@@ -640,7 +605,7 @@ def test_generate_idf_components_handles_dependency_cycle(
monkeypatch.setattr(IDFComponent, "download", fake_download)
monkeypatch.setattr(
- esphome.espidf.component,
+ esphome.platformio.library,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner,
@@ -697,7 +662,7 @@ def test_generate_idf_components_git_overrides_registry_warns(
monkeypatch.setattr(IDFComponent, "download", fake_download)
monkeypatch.setattr(
- esphome.espidf.component,
+ esphome.platformio.library,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner,
@@ -733,7 +698,7 @@ def test_generate_idf_components_missing_manifest_raises(
monkeypatch.setattr(IDFComponent, "download", fake_download)
monkeypatch.setattr(
- esphome.espidf.component,
+ esphome.platformio.library,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner,
@@ -777,7 +742,7 @@ def test_generate_idf_components_warns_on_noncanonical_duplicate(
monkeypatch.setattr(IDFComponent, "download", fake_download)
# Bare "shared" and "owner/shared" both resolve to canonical owner/shared.
monkeypatch.setattr(
- esphome.espidf.component,
+ esphome.platformio.library,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner or "owner",
@@ -810,7 +775,7 @@ def test_generate_idf_components_incompatible_top_level_raises(
monkeypatch.setattr(IDFComponent, "download", fake_download)
monkeypatch.setattr(
- esphome.espidf.component,
+ esphome.platformio.library,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner,
@@ -820,7 +785,7 @@ def test_generate_idf_components_incompatible_top_level_raises(
),
)
- with pytest.raises(RuntimeError, match="not compatible with ESP-IDF"):
+ with pytest.raises(RuntimeError, match="not compatible with espidf"):
generate_idf_components([Library("esphome/A", "1.0.0", None)])
@@ -846,7 +811,7 @@ def test_generate_idf_components_incompatible_dependency_skipped(
monkeypatch.setattr(IDFComponent, "download", fake_download)
monkeypatch.setattr(
- esphome.espidf.component,
+ esphome.platformio.library,
"_resolve_registry_version",
lambda owner, pkgname, requirements: (
owner,
@@ -892,7 +857,7 @@ def test_git_source_salt_scopes_domain(monkeypatch: pytest.MonkeyPatch) -> None:
return Path("/cloned"), None
monkeypatch.setattr(
- esphome.espidf.component.git, "clone_or_update", fake_clone_or_update
+ esphome.platformio.library.git, "clone_or_update", fake_clone_or_update
)
source = GitSource("https://github.com/esphome/noise-c.git", "v1.0")
diff --git a/tests/unit_tests/test_platformio_library.py b/tests/unit_tests/test_platformio_library.py
new file mode 100644
index 0000000000..55bc396c25
--- /dev/null
+++ b/tests/unit_tests/test_platformio_library.py
@@ -0,0 +1,229 @@
+"""Tests for the toolchain-agnostic PlatformIO library converter.
+
+Covers the shared download/parse/resolve/dependency-walk paths in
+``esphome.platformio.library`` directly (the ESP-IDF and Zephyr backends are
+exercised in their own test modules)."""
+
+import json
+import logging
+from pathlib import Path
+
+import pytest
+
+from esphome.core import Library
+import esphome.platformio.library as lib
+from esphome.platformio.library import (
+ ConvertedLibrary,
+ GitSource,
+ InvalidLibrary,
+ LibraryBackend,
+ Source,
+ URLSource,
+ _resolve_registry_version,
+ check_library_data,
+ convert_libraries,
+)
+
+
+def _backend(emit=lambda component: None) -> LibraryBackend:
+ return LibraryBackend(platform="espressif32", framework="espidf", emit=emit)
+
+
+def test_check_library_data_accepts_wildcards():
+ check_library_data({"platforms": "*", "frameworks": "*"}, "espressif32", "espidf")
+
+
+def test_check_library_data_accepts_missing_frameworks():
+ check_library_data({"platforms": "*"}, "espressif32", "espidf")
+
+
+def test_check_library_data_accepts_empty_manifest():
+ check_library_data({}, "espressif32", "espidf")
+
+
+def test_check_library_data_accepts_matching_platform():
+ check_library_data(
+ {"platforms": "espressif32", "frameworks": "*"}, "espressif32", "espidf"
+ )
+
+
+def test_check_library_data_accepts_matching_framework():
+ check_library_data(
+ {"platforms": "*", "frameworks": "espidf"}, "espressif32", "espidf"
+ )
+
+
+def test_check_library_data_rejects_unsupported_platform():
+ with pytest.raises(InvalidLibrary):
+ check_library_data(
+ {"platforms": ["other"], "frameworks": "*"}, "espressif32", "espidf"
+ )
+
+
+def test_check_library_data_warns_on_framework_mismatch(
+ caplog: pytest.LogCaptureFixture,
+):
+ # Framework mismatch is a warning, not a hard skip: the library is still
+ # included so manifests that only list "arduino" (but compile fine under the
+ # target framework) can be used without forking them.
+ with caplog.at_level(logging.WARNING, logger="esphome.platformio.library"):
+ check_library_data(
+ {"name": "lib", "platforms": "*", "frameworks": ["other"]},
+ "espressif32",
+ "espidf",
+ )
+ assert "do not include 'espidf'" in caplog.text
+
+
+def test_source_download_not_implemented():
+ with pytest.raises(NotImplementedError):
+ Source().download("x")
+
+
+def test_gitsource_str_includes_ref_when_present():
+ assert str(GitSource("http://git/repo.git", "main")) == "http://git/repo.git#main"
+ assert str(GitSource("http://git/repo.git", None)) == "http://git/repo.git"
+
+
+def test_urlsource_download_extracts_then_reuses_marker(setup_core, monkeypatch):
+ monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
+ dl_calls: list[list[str]] = []
+ monkeypatch.setattr(
+ lib, "download_from_mirrors", lambda urls, headers, f: dl_calls.append(urls)
+ )
+
+ def fake_extract(fileobj, path):
+ Path(path).mkdir(parents=True, exist_ok=True)
+
+ monkeypatch.setattr(lib, "archive_extract_all", fake_extract)
+
+ src = URLSource("http://example.test/lib.tar.gz")
+ out = src.download("mylib")
+
+ assert (out / ".esphome_extracted").is_file()
+ assert dl_calls == [["http://example.test/lib.tar.gz"]]
+
+ # The completion marker means a second download is skipped (cache hit).
+ out2 = src.download("mylib")
+ assert out2 == out
+ assert len(dl_calls) == 1
+
+
+def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
+ registry = lib._make_registry_client()
+ monkeypatch.setattr(
+ registry,
+ "fetch_registry_package",
+ lambda spec: {
+ "owner": {"username": spec.owner or "owner"},
+ "name": spec.name,
+ "versions": [{"name": "1.0.0", "files": [{}]}],
+ },
+ )
+ # A best version exists but none of its files is a compatible package.
+ monkeypatch.setattr(
+ registry, "pick_best_registry_version", lambda versions: versions[0]
+ )
+ monkeypatch.setattr(registry, "pick_compatible_pkg_file", lambda files: None)
+ monkeypatch.setattr(lib, "_make_registry_client", lambda: registry)
+
+ with pytest.raises(RuntimeError, match="No package file"):
+ _resolve_registry_version("owner", "pkg", set())
+
+
+def _patch_download_with_manifests(monkeypatch, tmp_path, manifests, *, properties=()):
+ """Fake ConvertedLibrary.download to materialize canned manifests on disk."""
+
+ def fake_download(self, force=False, salt=""):
+ self.path = tmp_path / self.get_sanitized_name().replace("/", "__")
+ self.path.mkdir(parents=True, exist_ok=True)
+ if self.name in properties:
+ (self.path / "library.properties").write_text(manifests[self.name])
+ else:
+ (self.path / "library.json").write_text(json.dumps(manifests[self.name]))
+
+ monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
+ monkeypatch.setattr(
+ lib,
+ "_resolve_registry_version",
+ lambda owner, pkgname, requirements: (
+ owner,
+ pkgname,
+ "1.0.0",
+ f"http://x/{pkgname}.tar.gz",
+ ),
+ )
+
+
+def test_convert_libraries_parses_library_properties(tmp_path, monkeypatch):
+ # A manifest provided as library.properties (Arduino style) instead of
+ # library.json must still be parsed and converted.
+ _patch_download_with_manifests(
+ monkeypatch,
+ tmp_path,
+ {"esphome/A": "name=A\nversion=1.0\n"},
+ properties=("esphome/A",),
+ )
+
+ emitted: list[ConvertedLibrary] = []
+ top = convert_libraries(
+ [Library("esphome/A", "1.0.0", None)], _backend(emitted.append)
+ )
+
+ assert [c.name for c in top] == ["esphome/A"]
+ assert top[0].data["name"] == "A"
+ assert emitted[0].data["version"] == "1.0"
+
+
+def test_convert_libraries_skips_dependency_without_version(tmp_path, monkeypatch):
+ # A dependency entry lacking a version is malformed and silently skipped.
+ _patch_download_with_manifests(
+ monkeypatch,
+ tmp_path,
+ {"esphome/A": {"name": "A", "dependencies": [{"name": "C"}]}},
+ )
+
+ # No version on the top-level spec exercises the "no requirement" path too.
+ top = convert_libraries([Library("esphome/A", None, None)], _backend())
+
+ assert top[0].dependencies == []
+
+
+def test_convert_libraries_handles_unparsable_dependency_version(tmp_path, monkeypatch):
+ # If the git/archive URL probe (urlparse) raises on a malformed value, the
+ # dependency is still kept and treated as a plain version spec.
+ _patch_download_with_manifests(
+ monkeypatch,
+ tmp_path,
+ {
+ "esphome/A": {
+ "name": "A",
+ # An unterminated IPv6 URL makes urlparse raise ValueError.
+ "dependencies": [{"name": "C", "version": "http://[::1"}],
+ },
+ "C": {"name": "C"},
+ },
+ )
+
+ top = convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
+
+ assert [d.name for d in top[0].dependencies] == ["C"]
+
+
+def test_convert_libraries_skips_incompatible_dependency(tmp_path, monkeypatch):
+ # A dependency that declares an incompatible platform is skipped (the
+ # top-level library still builds).
+ _patch_download_with_manifests(
+ monkeypatch,
+ tmp_path,
+ {
+ "esphome/A": {
+ "name": "A",
+ "dependencies": [{"name": "C", "version": "1.0", "platforms": ["avr"]}],
+ }
+ },
+ )
+
+ top = convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
+
+ assert top[0].dependencies == []