esphome/esphomeyaml/storage_json.py
Otto Winter 7c7032c59e
[Huge] Util Refactor, Dashboard Improvements, Hass.io Auth API, Better Validation Errors, Conditions, Custom Platforms, Substitutions (#234)
* Implement custom sensor platform

* Update

* Ethernet

* Lint

* Fix

* Login page

* Rename cookie secret

* Update manifest

* Update cookie check logic

* Favicon

* Fix

* Favicon manifest

* Fix

* Fix

* Fix

* Use hostname

* Message

* Temporary commit for screenshot

* Automatic board selection

* Undo temporary commit

* Update esphomeyaml-edge

* In-dashboard editing and hosting files locally

* Update esphomeyaml-edge

* Better ANSI color escaping

* Message

* Lint

* Download Efficiency

* Fix gitlab

* Fix

* Rename extra_libraries to libraries

* Add example

* Update README.md

* Update README.md

* Update README.md

* HassIO -> Hass.io

* Updates

* Add update available notice

* Update

* Fix substitutions

* Better error message

* Re-do dashboard ANSI colors

* Only include FastLED if user says so

* Autoscroll logs

* Remove old checks

* Use safer RedirectText

* Improvements

* Fix

* Use enviornment variable

* Use http://hassio/host/info

* Fix conditions

* Update platformio versions

* Revert "Use enviornment variable"

This reverts commit 7f038eb5d2.

* Fix

* README update

* Temp

* Better invalid config messages

* Platformio debug

* Improve error messages

* Debug

* Remove debug

* Multi Conf

* Update

* Better paths

* Remove unused

* Fixes

* Lint

* lib_ignore

* Try fix platformio colors

* Fix dashboard scrolling

* Revert

* Lint

* Revert
2018-12-05 21:22:06 +01:00

296 lines
12 KiB
Python

import binascii
import codecs
from datetime import datetime, timedelta
import json
import logging
import os
import threading
from esphomeyaml import const
from esphomeyaml.core import CORE
from esphomeyaml.helpers import mkdir_p
# pylint: disable=unused-import, wrong-import-order
from esphomeyaml.core import CoreType # noqa
from typing import Any, Dict, Optional # noqa
_LOGGER = logging.getLogger(__name__)
def storage_path(): # type: () -> str
return CORE.relative_path('.esphomeyaml', '{}.json'.format(CORE.config_filename))
def ext_storage_path(base_path, config_filename): # type: (str, str) -> str
return os.path.join(base_path, '.esphomeyaml', '{}.json'.format(config_filename))
def esphomeyaml_storage_path(base_path): # type: (str) -> str
return os.path.join(base_path, '.esphomeyaml', 'esphomeyaml.json')
# pylint: disable=too-many-instance-attributes
class StorageJSON(object):
def __init__(self, storage_version, name, esphomelib_version, esphomeyaml_version,
src_version, arduino_version, address, esp_platform, board, build_path,
firmware_bin_path, use_legacy_ota):
# Version of the storage JSON schema
assert storage_version is None or isinstance(storage_version, int)
self.storage_version = storage_version # type: int
# The name of the node
self.name = name # type: str
# The esphomelib version in use
assert esphomelib_version is None or isinstance(esphomelib_version, dict)
self.esphomelib_version = esphomelib_version # type: Dict[str, str]
# The esphomeyaml version this was compiled with
self.esphomeyaml_version = esphomeyaml_version # type: str
# 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 # type: int
# The version of the Arduino framework, the build files need to be cleared each time
# this changes
self.arduino_version = arduino_version # type: str
# Address of the ESP, for example livingroom.local or a static IP
self.address = address # type: str
# The type of ESP in use, either ESP32 or ESP8266
self.esp_platform = esp_platform # type: str
# The ESP board used, for example nodemcuv2
self.board = board # type: str
# The absolute path to the platformio project
self.build_path = build_path # type: str
# The absolute path to the firmware binary
self.firmware_bin_path = firmware_bin_path # type: str
# Whether to use legacy OTA, will be off after the first successful flash
self.use_legacy_ota = use_legacy_ota
def as_dict(self):
return {
'storage_version': self.storage_version,
'name': self.name,
'esphomelib_version': self.esphomelib_version,
'esphomeyaml_version': self.esphomeyaml_version,
'src_version': self.src_version,
'arduino_version': self.arduino_version,
'address': self.address,
'esp_platform': self.esp_platform,
'board': self.board,
'build_path': self.build_path,
'firmware_bin_path': self.firmware_bin_path,
'use_legacy_ota': self.use_legacy_ota,
}
def to_json(self):
return json.dumps(self.as_dict(), indent=2) + u'\n'
def save(self, path):
mkdir_p(os.path.dirname(path))
with codecs.open(path, 'w', encoding='utf-8') as f_handle:
f_handle.write(self.to_json())
@staticmethod
def from_esphomeyaml_core(esph, old): # type: (CoreType, Optional[StorageJSON]) -> StorageJSON
return StorageJSON(
storage_version=1,
name=esph.name,
esphomelib_version=esph.esphomelib_version,
esphomeyaml_version=const.__version__,
src_version=1,
arduino_version=esph.arduino_version,
address=esph.address,
esp_platform=esph.esp_platform,
board=esph.board,
build_path=esph.build_path,
firmware_bin_path=esph.firmware_bin,
use_legacy_ota=True if old is None else old.use_legacy_ota,
)
@staticmethod
def from_wizard(name, address, esp_platform, board):
# type: (str, str, str, str) -> StorageJSON
return StorageJSON(
storage_version=1,
name=name,
esphomelib_version=None,
esphomeyaml_version=const.__version__,
src_version=1,
arduino_version=None,
address=address,
esp_platform=esp_platform,
board=board,
build_path=None,
firmware_bin_path=None,
use_legacy_ota=False,
)
@staticmethod
def _load_impl(path): # type: (str) -> Optional[StorageJSON]
with codecs.open(path, 'r', encoding='utf-8') as f_handle:
text = f_handle.read()
storage = json.loads(text, encoding='utf-8')
storage_version = storage['storage_version']
name = storage.get('name')
esphomelib_version = storage.get('esphomelib_version')
esphomeyaml_version = storage.get('esphomeyaml_version')
src_version = storage.get('src_version')
arduino_version = storage.get('arduino_version')
address = storage.get('address')
esp_platform = storage.get('esp_platform')
board = storage.get('board')
build_path = storage.get('build_path')
firmware_bin_path = storage.get('firmware_bin_path')
use_legacy_ota = storage.get('use_legacy_ota')
return StorageJSON(storage_version, name, esphomelib_version, esphomeyaml_version,
src_version, arduino_version, address, esp_platform, board, build_path,
firmware_bin_path, use_legacy_ota)
@staticmethod
def load(path): # type: (str) -> Optional[StorageJSON]
try:
return StorageJSON._load_impl(path)
except Exception: # pylint: disable=broad-except
return None
def __eq__(self, o): # type: (Any) -> bool
return isinstance(o, StorageJSON) and self.as_dict() == o.as_dict()
class EsphomeyamlStorageJSON(object):
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 = storage_version # type: int
# The cookie secret for the dashboard
self.cookie_secret = cookie_secret # type: str
# The last time esphomeyaml checked for an update as an isoformat encoded str
self.last_update_check_str = last_update_check # type: str
# Cache of the version gotten in the last version check
self.remote_version = remote_version # type: Optional[str]
def as_dict(self): # type: () -> 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): # type: () -> 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): # type: (datetime) -> None
self.last_update_check_str = new.strftime("%Y-%m-%dT%H:%M:%S")
def to_json(self): # type: () -> dict
return json.dumps(self.as_dict(), indent=2) + u'\n'
def save(self, path): # type: (str) -> None
mkdir_p(os.path.dirname(path))
with codecs.open(path, 'w', encoding='utf-8') as f_handle:
f_handle.write(self.to_json())
@staticmethod
def _load_impl(path): # type: (str) -> Optional[EsphomeyamlStorageJSON]
with codecs.open(path, 'r', encoding='utf-8') as f_handle:
text = f_handle.read()
storage = json.loads(text, encoding='utf-8')
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 EsphomeyamlStorageJSON(storage_version, cookie_secret, last_update_check,
remote_version)
@staticmethod
def load(path): # type: (str) -> Optional[EsphomeyamlStorageJSON]
try:
return EsphomeyamlStorageJSON._load_impl(path)
except Exception: # pylint: disable=broad-except
return None
@staticmethod
def get_default(): # type: () -> EsphomeyamlStorageJSON
return EsphomeyamlStorageJSON(
storage_version=1,
cookie_secret=binascii.hexlify(os.urandom(64)),
last_update_check=None,
remote_version=None,
)
def __eq__(self, o): # type: (Any) -> bool
return isinstance(o, EsphomeyamlStorageJSON) and self.as_dict() == o.as_dict()
@property
def should_do_esphomeyaml_update_check(self): # type: () -> bool
if self.last_update_check is None:
return True
return self.last_update_check + timedelta(days=3) < datetime.utcnow()
class CheckForUpdateThread(threading.Thread):
def __init__(self, path):
threading.Thread.__init__(self)
self._path = path
@property
def docs_base(self):
return 'https://beta.esphomelib.com' if 'b' in const.__version__ else \
'https://esphomelib.com'
def fetch_remote_version(self):
import requests
storage = EsphomeyamlStorageJSON.load(self._path) or \
EsphomeyamlStorageJSON.get_default()
if not storage.should_do_esphomeyaml_update_check:
return storage
req = requests.get('{}/_static/version'.format(self.docs_base))
req.raise_for_status()
storage.remote_version = req.text.strip()
storage.last_update_check = datetime.utcnow()
storage.save(self._path)
return storage
@staticmethod
def format_version(ver):
vstr = '.'.join(map(str, ver.version))
if ver.prerelease:
vstr += ver.prerelease[0] + str(ver.prerelease[1])
return vstr
def cmp_versions(self, storage):
# pylint: disable=no-name-in-module, import-error
from distutils.version import StrictVersion
remote_version = StrictVersion(storage.remote_version)
self_version = StrictVersion(const.__version__)
if remote_version > self_version:
_LOGGER.warn("*" * 80)
_LOGGER.warn("A new version of esphomeyaml is available: %s (this is %s)",
self.format_version(remote_version), self.format_version(self_version))
_LOGGER.warn("Changelog: %s/esphomeyaml/changelog/index.html", self.docs_base)
_LOGGER.warn("Update Instructions: %s/esphomeyaml/guides/faq.html"
"#how-do-i-update-to-the-latest-version", self.docs_base)
_LOGGER.warn("*" * 80)
def run(self):
try:
storage = self.fetch_remote_version()
self.cmp_versions(storage)
except Exception: # pylint: disable=broad-except
pass
def start_update_check_thread(path):
# dummy call to strptime as python 2.7 has a bug with strptime when importing from threads
datetime.strptime('20180101', '%Y%m%d')
thread = CheckForUpdateThread(path)
thread.start()
return thread