esphome/esphome/storage_json.py
2023-01-17 10:28:09 +13:00

256 lines
8.7 KiB
Python

import binascii
import codecs
from datetime import datetime
import json
import logging
import os
from typing import Optional
from esphome import const
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() -> str:
return CORE.relative_internal_path(f"{CORE.config_filename}.json")
def ext_storage_path(base_path: str, config_filename: str) -> str:
return os.path.join(base_path, ".esphome", f"{config_filename}.json")
def esphome_storage_path(base_path: str) -> str:
return os.path.join(base_path, ".esphome", "esphome.json")
def trash_storage_path(base_path: str) -> str:
return os.path.join(base_path, ".esphome", "trash")
class StorageJSON:
def __init__(
self,
storage_version,
name,
friendly_name,
comment,
esphome_version,
src_version,
address,
web_port,
target_platform,
build_path,
firmware_bin_path,
loaded_integrations,
):
# Version of the storage JSON schema
assert storage_version is None or isinstance(storage_version, int)
self.storage_version: int = storage_version
# The name of the node
self.name: str = name
# The friendly name of the node
self.friendly_name: str = friendly_name
# The comment of the node
self.comment: str = comment
# The esphome version this was compiled with
self.esphome_version: str = 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: int = src_version
# Address of the ESP, for example livingroom.local or a static IP
self.address: str = address
# Web server port of the ESP, for example 80
assert web_port is None or isinstance(web_port, int)
self.web_port: int = web_port
# The type of hardware in use, like "ESP32", "ESP32C3", "ESP8266", etc.
self.target_platform: str = target_platform
# The absolute path to the platformio project
self.build_path: str = build_path
# The absolute path to the firmware binary
self.firmware_bin_path: str = firmware_bin_path
# A list of strings of names of loaded integrations
self.loaded_integrations: list[str] = loaded_integrations
self.loaded_integrations.sort()
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": self.build_path,
"firmware_bin_path": self.firmware_bin_path,
"loaded_integrations": self.loaded_integrations,
}
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: Optional["StorageJSON"]
) -> "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=list(esph.loaded_integrations),
)
@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=[],
)
@staticmethod
def _load_impl(path: str) -> Optional["StorageJSON"]:
with codecs.open(path, "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 = storage.get("build_path")
firmware_bin_path = storage.get("firmware_bin_path")
loaded_integrations = storage.get("loaded_integrations", [])
return StorageJSON(
storage_version,
name,
friendly_name,
comment,
esphome_version,
src_version,
address,
web_port,
esp_platform,
build_path,
firmware_bin_path,
loaded_integrations,
)
@staticmethod
def load(path: str) -> Optional["StorageJSON"]:
try:
return StorageJSON._load_impl(path)
except Exception: # pylint: disable=broad-except
return None
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: Optional[str] = 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) -> Optional[datetime]:
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) -> dict:
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) -> Optional["EsphomeStorageJSON"]:
with codecs.open(path, "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) -> Optional["EsphomeStorageJSON"]:
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()