Max Gramser
125346d94e
From the end-to-end test on the bench boards and an independent review: - Units spaced as Home Assistant does (54% / 54 % in German and French, 18°, 21.5 °C), in the firmware and the app. - Grouping from five digits where the language does (1234 but 12.345 in Spanish, Polish, Portuguese): the layout message carries group_min and percent_space; an unknown number style no longer refuses a layout. - Automatic follows Home Assistant's own language for the clock and numbers, also without a translation (Swedish: 24 hours, 1 234,5; pt-BR: 1.234,5): regions.json from the CLDR for all 63 languages Home Assistant offers. - 12-hour clock: the top bar says 3:48 PM, the clock card 3:48 (not 03:48), Last runs follow the clock. - The climate setpoint font has the decimal comma; the value fonts have ², ³ and µ (m³ was a box). - Weather days: at most three letters without the dot (mié, sáb, czw); the screen took three bytes. - Texts fall back per key through the base language (pt-BR -> pt -> en) in the firmware and the app. - Each screen gets the app's words in the language its firmware speaks; firmware from before the languages gets English and its own letters, not boxes. - set_language re-reads what it writes (indentation, comments, quoted keys, CRLF, !include) and every build writes the language first; the 24-hour pin no longer hits a fresh install and waits for the old clock switches. - Tests: region and notation, plural parity with the C++ rules, set_language edge cases, the pin, PUT /api/language. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
296 lines
14 KiB
Python
296 lines
14 KiB
Python
"""Firmware updates for paired screens: on demand, all at once, or nightly. One screen at a time."""
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
import changelog
|
|
from core import FIRMWARE_VERSION, parse_firmware
|
|
from i18n import Text, english, screen_t, shown, t
|
|
|
|
LOG = logging.getLogger('screen_manager')
|
|
NIGHT_HOURS = range(3, 6)
|
|
|
|
# Strict X.Y.Z, the one rule the feature gates follow too: an update goes by what the screen's sensor reports now.
|
|
parse_version = parse_firmware
|
|
|
|
TARGET = parse_version(FIRMWARE_VERSION)
|
|
|
|
def result_text(result):
|
|
"""A kept result's message: a Text from its key where it has one (app 0.2.90+), else the English it was kept in."""
|
|
key, params = result.get('key'), result.get('params')
|
|
if isinstance(key, str) and key:
|
|
return Text(str(result.get('message', '')), key, params if isinstance(params, dict) else {})
|
|
return result.get('message', '')
|
|
|
|
def shown_result(result):
|
|
"""A kept result as the editor shows it, its message in the editor's language."""
|
|
if not isinstance(result, dict) or not isinstance(result.get('key'), str):
|
|
return result
|
|
return {**result, 'message': shown(result_text(result))}
|
|
|
|
class Updater:
|
|
verify_timeout = 240
|
|
settle_seconds = 60
|
|
pause_seconds = 120
|
|
poll_seconds = 5
|
|
|
|
def __init__(self, manager, path):
|
|
self.manager, self.path = manager, Path(path)
|
|
self.auto, self.hosts, self.results, self.last_round = False, {}, {}, None
|
|
self.task, self.current, self.queue, self.phase = None, None, [], None
|
|
# What's new since a screen's firmware, for the Update badge (app 0.2.73).
|
|
self.changelog = changelog.load()
|
|
if self.path.exists():
|
|
raw = json.loads(self.path.read_text())
|
|
if raw.get('version') != 1:
|
|
raise ValueError('Unknown storage version for updates; data stays unchanged.')
|
|
self.auto = raw.get('auto') is True
|
|
self.hosts = {k: v for k, v in raw.get('hosts', {}).items() if isinstance(v, str)}
|
|
self.results = {k: v for k, v in raw.get('results', {}).items() if isinstance(v, dict)}
|
|
self.last_round = raw.get('last_round') if isinstance(raw.get('last_round'), str) else None
|
|
|
|
def save(self):
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temp = self.path.with_suffix('.tmp')
|
|
with open(temp, 'w', encoding='utf8') as handle:
|
|
os.chmod(temp, 0o600)
|
|
json.dump({'version': 1, 'auto': self.auto, 'hosts': self.hosts, 'results': self.results,
|
|
'last_round': self.last_round}, handle, ensure_ascii=False)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
temp.replace(self.path)
|
|
|
|
def busy(self):
|
|
return bool(self.task and not self.task.done())
|
|
|
|
def screen(self, inbox):
|
|
return self.manager.screen(inbox)
|
|
|
|
def current_id(self, inbox):
|
|
"""The id a screen reports under now; firmware 0.2.34 gave the inbox entity a new id."""
|
|
return getattr(self.manager, 'aliases', {}).get(inbox, inbox)
|
|
|
|
def renamed(self, old, new):
|
|
"""A screen's inbox entity got a new id: its address, last result and place in a running round follow."""
|
|
changed = False
|
|
for store in (self.hosts, self.results):
|
|
if old in store:
|
|
if new not in store:
|
|
store[new] = store[old]
|
|
del store[old]
|
|
changed = True
|
|
if changed:
|
|
self.save()
|
|
if self.current == old:
|
|
self.current = new
|
|
self.queue = [new if inbox == old else inbox for inbox in self.queue]
|
|
return changed
|
|
|
|
def resolve(self, screen, profiles=None):
|
|
"""Profile file and OTA address for a screen; None when the add-on cannot tell."""
|
|
if profiles is None:
|
|
profiles = self.manager.firmware.profile_names()
|
|
profile = next((f for f, p in profiles.items() if screen.get('node') and p['node'] == screen['node']), None)
|
|
if profile is None and screen.get('device'):
|
|
# Screens on firmware before 0.2.17 do not report their node name yet.
|
|
matches = [f for f, p in profiles.items() if p['friendly'] == screen['device']]
|
|
profile = matches[0] if len(matches) == 1 else None
|
|
return profile, screen.get('ip') or self.hosts.get(screen['id'])
|
|
|
|
def language_due(self, screen):
|
|
"""True when a screen speaks another language than Settings -> Language & region says (app 0.2.90): its
|
|
"Screen language" sensor, or English for firmware from before that sensor. A sensor without a state yet (the
|
|
screen restarts) says nothing."""
|
|
region = getattr(self.manager, 'region', None)
|
|
if not region or (screen.get('language_sensor') and not screen.get('language')):
|
|
return False
|
|
return (screen.get('language') or 'en') != region.language()
|
|
|
|
def speaks_region(self, screen):
|
|
"""Whether the screen says it speaks the language of Settings -> Language & region: what an update waits for."""
|
|
region = getattr(self.manager, 'region', None)
|
|
if not region:
|
|
return True
|
|
return (screen.get('language') if screen.get('language_sensor') else 'en') == region.language()
|
|
|
|
def state_for(self, screen, profiles=None):
|
|
version = parse_version(screen.get('firmware'))
|
|
language = self.language_due(screen)
|
|
profile, host = self.resolve(screen, profiles)
|
|
if screen['id'] == self.current:
|
|
state = 'running'
|
|
elif screen['id'] in self.queue:
|
|
state = 'queued'
|
|
else:
|
|
state = 'idle'
|
|
return {'available': bool(version and TARGET and version < TARGET) or (bool(version) and language),
|
|
'language': bool(version) and language, 'target': FIRMWARE_VERSION,
|
|
'profile': profile, 'host': host, 'state': state, 'phase': self.phase if state == 'running' else None,
|
|
'result': shown_result(self.results.get(screen['id']))}
|
|
|
|
def summary(self, screens=None, profiles=None):
|
|
# The changelog goes with the full inventory only (app 0.2.78): this summary is in every live update of the page.
|
|
return {'auto': self.auto, 'target': FIRMWARE_VERSION, 'busy': self.current,
|
|
'pending': len(self.pending(screens, profiles)), 'last_round': self.last_round}
|
|
|
|
def pending(self, screens=None, profiles=None):
|
|
# Callers that already hold the inventory and profile list pass them in; the inventory
|
|
# handler otherwise re-reads every ESPHome profile several times per request.
|
|
if screens is None:
|
|
screens = self.manager.screens()
|
|
if profiles is None:
|
|
profiles = self.manager.firmware.profile_names()
|
|
return [s['id'] for s in screens
|
|
if s['online'] and self.state_for(s, profiles)['available'] and all(self.resolve(s, profiles))]
|
|
|
|
def set_auto(self, enabled):
|
|
if not isinstance(enabled, bool):
|
|
raise ValueError(t('addon.errors.updates.automatic'))
|
|
self.auto = enabled
|
|
self.save()
|
|
|
|
def start(self, inbox, host=None):
|
|
if self.busy():
|
|
raise ValueError(t('addon.errors.updates.busy'))
|
|
screen = self.screen(inbox)
|
|
inbox = self.current_id(inbox)
|
|
if not screen:
|
|
raise ValueError(t('addon.errors.updates.unknown_screen'))
|
|
if not screen['online']:
|
|
raise ValueError(t('addon.errors.updates.offline'))
|
|
if not self.state_for(screen)['available']:
|
|
raise ValueError(t('addon.errors.updates.latest'))
|
|
if host is not None:
|
|
if not isinstance(host, str) or not re.fullmatch(r'[a-zA-Z0-9][a-zA-Z0-9.-]{0,252}', host.strip()):
|
|
raise ValueError(t('addon.errors.updates.enter_ip'))
|
|
self.hosts[inbox] = host.strip()
|
|
self.save()
|
|
profile, host = self.resolve(screen)
|
|
if not profile:
|
|
raise ValueError(t('addon.errors.updates.no_profile'))
|
|
if not host:
|
|
raise ValueError(t('addon.errors.updates.ip_unknown'))
|
|
self.launch([inbox])
|
|
return self.state_for(screen)
|
|
|
|
def start_all(self):
|
|
if self.busy():
|
|
raise ValueError(t('addon.errors.updates.busy'))
|
|
pending = self.pending()
|
|
if not pending:
|
|
raise ValueError(t('addon.errors.updates.all_updated'))
|
|
self.launch(pending)
|
|
return pending
|
|
|
|
def launch(self, inboxes, automatic=False):
|
|
# Mark the first screen busy right away so the page shows progress before the task runs.
|
|
self.current, self.queue, self.phase = inboxes[0], inboxes[1:], 'install'
|
|
self.task = asyncio.create_task(self.run_round(inboxes, automatic))
|
|
|
|
def record(self, inbox, state, message):
|
|
"""A screen's last result. `message` keeps its key and params (a Text) next to the English, so the editor shows it
|
|
in its own language and an app from before 0.2.90 still reads the English."""
|
|
result = {'time': time.time(), 'state': state, 'message': str(message), 'version': FIRMWARE_VERSION}
|
|
if isinstance(message, Text) and message.key:
|
|
result.update(message=message.into('en'), key=message.key, params=message.params)
|
|
self.results[self.current_id(inbox)] = result
|
|
self.save()
|
|
|
|
async def run_round(self, inboxes, automatic=False):
|
|
"""Update screens one by one; a failure ends the round so a bad build never reaches the next screen."""
|
|
try:
|
|
for index, inbox in enumerate(inboxes):
|
|
inbox = self.current_id(inbox)
|
|
self.current, self.queue = inbox, [self.current_id(i) for i in inboxes[index+1:]]
|
|
outcome = await self.update_one(inbox)
|
|
if outcome == 'failed':
|
|
if automatic:
|
|
# For whoever reads Home Assistant: in the screens' language, Home Assistant's own unless another was
|
|
# chosen (app 0.2.90).
|
|
await self.notify(screen_t('addon.notify.update_stopped', name=self.name(inbox),
|
|
reason=result_text(self.results[self.current_id(inbox)])))
|
|
break
|
|
if self.queue and outcome == 'success':
|
|
await asyncio.sleep(self.pause_seconds)
|
|
finally:
|
|
self.current, self.queue, self.phase = None, [], None
|
|
|
|
def name(self, inbox):
|
|
screen = self.screen(inbox)
|
|
return screen['name'] if screen else inbox
|
|
|
|
async def update_one(self, inbox):
|
|
screen = self.screen(inbox)
|
|
if not screen or not screen['online']:
|
|
self.record(inbox, 'skipped', english('addon.updates.offline'))
|
|
return 'skipped'
|
|
profile, host = self.resolve(screen)
|
|
if not profile or not host:
|
|
self.record(inbox, 'skipped', english('addon.updates.unknown_target'))
|
|
return 'skipped'
|
|
self.phase = 'install'
|
|
try:
|
|
self.manager.firmware.start({'file': profile, 'action': 'install', 'target': host})
|
|
await self.manager.firmware.task
|
|
except ValueError as error:
|
|
# The sentence firmware.start refused with, kept with its key.
|
|
self.record(inbox, 'failed', error.args[0] if len(error.args) == 1 else str(error))
|
|
return 'failed'
|
|
if self.manager.firmware.job.get('state') != 'success':
|
|
self.record(inbox, 'failed', english('addon.updates.build_failed'))
|
|
return 'failed'
|
|
self.phase = 'verify'
|
|
if not await self.wait_for_target(inbox):
|
|
self.record(inbox, 'failed', english('addon.updates.no_report', version=FIRMWARE_VERSION))
|
|
return 'failed'
|
|
self.phase = 'settle'
|
|
await asyncio.sleep(self.settle_seconds)
|
|
current = self.screen(inbox)
|
|
if not current or not current['online']:
|
|
self.record(inbox, 'failed', english('addon.updates.dropped_off'))
|
|
return 'failed'
|
|
self.record(inbox, 'success', english('addon.updates.updated', version=FIRMWARE_VERSION))
|
|
return 'success'
|
|
|
|
async def wait_for_target(self, inbox):
|
|
deadline = time.monotonic() + self.verify_timeout
|
|
while time.monotonic() < deadline:
|
|
screen = self.screen(inbox)
|
|
version = parse_version(screen.get('firmware')) if screen else None
|
|
if screen and screen['online'] and version and version >= TARGET and self.speaks_region(screen):
|
|
return True
|
|
await asyncio.sleep(self.poll_seconds)
|
|
return False
|
|
|
|
async def notify(self, message):
|
|
try:
|
|
await self.manager.ha.request('call_service', domain='persistent_notification', service='create',
|
|
service_data={'notification_id': 'esp_screens_update', 'title': 'ESP Screens', 'message': message})
|
|
except Exception as error:
|
|
LOG.warning('Notifying Home Assistant failed (%s)', type(error).__name__)
|
|
|
|
def due(self, now):
|
|
return self.auto and now.hour in NIGHT_HOURS and self.last_round != now.date().isoformat()
|
|
|
|
async def run(self):
|
|
while True:
|
|
await asyncio.sleep(60)
|
|
try:
|
|
if self.busy() or not self.manager.ha.online:
|
|
continue
|
|
now = datetime.now(getattr(self.manager.ha, 'time_zone', None) or timezone.utc)
|
|
if not self.due(now):
|
|
continue
|
|
self.last_round = now.date().isoformat()
|
|
self.save()
|
|
pending = self.pending()
|
|
if pending:
|
|
LOG.info('Nightly firmware round: %d screen(s)', len(pending))
|
|
self.launch(pending, automatic=True)
|
|
except Exception as error:
|
|
LOG.warning('Nightly update check skipped (%s)', type(error).__name__)
|