mirror of
https://github.com/esphome/esphome.git
synced 2026-09-24 13:34:07 +00:00
J. Nick Koston
a19e817d28
apply_to_core was over-populating: it restored friendly_name, loaded_integrations, and loaded_platforms even though every consumer of those three lives inside a component validator (esp32_camera, esp32, deep_sleep, zigbee, lvgl, zephyr_mcumgr), and the whole point of the fast path is to skip validation. Drop them. CORE.__init__ already leaves all three at safe defaults (None / empty set) for any incidental reader. What's left is exactly what upload/logs walk: - CORE.name (api.client.run_logs, firmware_bin path, mDNS) - CORE.build_path (firmware_bin / partition_table_bin / bootloader_bin) - CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] (module dispatch, .is_esp32 etc) - CORE.data[KEY_CORE][KEY_TARGET_FRAMEWORK] (.is_arduino, firmware_bin branch) Method body shrinks from 9 statements to 4; setdefault + two conditional inserts collapse into one dict literal; the function-local import moves to module top. Drift surface drops from 7 paired fields to 4. The wizard-only-sidecar None case is gated once at the load_compiled_config boundary so apply_to_core no longer has to defend against it.
357 lines
12 KiB
Python
357 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import binascii
|
|
from datetime import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from esphome import const
|
|
from esphome.const import (
|
|
CONF_DISABLED,
|
|
CONF_MDNS,
|
|
KEY_CORE,
|
|
KEY_TARGET_FRAMEWORK,
|
|
KEY_TARGET_PLATFORM,
|
|
)
|
|
from esphome.core import CORE
|
|
from esphome.helpers import write_file_if_changed
|
|
from esphome.types import CoreType
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
def storage_path() -> Path:
|
|
return CORE.data_dir / "storage" / f"{CORE.config_filename}.json"
|
|
|
|
|
|
def ext_storage_path(config_filename: str) -> Path:
|
|
"""Path to the per-config StorageJSON sidecar.
|
|
|
|
Used by:
|
|
- device-builder (esphome/device-builder) — locates the sidecar
|
|
to read board / framework / firmware-bin / loaded_integrations
|
|
info for the dashboard. Coordinate before changing the path
|
|
shape; device-builder reads the same file on disk.
|
|
"""
|
|
return CORE.data_dir / "storage" / f"{config_filename}.json"
|
|
|
|
|
|
def esphome_storage_path() -> Path:
|
|
return CORE.data_dir / "esphome.json"
|
|
|
|
|
|
def ignored_devices_storage_path() -> Path:
|
|
"""Path to the dashboard's ignored-devices list.
|
|
|
|
Used by:
|
|
- device-builder (esphome/device-builder) — reads the same
|
|
``ignored-devices.json`` so the new dashboard's "ignore" toggle
|
|
stays compatible with the legacy one. Don't change the file
|
|
shape without coordinating.
|
|
"""
|
|
return CORE.data_dir / "ignored-devices.json"
|
|
|
|
|
|
def trash_storage_path() -> Path:
|
|
return CORE.relative_config_path("trash")
|
|
|
|
|
|
def archive_storage_path() -> Path:
|
|
return CORE.relative_config_path("archive")
|
|
|
|
|
|
def _to_path_if_not_none(value: str | None) -> Path | None:
|
|
"""Convert a string to Path if it's not None."""
|
|
return Path(value) if value is not None else None
|
|
|
|
|
|
class StorageJSON:
|
|
"""Persisted device metadata sidecar.
|
|
|
|
Used by:
|
|
- esphome.dashboard (legacy dashboard)
|
|
- device-builder (esphome/device-builder) — reads/writes the same
|
|
JSON file as the legacy dashboard so a single config_dir can be
|
|
shared between the two during the transition. The schema
|
|
(``storage_version``, field names, types) must stay backwards
|
|
compatible — coordinate with the device-builder team before
|
|
adding required fields or changing semantics of existing ones.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
storage_version: int,
|
|
name: str,
|
|
friendly_name: str,
|
|
comment: str | None,
|
|
esphome_version: str | None,
|
|
src_version: int | None,
|
|
address: str,
|
|
web_port: int | None,
|
|
target_platform: str,
|
|
build_path: Path | None,
|
|
firmware_bin_path: Path | None,
|
|
loaded_integrations: set[str],
|
|
loaded_platforms: set[str],
|
|
no_mdns: bool,
|
|
framework: str | None = None,
|
|
core_platform: str | None = None,
|
|
) -> None:
|
|
# Version of the storage JSON schema
|
|
assert storage_version is None or isinstance(storage_version, int)
|
|
self.storage_version = storage_version
|
|
# The name of the node
|
|
self.name = name
|
|
# The friendly name of the node
|
|
self.friendly_name = friendly_name
|
|
# The comment of the node
|
|
self.comment = comment
|
|
# The esphome version this was compiled with
|
|
self.esphome_version = esphome_version
|
|
# The version of the file in src/main.cpp - Used to migrate the file
|
|
assert src_version is None or isinstance(src_version, int)
|
|
self.src_version = src_version
|
|
# Address of the ESP, for example livingroom.local or a static IP
|
|
self.address = address
|
|
# Web server port of the ESP, for example 80
|
|
assert web_port is None or isinstance(web_port, int)
|
|
self.web_port = web_port
|
|
# The type of hardware in use, like "ESP32", "ESP32C3", "ESP8266", etc.
|
|
self.target_platform = target_platform
|
|
# The absolute path to the platformio project
|
|
self.build_path = build_path
|
|
# The absolute path to the firmware binary
|
|
self.firmware_bin_path = firmware_bin_path
|
|
# A set of strings of names of loaded integrations
|
|
self.loaded_integrations = loaded_integrations
|
|
# A set of strings for platform/integration combos
|
|
self.loaded_platforms = loaded_platforms
|
|
# Is mDNS disabled
|
|
self.no_mdns = no_mdns
|
|
# The framework used to compile the firmware
|
|
self.framework = framework
|
|
# The core platform of this firmware. Like "esp32", "rp2040", "host" etc.
|
|
self.core_platform = core_platform
|
|
|
|
def as_dict(self):
|
|
return {
|
|
"storage_version": self.storage_version,
|
|
"name": self.name,
|
|
"friendly_name": self.friendly_name,
|
|
"comment": self.comment,
|
|
"esphome_version": self.esphome_version,
|
|
"src_version": self.src_version,
|
|
"address": self.address,
|
|
"web_port": self.web_port,
|
|
"esp_platform": self.target_platform,
|
|
"build_path": str(self.build_path),
|
|
"firmware_bin_path": str(self.firmware_bin_path),
|
|
"loaded_integrations": sorted(self.loaded_integrations),
|
|
"loaded_platforms": sorted(self.loaded_platforms),
|
|
"no_mdns": self.no_mdns,
|
|
"framework": self.framework,
|
|
"core_platform": self.core_platform,
|
|
}
|
|
|
|
def to_json(self):
|
|
return f"{json.dumps(self.as_dict(), indent=2)}\n"
|
|
|
|
def save(self, path):
|
|
write_file_if_changed(path, self.to_json())
|
|
|
|
@staticmethod
|
|
def from_esphome_core(esph: CoreType, old: StorageJSON | None) -> StorageJSON:
|
|
hardware = esph.target_platform.upper()
|
|
if esph.is_esp32:
|
|
from esphome.components import esp32
|
|
|
|
hardware = esp32.get_esp32_variant(esph)
|
|
return StorageJSON(
|
|
storage_version=1,
|
|
name=esph.name,
|
|
friendly_name=esph.friendly_name,
|
|
comment=esph.comment,
|
|
esphome_version=const.__version__,
|
|
src_version=1,
|
|
address=esph.address,
|
|
web_port=esph.web_port,
|
|
target_platform=hardware,
|
|
build_path=esph.build_path,
|
|
firmware_bin_path=esph.firmware_bin,
|
|
loaded_integrations=esph.loaded_integrations,
|
|
loaded_platforms=esph.loaded_platforms,
|
|
no_mdns=(
|
|
CONF_MDNS in esph.config
|
|
and CONF_DISABLED in esph.config[CONF_MDNS]
|
|
and esph.config[CONF_MDNS][CONF_DISABLED] is True
|
|
),
|
|
framework=esph.target_framework,
|
|
core_platform=esph.target_platform,
|
|
)
|
|
|
|
@staticmethod
|
|
def from_wizard(
|
|
name: str, friendly_name: str, address: str, platform: str
|
|
) -> StorageJSON:
|
|
return StorageJSON(
|
|
storage_version=1,
|
|
name=name,
|
|
friendly_name=friendly_name,
|
|
comment=None,
|
|
esphome_version=None,
|
|
src_version=1,
|
|
address=address,
|
|
web_port=None,
|
|
target_platform=platform,
|
|
build_path=None,
|
|
firmware_bin_path=None,
|
|
loaded_integrations=set(),
|
|
loaded_platforms=set(),
|
|
no_mdns=False,
|
|
framework=None,
|
|
core_platform=platform.lower(),
|
|
)
|
|
|
|
@staticmethod
|
|
def _load_impl(path: Path) -> StorageJSON | None:
|
|
with path.open("r", encoding="utf-8") as f_handle:
|
|
storage = json.load(f_handle)
|
|
storage_version = storage["storage_version"]
|
|
name = storage.get("name")
|
|
friendly_name = storage.get("friendly_name")
|
|
comment = storage.get("comment")
|
|
esphome_version = storage.get(
|
|
"esphome_version", storage.get("esphomeyaml_version")
|
|
)
|
|
src_version = storage.get("src_version")
|
|
address = storage.get("address")
|
|
web_port = storage.get("web_port")
|
|
esp_platform = storage.get("esp_platform")
|
|
build_path = _to_path_if_not_none(storage.get("build_path"))
|
|
firmware_bin_path = _to_path_if_not_none(storage.get("firmware_bin_path"))
|
|
loaded_integrations = set(storage.get("loaded_integrations", []))
|
|
loaded_platforms = set(storage.get("loaded_platforms", []))
|
|
no_mdns = storage.get("no_mdns", False)
|
|
framework = storage.get("framework")
|
|
core_platform = storage.get("core_platform")
|
|
return StorageJSON(
|
|
storage_version,
|
|
name,
|
|
friendly_name,
|
|
comment,
|
|
esphome_version,
|
|
src_version,
|
|
address,
|
|
web_port,
|
|
esp_platform,
|
|
build_path,
|
|
firmware_bin_path,
|
|
loaded_integrations,
|
|
loaded_platforms,
|
|
no_mdns,
|
|
framework,
|
|
core_platform,
|
|
)
|
|
|
|
@staticmethod
|
|
def load(path: Path) -> StorageJSON | None:
|
|
try:
|
|
return StorageJSON._load_impl(path)
|
|
except Exception: # pylint: disable=broad-except
|
|
return None
|
|
|
|
def apply_to_core(self) -> None:
|
|
"""Populate CORE with the metadata upload/logs read.
|
|
|
|
Inverse of :meth:`from_esphome_core`'s CORE→StorageJSON
|
|
projection. Keep paired -- a new CORE attribute the
|
|
``upload`` / ``logs`` fast path needs has to be captured by
|
|
``from_esphome_core`` too. Validators (``loaded_integrations``,
|
|
``loaded_platforms``, ``friendly_name``) are deliberately not
|
|
restored: they're consumed by component validation, which the
|
|
fast path skips, and ``CORE.__init__`` already leaves them at
|
|
safe defaults.
|
|
"""
|
|
CORE.name = self.name
|
|
CORE.build_path = self.build_path
|
|
CORE.data[KEY_CORE] = {
|
|
KEY_TARGET_PLATFORM: self.core_platform or self.target_platform.lower(),
|
|
KEY_TARGET_FRAMEWORK: self.framework,
|
|
}
|
|
|
|
def __eq__(self, o) -> bool:
|
|
return isinstance(o, StorageJSON) and self.as_dict() == o.as_dict()
|
|
|
|
|
|
class EsphomeStorageJSON:
|
|
def __init__(
|
|
self, storage_version, cookie_secret, last_update_check, remote_version
|
|
):
|
|
# Version of the storage JSON schema
|
|
assert storage_version is None or isinstance(storage_version, int)
|
|
self.storage_version: int = storage_version
|
|
# The cookie secret for the dashboard
|
|
self.cookie_secret: str = cookie_secret
|
|
# The last time ESPHome checked for an update as an isoformat encoded str
|
|
self.last_update_check_str: str = last_update_check
|
|
# Cache of the version gotten in the last version check
|
|
self.remote_version: str | None = remote_version
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"storage_version": self.storage_version,
|
|
"cookie_secret": self.cookie_secret,
|
|
"last_update_check": self.last_update_check_str,
|
|
"remote_version": self.remote_version,
|
|
}
|
|
|
|
@property
|
|
def last_update_check(self) -> datetime | None:
|
|
try:
|
|
return datetime.strptime(self.last_update_check_str, "%Y-%m-%dT%H:%M:%S")
|
|
except Exception: # pylint: disable=broad-except
|
|
return None
|
|
|
|
@last_update_check.setter
|
|
def last_update_check(self, new: datetime) -> None:
|
|
self.last_update_check_str = new.strftime("%Y-%m-%dT%H:%M:%S")
|
|
|
|
def to_json(self) -> str:
|
|
return f"{json.dumps(self.as_dict(), indent=2)}\n"
|
|
|
|
def save(self, path: str) -> None:
|
|
write_file_if_changed(path, self.to_json())
|
|
|
|
@staticmethod
|
|
def _load_impl(path: str) -> EsphomeStorageJSON | None:
|
|
with Path(path).open("r", encoding="utf-8") as f_handle:
|
|
storage = json.load(f_handle)
|
|
storage_version = storage["storage_version"]
|
|
cookie_secret = storage.get("cookie_secret")
|
|
last_update_check = storage.get("last_update_check")
|
|
remote_version = storage.get("remote_version")
|
|
return EsphomeStorageJSON(
|
|
storage_version, cookie_secret, last_update_check, remote_version
|
|
)
|
|
|
|
@staticmethod
|
|
def load(path: str) -> EsphomeStorageJSON | None:
|
|
try:
|
|
return EsphomeStorageJSON._load_impl(path)
|
|
except Exception: # pylint: disable=broad-except
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_default() -> EsphomeStorageJSON:
|
|
return EsphomeStorageJSON(
|
|
storage_version=1,
|
|
cookie_secret=binascii.hexlify(os.urandom(64)).decode(),
|
|
last_update_check=None,
|
|
remote_version=None,
|
|
)
|
|
|
|
def __eq__(self, o) -> bool:
|
|
return isinstance(o, EsphomeStorageJSON) and self.as_dict() == o.as_dict()
|