diff --git a/esphome/__main__.py b/esphome/__main__.py index 561391708e..33f1727696 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -2117,6 +2117,16 @@ def parse_args(argv): help="Upload as bootloader (OTA).", action="store_true", ) + parser_upload.add_argument( + "--from-storage-json", + action="store_true", + help=( + "Skip YAML schema validation by loading device metadata from " + "the .esphome/storage/.json sidecar produced by the last " + "successful compile. Falls back to a full validation pass when " + "the sidecar is missing or older than the YAML." + ), + ) parser_logs = subparsers.add_parser( "logs", @@ -2144,6 +2154,16 @@ def parse_args(argv): action="store_true", help="Do not show entity state changes in log output.", ) + parser_logs.add_argument( + "--from-storage-json", + action="store_true", + help=( + "Skip YAML schema validation by loading device metadata from " + "the .esphome/storage/.json sidecar produced by the last " + "successful compile. Falls back to a full validation pass when " + "the sidecar is missing or older than the YAML." + ), + ) parser_discover = subparsers.add_parser( "discover", @@ -2417,10 +2437,39 @@ def run_esphome(argv): # Commands that don't need fresh external components: logs just connects # to the device, and clean is about to delete the build directory. skip_external = args.command in ("logs", "clean") - config = read_config( - dict(args.substitution) if args.substitution else {}, - skip_external_update=skip_external, - ) + command_line_substitutions = dict(args.substitution) if args.substitution else {} + + # Fast path for `upload --from-storage-json` and `logs --from-storage-json`: + # the caller already has a binary on disk and only needs the CLI to ship + # bytes to a device or stream logs back. Re-running the full + # `read_config()` pipeline (parse + schema validate + final-validate + + # external component refresh) for those two subcommands is dead work and + # produces a wall of "Reading configuration ..." log lines for every + # remote install. Source the platform / build metadata from the + # StorageJSON sidecar instead; fall back to full validation if it's + # missing or stale so a cold cache never produces a worse outcome. + config = None + if getattr(args, "from_storage_json", False) and args.command in ( + "upload", + "logs", + ): + from esphome.lite_config import load_lite_config_from_storage + + config = load_lite_config_from_storage(conf_path, command_line_substitutions) + if config is None: + _LOGGER.warning( + "StorageJSON sidecar missing or stale for %s; falling back to " + "full config validation.", + conf_path, + ) + else: + _LOGGER.info("Loaded device metadata from StorageJSON sidecar.") + + if config is None: + config = read_config( + command_line_substitutions, + skip_external_update=skip_external, + ) if config is None: return 2 CORE.config = config diff --git a/esphome/lite_config.py b/esphome/lite_config.py new file mode 100644 index 0000000000..89816a00c7 --- /dev/null +++ b/esphome/lite_config.py @@ -0,0 +1,197 @@ +"""Lite-config loader for ``upload`` / ``logs`` fast paths. + +When the caller (typically a dashboard) already has a known-good +firmware binary on disk and only needs the CLI to ship bytes to a +device or stream logs back, running the full ``read_config()`` +pipeline is dead work. It re-parses every ``!include``, runs every +component's schema validator, executes all final-validate hooks, and +fetches/refreshes external components -- producing a fully validated +``Config`` object that the upload / logs subcommands only consult for a +handful of leaf keys. + +This module loads the per-config ``StorageJSON`` sidecar that was +written by the last successful ``compile`` run, populates ``CORE`` with +the platform / build metadata from it, and re-parses just enough of the +YAML head (substitutions + packages, no schema validation) to recover +``api:`` / ``logger:`` / ``ota:`` / ``mqtt:`` / network blocks for the +subcommand callers. + +If the sidecar is missing or older than the YAML, this returns ``None`` +so the dispatcher can fall back to the full ``read_config()`` path. +That makes the flag a pure optimisation: a cold cache never produces a +worse outcome than today. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +from esphome.const import ( + CONF_API, + CONF_ESPHOME, + CONF_ETHERNET, + CONF_FRIENDLY_NAME, + CONF_LOGGER, + CONF_MQTT, + CONF_NAME, + CONF_OPENTHREAD, + CONF_OTA, + CONF_USE_ADDRESS, + CONF_WEB_SERVER, + CONF_WIFI, + KEY_CORE, + KEY_TARGET_FRAMEWORK, + KEY_TARGET_PLATFORM, +) +from esphome.core import CORE +from esphome.storage_json import StorageJSON, ext_storage_path +from esphome.types import ConfigType + +_LOGGER = logging.getLogger(__name__) + +# Top-level keys we copy from the raw YAML into the lite config dict. +# Anything outside this list is irrelevant to ``upload`` / ``logs``. +_LITE_TOP_LEVEL_KEYS: tuple[str, ...] = ( + CONF_ESPHOME, + CONF_API, + CONF_LOGGER, + CONF_OTA, + CONF_MQTT, + CONF_WIFI, + CONF_ETHERNET, + CONF_OPENTHREAD, + CONF_WEB_SERVER, +) + + +def _parse_yaml_head( + conf_path: Path, command_line_substitutions: dict[str, Any] | None +) -> dict[str, Any] | None: + """Load and lightly process the YAML, without schema validation. + + Resolves ``!include`` / ``!secret`` (via ``yaml_util``), + substitutions, and ``packages:`` -- the three passes whose output + the upload / logs subcommands need. Returns ``None`` on any error + so the caller can fall back to the full ``read_config()`` path. + """ + from esphome import yaml_util + from esphome.components.packages import resolve_packages + from esphome.components.substitutions import do_substitution_pass + + try: + config = yaml_util.load_yaml(conf_path) + except Exception: # pylint: disable=broad-except + return None + + try: + config = do_substitution_pass(config, command_line_substitutions) + except Exception: # pylint: disable=broad-except + return None + + try: + config = resolve_packages( + config, command_line_substitutions=command_line_substitutions + ) + except Exception: # pylint: disable=broad-except + return None + + return config + + +def _build_lite_config(raw_config: dict[str, Any]) -> ConfigType: + """Project the raw YAML down to the keys upload / logs read.""" + return {key: raw_config[key] for key in _LITE_TOP_LEVEL_KEYS if key in raw_config} + + +def _populate_core_from_storage(storage: StorageJSON) -> None: + """Populate ``CORE`` fields the subcommands read off the platform.""" + CORE.name = storage.name + CORE.friendly_name = storage.friendly_name + CORE.build_path = storage.build_path + CORE.loaded_integrations = set(storage.loaded_integrations) + CORE.loaded_platforms = set(storage.loaded_platforms) + + core_platform = storage.core_platform or ( + storage.target_platform.lower() if storage.target_platform else None + ) + CORE.data.setdefault(KEY_CORE, {}) + if core_platform is not None: + CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = core_platform + if storage.framework is not None: + CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] = storage.framework + + +def _ensure_address_in_config(config: ConfigType, storage: StorageJSON) -> None: + """Make sure ``CORE.address`` resolves once the lite config is set. + + ``CORE.address`` reads ``use_address`` off the wifi / ethernet / + openthread block; if the dashboard rewrote the YAML between the + last compile and now those blocks may be missing from the lite + parse. As a backstop, lift the address from the StorageJSON when + we have one and the YAML doesn't supply one already. + """ + if storage.address is None: + return + for network_type in (CONF_WIFI, CONF_ETHERNET, CONF_OPENTHREAD): + if network_type in config and CONF_USE_ADDRESS in config[network_type]: + return + # Fall back to a synthetic wifi block so ``CORE.address`` resolves. + config.setdefault(CONF_WIFI, {})[CONF_USE_ADDRESS] = storage.address + + +def load_lite_config_from_storage( + conf_path: Path, command_line_substitutions: dict[str, Any] | None = None +) -> ConfigType | None: + """Build a minimal config + populate CORE from the StorageJSON sidecar. + + Returns the lite config dict on success, or ``None`` when the + dispatcher should fall back to ``read_config()`` (sidecar missing, + stale, unreadable, or required fields absent). + """ + storage_path = ext_storage_path(conf_path.name) + try: + yaml_stat = conf_path.stat() + except OSError: + return None + try: + storage_stat = storage_path.stat() + except OSError: + return None + + # The sidecar is written by `compile`; if the YAML has been edited + # since, the cached platform / loaded_integrations may no longer + # describe what the binary on disk was built from. Fall back to a + # full validation pass in that case. + if storage_stat.st_mtime < yaml_stat.st_mtime: + return None + + storage = StorageJSON.load(storage_path) + if storage is None: + return None + if not storage.target_platform and not storage.core_platform: + # An incomplete sidecar (e.g. from a wizard run that never + # compiled) can't drive the upload / logs subcommands. + return None + + raw_config = _parse_yaml_head(conf_path, command_line_substitutions) + if raw_config is None: + return None + if CONF_ESPHOME not in raw_config or CONF_NAME not in raw_config[CONF_ESPHOME]: + return None + + config = _build_lite_config(raw_config) + + # Backfill `esphome:` block fields from the sidecar so downstream + # callers that read `config["esphome"]["name"]` work even when the + # YAML uses substitutions that didn't survive the light parse. + esphome_block = config.setdefault(CONF_ESPHOME, {}) + esphome_block.setdefault(CONF_NAME, storage.name) + if storage.friendly_name is not None: + esphome_block.setdefault(CONF_FRIENDLY_NAME, storage.friendly_name) + + _populate_core_from_storage(storage) + _ensure_address_in_config(config, storage) + + return config diff --git a/tests/unit_tests/test_main_lite_config.py b/tests/unit_tests/test_main_lite_config.py new file mode 100644 index 0000000000..cf0fa3c086 --- /dev/null +++ b/tests/unit_tests/test_main_lite_config.py @@ -0,0 +1,280 @@ +"""Tests for the StorageJSON-backed lite config fast path.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from esphome.__main__ import run_esphome +from esphome.const import ( + CONF_API, + CONF_ESPHOME, + CONF_LOGGER, + CONF_NAME, + CONF_OTA, + CONF_USE_ADDRESS, + CONF_WIFI, + KEY_CORE, + KEY_TARGET_FRAMEWORK, + KEY_TARGET_PLATFORM, +) +from esphome.core import CORE +from esphome.lite_config import load_lite_config_from_storage + +# A YAML the lite parser can load. Contains every block ``upload`` / +# ``logs`` cares about (api, logger, ota, wifi) plus a substitution so +# the tests exercise the substitution pass too. +_SAMPLE_YAML = """\ +substitutions: + device_name: lite_test + encryption_key: 6dGhpcyBpcyBhIHRlc3Q= + +esphome: + name: ${device_name} + friendly_name: Lite Test Device + +esp32: + board: nodemcu-32s + +logger: + +api: + encryption: + key: ${encryption_key} + +ota: + - platform: esphome + +wifi: + ssid: "ssid" + password: "password" + use_address: 192.168.1.42 +""" + + +def _write_storage( + storage_path: Path, + *, + name: str = "lite_test", + friendly_name: str | None = "Lite Test Device", + address: str | None = "192.168.1.42", + target_platform: str = "ESP32", + core_platform: str | None = "esp32", + framework: str | None = "arduino", + build_path: str | None = "/build/lite_test", + firmware_bin_path: str | None = "/build/lite_test/firmware.bin", + loaded_integrations: list[str] | None = None, + loaded_platforms: list[str] | None = None, +) -> None: + """Write a StorageJSON sidecar to ``storage_path`` for the tests.""" + storage_path.parent.mkdir(parents=True, exist_ok=True) + data = { + "storage_version": 1, + "name": name, + "friendly_name": friendly_name, + "comment": None, + "esphome_version": "2026.1.0", + "src_version": 1, + "address": address, + "web_port": None, + "esp_platform": target_platform, + "build_path": build_path, + "firmware_bin_path": firmware_bin_path, + "loaded_integrations": loaded_integrations or ["api", "logger", "ota", "wifi"], + "loaded_platforms": loaded_platforms or [], + "no_mdns": False, + "framework": framework, + "core_platform": core_platform, + } + storage_path.write_text(json.dumps(data)) + + +@pytest.fixture +def lite_config_files(tmp_path: Path) -> tuple[Path, Path]: + """Create a YAML + matching StorageJSON for the lite-config tests. + + The YAML lives at ``/lite_test.yaml`` so ``CORE.data_dir`` + points into ``/.esphome``. The storage sidecar is written + after the YAML and then back-dated, so callers can rely on + ``storage.mtime >= yaml.mtime`` matching the production happy + path (compile writes the sidecar last). + """ + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text(_SAMPLE_YAML) + CORE.config_path = yaml_path + + storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" + _write_storage(storage_path) + # Storage is younger than the YAML by virtue of being created last, + # but make the relationship explicit for tests that depend on it. + yaml_stat = yaml_path.stat() + os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) + + return yaml_path, storage_path + + +def test_lite_config_happy_path(lite_config_files: tuple[Path, Path]) -> None: + """Storage + YAML in sync: lite config returns a populated dict and CORE.""" + yaml_path, _ = lite_config_files + + config = load_lite_config_from_storage(yaml_path, {}) + + assert config is not None + assert config[CONF_ESPHOME][CONF_NAME] == "lite_test" + assert CONF_LOGGER in config + assert CONF_API in config + assert config[CONF_API]["encryption"]["key"] == "6dGhpcyBpcyBhIHRlc3Q=" + assert CONF_OTA in config + assert config[CONF_OTA][0]["platform"] == "esphome" + # Lite config should NOT contain bulky non-essential keys like esp32. + assert "esp32" not in config + + # CORE populated from storage. + assert CORE.name == "lite_test" + assert CORE.friendly_name == "Lite Test Device" + assert CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] == "esp32" + assert CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] == "arduino" + assert CORE.build_path == Path("/build/lite_test") + assert "api" in CORE.loaded_integrations + + +def test_lite_config_missing_storage_returns_none(tmp_path: Path) -> None: + """No StorageJSON sidecar → return None so caller falls back.""" + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text(_SAMPLE_YAML) + CORE.config_path = yaml_path + + assert load_lite_config_from_storage(yaml_path, {}) is None + + +def test_lite_config_stale_storage_returns_none(tmp_path: Path) -> None: + """StorageJSON older than YAML → return None. + + The YAML may have grown a new ``api:`` key since the binary was + built; falling back to ``read_config()`` catches that. + """ + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text(_SAMPLE_YAML) + CORE.config_path = yaml_path + + storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" + _write_storage(storage_path) + + yaml_stat = yaml_path.stat() + # Storage is one minute older than the YAML. + os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime - 60)) + + assert load_lite_config_from_storage(yaml_path, {}) is None + + +def test_lite_config_address_backfilled_from_storage(tmp_path: Path) -> None: + """No wifi/ethernet block in YAML → address comes from storage. + + Some users gate their network blocks behind substitutions that + don't fire in the lite parse. The loader should fall back to the + sidecar's ``address`` so ``CORE.address`` still resolves. + """ + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text( + "esphome:\n name: lite_test\nesp32:\n board: nodemcu-32s\n" + "logger:\napi:\nota:\n - platform: esphome\n" + ) + CORE.config_path = yaml_path + + storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" + _write_storage(storage_path, address="10.0.0.5") + + yaml_stat = yaml_path.stat() + os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) + + config = load_lite_config_from_storage(yaml_path, {}) + assert config is not None + assert config[CONF_WIFI][CONF_USE_ADDRESS] == "10.0.0.5" + + +def test_lite_config_corrupt_storage_returns_none(tmp_path: Path) -> None: + """A malformed sidecar (StorageJSON.load returns None) → fall back.""" + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text(_SAMPLE_YAML) + CORE.config_path = yaml_path + + storage_path = tmp_path / ".esphome" / "storage" / "lite_test.yaml.json" + storage_path.parent.mkdir(parents=True, exist_ok=True) + storage_path.write_text("not valid json{") + + yaml_stat = yaml_path.stat() + os.utime(storage_path, (yaml_stat.st_atime, yaml_stat.st_mtime + 5)) + + assert load_lite_config_from_storage(yaml_path, {}) is None + + +@pytest.mark.parametrize("command", ["upload", "logs"]) +def test_run_esphome_from_storage_json_skips_read_config( + tmp_path: Path, + command: str, + lite_config_files: tuple[Path, Path], +) -> None: + """`--from-storage-json` makes the dispatcher skip read_config().""" + yaml_path, _ = lite_config_files + + # Stub out the subcommand handler so the test doesn't try to open + # a network connection or run platform-specific upload logic. + with ( + patch("esphome.__main__.read_config") as mock_read, + patch.dict( + "esphome.__main__.POST_CONFIG_ACTIONS", + {command: lambda args, config: 0}, + ), + ): + result = run_esphome( + ["esphome", command, "--from-storage-json", str(yaml_path)] + ) + + mock_read.assert_not_called() + assert result == 0 + + +@pytest.mark.parametrize("command", ["upload", "logs"]) +def test_run_esphome_from_storage_json_falls_back_when_missing( + tmp_path: Path, command: str +) -> None: + """With no sidecar on disk, the dispatcher falls back to read_config().""" + yaml_path = tmp_path / "lite_test.yaml" + yaml_path.write_text(_SAMPLE_YAML) + + with ( + patch("esphome.__main__.read_config", return_value=None) as mock_read, + patch.dict( + "esphome.__main__.POST_CONFIG_ACTIONS", + {command: lambda args, config: 0}, + ), + ): + result = run_esphome( + ["esphome", command, "--from-storage-json", str(yaml_path)] + ) + + mock_read.assert_called_once() + # read_config returned None → dispatcher exits with 2. + assert result == 2 + + +def test_run_esphome_without_flag_still_calls_read_config( + tmp_path: Path, lite_config_files: tuple[Path, Path] +) -> None: + """Sanity: omitting the flag preserves the current behaviour.""" + yaml_path, _ = lite_config_files + + with ( + patch("esphome.__main__.read_config", return_value=None) as mock_read, + patch.dict( + "esphome.__main__.POST_CONFIG_ACTIONS", + {"upload": lambda args, config: 0}, + ), + ): + run_esphome(["esphome", "upload", str(yaml_path)]) + + mock_read.assert_called_once()