esphome/esphome/storage_json.py

from __future__ import annotations

import binascii
import json
import logging
import os
from pathlib import Path
from typing import TYPE_CHECKING

from esphome import const
from esphome.const import (
    CONF_DISABLED,
    CONF_MDNS,
    KEY_CORE,
    KEY_ESP32,
    KEY_FRAMEWORK_VERSION,
    KEY_IDF_VERSION,
    KEY_TARGET_FRAMEWORK,
    KEY_TARGET_PLATFORM,
    KEY_VARIANT,
    Toolchain,
)
from esphome.core import CORE, EsphomeError, Version
from esphome.helpers import write_file_if_changed
from esphome.types import CoreType

if TYPE_CHECKING:
    from datetime import datetime

_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; None and the legacy "None" both map to None.

    Sidecars written before as_dict skipped unset paths hold str(None).
    """
    return Path(value) if value is not None and value != "None" else None


def _parse_framework_version(framework_version: str) -> Version:
    try:
        return Version.parse(framework_version)
    except ValueError as err:
        raise EsphomeError(
            f"Could not parse the framework version "
            f"{framework_version!r} from {storage_path()}. "
            f"Please clean the build files and recompile."
        ) from err


class StorageJSON:
    """Persisted device metadata sidecar.

    Used by device-builder (esphome/device-builder), which reads/writes this
    JSON file. 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,
        toolchain: str | None = None,
        area: str | None = None,
        framework_version: 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", "rp2", "host" etc.
        self.core_platform = core_platform
        # The toolchain used for the build ("platformio" / "esp-idf")
        self.toolchain = toolchain
        # The area of the node
        self.area = area
        # The framework version the build used (for esp32, the resolved ESP-IDF version)
        self.framework_version = framework_version

    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) if self.build_path else None,
            "firmware_bin_path": (
                str(self.firmware_bin_path) if self.firmware_bin_path else None
            ),
            "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,
            "toolchain": self.toolchain,
            "area": self.area,
            "framework_version": self.framework_version,
        }

    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, *, claim_build: bool = True
    ) -> StorageJSON:
        """Build a sidecar from post-validation CORE state.

        claim_build=False (the upload/logs fallback, which runs no build)
        carries the build-artifact fields (esphome_version,
        firmware_bin_path) from *old* instead of asserting this run built
        firmware. Validation-derived fields (platform, framework_version,
        toolchain, build_path) always stamp; storage_should_clean compares
        them against the next compile.
        """
        hardware = esph.target_platform.upper()
        framework_version: str | None = None
        if esph.is_esp32:
            from esphome.components import esp32

            hardware = esp32.get_esp32_variant(esph)
            framework_version = str(esp32.idf_version())
        elif esph.is_nrf52:
            framework_version = str(esph.data[KEY_CORE][KEY_FRAMEWORK_VERSION])
        return StorageJSON(
            storage_version=1,
            name=esph.name,
            friendly_name=esph.friendly_name,
            comment=esph.comment,
            esphome_version=(
                const.__version__
                if claim_build
                else (old.esphome_version if old else None)
            ),
            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
                if claim_build
                else (old.firmware_bin_path if old else None)
            ),
            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,
            toolchain=esph.toolchain.value if esph.toolchain is not None else None,
            area=esph.area,
            framework_version=framework_version,
        )

    @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")
        toolchain = storage.get("toolchain")
        area = storage.get("area")
        framework_version = storage.get("framework_version")
        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,
            toolchain,
            area,
            framework_version,
        )

    @staticmethod
    def load(path: Path) -> StorageJSON | None:
        try:
            return StorageJSON._load_impl(path)
        except Exception:  # noqa: BLE001  # pylint: disable=broad-except
            return None

    @staticmethod
    def load_strict(path: Path) -> StorageJSON | None:
        """Like load, but None only means missing; an unreadable file raises."""
        if not path.is_file():
            return None
        return StorageJSON._load_impl(path)

    def can_apply_to_core(self) -> bool:
        """True when the sidecar carries everything apply_to_core hands CORE.

        Wizard-written sidecars leave build_path unset (older wizards also
        the platform fields) and can't drive upload/logs.
        """
        return bool((self.core_platform or self.target_platform) and self.build_path)

    def apply_to_core(self) -> None:
        """Populate CORE with the metadata upload/logs read.

        Inverse of :meth:`from_esphome_core`. Keep paired -- a new
        attribute upload/logs needs has to be captured there too and
        reflected in :meth:`can_apply_to_core`.
        Validator-only fields (loaded_integrations/platforms,
        friendly_name) are skipped; the fast path doesn't run
        validation and CORE.__init__ defaults them.
        """
        CORE.name = self.name
        CORE.build_path = self.build_path
        # Restore toolchain so upload/logs picks the right firmware_bin path.
        # An unknown value (corrupt sidecar, or written by a newer ESPHome)
        # just leaves CORE.toolchain None — the fallback then picks PlatformIO.
        if self.toolchain and CORE.toolchain is None:
            try:
                CORE.toolchain = Toolchain(self.toolchain)
            except ValueError:
                _LOGGER.debug(
                    "Ignoring unknown toolchain %r from %s",
                    self.toolchain,
                    storage_path(),
                )
        target_platform = self.core_platform or self.target_platform.lower()
        CORE.data[KEY_CORE] = {
            KEY_TARGET_PLATFORM: target_platform,
            KEY_TARGET_FRAMEWORK: self.framework,
        }
        # The compile pipeline populates CORE.data[KEY_ESP32] when esp32's
        # validator runs; on the cache fast path that validator is skipped,
        # so populate the variant upload_using_esptool reads from
        # CORE.data[KEY_ESP32][KEY_VARIANT]. target_platform on disk is the
        # variant (e.g. "ESP32S3"); core_platform is the family (e.g. "esp32").
        if target_platform == const.PLATFORM_ESP32:
            esp32_data = {KEY_VARIANT: self.target_platform}
            if self.framework_version:
                esp32_data[KEY_IDF_VERSION] = _parse_framework_version(
                    self.framework_version
                )
            CORE.data[KEY_ESP32] = esp32_data
        elif target_platform == const.PLATFORM_NRF52 and self.framework_version:
            CORE.data[KEY_CORE][KEY_FRAMEWORK_VERSION] = _parse_framework_version(
                self.framework_version
            )

    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:
        # Deferred: this module is on the upload/logs fast path; only the
        # dashboard's update check touches these accessors.
        from datetime import datetime

        try:
            # Stored format is naive ISO without %z; preserved for backward compat.
            return datetime.strptime(  # noqa: DTZ007
                self.last_update_check_str, "%Y-%m-%dT%H:%M:%S"
            )
        except Exception:  # noqa: BLE001  # 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:  # noqa: BLE001  # 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()