[esp8266] Use our own compiler builds for the native toolchain (#19849)

esphome/__main__.py

@@ -838,7 +838,7 @@ def compile_program(args: ArgsProtocol, config: ConfigType) -> int:
 
     # Keep this here, NOT in codegen: config-hash and --only-generate must keep
     # working on machines that cannot run the toolchain.
-    if CORE.is_esp8266:
+    if CORE.is_esp8266 and CORE.using_toolchain_platformio:
         from esphome.components.esp8266 import check_rosetta
 
         check_rosetta()

esphome/arduino8266/framework.py

@@ -4,11 +4,12 @@ Artifacts land in a machine-global cache (shared across projects, like the
 ESP-IDF install in ``esphome.espidf.framework``):
 
     <cache>/arduino8266/frameworks/<version>/   framework-arduinoespressif8266
-    <cache>/arduino8266/toolchains/<version>/   toolchain-xtensa (gcc 10.3)
+    <cache>/arduino8266/toolchains/<version>/   xtensa-lx106-elf gcc 10.3
 
-Packages come from the PlatformIO registry (identical bits to the PlatformIO
-backend); ``ESPHOME_ARDUINO8266_*_MIRRORS`` overrides the URLs. ninja comes
-from PATH or the ninja PyPI wheel.
+The framework comes from the PlatformIO registry, the toolchain from
+https://github.com/esphome-libs/xtensa-lx106-elf-toolchain/releases/;
+``ESPHOME_ARDUINO8266_*_MIRRORS`` overrides the URLs. ninja comes from PATH
+or the ninja PyPI wheel.
 """
 
 from __future__ import annotations
@@ -22,13 +23,45 @@ from esphome.build_helpers.ninja import find_ninja
 from esphome.build_helpers.tools_cache import ARDUINO8266_TOOLS_CACHE, tools_cache_path
 from esphome.core import EsphomeError, Version
 from esphome.framework_helpers import str_to_lst_of_str
-from esphome.platformio.registry import install_package, prefetch_packages
+from esphome.platformio.registry import (
+    Download,
+    Resolver,
+    get_systype,
+    install_package,
+    prefetch_packages,
+)
 
 FRAMEWORK_PACKAGE = "framework-arduinoespressif8266"
-TOOLCHAIN_PACKAGE = "toolchain-xtensa"
+TOOLCHAIN_PACKAGE = "toolchain-xtensa-lx106-elf"
 # gcc 10.3, the toolchain Arduino core 3.x builds with; the build
 # generator's compile flags are tuned to it.
-TOOLCHAIN_VERSION = "2.100300.220621"
+TOOLCHAIN_VERSION = "10.3.0-esphome.2"
+_TOOLCHAIN_RELEASES = (
+    "https://github.com/esphome-libs/xtensa-lx106-elf-toolchain/releases/"
+)
+# Registry system tag -> (sha256, size) of that host's archive
+TOOLCHAIN_BUILDS: dict[str, tuple[str, int]] = {
+    "darwin_arm64": (
+        "849cede44d4d5c6ea0f14099783239f559f46327bea314281814f2652b486201",
+        60830321,
+    ),
+    "darwin_x86_64": (
+        "ca69904daabf0c5983b372423e5e62f49182a793e992c052e94666852470c897",
+        64149487,
+    ),
+    "linux_aarch64": (
+        "60a49a4f082bf246544bd409a9517dbbcab19bb30ac9decbee544b896aaccbd6",
+        67573397,
+    ),
+    "linux_x86_64": (
+        "1fba33ca1494ec79f2776e0e37eca93282d30f8bb9992f5f4f9a655d6fff1db4",
+        68431336,
+    ),
+    "windows_amd64": (
+        "af9066b0e5bf036f04f2bd9d08b89b81a7f183c57dac0abcaff71dd861cf5f3b",
+        67664137,
+    ),
+}
 
 ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS = str_to_lst_of_str(
     os.environ.get("ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS", "")
@@ -74,6 +107,22 @@ def get_toolchain_path() -> Path:
     return get_arduino8266_tools_path() / "toolchains" / TOOLCHAIN_VERSION
 
 
+def toolchain_download() -> Download:
+    """The toolchain archive for the current host."""
+    systype = get_systype()
+    if (build := TOOLCHAIN_BUILDS.get(systype)) is None:
+        raise EsphomeError(
+            f"There is no ESP8266 toolchain for this system ({systype}); "
+            f"supported systems are {', '.join(sorted(TOOLCHAIN_BUILDS))}. "
+            "Either set 'toolchain: platformio' under 'esp8266:', or point "
+            "ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS at a toolchain archive"
+        )
+    sha256, size = build
+    archive = f"{TOOLCHAIN_PACKAGE}-{TOOLCHAIN_VERSION}-{systype}.tar.gz"
+    url = f"{_TOOLCHAIN_RELEASES}download/{TOOLCHAIN_VERSION}/{archive}"
+    return Download(url, sha256, size)
+
+
 class InstalledPaths(NamedTuple):
     """Locations of the installed framework, toolchain, and ninja binary."""
 
@@ -115,10 +164,23 @@ def check_and_install(framework_version: Version) -> InstalledPaths:
             ("bin", "xtensa-lx106-elf"),
         ),
     )
+    # Resolved only when a download is needed, so an installed toolchain
+    # keeps working on a host without a build; a mirror override wins
+    resolvers: dict[str, Resolver] = {}
+    if not ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS:
+        resolvers[TOOLCHAIN_PACKAGE] = toolchain_download
     # Fetch both archives at once; the installs below verify and extract
-    prefetch_packages([spec[:4] for spec in specs], downloads_dir)
+    prefetch_packages([spec[:4] for spec in specs], downloads_dir, resolvers)
     for name, version, dest, mirrors, expect in specs:
-        install_package(name, version, dest, mirrors, downloads_dir, expect=expect)
+        install_package(
+            name,
+            version,
+            dest,
+            mirrors,
+            downloads_dir,
+            expect=expect,
+            resolve=resolvers.get(name),
+        )
     return InstalledPaths(
         framework=framework_path, toolchain=toolchain_path, ninja=ninja_path
     )

esphome/components/esp8266/__init__.py

@@ -314,8 +314,8 @@ CONFIG_SCHEMA = cv.All(
 def check_rosetta() -> None:
     """Fail fast when the x86_64 ESP8266 toolchain cannot run on this Mac.
 
-    There is no native arm64 build of the xtensa-lx106 toolchain; on Apple
-    Silicon it runs under Rosetta 2, which macOS updates can remove.
+    PlatformIO's xtensa-lx106 toolchain has no arm64 build; on Apple Silicon
+    it runs under Rosetta 2, which macOS updates can remove.
     """
     if not IS_MACOS or platform.machine() != "arm64":
         return
@@ -330,9 +330,10 @@ def check_rosetta() -> None:
         return  # arch(1) unavailable; let the build proceed
     if result.returncode != 0:
         raise EsphomeError(
-            "ESP8266 builds on Apple Silicon Macs use an Intel (x86_64) "
-            "compiler that requires Rosetta 2, which is not installed on "
-            "this system. Install it with:\n"
+            "ESP8266 builds with PlatformIO on Apple Silicon Macs use an Intel "
+            "(x86_64) compiler that requires Rosetta 2, which is not installed "
+            "on this system. Either set 'toolchain: arduino' under 'esp8266:', "
+            "which needs no Rosetta, or install it with:\n"
             "  softwareupdate --install-rosetta --agree-to-license"
         )
 

esphome/platformio/registry.py

@@ -31,6 +31,18 @@ _REGISTRY_URL = (
 )
 
 
+class Download(NamedTuple):
+    """A package archive to fetch and verify."""
+
+    url: str
+    sha256: str
+    size: int | None
+
+
+# Looks a package's download up; called only when a download is needed
+Resolver = Callable[[], Download]
+
+
 def get_systype() -> str:
     """The registry system tag for the current host.
 
@@ -56,7 +68,7 @@ def get_systype() -> str:
 
 
 @cache
-def registry_download(package: str, version: str) -> tuple[str, str, int | None]:
+def registry_download(package: str, version: str) -> Download:
     """Resolve a package's download URL, sha256, and size via the registry.
 
     The metadata fetch goes through ``http_request``/``fetch_with_retry``
@@ -144,7 +156,7 @@ def registry_download(package: str, version: str) -> tuple[str, str, int | None]
                         f"The package registry returned no download URL for "
                         f"{package} {version}"
                     )
-                return (url, sha256, file.get("size"))
+                return Download(url, sha256, file.get("size"))
         raise EsphomeError(
             f"No {package} {version} build for this platform ({systype})"
         )
@@ -177,17 +189,20 @@ def _archive_path(downloads_dir: Path, name: str, version: str) -> Path:
     return downloads_dir / f"{name}-{version}"
 
 
-def _already_installed(dest: Path) -> bool:
+def is_installed(dest: Path) -> bool:
     """Whether ``dest`` holds a completed install (extraction marker)."""
     return (dest / ".esphome_extracted").is_file()
 
 
 def prefetch_packages(
-    packages: list[tuple[str, str, Path, list[str]]], downloads_dir: Path
+    packages: list[tuple[str, str, Path, list[str]]],
+    downloads_dir: Path,
+    resolvers: dict[str, Resolver] | None = None,
 ) -> None:
     """Download pending package archives in parallel under one combined bar.
 
-    ``packages`` holds ``(name, version, dest, mirrors)`` per package. Purely
+    ``packages`` holds ``(name, version, dest, mirrors)`` per package;
+    ``resolvers`` replaces the registry lookup by name. Purely
     an optimization: ``install_package`` verifies every archive and
     re-downloads anything this pass left unfinished. Mirror overrides and
     registry entries without a size stay on the sequential path so its
@@ -200,15 +215,18 @@ def prefetch_packages(
     pending: list[_PendingArchive] = []
     seen: set[Path] = set()
     for name, version, dest, mirrors in packages:
-        if mirrors or (dest / ".esphome_extracted").is_file():
+        if mirrors or is_installed(dest):
             continue
         archive = _archive_path(downloads_dir, name, version)
         if archive in seen:
             # A duplicate entry would race itself between two workers
             continue
         seen.add(archive)
+        resolve = (resolvers or {}).get(name) or partial(
+            registry_download, name, version
+        )
         try:
-            url, sha256, size = registry_download(name, version)
+            url, sha256, size = resolve()
         except EsphomeError as err:
             # The sequential install reports the real failure with context
             _LOGGER.debug("Prefetch resolve for %s failed: %s", name, err)
@@ -234,7 +252,7 @@ def prefetch_packages(
             if done := downloaded_bytes(entry.archive, entry.size):
                 return done
             # The holder deletes the archive once it has installed it
-            return entry.size if _already_installed(entry.dest) else 0
+            return entry.size if is_installed(entry.dest) else 0
 
         lock = FileLock(f"{entry.dest}.lock", fallback_to_soft=False)
         try:
@@ -245,7 +263,7 @@ def prefetch_packages(
             _LOGGER.debug("Leaving %s to its current downloader", entry.name)
             return
         try:
-            if _already_installed(entry.dest):
+            if is_installed(entry.dest):
                 # A concurrent build installed it while we waited; a
                 # re-download would orphan a fresh copy in downloads_dir
                 tracker(entry.size)
@@ -282,13 +300,15 @@ def install_package(
     mirrors: list[str],
     downloads_dir: Path,
     expect: Collection[str],
+    resolve: Resolver | None = None,
 ) -> 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.
+    archive between runs so an interrupted download resumes. ``resolve``
+    replaces the registry lookup.
     """
     if not expect:
         # Layout validation before marker.touch() is the only guard against
@@ -323,7 +343,9 @@ def install_package(
                 mirrors, {"VERSION": version, "SYSTEM": get_systype()}, archive
             )
         else:
-            url, sha256, size = registry_download(name, version)
+            url, sha256, size = (
+                resolve() if resolve else 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")

tests/unit_tests/test_arduino8266_framework.py

@@ -54,6 +54,91 @@ def test_tools_path_default_and_prefix(tmp_path: Path) -> None:
     assert path != Path.cwd()
 
 
+def test_toolchain_builds_are_pinned() -> None:
+    """A release must not be pinned without its checksums."""
+    for sha256, size in framework.TOOLCHAIN_BUILDS.values():
+        assert len(sha256) == 64
+        assert size > 0
+
+
+def test_toolchain_download() -> None:
+    sha256, size = framework.TOOLCHAIN_BUILDS["darwin_arm64"]
+    with (
+        patch.object(framework, "TOOLCHAIN_VERSION", "1.2.3"),
+        patch.object(framework, "get_systype", return_value="darwin_arm64"),
+    ):
+        download = framework.toolchain_download()
+    releases = "https://github.com/esphome-libs/xtensa-lx106-elf-toolchain/releases/"
+    archive = "toolchain-xtensa-lx106-elf-1.2.3-darwin_arm64.tar.gz"
+    assert download == (f"{releases}download/1.2.3/{archive}", sha256, size)
+
+
+def test_toolchain_download_unsupported_system() -> None:
+    with (
+        patch.object(framework, "get_systype", return_value="linux_armv7l"),
+        pytest.raises(
+            EsphomeError, match=r"linux_armv7l.*darwin_arm64.*toolchain: platformio"
+        ),
+    ):
+        framework.toolchain_download()
+
+
+def _fake_framework(tmp_path: Path) -> None:
+    """The layout install_package expects of an installed framework."""
+    for sub in ("cores/esp8266", "tools/sdk", "libraries"):
+        (tmp_path / "frameworks" / "3.30102.0" / sub).mkdir(parents=True)
+    (tmp_path / "frameworks" / "3.30102.0" / ".esphome_extracted").touch()
+
+
+def test_check_and_install_mirror_skips_pinned_toolchain(tmp_path: Path) -> None:
+    """With a mirror override an unsupported host can bring its own toolchain."""
+    with (
+        patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
+        patch.object(framework, "ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS", ["http://m"]),
+        patch.object(framework, "install_package") as mock_install,
+        patch.object(framework, "prefetch_packages") as mock_prefetch,
+        patch.object(framework, "find_ninja", return_value=tmp_path / "ninja"),
+    ):
+        framework.check_and_install(cv.Version(3, 1, 2))
+    assert mock_prefetch.call_args.args[2] == {}
+    assert mock_install.call_args_list[1].kwargs["resolve"] is None
+
+
+def test_check_and_install_installed_toolchain_on_unsupported_host(
+    tmp_path: Path,
+) -> None:
+    """The host is only checked when the toolchain must be downloaded."""
+    _fake_framework(tmp_path)
+    toolchain = tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION
+    for sub in ("bin", "xtensa-lx106-elf"):
+        (toolchain / sub).mkdir(parents=True)
+    (toolchain / ".esphome_extracted").touch()
+    with (
+        patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
+        patch.object(framework, "get_systype", return_value="linux_armv7l"),
+        patch.object(framework, "prefetch_packages"),
+        patch.object(framework, "find_ninja", return_value=tmp_path / "ninja"),
+    ):
+        paths = framework.check_and_install(cv.Version(3, 1, 2))
+    assert paths.toolchain == toolchain
+
+
+def test_check_and_install_unsupported_host_without_toolchain_raises(
+    tmp_path: Path,
+) -> None:
+    """Without an installed toolchain the download is needed, and the host
+    error names the ways out."""
+    _fake_framework(tmp_path)
+    with (
+        patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
+        patch.object(framework, "get_systype", return_value="linux_armv7l"),
+        patch.object(framework, "prefetch_packages"),
+        patch.object(framework, "find_ninja", return_value=tmp_path / "ninja"),
+        pytest.raises(EsphomeError, match=r"linux_armv7l.*toolchain: platformio"),
+    ):
+        framework.check_and_install(cv.Version(3, 1, 2))
+
+
 def test_check_and_install_returns_paths(tmp_path: Path) -> None:
     with (
         patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
@@ -76,7 +161,10 @@ def test_check_and_install_returns_paths(tmp_path: Path) -> None:
         framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
         tmp_path / "downloads",
     )
-    assert fw_call.kwargs["expect"] == ("cores/esp8266", "tools/sdk", "libraries")
+    assert fw_call.kwargs == {
+        "expect": ("cores/esp8266", "tools/sdk", "libraries"),
+        "resolve": None,
+    }
     assert tc_call.args == (
         framework.TOOLCHAIN_PACKAGE,
         framework.TOOLCHAIN_VERSION,
@@ -84,7 +172,10 @@ def test_check_and_install_returns_paths(tmp_path: Path) -> None:
         framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
         tmp_path / "downloads",
     )
-    assert tc_call.kwargs["expect"] == ("bin", "xtensa-lx106-elf")
+    assert tc_call.kwargs == {
+        "expect": ("bin", "xtensa-lx106-elf"),
+        "resolve": framework.toolchain_download,
+    }
     # The prefetch sees the same package specs as the installs
     assert mock_prefetch.call_args.args == (
         [
@@ -102,6 +193,7 @@ def test_check_and_install_returns_paths(tmp_path: Path) -> None:
             ),
         ],
         tmp_path / "downloads",
+        {framework.TOOLCHAIN_PACKAGE: framework.toolchain_download},
     )
 
 

tests/unit_tests/test_main.py

@@ -5999,6 +5999,26 @@ def test_compile_program_esp8266_runs_rosetta_check(tmp_path: Path) -> None:
     mock_check.assert_called_once()
 
 
+def test_compile_program_esp8266_native_skips_rosetta_check(tmp_path: Path) -> None:
+    """The native toolchain has an arm64 macOS build, so nothing is probed."""
+    setup_core(platform=PLATFORM_ESP8266, tmp_path=tmp_path, name="test_device")
+    CORE.toolchain = Toolchain.ARDUINO
+
+    config: dict[str, Any] = {CONF_ESPHOME: {CONF_NAME: "test_device"}}
+
+    with (
+        patch("esphome.components.esp8266.check_rosetta") as mock_check,
+        patch(
+            "esphome.components.esp8266.run_compile",
+            side_effect=EsphomeError("compile reached"),
+        ),
+        pytest.raises(EsphomeError, match="compile reached"),
+    ):
+        compile_program(MockArgs(), config)
+
+    mock_check.assert_not_called()
+
+
 def test_compile_program_skips_rosetta_check_on_other_platforms(
     tmp_path: Path,
     mock_compile_build_info_run_compile: Mock,

tests/unit_tests/test_platformio_registry.py

@@ -311,6 +311,56 @@ def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
     assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
 
 
+def test_install_package_downloads_pinned(tmp_path: Path) -> None:
+    """A resolver's download is verified like a registry one, with no lookup."""
+    dest = tmp_path / "pkg"
+    pinned = registry.Download("http://y/pinned.tar.gz", "def456", 7)
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "registry_download") as mock_registry,
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg",
+            "1.0.0",
+            dest,
+            [],
+            tmp_path / "dl",
+            expect=("payload",),
+            resolve=lambda: pinned,
+        )
+    mock_registry.assert_not_called()
+    assert mock_download.call_args[0][0] == "http://y/pinned.tar.gz"
+    assert mock_download.call_args[1] == {"sha256": "def456", "size": 7}
+
+
+def test_install_package_mirror_wins_over_pinned(tmp_path: Path) -> None:
+    """A mirror override replaces the pinned download."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors") as mock_mirrors,
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg",
+            "1.0.0",
+            dest,
+            ["http://mirror/{VERSION}"],
+            tmp_path / "dl",
+            expect=("payload",),
+            resolve=lambda: registry.Download("http://y/pinned.tar.gz", "def456", 7),
+        )
+    mock_mirrors.assert_called_once()
+    mock_download.assert_not_called()
+
+
 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"
@@ -535,6 +585,29 @@ def test_prefetch_packages_downloads_pending_in_parallel(tmp_path: Path) -> None
         assert callable(call[1]["progress"])
 
 
+def test_prefetch_packages_uses_pinned_download(tmp_path: Path) -> None:
+    """A package with a resolver skips the registry; the rest use it."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10})
+        ) as mock_registry,
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+            {"b": lambda: registry.Download("http://y/b.tar.gz", "def456", 20)},
+        )
+    mock_registry.assert_called_once_with("a", "1.0")
+    calls = sorted(mock_download.call_args_list, key=lambda c: c[0][0])
+    assert [c[0][0] for c in calls] == ["http://x/a.tar.gz", "http://y/b.tar.gz"]
+    assert calls[1][1]["sha256"] == "def456"
+    assert calls[1][1]["size"] == 20
+
+
 def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
     """A dest whose marker appeared while the worker waited on the lock is
     already installed; re-downloading would orphan an archive copy."""
@@ -620,13 +693,13 @@ def test_prefetch_packages_leaves_a_long_held_lock_to_its_holder(
     mock_download.assert_not_called()
 
 
-def test_already_installed_probe(tmp_path: Path) -> None:
+def test_is_installed_probe(tmp_path: Path) -> None:
     """Both arms of the marker probe the prefetch worker keys on."""
     dest = tmp_path / "pkg"
     dest.mkdir()
-    assert registry._already_installed(dest) is False
+    assert registry.is_installed(dest) is False
     (dest / ".esphome_extracted").touch()
-    assert registry._already_installed(dest) is True
+    assert registry.is_installed(dest) is True
 
 
 def test_prefetch_packages_dedupes_duplicate_entries(tmp_path: Path) -> None: