mirror of
https://github.com/esphome/esphome.git
synced 2026-08-22 22:26:21 +00:00
779 lines
29 KiB
Python
779 lines
29 KiB
Python
"""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 ``<data_dir>/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
|
|
from urllib.parse import 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/> -<example/> -<examples/> -<test/> -<tests/>"
|
|
)
|
|
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 = "", namespace: 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 = "", namespace: str = ""
|
|
) -> Path:
|
|
# Namespace the cache per backend (e.g. pio_components/idf, .../zephyr) so
|
|
# the build files each backend writes into the library dir can't collide.
|
|
base_dir = Path(CORE.data_dir) / DOMAIN
|
|
if namespace:
|
|
base_dir = base_dir / namespace
|
|
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 = "", namespace: str = ""
|
|
) -> Path:
|
|
domain = DOMAIN
|
|
if namespace:
|
|
domain = f"{domain}/{namespace}"
|
|
if salt:
|
|
domain = f"{domain}/{salt}"
|
|
path, _ = git.clone_or_update(
|
|
url=self.url,
|
|
ref=self.ref,
|
|
refresh=git.NEVER_REFRESH if not force else None,
|
|
domain=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 = "", namespace: 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``). ``namespace`` keeps each backend's cache separate.
|
|
"""
|
|
self.path = self.source.download(
|
|
self.get_sanitized_name(), force=force, salt=salt, namespace=namespace
|
|
)
|
|
|
|
|
|
@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``).
|
|
``cache_key`` namespaces the download cache (``pio_components/<cache_key>/``)
|
|
so the differing build files two backends emit into a library dir never
|
|
collide when the same config dir hosts both an ESP-IDF and a Zephyr build.
|
|
"""
|
|
|
|
platform: str | None
|
|
framework: str
|
|
emit: Callable[["ConvertedLibrary"], None]
|
|
cache_key: str
|
|
|
|
|
|
def ensure_list[T](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])
|
|
|
|
# glob keeps the pattern's literal separators for non-wildcard path
|
|
# components, so on Windows the same file can surface with different
|
|
# separators depending on where the wildcards sit; normalize so the
|
|
# include/exclude set operations below compare equal paths.
|
|
matched = [os.path.normpath(m) for m in matched]
|
|
|
|
# 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 | None, 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``). ``None`` skips the platform check entirely — useful
|
|
for targets (e.g. Zephyr) where PIO manifests rarely declare the
|
|
platform yet portable libraries still build.
|
|
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 = platform is None or "*" 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", "<unknown>"),
|
|
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 _url_or_none(value: Any) -> str | None:
|
|
"""Return ``value`` if it parses as a URL (scheme and host), else None."""
|
|
if not value or not isinstance(value, str):
|
|
return None
|
|
try:
|
|
parsed = urlsplit(value)
|
|
except ValueError:
|
|
return None
|
|
return value if parsed.scheme and parsed.netloc else None
|
|
|
|
|
|
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.
|
|
|
|
PlatformIO's Library Manager also accepted a git URL in the *name*
|
|
position (``add_library("https://github.com/x/y", None)``), including the
|
|
``git+`` VCS prefix and the ``CustomName=URL`` form; recognize those here
|
|
so such specs resolve as git sources instead of failing a registry lookup.
|
|
"""
|
|
if not repository and name and "://" in name:
|
|
# Try the whole name first so a bare URL whose query contains ``=``
|
|
# stays intact; fall back to the ``CustomName=URL`` form, where the
|
|
# key derives from the URL path and the custom name is irrelevant.
|
|
repository = _url_or_none(name) or _url_or_none(name.split("=", 1)[-1])
|
|
if repository is None:
|
|
# Anything with ``://`` was meant to be a URL; failing it fast
|
|
# beats a confusing registry "package not found" error.
|
|
raise RuntimeError(f"Invalid PIO library URL: {name}")
|
|
if repository:
|
|
split_result = urlsplit(repository.removeprefix("git+"))
|
|
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, namespace=backend.cache_key)
|
|
|
|
library_json_path = component.path / "library.json"
|
|
library_properties_path = component.path / "library.properties"
|
|
has_json = library_json_path.is_file()
|
|
has_properties = library_properties_path.is_file()
|
|
if not has_json and not has_properties:
|
|
# The shared cache can hold a broken copy (e.g. a clone or an
|
|
# extraction interrupted by a killed process). Force one
|
|
# re-download so a bad cache entry self-heals instead of failing
|
|
# every build until the user runs a full clean.
|
|
_LOGGER.warning(
|
|
"Library %s at %s is missing library.json and library.properties; "
|
|
"re-downloading",
|
|
key,
|
|
component.path,
|
|
)
|
|
component.download(force=True, salt=salt, namespace=backend.cache_key)
|
|
has_json = library_json_path.is_file()
|
|
has_properties = library_properties_path.is_file()
|
|
if has_json:
|
|
component.data = _parse_library_json(library_json_path)
|
|
elif has_properties:
|
|
component.data = _parse_library_properties(library_properties_path)
|
|
else:
|
|
raise RuntimeError(
|
|
f"Invalid PIO library {key}: missing library.json and "
|
|
f"library.properties in {component.path}"
|
|
)
|
|
|
|
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 = _url_or_none(dep_version)
|
|
if dep_url is not None:
|
|
dep_version = None
|
|
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]
|