Add the shared registry, ninja, ccache, and tools-cache infrastructure for native toolchains

esphome/build_helpers/ccache.py

@@ -0,0 +1,98 @@
+"""Shared ccache policy for build backends.
+
+One place for the probe, the enable/override rules, and the ``CCACHE_*``
+defaults, so the backends cannot drift apart.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+import subprocess
+
+from esphome.framework_helpers import strip_win_long_path_prefix
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _ccache_runs(ccache: str) -> bool:
+    """Return True when the ``ccache`` found on PATH actually runs.
+
+    ``shutil.which`` proves existence, not runnability: on Windows it also
+    matches ``.bat``/``.cmd`` wrappers and stale package-manager shims whose
+    target is gone. Wrapping compiles around such a find fails every compile
+    step with an opaque OS error, so probe once and fall back to compiling
+    without ccache when the probe fails.
+    """
+    try:
+        subprocess.run(
+            [ccache, "--version"],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=15,
+            close_fds=False,
+        )
+    except (OSError, subprocess.SubprocessError):
+        _LOGGER.warning(
+            "Ignoring ccache at %s because it failed to run; compiling without ccache",
+            ccache,
+        )
+        return False
+    return True
+
+
+def resolve_ccache_path() -> str | None:
+    """The ccache binary to wrap compiles with, or None when disabled.
+
+    Shared policy for every backend: on by default when a runnable ccache is
+    on PATH, ``ESPHOME_CCACHE_ENABLE=0`` opts out, and an explicit ``=1``
+    warns when no binary is found and skips the runnability probe. The
+    Windows extended-length prefix is stripped before probing so the probe
+    validates the exact string the build will execute (#18399).
+    """
+    import shutil
+
+    from esphome.helpers import get_bool_env
+
+    explicit = "ESPHOME_CCACHE_ENABLE" in os.environ
+    if explicit and not get_bool_env("ESPHOME_CCACHE_ENABLE"):
+        return None
+    ccache = shutil.which("ccache")
+    if ccache is None:
+        if explicit:
+            _LOGGER.warning(
+                "ESPHOME_CCACHE_ENABLE is set but no ccache binary is on PATH; "
+                "compiling without ccache"
+            )
+        return None
+    ccache = strip_win_long_path_prefix(ccache)
+    if not explicit and not _ccache_runs(ccache):
+        return None
+    return ccache
+
+
+def ccache_defaults_env(cache_dir: Path) -> dict[str, str]:
+    """Default ``CCACHE_*`` values for a build subprocess (not os.environ).
+
+    Values the user already set in the environment are respected. Depend
+    mode is on: both native backends emit depfiles (-MMD / CMake), which
+    keeps cache-miss overhead low.
+    """
+    from esphome.core import CORE
+
+    # build_path is set during preload for every config-loading command; unset
+    # means the caller built the environment too early. Fail loudly rather
+    # than silently drop CCACHE_BASEDIR (losing cross-device cache hits).
+    if CORE.build_path is None:
+        raise ValueError(
+            "CORE.build_path must be set before constructing the build environment"
+        )
+    defaults = {
+        "CCACHE_DIR": str(cache_dir),
+        "CCACHE_NOHASHDIR": "true",
+        "CCACHE_DEPEND": "1",
+        "CCACHE_BASEDIR": str(Path(CORE.build_path).resolve()),
+    }
+    return {k: v for k, v in defaults.items() if k not in os.environ}

esphome/build_helpers/ninja.py

@@ -0,0 +1,33 @@
+"""Platform-neutral helpers for ninja-driven native builds."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import shutil
+
+from esphome.core import EsphomeError
+
+
+def find_ninja() -> Path:
+    """Locate the ninja binary: PATH first, else the ninja PyPI wheel.
+
+    The wheel is a requirements.txt dependency, so pip has already
+    integrity-checked it; no download logic is needed here.
+    """
+    if binary := shutil.which("ninja"):
+        return Path(binary)
+    try:
+        import ninja
+    except ImportError:
+        wheel_binary = None
+    else:
+        wheel_binary = Path(ninja.BIN_DIR) / (
+            "ninja.exe" if os.name == "nt" else "ninja"
+        )
+    if wheel_binary is None or not wheel_binary.is_file():
+        raise EsphomeError(
+            "ninja not found on PATH or in the ninja package; reinstall the "
+            "esphome Python environment"
+        )
+    return wheel_binary

esphome/build_helpers/tools_cache.py

@@ -0,0 +1,22 @@
+"""Machine-global tools cache location shared by the native backends."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def tools_cache_path(env_var: str, subdir: str) -> Path:
+    """A backend's machine-global tools directory, with an env override.
+
+    A blank/whitespace override is treated as unset: ``Path("")`` resolves
+    to the CWD, which ``clean-all`` would then delete.
+    """
+    import platformdirs
+
+    from esphome.helpers import get_str_env
+
+    if prefix := get_str_env(env_var, "").strip():
+        return Path(prefix).expanduser().resolve()
+    return (
+        Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / subdir
+    ).resolve()

esphome/components/nrf52/framework.py

@@ -7,8 +7,7 @@ import shutil
 import sys
 import tempfile
 
-import platformdirs
-
+from esphome.build_helpers.tools_cache import tools_cache_path
 import esphome.config_validation as cv
 from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION
 from esphome.core import CORE, EsphomeError
@@ -21,7 +20,6 @@ from esphome.framework_helpers import (
     run_command_ok,
     str_to_lst_of_str,
 )
-from esphome.helpers import get_str_env
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -51,15 +49,9 @@ SDK_NG_MINIMAL_MIRRORS = str_to_lst_of_str(
 
 
 def get_sdk_nrf_tools_path() -> Path:
-    # A blank ESPHOME_SDK_NRF_PREFIX must be treated as unset: Path("")
-    # resolves to the CWD, which clean-all would then delete.
-    if prefix := get_str_env("ESPHOME_SDK_NRF_PREFIX", "").strip():
-        path = Path(prefix).expanduser()
-    else:
-        # Machine-global (OS user cache dir) so all projects share one install;
-        # see espidf.framework.get_idf_tools_path for the location rationale.
-        path = Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "sdk-nrf"
-    return path.resolve()
+    # Machine-global (OS user cache dir) so all projects share one install;
+    # see espidf.framework.get_idf_tools_path for the location rationale.
+    return tools_cache_path("ESPHOME_SDK_NRF_PREFIX", "sdk-nrf")
 
 
 def _needs_venv_rebuild(

esphome/espidf/framework.py

@@ -11,9 +11,9 @@ import re
 import shutil
 from typing import Any, NoReturn
 
-import platformdirs
-
-from esphome.core import CORE, Version
+from esphome.build_helpers.ccache import ccache_defaults_env
+from esphome.build_helpers.tools_cache import tools_cache_path
+from esphome.core import Version
 from esphome.framework_helpers import (
     PathType,
     archive_extract_all,
@@ -27,7 +27,7 @@ from esphome.framework_helpers import (
     run_command_ok,
     str_to_lst_of_str,
 )
-from esphome.helpers import get_bool_env, get_str_env, write_file_if_changed
+from esphome.helpers import get_bool_env, write_file_if_changed
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -88,22 +88,10 @@ def get_idf_tools_path() -> Path:
     Returns:
         Path object pointing to the ESP-IDF tools directory
     """
-    # Treat an empty/whitespace ESPHOME_ESP_IDF_PREFIX as unset: Path("")
-    # resolves to the CWD, which would install into (and let clean-all delete)
-    # the working directory by accident.
-    if prefix := get_str_env("ESPHOME_ESP_IDF_PREFIX", "").strip():
-        path = Path(prefix).expanduser()
-    else:
-        # Machine-global so all projects share the multi-GB install instead of
-        # a per-config-directory copy. The user cache dir (not ~/.esphome)
-        # avoids colliding with data_dir when configs live in the home dir.
-        # appauthor=False drops the redundant <author>\ segment on Windows
-        # (which otherwise repeats "esphome\esphome\") to keep the path short.
-        path = Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "idf"
-    # Resolve so an unnormalized config path (e.g. compiling ``../config/x.yaml``)
-    # doesn't leave ``..`` segments in the IDF_TOOLS_PATH handed to idf.py, which
-    # otherwise warns that the venv interpreter path doesn't match the install.
-    return path.resolve()
+    # Machine-global so all projects share the multi-GB install instead of
+    # a per-config-directory copy; see build_helpers.tools_cache.tools_cache_path
+    # for the env-override and normalization rules.
+    return tools_cache_path("ESPHOME_ESP_IDF_PREFIX", "idf")
 
 
 # Windows' default MAX_PATH is 260 characters. ESP-IDF toolchains nest deeply
@@ -1169,25 +1157,12 @@ def _ccache_env() -> dict[str, str]:
         # ESP-IDF silently skips ccache without the binary; don't enable it.
         return {}
 
-    # ccache is enabled past here. build_path is set during preload for every
-    # config-loading command, so it being unset means a caller built the IDF env
-    # too early -- fail loudly rather than silently drop CCACHE_BASEDIR (which
-    # would quietly cost cross-device cache hits).
-    if CORE.build_path is None:
-        raise ValueError(
-            "CORE.build_path must be set before constructing the ESP-IDF build "
-            "environment"
-        )
-
-    defaults = {
-        "IDF_CCACHE_ENABLE": "1",
-        "CCACHE_DIR": str(get_idf_tools_path() / "ccache"),
-        "CCACHE_NOHASHDIR": "true",
-        "CCACHE_DEPEND": "1",
-        "CCACHE_BASEDIR": str(Path(CORE.build_path).resolve()),
-    }
-    # Don't override CCACHE_* values the user already set in their environment.
-    return {k: v for k, v in defaults.items() if k not in os.environ}
+    # ccache is enabled past here; the shared helper carries the CCACHE_*
+    # policy (and the fail-loud build_path guard).
+    env = ccache_defaults_env(get_idf_tools_path() / "ccache")
+    if "IDF_CCACHE_ENABLE" not in os.environ:
+        env["IDF_CCACHE_ENABLE"] = "1"
+    return env
 
 
 def get_framework_env(

esphome/framework_helpers.py

@@ -1169,3 +1169,37 @@ def download_from_mirrors(
             f"No mirror URL template matched the provided substitutions:{details}"
         )
     raise ValueError("download_from_mirrors called with an empty mirrors list")
+
+
+def strip_win_long_path_prefix(path: str) -> str:
+    r"""Strip the Windows extended-length path prefix from ``path``.
+
+    Handles both forms documented at
+    https://learn.microsoft.com/windows/win32/fileio/naming-a-file:
+
+    * ``\\?\C:\path\to\file`` -> ``C:\path\to\file``
+    * ``\\?\UNC\server\share\path`` -> ``\\server\share\path``
+
+    The NSIS-installed ``esphome.exe`` launcher on Windows starts Python with
+    ``sys.executable`` already prefixed with ``\\?\``. That prefix propagates
+    into PlatformIO's ``$PYTHONEXE`` (PlatformIO reads ``PYTHONEXEPATH`` from
+    the environment, falling back to ``os.path.normpath(sys.executable)``)
+    and ends up baked into SCons-emitted command lines for build steps such
+    as the esp8266 ``elf2bin`` invocation. ``cmd.exe`` does not understand
+    the ``\\?\`` prefix, so the build fails with
+    "The system cannot find the path specified." Stripping the prefix early
+    keeps the path shell-quotable.
+
+    Also applied to the ccache path exported by the ccache helpers, which
+    ``shutil.which`` can return with the same prefix.
+
+    No-op on non-Windows platforms.
+    """
+    if sys.platform != "win32":
+        return path
+    if path.startswith("\\\\?\\UNC\\"):
+        # \\?\UNC\server\share\... -> \\server\share\...
+        return "\\\\" + path[len("\\\\?\\UNC\\") :]
+    if path.startswith("\\\\?\\"):
+        return path[len("\\\\?\\") :]
+    return path

esphome/platformio/registry.py

@@ -0,0 +1,159 @@
+"""Install packages from the PlatformIO registry without PlatformIO.
+
+Native toolchains install the exact registry packages the PlatformIO backend
+uses, so the bits are identical, but resolve and verify them with esphome's
+own download machinery instead of importing the platformio package.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+import io
+import json
+import logging
+import os
+from pathlib import Path
+import platform
+
+from esphome.core import EsphomeError
+from esphome.framework_helpers import (
+    archive_extract_all,
+    download_from_mirrors,
+    download_with_resume,
+    rmdir,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+_REGISTRY_URL = (
+    "https://api.registry.platformio.org/v3/packages/platformio/tool/{package}"
+)
+
+
+def get_systype() -> str:
+    """The registry system tag for the current host.
+
+    A transliteration of ``platformio.util.get_systype()``, honoring the same
+    ``PLATFORMIO_SYSTEM_TYPE`` override, so this module never imports the
+    platformio package. One deviation: windows-arm64 maps straight to
+    ``windows_amd64``: the registry ships no arm64 toolchains and those hosts
+    run x86 binaries via emulation, which upstream leaves to the override.
+    """
+    if systype := os.environ.get("PLATFORMIO_SYSTEM_TYPE"):
+        return systype
+    system = platform.system().lower()
+    arch = platform.machine().lower()
+    if system == "windows":
+        if not arch:  # same fallback as upstream (platformio issue #4353)
+            arch = "x86_" + platform.architecture()[0]
+        if "x86" in arch:
+            arch = "amd64" if "64" in arch else "x86"
+        elif arch == "arm64":
+            arch = "amd64"
+    if arch == "aarch64" and platform.architecture()[0] == "32bit":
+        # 64-bit kernel with a 32-bit userland (e.g. 32-bit Raspberry Pi OS)
+        arch = "armv7l"
+    return f"{system}_{arch}" if arch else system
+
+
+def registry_download(package: str, version: str) -> tuple[str, str, int | None]:
+    """Resolve a package's download URL, sha256, and size via the registry.
+
+    The metadata fetch goes through ``download_from_mirrors`` so it shares
+    the retry, backoff, and error reporting of every other download here.
+    """
+    buf = io.BytesIO()
+    download_from_mirrors([_REGISTRY_URL], {"package": package}, buf)
+    try:
+        data = json.loads(buf.getvalue())
+    except ValueError as err:
+        raise EsphomeError(
+            f"The package registry returned invalid JSON for {package}: {err}"
+        ) from err
+    systype = get_systype()
+    for ver in data.get("versions", []):
+        if ver.get("name") != version:
+            continue
+        for file in ver.get("files", []):
+            # A bare string would make ``in`` a substring test
+            systems = file.get("system") or "*"
+            if isinstance(systems, str):
+                systems = [systems]
+            if "*" in systems or systype in systems:
+                sha256 = (file.get("checksum") or {}).get("sha256")
+                if not sha256:
+                    # Never extract an unverified archive; the registry
+                    # publishes a checksum for every package file.
+                    raise EsphomeError(
+                        f"The package registry returned no sha256 for "
+                        f"{package} {version}; refusing the unverified download"
+                    )
+                return (file["download_url"], sha256, file.get("size"))
+        raise EsphomeError(
+            f"No {package} {version} build for this platform ({systype})"
+        )
+    raise EsphomeError(f"{package} {version} not found in the package registry")
+
+
+def install_package(
+    name: str,
+    version: str,
+    dest: Path,
+    mirrors: list[str],
+    downloads_dir: Path,
+    expect: Collection[str] = (),
+) -> None:
+    """Download, verify, and extract one package if not already installed.
+
+    The registry path is integrity-checked against the sha256 the registry
+    publishes; a mirror override (URL templates with ``{VERSION}``/``{SYSTEM}``
+    substitution) is trusted as configured. ``downloads_dir`` holds the
+    archive between runs so an interrupted download resumes.
+    """
+    marker = dest / ".esphome_extracted"
+    if marker.is_file():
+        return
+    from filelock import FileLock
+
+    # The cache is machine-global; serialize concurrent cold builds so one
+    # process cannot wipe the directory another is extracting into (same
+    # filelock pattern as platformio/toolchain.py and git.py).
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    # fallback_to_soft would silently degrade to an existence lock on a
+    # flock-less filesystem; a hard-killed run would then hang every later
+    # build forever (same hazard git.py documents).
+    with FileLock(f"{dest}.lock", fallback_to_soft=False):
+        if marker.is_file():
+            # Another process finished the install while we waited
+            return
+        rmdir(dest, msg=f"Clean up incomplete {name} install")
+        # A persistent download location (not a temp dir) so an interrupted
+        # download resumes across esphome runs via download_with_resume's
+        # .part file, mirroring the espidf dist/ convention.
+        downloads_dir.mkdir(parents=True, exist_ok=True)
+        archive = downloads_dir / f"{name}-{version}"
+        _LOGGER.info("Downloading %s %s ...", name, version)
+        if mirrors:
+            _LOGGER.warning(
+                "Downloading %s from a mirror override; checksum verification "
+                "is skipped for mirrors",
+                name,
+            )
+            download_from_mirrors(
+                mirrors, {"VERSION": version, "SYSTEM": get_systype()}, archive
+            )
+        else:
+            url, sha256, size = registry_download(name, version)
+            download_with_resume(url, archive, sha256=sha256, size=size)
+        _LOGGER.info("Extracting %s ...", name)
+        archive_extract_all(archive, dest, progress_header="Extracting")
+        # Validate the layout before recording success, so an unexpected
+        # package is never cached as a working install.
+        for rel in expect:
+            if not (dest / rel).is_dir():
+                raise EsphomeError(
+                    f"{name} {version} extracted without the expected {rel} "
+                    "directory; run 'esphome clean-all' and retry"
+                )
+        marker.touch()
+        archive.unlink(missing_ok=True)

esphome/platformio/toolchain.py

@@ -4,19 +4,18 @@ import logging
 import os
 from pathlib import Path
 import re
-import shutil
-import subprocess
 import sys
 from typing import TYPE_CHECKING, Any
 
 import platformdirs
 
+from esphome.build_helpers.ccache import resolve_ccache_path
 from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE
 from esphome.core import CORE, EsphomeError
+from esphome.framework_helpers import strip_win_long_path_prefix
 from esphome.helpers import (
     add_git_ceiling_directory,
     copy_file_if_changed,
-    get_bool_env,
     rmtree,
     write_file,
 )
@@ -41,40 +40,6 @@ _PIO_PYTHON_STAMP_LOCK = ".esphome.pio.stamp.lock"
 _PIO_PYTHON_STAMP_SCHEMA = "0"
 
 
-def _strip_win_long_path_prefix(path: str) -> str:
-    r"""Strip the Windows extended-length path prefix from ``path``.
-
-    Handles both forms documented at
-    https://learn.microsoft.com/windows/win32/fileio/naming-a-file:
-
-    * ``\\?\C:\path\to\file`` -> ``C:\path\to\file``
-    * ``\\?\UNC\server\share\path`` -> ``\\server\share\path``
-
-    The NSIS-installed ``esphome.exe`` launcher on Windows starts Python with
-    ``sys.executable`` already prefixed with ``\\?\``. That prefix propagates
-    into PlatformIO's ``$PYTHONEXE`` (PlatformIO reads ``PYTHONEXEPATH`` from
-    the environment, falling back to ``os.path.normpath(sys.executable)``)
-    and ends up baked into SCons-emitted command lines for build steps such
-    as the esp8266 ``elf2bin`` invocation. ``cmd.exe`` does not understand
-    the ``\\?\`` prefix, so the build fails with
-    "The system cannot find the path specified." Stripping the prefix early
-    keeps the path shell-quotable.
-
-    Also applied to the ccache path exported by ``_ccache_env()``, which
-    ``shutil.which`` can return with the same prefix.
-
-    No-op on non-Windows platforms.
-    """
-    if sys.platform != "win32":
-        return path
-    if path.startswith("\\\\?\\UNC\\"):
-        # \\?\UNC\server\share\... -> \\server\share\...
-        return "\\\\" + path[len("\\\\?\\UNC\\") :]
-    if path.startswith("\\\\?\\"):
-        return path[len("\\\\?\\") :]
-    return path
-
-
 def get_platformio_config() -> "ProjectConfig | None":
     """Return PlatformIO's ``ProjectConfig``, or None when PlatformIO is absent."""
     try:
@@ -238,33 +203,6 @@ def _check_platformio_python_stamp(config: "ProjectConfig") -> None:
         _write_pio_stamp_python(stamp_file, current)
 
 
-def _ccache_runs(ccache: str) -> bool:
-    """Return True when the ``ccache`` found on PATH actually runs.
-
-    ``shutil.which`` proves existence, not runnability: on Windows it also
-    matches ``.bat``/``.cmd`` wrappers and stale package-manager shims whose
-    target is gone. Wrapping compiles around such a find fails every compile
-    step with an opaque OS error, so probe once and fall back to compiling
-    without ccache when the probe fails.
-    """
-    try:
-        subprocess.run(
-            [ccache, "--version"],
-            check=True,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            timeout=15,
-            close_fds=False,
-        )
-    except (OSError, subprocess.SubprocessError):
-        _LOGGER.warning(
-            "Ignoring ccache at %s because it failed to run; compiling without ccache",
-            ccache,
-        )
-        return False
-    return True
-
-
 def _ccache_env() -> dict[str, str]:
     r"""Return ccache settings for PlatformIO builds.
 
@@ -283,7 +221,7 @@ def _ccache_env() -> dict[str, str]:
     runs fine through ``CreateProcess``, which is how ESP-IDF invokes it,
     but SCons runs every compile through ``cmd.exe``, which fails on it with
     "The system cannot find the path specified." (#18399), so the prefix is
-    stripped here with ``_strip_win_long_path_prefix()`` before the
+    stripped here with ``strip_win_long_path_prefix()`` before the
     runnability probe, which therefore validates the exact string the build
     will execute.
     ``ESPHOME_CCACHE_PATH`` is an internal channel, not a user setting: the
@@ -309,22 +247,8 @@ def _ccache_env() -> dict[str, str]:
     build dir. The other ``CCACHE_*`` values the user already set in the
     environment are respected.
     """
-    explicit = "ESPHOME_CCACHE_ENABLE" in os.environ
-    if explicit and not get_bool_env("ESPHOME_CCACHE_ENABLE"):
-        return {"ESPHOME_CCACHE_ENABLE": "0"}
-    ccache_path = shutil.which("ccache")
+    ccache_path = resolve_ccache_path()
     if ccache_path is None:
-        if explicit:
-            _LOGGER.warning(
-                "ESPHOME_CCACHE_ENABLE is set but no ccache binary is on PATH; "
-                "compiling without ccache"
-            )
-        return {"ESPHOME_CCACHE_ENABLE": "0"}
-    # Strip before probing so the probe validates (and the failure warning
-    # names) the exact string the build will execute through cmd.exe.
-    ccache_path = _strip_win_long_path_prefix(ccache_path)
-    # An explicit opt-in skips the runnability probe.
-    if not explicit and not _ccache_runs(ccache_path):
         return {"ESPHOME_CCACHE_ENABLE": "0"}
     env = {
         "ESPHOME_CCACHE_ENABLE": "1",
@@ -386,7 +310,7 @@ def run_platformio_cli(*args, **kwargs) -> str | int:
     # Strip the Windows extended-length path prefix from sys.executable so it
     # doesn't propagate into PlatformIO's $PYTHONEXE and break SCons-emitted
     # command lines run through cmd.exe.
-    python_exe = _strip_win_long_path_prefix(sys.executable)
+    python_exe = strip_win_long_path_prefix(sys.executable)
     if python_exe != sys.executable:
         # Only override PYTHONEXEPATH when we actually stripped a prefix.
         # PlatformIO's get_pythonexe_path() reads this and falls back to

requirements.txt

@@ -28,6 +28,7 @@ smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
 platformdirs==4.11.3  # native esp-idf toolchain global cache dir
+ninja==1.13.0  # native esp8266 arduino toolchain build driver
 filelock==3.32.3  # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg
 
 # esp-idf >= 5.0 requires this

tests/unit_tests/build_helpers/test_ninja.py

@@ -0,0 +1,50 @@
+"""Tests for esphome.build_helpers.ninja."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.build_helpers import ninja as ninja_helper
+from esphome.core import EsphomeError
+
+
+def test_find_ninja_prefers_path(tmp_path: Path) -> None:
+    with patch("shutil.which", return_value=str(tmp_path / "ninja")):
+        assert ninja_helper.find_ninja() == tmp_path / "ninja"
+
+
+def test_find_ninja_falls_back_to_wheel(tmp_path: Path) -> None:
+    """Without a PATH entry, the ninja PyPI wheel's binary is used."""
+    binary_name = "ninja.exe" if os.name == "nt" else "ninja"
+    (tmp_path / binary_name).touch()
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / binary_name
+
+
+def test_find_ninja_package_not_installed() -> None:
+    """A missing ninja package raises the actionable message, not ImportError."""
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": None}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        ninja_helper.find_ninja()
+
+
+def test_find_ninja_missing_everywhere(tmp_path: Path) -> None:
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        ninja_helper.find_ninja()

tests/unit_tests/test_espidf_framework.py

@@ -1400,8 +1400,9 @@ def _ccache_patches(tmp_path: Path, which: str | None, build_path: Path | None):
             "esphome.espidf.framework.get_idf_tools_path",
             return_value=tmp_path / "tools",
         ),
+        # ccache_defaults_env (framework_helpers) reads CORE at call time
         patch(
-            "esphome.espidf.framework.CORE",
+            "esphome.core.CORE",
             SimpleNamespace(build_path=build_path),
         ),
     )

tests/unit_tests/test_platformio_registry.py

@@ -0,0 +1,310 @@
+"""Tests for esphome.platformio.registry (PIO-registry package installs)."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+import json
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.core import EsphomeError
+from esphome.platformio import registry
+
+
+@pytest.mark.parametrize(
+    ("system", "machine", "expected"),
+    [
+        ("Darwin", "arm64", "darwin_arm64"),
+        ("Darwin", "x86_64", "darwin_x86_64"),
+        ("Windows", "AMD64", "windows_amd64"),
+        # Deviation from upstream: auto-mapped to the emulated-x86 packages
+        ("Windows", "ARM64", "windows_amd64"),
+        ("Windows", "x86", "windows_x86"),
+        ("Linux", "x86_64", "linux_x86_64"),
+        ("Linux", "aarch64", "linux_aarch64"),
+        ("Linux", "i686", "linux_i686"),
+        ("Linux", "armv7l", "linux_armv7l"),
+        # Unknown hosts pass through like upstream; the registry lookup
+        # then fails naming the tag
+        ("FreeBSD", "amd64", "freebsd_amd64"),
+    ],
+)
+def test_get_systype(system: str, machine: str, expected: str) -> None:
+    with (
+        patch("platform.system", return_value=system),
+        patch("platform.machine", return_value=machine),
+        patch("platform.architecture", return_value=("64bit", "")),
+    ):
+        assert registry.get_systype() == expected
+
+
+def test_get_systype_env_override() -> None:
+    """PLATFORMIO_SYSTEM_TYPE wins, exactly as in upstream get_systype()."""
+    with patch.dict(os.environ, {"PLATFORMIO_SYSTEM_TYPE": "windows_amd64"}):
+        assert registry.get_systype() == "windows_amd64"
+
+
+def test_get_systype_aarch64_32bit_userland() -> None:
+    """A 32-bit userland on a 64-bit arm kernel gets armv7l binaries."""
+    with (
+        patch("platform.system", return_value="Linux"),
+        patch("platform.machine", return_value="aarch64"),
+        patch("platform.architecture", return_value=("32bit", "")),
+    ):
+        assert registry.get_systype() == "linux_armv7l"
+
+
+def test_get_systype_windows_empty_machine() -> None:
+    """An empty machine string falls back to the architecture bits."""
+    with (
+        patch("platform.system", return_value="Windows"),
+        patch("platform.machine", return_value=""),
+        patch("platform.architecture", return_value=("64bit", "")),
+    ):
+        assert registry.get_systype() == "windows_amd64"
+
+
+def _registry_response(files: list[dict]):
+    """Patch the shared downloader to serve a canned registry response."""
+    payload = {"versions": [{"name": "1.0.0", "files": files}]}
+
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(json.dumps(payload).encode())
+        return mirrors[0].format(**substitutions)
+
+    return patch.object(registry, "download_from_mirrors", side_effect=fake_download)
+
+
+def test_registry_download_uses_shared_downloader() -> None:
+    """The metadata fetch delegates its retries and error reporting to
+    download_from_mirrors; failures surface unchanged."""
+    with (
+        patch.object(
+            registry,
+            "download_from_mirrors",
+            side_effect=EsphomeError("Failed to download from all mirrors"),
+        ) as mock_download,
+        pytest.raises(EsphomeError, match="Failed to download from all mirrors"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+    (mirrors, substitutions, _), _ = mock_download.call_args
+    assert mirrors == [registry._REGISTRY_URL]
+    assert substitutions == {"package": "pkg"}
+
+
+def test_registry_download_invalid_json_is_clean() -> None:
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(b"<html>not json</html>")
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="invalid JSON"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_matches_system() -> None:
+    with (
+        _registry_response(
+            [
+                {"system": ["windows_amd64"], "download_url": "http://x/win"},
+                {
+                    "system": ["linux_x86_64"],
+                    "download_url": "http://x/linux",
+                    "checksum": {"sha256": "abc123"},
+                    "size": 42,
+                },
+            ]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == (
+            "http://x/linux",
+            "abc123",
+            42,
+        )
+
+
+def test_registry_download_bare_string_system() -> None:
+    """A bare-string system tag is an exact match, not a substring test."""
+    with (
+        _registry_response(
+            [
+                {"system": "linux_x86", "download_url": "http://x/x86"},
+                {
+                    "system": "linux_x86_64",
+                    "download_url": "http://x/x86_64",
+                    "checksum": {"sha256": "abc"},
+                },
+            ]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        assert registry.registry_download("pkg", "1.0.0")[0] == "http://x/x86_64"
+
+
+def test_registry_download_wildcard_system() -> None:
+    with _registry_response(
+        [
+            {
+                "system": "*",
+                "download_url": "http://x/any",
+                "checksum": {"sha256": "abc"},
+                "size": 7,
+            }
+        ]
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == (
+            "http://x/any",
+            "abc",
+            7,
+        )
+
+
+def test_registry_download_missing_checksum_raises() -> None:
+    """An unverifiable archive is refused, never silently extracted."""
+    with (
+        _registry_response([{"system": "*", "download_url": "http://x/any"}]),
+        pytest.raises(EsphomeError, match="no sha256"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_no_system_match() -> None:
+    with (
+        _registry_response(
+            [{"system": ["windows_amd64"], "download_url": "http://x/win"}]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_version_not_found() -> None:
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(
+            json.dumps({"versions": [{"name": "2.0.0", "files": []}]}).encode()
+        )
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="not found"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    dest.mkdir()
+    (dest / ".esphome_extracted").touch()
+    with patch.object(registry, "download_from_mirrors") as mock_download:
+        registry.install_package("pkg", "1.0.0", dest, [], tmp_path / "dl")
+    mock_download.assert_not_called()
+
+
+def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
+    with (
+        patch.object(registry, "download_from_mirrors") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        # Extraction is expected to create the directory
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
+        registry.install_package("pkg", "1.0.0", dest, mirrors, tmp_path / "dl")
+    assert mock_download.call_args[0][0] is mirrors
+    assert mock_download.call_args[0][1] == {
+        "VERSION": "1.0.0",
+        "SYSTEM": "linux_x86_64",
+    }
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
+    """The registry path downloads with the registry's sha256 and size."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(
+            registry,
+            "registry_download",
+            return_value=("http://x/pkg.tar.gz", "abc123", 42),
+        ),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
+        registry.install_package("pkg", "1.0.0", dest, [], tmp_path / "dl")
+    assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
+    assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
+
+
+def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
+    """The success marker is only written when the extracted tree is usable."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "bin").mkdir(parents=True)
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
+        )
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="without the expected bin"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir()
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
+        )
+    assert not (dest / ".esphome_extracted").exists()
+
+
+def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None:
+    """A concurrent install finishing while we wait for the lock is detected."""
+    dest = tmp_path / "pkg"
+    marker = dest / ".esphome_extracted"
+
+    @contextmanager
+    def _fake_lock(*_a, **_kw):
+        dest.mkdir(parents=True, exist_ok=True)
+        marker.touch()
+        yield
+
+    with (
+        patch("filelock.FileLock", _fake_lock),
+        patch.object(registry, "download_from_mirrors") as mock_download,
+        patch.object(registry, "rmdir") as mock_rmdir,
+    ):
+        registry.install_package("pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl")
+    mock_download.assert_not_called()
+    mock_rmdir.assert_not_called()
+
+
+def test_install_package_uses_hard_lock(tmp_path: Path) -> None:
+    """The install lock must never degrade to a soft (existence) lock."""
+    dest = tmp_path / "pkg"
+    with (
+        patch("filelock.FileLock") as mock_lock,
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: dest.mkdir(exist_ok=True)
+        registry.install_package("pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl")
+    assert mock_lock.call_args.kwargs["fallback_to_soft"] is False

tests/unit_tests/test_platformio_toolchain.py

@@ -431,8 +431,8 @@ def test_ccache_env_enabled_by_default(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -469,7 +469,7 @@ def test_ccache_env_disabled_without_binary(
 
     with (
         patch.dict(os.environ, env_vars, clear=True),
-        patch.object(toolchain.shutil, "which", return_value=None),
+        patch("shutil.which", return_value=None),
         caplog.at_level("WARNING"),
     ):
         env = toolchain._ccache_env()
@@ -494,8 +494,8 @@ def test_ccache_env_disabled_when_probe_fails(
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run", side_effect=probe_error),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=probe_error),
     ):
         env = toolchain._ccache_env()
 
@@ -508,8 +508,8 @@ def test_ccache_env_forced_on_skips_probe(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run") as mock_probe,
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -538,8 +538,8 @@ def test_ccache_env_strips_win_long_path_prefix(setup_core: Path) -> None:
         # shutil.which is patched, so the win32 code path of the real
         # implementation (which crashes on a POSIX host) is never reached.
         patch("esphome.platformio.toolchain.sys.platform", "win32"),
-        patch.object(toolchain.shutil, "which", return_value=prefixed),
-        patch.object(toolchain.subprocess, "run") as mock_probe,
+        patch("shutil.which", return_value=prefixed),
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -555,7 +555,7 @@ def test_ccache_env_opt_out(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
     ):
         env = toolchain._ccache_env()
 
@@ -568,7 +568,7 @@ def test_ccache_env_normalizes_enable_value(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "yes"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
     ):
         env = toolchain._ccache_env()
 
@@ -587,8 +587,8 @@ def test_ccache_env_respects_user_values_and_refreshes_basedir(
 
     with (
         patch.dict(os.environ, user_env, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -606,8 +606,8 @@ def test_run_platformio_cli_passes_ccache_env_to_subprocess_only(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         mock_run_external_process.return_value = 0
@@ -628,8 +628,8 @@ def test_ccache_env_requires_build_path(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
         pytest.raises(ValueError, match="CORE.build_path must be set"),
     ):
         toolchain._ccache_env()
@@ -642,8 +642,8 @@ def test_run_platformio_cli_merges_caller_env(
     CORE.build_path = str(setup_core / "build" / "test")
 
     with (
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         mock_run_external_process.return_value = 0
         toolchain.run_platformio_cli(
@@ -800,9 +800,7 @@ def test_ccache_env_real_probe_runs_stripped_path(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch.object(
-            toolchain.shutil, "which", return_value="\\\\?\\" + sys.executable
-        ),
+        patch("shutil.which", return_value="\\\\?\\" + sys.executable),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         env = toolchain._ccache_env()
@@ -874,7 +872,7 @@ def test_strip_win_long_path_prefix(
 ) -> None:
     r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
     with patch("esphome.platformio.toolchain.sys.platform", platform):
-        assert toolchain._strip_win_long_path_prefix(input_path) == expected
+        assert toolchain.strip_win_long_path_prefix(input_path) == expected
 
 
 def test_run_platformio_cli_strips_win_long_path_prefix(