mirror of
https://github.com/esphome/esphome.git
synced 2026-08-22 22:26:21 +00:00
1348 lines
52 KiB
Python
1348 lines
52 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, Iterable
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass, field
|
|
import glob
|
|
import hashlib
|
|
import itertools
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import re
|
|
import tempfile
|
|
from typing import Any
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
from urllib.request import url2pathname
|
|
|
|
from esphome import git
|
|
from esphome.core import CORE, EsphomeError, Library
|
|
from esphome.framework_helpers import (
|
|
BatchDownloadProgress,
|
|
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 = []
|
|
# Suffix -> compiler kind (PlatformIO's CSUFFIXES/CXXSUFFIXES/ASSUFFIXES).
|
|
# "asm" merges SCons's AS and ASPP sets: all compile as assembler-with-cpp.
|
|
SOURCE_KIND_FOR_SUFFIX: dict[str, str] = {
|
|
".c": "c",
|
|
".cpp": "cxx",
|
|
".cc": "cxx",
|
|
".cxx": "cxx",
|
|
".c++": "cxx",
|
|
".C": "cxx",
|
|
".C++": "cxx",
|
|
".S": "asm",
|
|
".spp": "asm",
|
|
".SPP": "asm",
|
|
".sx": "asm",
|
|
".s": "asm",
|
|
".asm": "asm",
|
|
".ASM": "asm",
|
|
}
|
|
SRC_FILE_EXTENSIONS = list(SOURCE_KIND_FOR_SUFFIX)
|
|
# Suffixes that count as headers when probing whether a library has any
|
|
# usable files at all (compare against Path.suffix.lower())
|
|
LIBRARY_HEADER_SUFFIXES = frozenset(
|
|
{".h", ".hpp", ".hh", ".hxx", ".inc", ".ipp", ".tcc"}
|
|
)
|
|
|
|
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 = "",
|
|
progress: Callable[[int], None] | None = None,
|
|
) -> Path:
|
|
raise NotImplementedError
|
|
|
|
def source_root(self, build_path: Path) -> Path:
|
|
"""Directory holding the library's own files (manifest + sources).
|
|
|
|
Defaults to the downloaded build directory; a source that references its
|
|
files in place (:class:`LocalSource`) overrides this to point elsewhere.
|
|
"""
|
|
return build_path
|
|
|
|
|
|
class URLSource(Source):
|
|
def __init__(self, url: str):
|
|
self.url = url
|
|
|
|
def _cache_dir(self, dir_suffix: str, 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())
|
|
return base_dir / h.hexdigest()[:8] / dir_suffix
|
|
|
|
def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool:
|
|
"""Whether a completed extraction already exists for this source."""
|
|
return (
|
|
self._cache_dir(dir_suffix, salt, namespace) / ".esphome_extracted"
|
|
).is_file()
|
|
|
|
def download(
|
|
self,
|
|
dir_suffix: str,
|
|
force: bool = False,
|
|
salt: str = "",
|
|
namespace: str = "",
|
|
progress: Callable[[int], None] | None = None,
|
|
) -> Path:
|
|
path = self._cache_dir(dir_suffix, salt, namespace)
|
|
# 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:
|
|
if progress is None:
|
|
# A batch caller draws one combined bar and logs the list
|
|
_LOGGER.info("Downloading %s ...", self.url)
|
|
_LOGGER.debug("Location: %s", path)
|
|
|
|
download_from_mirrors([self.url], {}, tmp.file, progress=progress)
|
|
|
|
_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 = "",
|
|
progress: Callable[[int], None] | None = None,
|
|
) -> 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,
|
|
init_submodules=True,
|
|
subpath=Path(dir_suffix),
|
|
)
|
|
return path
|
|
|
|
def __str__(self):
|
|
return f"{self.url}#{self.ref}" if self.ref else self.url
|
|
|
|
|
|
class LocalSource(Source):
|
|
"""A library that already exists as a directory on the local filesystem.
|
|
|
|
Referenced with a ``file://`` URL (PlatformIO's spelling for a local library
|
|
folder). Nothing is copied: the backend generates its build files into an
|
|
otherwise empty cache directory and references the library's own sources in
|
|
place by absolute path (via :meth:`source_root`). So the user's source tree
|
|
stays untouched and edits are picked up on the next build without syncing.
|
|
"""
|
|
|
|
def __init__(self, path: str):
|
|
self.local_path = path
|
|
|
|
def download(
|
|
self,
|
|
dir_suffix: str,
|
|
force: bool = False,
|
|
salt: str = "",
|
|
namespace: str = "",
|
|
progress: Callable[[int], None] | None = None,
|
|
) -> Path:
|
|
src = Path(self.local_path)
|
|
if not src.is_dir():
|
|
# EsphomeError (not InvalidLibrary) so the CLI prints a clean message
|
|
# instead of a traceback -- pointing a file:// at a missing folder is
|
|
# the most common first mistake with a local library.
|
|
raise EsphomeError(
|
|
f"Local library directory does not exist: {self.local_path}"
|
|
)
|
|
base_dir = Path(CORE.data_dir) / DOMAIN
|
|
if namespace:
|
|
base_dir = base_dir / namespace
|
|
h = hashlib.new("sha256")
|
|
h.update(str(src.resolve()).encode())
|
|
if salt:
|
|
h.update(salt.encode())
|
|
# Only the generated build files live here; the library's own sources
|
|
# are referenced in place from source_root().
|
|
path = base_dir / h.hexdigest()[:8] / dir_suffix
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
def source_root(self, build_path: Path) -> Path:
|
|
return Path(self.local_path)
|
|
|
|
def __str__(self):
|
|
path = Path(self.local_path)
|
|
# as_uri() needs an absolute path; _node_key rejects relative file://
|
|
# URLs, but guard anyway so a diagnostic can't itself raise.
|
|
return path.as_uri() if path.is_absolute() else f"file://{self.local_path}"
|
|
|
|
|
|
class InvalidLibrary(Exception):
|
|
pass
|
|
|
|
|
|
class IncompatiblePlatform(InvalidLibrary):
|
|
"""The manifest's platform filter rejected the target platform.
|
|
|
|
A distinct type so callers can treat the routine cross-platform skip
|
|
differently from other manifest problems without matching message text.
|
|
"""
|
|
|
|
|
|
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
|
|
# Where the library's own files live (manifest + sources). Set by
|
|
# download(); equals path for registry/git, the user's dir for local.
|
|
self.source_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
|
|
|
|
@property
|
|
def source_dir(self) -> Path:
|
|
"""Directory the library's own files (manifest + sources) are read from.
|
|
|
|
The build dir for a registry/git source; the user's directory for a
|
|
local library. Backends read sources from here and emit their build
|
|
files into ``path``.
|
|
"""
|
|
return self.source_path or self.path
|
|
|
|
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 = "",
|
|
progress: Callable[[int], None] | None = None,
|
|
):
|
|
"""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,
|
|
progress=progress,
|
|
)
|
|
self.source_path = self.source.source_root(self.path)
|
|
|
|
|
|
@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
|
|
# Owner-less dependency names this returns True for are skipped by the
|
|
# walk; the backend supplies them outside the registry (e.g. core-bundled
|
|
# libraries).
|
|
provides: Callable[[str], bool] | None = None
|
|
|
|
|
|
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 | None):
|
|
"""
|
|
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. ``None`` skips
|
|
the framework check (and its warning), mirroring ``platform``.
|
|
|
|
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 IncompatiblePlatform(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 = framework is None or "*" 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 split_flag_entry(entry: Any, owner: str) -> list[str]:
|
|
"""``shlex.split`` with a clean error naming the offending flags entry."""
|
|
# Late import: shlex is only needed when actually lexing flags
|
|
import shlex
|
|
|
|
try:
|
|
return shlex.split(entry)
|
|
except (ValueError, AttributeError, TypeError) as err:
|
|
# AttributeError/TypeError: a dict or number from a third-party
|
|
# manifest; name the entry instead of an opaque shlex traceback
|
|
raise EsphomeError(f"Malformed build flag {entry!r} in {owner}: {err}") from err
|
|
|
|
|
|
def lex_build_flags(entries: str | list[str], owner: str) -> list[str]:
|
|
"""Shell-lex ``build.flags`` entries the way PlatformIO's ParseFlags
|
|
does; bare -I/-L/-l/-D tokens re-glue to their argument."""
|
|
# Join per entry, as SCons's ParseFlags lexes each string independently:
|
|
# a dangling -I ending one entry must warn, not absorb the next entry's
|
|
# first token.
|
|
return [
|
|
token
|
|
for entry in ensure_list(entries)
|
|
for token in join_flag_args(split_flag_entry(entry, owner), owner)
|
|
]
|
|
|
|
|
|
# Flags whose argument may follow as a separate token; ParseFlags glues them
|
|
BARE_ARG_FLAGS = frozenset({"-I", "-L", "-l", "-D"})
|
|
|
|
|
|
def join_flag_args(tokens: Iterable[str], owner: str) -> list[str]:
|
|
"""Join a bare ``-I``/``-L``/``-l``/``-D`` with its following token,
|
|
the way PlatformIO's ParseFlags lexes them."""
|
|
out: list[str] = []
|
|
it = iter(tokens)
|
|
for tok in it:
|
|
if tok in BARE_ARG_FLAGS:
|
|
arg = next(it, None)
|
|
if arg is None:
|
|
_LOGGER.warning("Ignoring trailing '%s' in %s build flags", tok, owner)
|
|
break
|
|
tok += arg
|
|
out.append(tok)
|
|
return out
|
|
|
|
|
|
def warn_properties_depends(name: str, data: object) -> None:
|
|
"""Warn when a manifest declares dependencies only as ``depends=``.
|
|
|
|
The dependency walk reads the JSON ``dependencies`` key; the raw
|
|
``library.properties`` spelling would otherwise drop silently.
|
|
"""
|
|
if isinstance(data, dict) and not data.get("dependencies") and data.get("depends"):
|
|
_LOGGER.warning(
|
|
"Library %s declares dependencies via library.properties "
|
|
"depends=, which are not resolved automatically; add them with "
|
|
"add_library() if needed",
|
|
name,
|
|
)
|
|
|
|
|
|
def dependency_is_usable(
|
|
dep: dict, platform: str | None, framework: str, requester: str
|
|
) -> bool:
|
|
"""Compatibility filter for a manifest dependency: platform mismatches
|
|
skip at debug, any other ``InvalidLibrary`` warns naming the requester."""
|
|
try:
|
|
check_library_data(dep, platform, framework)
|
|
except IncompatiblePlatform as e:
|
|
_LOGGER.debug("Skip dependency %s of %s: %s", dep.get("name"), requester, e)
|
|
return False
|
|
except InvalidLibrary as e:
|
|
_LOGGER.warning(
|
|
"Skipping dependency %s of %s: %s", dep.get("name"), requester, e
|
|
)
|
|
return False
|
|
return True
|
|
|
|
|
|
def _valid_dependency_entry(entry: dict, manifest_name: str) -> bool:
|
|
"""Whether a normalized entry carries a usable name and version.
|
|
|
|
The name must be a non-empty string (every consumer indexes or joins
|
|
it); a present version must be a string (a container would raise from
|
|
``set.add()``, an int fails opaquely inside the registry resolution).
|
|
Invalid entries warn naming the manifest.
|
|
"""
|
|
name = entry.get("name")
|
|
if (
|
|
isinstance(name, str)
|
|
and name
|
|
and ("version" not in entry or isinstance(entry["version"], str))
|
|
):
|
|
return True
|
|
_LOGGER.warning(
|
|
"Ignoring unrecognized dependency entry %r of %s", entry, manifest_name
|
|
)
|
|
return False
|
|
|
|
|
|
def normalize_dependencies(
|
|
dependencies: Any, manifest_name: str = "manifest"
|
|
) -> list[dict]:
|
|
"""Normalize a library manifest's ``dependencies`` to a list of dicts.
|
|
|
|
PIO's library.json accepts the list-of-dicts form, the shorthand dict
|
|
form (``{"owner/Name": "version_spec"}``), bare name strings inside the
|
|
list, and a plain (possibly comma-separated) string; normalize them all
|
|
so callers see a uniform list. ``manifest_name`` names the manifest in the
|
|
warning for entries that cannot be normalized.
|
|
"""
|
|
if not dependencies:
|
|
return []
|
|
if isinstance(dependencies, str):
|
|
# A plain string is one or more comma-separated names; iterating it
|
|
# as a list would shred it into one-character "libraries"
|
|
return [{"name": n.strip()} for n in dependencies.split(",") if n.strip()]
|
|
if isinstance(dependencies, dict):
|
|
normalized = []
|
|
for raw_name, spec in dependencies.items():
|
|
if isinstance(raw_name, str) and "/" 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
|
|
if _valid_dependency_entry(entry, manifest_name):
|
|
normalized.append(entry)
|
|
return normalized
|
|
if not isinstance(dependencies, (list, tuple)):
|
|
_LOGGER.warning(
|
|
"Ignoring unrecognized dependencies %r of %s",
|
|
dependencies,
|
|
manifest_name,
|
|
)
|
|
return []
|
|
normalized = []
|
|
for entry in dependencies:
|
|
if isinstance(entry, dict):
|
|
if _valid_dependency_entry(entry, manifest_name):
|
|
normalized.append(entry)
|
|
elif isinstance(entry, str) and entry:
|
|
# PIO also accepts a bare list of names ("dependencies": ["Wire"])
|
|
normalized.append({"name": entry})
|
|
else:
|
|
_LOGGER.warning(
|
|
"Ignoring unrecognized dependency entry %r of %s",
|
|
entry,
|
|
manifest_name,
|
|
)
|
|
return normalized
|
|
|
|
|
|
@dataclass
|
|
class _LibNode:
|
|
"""A node in the library dependency graph being resolved as a batch."""
|
|
|
|
key: str
|
|
is_git: bool
|
|
is_local: bool = False
|
|
is_registry: bool = False
|
|
owner: str | None = None
|
|
pkgname: str | None = None
|
|
requirements: set[str] = field(default_factory=set)
|
|
url: str | None = None
|
|
ref: str | None = None
|
|
local_path: 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, str, tuple[str | None, str | None]]:
|
|
"""Return ``(key, kind, locator)`` for a library or dependency spec.
|
|
|
|
``kind`` is one of:
|
|
|
|
- ``"registry"`` -- ``locator`` is ``(owner, pkgname)``.
|
|
- ``"git"`` -- ``locator`` is ``(url, ref)``.
|
|
- ``"local"`` -- a ``file://`` directory; ``locator`` is ``(path, None)``.
|
|
|
|
The key is derived from the *input* spec (the registry name as written, the
|
|
git URL path, or the custom name / directory name for a local folder), 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 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 (or local) sources instead of failing a registry
|
|
lookup. A plain ``file://`` URL is PlatformIO's spelling for a local library
|
|
folder, so it resolves as a local directory; ``git+file://`` stays a git
|
|
source.
|
|
"""
|
|
if not repository and name and "://" in name:
|
|
# Split a ``CustomName=URL`` name, but only when the whole string isn't
|
|
# itself a valid URL (a bare URL whose query contains ``=`` must stay
|
|
# intact).
|
|
custom_name, candidate = None, name
|
|
if "=" in name and _url_or_none(name) is None:
|
|
custom_name, candidate = name.split("=", 1)
|
|
try:
|
|
scheme = urlsplit(candidate).scheme
|
|
except ValueError:
|
|
scheme = ""
|
|
if scheme == "file" or _url_or_none(candidate):
|
|
name, repository = custom_name, candidate
|
|
else:
|
|
# 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:
|
|
is_git_prefixed = repository.startswith("git+")
|
|
split_result = urlsplit(repository.removeprefix("git+"))
|
|
if split_result.scheme == "file" and not is_git_prefixed:
|
|
# A plain file:// URL points at a local library directory. A local
|
|
# file URL is written file:///absolute/path (empty host) or, less
|
|
# commonly, file://localhost/path. Anything else -- a real host, or
|
|
# a relative path whose first segment parses as the host -- is
|
|
# rejected rather than silently resolved to the wrong directory.
|
|
if split_result.netloc not in ("", "localhost"):
|
|
raise RuntimeError(
|
|
f"Unsupported host in file:// library URL '{repository}'; "
|
|
"use an absolute path, e.g. file:///path/to/lib"
|
|
)
|
|
# Validate the URL path itself (always POSIX-style, leading slash),
|
|
# not the OS path: on Windows a "/foo" path is not is_absolute()
|
|
# without a drive, which would wrongly reject a valid file:/// URL.
|
|
# Reject a relative path (``file:lib_dev``) or a bare root
|
|
# (``file:///``, which has no final segment).
|
|
url_path = split_result.path
|
|
if not url_path.startswith("/") or not PurePosixPath(url_path).name:
|
|
raise RuntimeError(
|
|
f"file:// library URL '{repository}' must be an absolute "
|
|
"directory path, e.g. file:///path/to/lib"
|
|
)
|
|
path = url2pathname(url_path)
|
|
return (name or PurePosixPath(url_path).name), "local", (path, None)
|
|
key = str(split_result.path).strip("/").removesuffix(".git")
|
|
ref = split_result.fragment.strip() or None
|
|
url = urlunsplit(split_result._replace(fragment=""))
|
|
return key, "git", (url, ref)
|
|
if name and "/" in name:
|
|
owner, pkgname = name.split("/", 1)
|
|
else:
|
|
owner, pkgname = None, name
|
|
return name, "registry", (owner, pkgname)
|
|
|
|
|
|
def lib_ignore_set() -> set[str]:
|
|
"""The ``lib_ignore`` names from ``esphome->platformio_options``,
|
|
normalized to lowercase short names (the part after the ``/``)."""
|
|
return {
|
|
name.split("/")[-1].lower()
|
|
for name in CORE.platformio_options.get("lib_ignore", [])
|
|
}
|
|
|
|
|
|
def is_lib_ignored(name: str | None, lib_ignore: set[str]) -> bool:
|
|
"""Whether ``name`` matches the normalized ``lib_ignore`` set."""
|
|
return (
|
|
bool(lib_ignore)
|
|
and name is not None
|
|
and (name.split("/")[-1].lower() in lib_ignore)
|
|
)
|
|
|
|
|
|
def _warn_unsatisfied_versionless(
|
|
skipped_versionless: list[tuple[Any, Any, str]],
|
|
components: dict[str, ConvertedLibrary],
|
|
backend: LibraryBackend,
|
|
) -> None:
|
|
"""Warn for version-less deps nothing satisfied (request key, manifest
|
|
name, or backend provides()); a silent drop surfaces as link errors far
|
|
from the cause."""
|
|
resolved_manifest_names = {c.data.get("name") for c in components.values()}
|
|
warned: set[str] = set()
|
|
for dep_name, dep_owner, requester in skipped_versionless:
|
|
if not isinstance(dep_name, str) or not dep_name or dep_name in warned:
|
|
continue
|
|
if dep_name in components:
|
|
# A version-less dep's request key is the name itself
|
|
continue
|
|
if dep_name in resolved_manifest_names:
|
|
# Name-only evidence: any resolved component with this manifest
|
|
# name counts, not just ones the requester can reach
|
|
_LOGGER.debug(
|
|
"Version-less dependency %s of %s satisfied by manifest name only",
|
|
dep_name,
|
|
requester,
|
|
)
|
|
continue
|
|
if (
|
|
not dep_owner
|
|
and backend.provides is not None
|
|
and backend.provides(dep_name)
|
|
):
|
|
# provides() only satisfies owner-less names: the walk's
|
|
# backend-provided skip has the same owner guard, so an
|
|
# owner-qualified version-less dep was added by nobody
|
|
continue
|
|
warned.add(dep_name)
|
|
_LOGGER.warning(
|
|
"Dependency %s of %s has no version to resolve and nothing "
|
|
"provides it; skipping",
|
|
dep_name,
|
|
requester,
|
|
)
|
|
|
|
|
|
# A few streams saturate most links without hammering the registry
|
|
_DOWNLOAD_WORKERS = 4
|
|
|
|
|
|
def _content_lengths(urls: list[str]) -> list[int | None]:
|
|
"""Content-Length per URL via HEAD requests; None when unknown."""
|
|
import requests
|
|
|
|
def head(url: str) -> int | None:
|
|
try:
|
|
resp = requests.head(url, timeout=10, allow_redirects=True)
|
|
if not resp.ok:
|
|
_LOGGER.debug("HEAD %s returned %s", url, resp.status_code)
|
|
return None
|
|
return int(resp.headers.get("content-length", 0)) or None
|
|
except (requests.RequestException, ValueError) as err:
|
|
_LOGGER.debug("HEAD %s failed: %s", url, err)
|
|
return None
|
|
|
|
with ThreadPoolExecutor(max_workers=min(_DOWNLOAD_WORKERS, len(urls))) as ex:
|
|
return list(ex.map(head, urls))
|
|
|
|
|
|
def _prefetch_wave(
|
|
wave: list[tuple[str, ConvertedLibrary]], salt: str, namespace: str
|
|
) -> None:
|
|
"""Best-effort parallel download of a wave's registry archives.
|
|
|
|
The walk's own ``download()`` call stays authoritative (it surfaces real
|
|
failures, with resume); bars are suppressed since parallel bars would
|
|
interleave. Duplicate URLs prefetch once so two threads never extract
|
|
into the same cache directory.
|
|
"""
|
|
components: list[ConvertedLibrary] = []
|
|
seen: set[str] = set()
|
|
for _key, component in wave:
|
|
if not isinstance(component.source, URLSource):
|
|
continue
|
|
if component.source.url in seen:
|
|
continue
|
|
seen.add(component.source.url)
|
|
try:
|
|
cached = component.source.is_cached(
|
|
component.get_sanitized_name(), salt=salt, namespace=namespace
|
|
)
|
|
except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught
|
|
# Best-effort: a failing probe prefetches (and re-downloads)
|
|
_LOGGER.debug("Cache probe for %s failed: %s", component.name, err)
|
|
cached = False
|
|
if cached:
|
|
# A completed extraction downloads nothing; a warm build must
|
|
# stay silent
|
|
continue
|
|
components.append(component)
|
|
if len(components) < 2:
|
|
return
|
|
_LOGGER.info(
|
|
"Downloading %d libraries: %s",
|
|
len(components),
|
|
", ".join(c.name for c in components),
|
|
)
|
|
# One combined bar over the batch, sized by HEAD requests. An unknown
|
|
# size would mean a silent multi-MB download; fall back to sequential
|
|
# downloads with their per-file bars instead.
|
|
sizes = _content_lengths([c.source.url for c in components])
|
|
if not all(sizes):
|
|
return
|
|
progress = BatchDownloadProgress("Downloading libraries", sum(sizes))
|
|
|
|
def _fetch(component: ConvertedLibrary) -> None:
|
|
tracker = progress.tracker()
|
|
try:
|
|
component.download(salt=salt, namespace=namespace, progress=tracker)
|
|
except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught
|
|
# The sequential call below retries and reports the failure
|
|
_LOGGER.debug("Prefetch of %s failed: %s", component.name, err)
|
|
tracker(0)
|
|
|
|
ex = ThreadPoolExecutor(max_workers=min(_DOWNLOAD_WORKERS, len(components)))
|
|
try:
|
|
for future in [ex.submit(_fetch, component) for component in components]:
|
|
future.result()
|
|
finally:
|
|
# On Ctrl-C drop the queued archives instead of downloading them
|
|
# all before the process can exit; in-flight ones still finish.
|
|
ex.shutdown(wait=True, cancel_futures=True)
|
|
progress.done()
|
|
|
|
|
|
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 = lib_ignore_set()
|
|
|
|
# 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 add_spec(name: str | None, version: str | None, repository: str | None) -> str:
|
|
key, kind, locator = _node_key(name, version, repository)
|
|
node = nodes.get(key) or _LibNode(key=key, is_git=kind == "git")
|
|
nodes[key] = node
|
|
# The same key requested from two different kinds of source (or two
|
|
# different local paths) is a config mistake: one silently wins. Warn so
|
|
# it isn't a surprise. (git-vs-registry is reported separately below.)
|
|
if kind == "git":
|
|
if node.is_local:
|
|
_LOGGER.warning(
|
|
"Library %s is requested as both a local directory and a git "
|
|
"source; using the git source.",
|
|
key,
|
|
)
|
|
node.is_git = True
|
|
node.url, node.ref = locator
|
|
elif kind == "local":
|
|
new_path = locator[0]
|
|
if node.is_git:
|
|
# git wins (checked first when building the source); leave the
|
|
# node as a git source.
|
|
_LOGGER.warning(
|
|
"Library %s is requested as both a local directory and a git "
|
|
"source; using the git source.",
|
|
key,
|
|
)
|
|
else:
|
|
if node.is_local and node.local_path != new_path:
|
|
_LOGGER.warning(
|
|
"Library %s is requested from two local directories (%s "
|
|
"and %s); using %s.",
|
|
key,
|
|
node.local_path,
|
|
new_path,
|
|
new_path,
|
|
)
|
|
node.is_local = True
|
|
node.local_path = new_path
|
|
else:
|
|
node.is_registry = True
|
|
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_lib_ignored(library.name, lib_ignore)
|
|
]
|
|
|
|
# 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)
|
|
# (name, owner, requester) reconciled against the final resolution set
|
|
skipped_versionless: list[tuple[Any, Any, str]] = []
|
|
worklist = deque(dict.fromkeys(top_level))
|
|
while worklist:
|
|
# Drain the frontier sequentially (spec resolution mutates shared
|
|
# node state), then prefetch the wave's registry archives in
|
|
# parallel; the per-component download() below stays authoritative.
|
|
wave: list[tuple[str, ConvertedLibrary]] = []
|
|
while worklist:
|
|
key = worklist.popleft()
|
|
node = nodes[key]
|
|
|
|
# Re-resolve only when the requirement set grew; requirements
|
|
# only ever grow, so the fixpoint converges and cycles terminate
|
|
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))
|
|
elif node.is_local:
|
|
component = ConvertedLibrary(key, "*", LocalSource(node.local_path))
|
|
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)
|
|
)
|
|
wave.append((key, component))
|
|
_prefetch_wave(wave, salt, backend.cache_key)
|
|
for key, component in wave:
|
|
node = nodes[key]
|
|
component.download(salt=salt, namespace=backend.cache_key)
|
|
|
|
source_dir = component.source_dir
|
|
library_json_path = source_dir / "library.json"
|
|
library_properties_path = source_dir / "library.properties"
|
|
has_json = library_json_path.is_file()
|
|
has_properties = library_properties_path.is_file()
|
|
if not has_json and not has_properties and not node.is_local:
|
|
# An interrupted clone/extraction self-heals with one forced
|
|
# re-download; a local source has nothing to re-download
|
|
_LOGGER.warning(
|
|
"Library %s at %s is missing library.json and library.properties; "
|
|
"re-downloading",
|
|
key,
|
|
source_dir,
|
|
)
|
|
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:
|
|
# Local sources are user input (EsphomeError); a registry/git
|
|
# miss means a corrupt cache (RuntimeError)
|
|
error_cls = EsphomeError if node.is_local else RuntimeError
|
|
raise error_cls(
|
|
f"Invalid PIO library {key}: missing library.json and "
|
|
f"library.properties in {source_dir}"
|
|
)
|
|
|
|
if not isinstance(component.data, dict) or not isinstance(
|
|
component.data.get("build", {}), dict
|
|
):
|
|
# A bare json.load imposes no shape; every backend dereferences
|
|
# data/build, so validate once here and name the library
|
|
raise EsphomeError(f"Library {key} has a malformed manifest")
|
|
warn_properties_depends(component.name, component.data)
|
|
|
|
try:
|
|
check_library_data(component.data, backend.platform, backend.framework)
|
|
except InvalidLibrary as e:
|
|
# An explicitly requested library fails fast; the routine
|
|
# cross-platform skip stays at debug, other causes warn
|
|
if key in top_level_keys:
|
|
raise RuntimeError(
|
|
f"Requested library {key} is not compatible with "
|
|
f"{backend.framework}: {e}"
|
|
) from e
|
|
if isinstance(e, IncompatiblePlatform):
|
|
_LOGGER.debug("Skip incompatible dependency %s: %s", key, str(e))
|
|
else:
|
|
_LOGGER.warning("Skipping 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"), component.name
|
|
):
|
|
if "version" not in dependency:
|
|
# Version-less deps cannot resolve from the registry; the
|
|
# post-emit reconciliation owns the drop warning. An
|
|
# is_lib_ignored name is deliberately excluded, not a drop.
|
|
_LOGGER.debug(
|
|
"Skip version-less dependency %r of %s",
|
|
dependency.get("name"),
|
|
component.name,
|
|
)
|
|
if not is_lib_ignored(
|
|
dependency.get("name"), lib_ignore
|
|
) and dependency_is_usable(
|
|
dependency, backend.platform, backend.framework, component.name
|
|
):
|
|
# A platform-filtered or ignored dep is deliberately
|
|
# absent, not a drop to reconcile
|
|
skipped_versionless.append(
|
|
(
|
|
dependency.get("name"),
|
|
dependency.get("owner"),
|
|
component.name,
|
|
)
|
|
)
|
|
continue
|
|
if not dependency_is_usable(
|
|
dependency, backend.platform, backend.framework, component.name
|
|
):
|
|
continue
|
|
dep_name = _owner_pkgname_to_name(
|
|
dependency.get("owner"), dependency.get("name")
|
|
)
|
|
if is_lib_ignored(dep_name, lib_ignore):
|
|
_LOGGER.debug("Skip ignored dependency %s", dep_name)
|
|
continue
|
|
# The version field may actually be a URL (git/archive
|
|
# dependency), which names one specific source; it must not
|
|
# be substituted with a same-named bundled library below.
|
|
dep_version = dependency["version"]
|
|
dep_url = _url_or_none(dep_version)
|
|
if dep_url is not None:
|
|
dep_version = None
|
|
elif (
|
|
backend.provides is not None
|
|
and not dependency.get("owner")
|
|
and backend.provides(dep_name)
|
|
):
|
|
# The backend adds it from its own tree; resolving it here
|
|
# would fetch a same-named registry package instead
|
|
if dep_version and dep_version != "*":
|
|
# The version pin is discarded for the bundled copy;
|
|
# make the substitution visible
|
|
_LOGGER.warning(
|
|
"Dependency %s pins version %s; using the library "
|
|
"bundled with the framework instead",
|
|
dep_name,
|
|
dep_version,
|
|
)
|
|
else:
|
|
_LOGGER.debug("Skip backend-provided dependency %s", dep_name)
|
|
continue
|
|
dep_key = add_spec(dep_name, dep_version, dep_url)
|
|
node.edges.add(dep_key)
|
|
worklist.append(dep_key)
|
|
|
|
# A git or local source wins over the same component requested from the
|
|
# registry. That's intentional, but warn so the dropped registry spec isn't
|
|
# a silent surprise -- including when it carried no version pin (a bare
|
|
# cg.add_library("Foo"), which is how most components add libraries).
|
|
for node in nodes.values():
|
|
if (node.is_git or node.is_local) and (node.is_registry or node.requirements):
|
|
source = "git" if node.is_git else "local"
|
|
registry = (
|
|
f"registry version(s) {sorted(node.requirements)}"
|
|
if node.requirements
|
|
else "a registry package"
|
|
)
|
|
_LOGGER.warning(
|
|
"Library %s is requested both from a %s source (%s) and as %s; "
|
|
"using the %s source.",
|
|
node.key,
|
|
source,
|
|
node.url if node.is_git else node.local_path,
|
|
registry,
|
|
source,
|
|
)
|
|
|
|
# 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)
|
|
|
|
_warn_unsatisfied_versionless(skipped_versionless, components, backend)
|
|
|
|
return [components[key] for key in top_level if key in components]
|