Files
homeassistant_espscreen/screen_manager/app/core.py
T
Max Gramser 0120175dce Show a green dot for online screens in the screen list (0.2.30)
The status line under each screen name used the muted text colour for the
online dot, so online and offline screens looked alike. The dot is now its
own span and turns green when the screen is online; offline keeps the grey
open circle. Firmware 0.2.25 for both boards carries only the new version.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-14 11:50:56 +02:00

426 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Pure validation, firmware generation and bounded display protocol."""
import base64
from datetime import datetime, timedelta, timezone
import json
import math
import re
import secrets
import tile_icons
DOMAINS = frozenset('light switch input_boolean scene script climate vacuum fan cover sensor binary_sensor input_select select number input_number weather media_player button input_button sun timer person screen'.split())
# Built-in cards without a Home Assistant entity; firmware 0.2.14+ renders them.
BUILTIN = {'screen.clock': 'Klok'}
NEW_DOMAINS = frozenset('sun timer person screen'.split())
WEEKDAYS = ['ma', 'di', 'wo', 'do', 'vr', 'za', 'zo']
REPO = 'https://github.com/MaxGramser/homeassistant_espscreen'
REFS = {'cyd': 'main', 'guition': 'main'}
# Firmware shipped with this app release; screens below it get an update offer.
FIRMWARE_VERSION = '0.2.25'
ATTRS = frozenset('brightness percentage current_position current_temperature temperature current_humidity min_temp max_temp target_temp_step supported_color_modes hvac_modes hvac_action hs_color color_temp_kelvin min_color_temp_kelvin max_color_temp_kelvin fan_speed_list unit_of_measurement battery_level fan_speed volume_level is_volume_muted media_title options min max step temperature_unit supported_features device_class next_rising next_setting finishes_at duration remaining humidity wind_speed wind_speed_unit apparent_temperature fan_modes swing_modes fan_mode swing_mode'.split())
# Attributes whose boolean value the screen needs; every other bool stays behind.
BOOL_ATTRS = frozenset(['is_volume_muted'])
TILE_BACKGROUNDS = {
'auto': {'label': 'Standaard', 'color': None},
# No card behind the tile: contents keep their size and place on the screen background.
'none': {'label': 'Geen', 'color': None},
'red': {'label': 'Rood', 'color': '#FADADD'},
'orange': {'label': 'Oranje', 'color': '#FFE1C6'},
'yellow': {'label': 'Geel', 'color': '#FFF0C2'},
'green': {'label': 'Groen', 'color': '#D9EEDC'},
'mint': {'label': 'Mint', 'color': '#D5F0EA'},
'blue': {'label': 'Blauw', 'color': '#D9EAFB'},
'purple': {'label': 'Paars', 'color': '#E9DDF5'},
'pink': {'label': 'Roze', 'color': '#F7DDEC'},
'gray': {'label': 'Grijs', 'color': '#E5E7EB'},
}
# Display modes per domain; everything else offers standard and watch (large value).
DISPLAYS = {'weather': ('standard', 'watch', 'forecast'), 'sensor': ('standard', 'watch', 'graph'), 'screen': ('digital', 'analog'), 'sun': ('standard', 'watch', 'sunpath')}
# Displays that only work on a double-width card.
WIDE_ONLY = ('forecast', 'sunpath')
# Direct controls on the right half of a double-width card (firmware 0.2.19+), like
# Home Assistant's own entity rows. The first choice is what a wide card shows
# when the tile has no explicit choice; 'none' keeps the plain card.
CONTROLS = {
'climate': (('setpoint', 'Temperatuur − / +'), ('mode', 'Uit, verwarmen, koelen')),
'switch': (('toggle', 'Aan/uit-schakelaar'),),
'input_boolean': (('toggle', 'Aan/uit-schakelaar'),),
'light': (('toggle', 'Aan/uit-schakelaar'), ('brightness', 'Helderheidsschuif')),
'fan': (('toggle', 'Aan/uit-schakelaar'), ('speed', 'Snelheidsschuif')),
'vacuum': (('buttons', 'Start, stop, naar dock'),),
'cover': (('buttons', 'Open, stop, dicht'), ('position', 'Positieschuif')),
'media_player': (('volume', 'Volume en dempen'), ('playback', 'Vorige, play/pauze, volgende')),
'number': (('stepper', 'Waarde − / +'), ('slider', 'Schuif')),
'input_number': (('stepper', 'Waarde − / +'), ('slider', 'Schuif')),
'select': (('stepper', 'Vorige / volgende keuze'),),
'input_select': (('stepper', 'Vorige / volgende keuze'),),
'timer': (('buttons', 'Start/pauze en annuleren'),),
'scene': (('run', 'Knop Activeren'),),
'script': (('run', 'Knop Uitvoeren'),),
'button': (('run', 'Knop Indrukken'),),
'input_button': (('run', 'Knop Indrukken'),),
}
def controls_catalogue():
"""Editor choices per domain: the default first, then 'none'."""
return {domain: {'default': choices[0][0], 'choices': [{'key': key, 'label': label} for key, label in choices] + [{'key': 'none', 'label': 'Geen'}]}
for domain, choices in CONTROLS.items()}
def resolve_controls(tile):
"""Control set a card shows on the screen, or None: only wide cards in the standard layout have room for one."""
options = tile.get('options', {})
domain = tile['entity'].split('.')[0]
if domain not in CONTROLS or options.get('size') != 'wide' or options.get('display', 'standard') != 'standard' or options.get('inline') == 'slider':
return None
choice = options.get('controls', CONTROLS[domain][0][0])
return None if choice == 'none' else choice
# Additive schema 1 extension. An absent object retains old firmware/YAML defaults.
SETTING_RULES = {
'standby_enabled': (True, None, None),
'standby_seconds': (600, 60, 86400),
'brightness': (100, 5, 100),
'standby_brightness': (20, 0, 100),
'night_enabled': (True, None, None),
'night_start': (1320, 0, 1439),
'night_end': (420, 0, 1439),
'night_brightness': (10, 0, 100),
'show_clock': (True, None, None),
'clock_24h': (True, None, None),
'home_on_standby': (False, None, None),
'swipe_pages': (False, None, None),
'rotation': (0, 0, 270),
}
def validate_settings(data):
if not isinstance(data, dict) or set(data) - SETTING_RULES.keys():
raise ValueError('Onbekende scherminstellingen; vernieuw de beheerpagina.')
clean = {}
for key, (default, minimum, maximum) in SETTING_RULES.items():
value = data.get(key, default)
if minimum is None:
valid = type(value) is bool
else:
valid = type(value) is int and minimum <= value <= maximum
if key == "rotation": valid = valid and value in (0, 90, 180, 270)
if not valid:
raise ValueError(f'Ongeldige waarde voor {key}.')
clean[key] = value
if max(clean['standby_brightness'], clean['night_brightness']) > clean['brightness']:
raise ValueError('Helderheid in standby en nacht mag niet hoger zijn dan normaal.')
return clean
def entity_id(value):
if not (isinstance(value, str) and len(value) <= 120 and re.fullmatch(r'[a-z0-9_]+\.[a-z0-9_]+', value)):
return False
return value in BUILTIN if value.startswith('screen.') else value.split('.')[0] in DOMAINS
def min_firmware(layout):
"""Oldest firmware that still accepts this layout; None when any version works."""
if any(t.get('options', {}).get('background') == 'none' for t in layout['tiles']):
return (0, 2, 16)
if any(t['entity'].split('.')[0] in NEW_DOMAINS for t in layout['tiles']):
return (0, 2, 14)
if len(layout['tiles']) > 10:
return (0, 2, 7)
return None
def short(value, limit):
return str(value).encode('utf-8')[:limit].decode('utf-8', errors='ignore')
def validate_layout(data):
if not isinstance(data, dict):
raise ValueError('Ongeldige indeling.')
title, tiles = data.get('title'), data.get('tiles')
if not isinstance(title, str) or not title.strip() or len(title.encode()) > 96:
raise ValueError('Geef het scherm een titel van maximaal 96 bytes.')
if not isinstance(tiles, list) or len(tiles) > 20:
raise ValueError('Kies maximaal 20 tegels.')
clean, seen = [], set()
for tile in tiles:
if not isinstance(tile, dict) or not entity_id(tile.get('entity')):
raise ValueError('Deze entiteit wordt niet ondersteund.')
if tile['entity'] in seen:
raise ValueError('Een entiteit kan maar één keer op een scherm staan.')
name = tile.get('name', '')
if not isinstance(name, str) or len(name.encode()) > 80:
raise ValueError('Een tegelnaam mag maximaal 80 bytes bevatten.')
seen.add(tile['entity'])
item = {'entity': tile['entity'], 'name': name.strip()}
if 'options' in tile:
options = tile['options']
if not isinstance(options, dict) or set(options) - {'tap', 'display', 'inline', 'history_hours', 'background', 'size', 'icon', 'controls'}:
raise ValueError('Onbekende tegelinstellingen.')
if 'background' in options and (not isinstance(options['background'],str) or options['background'] not in TILE_BACKGROUNDS):
raise ValueError('Kies een pastel achtergrondkleur uit het palet.')
if 'icon' in options and not (options['icon'] == 'auto' or isinstance(options['icon'], str) and options['icon'] in tile_icons.ICONS):
raise ValueError('Kies een icoon uit de lijst.')
domain = tile['entity'].split('.')[0]
displays = DISPLAYS.get(domain, ('standard', 'watch'))
choices = {'tap': ('auto', 'detail', 'toggle', 'none'), 'display': displays, 'inline': ('none', 'slider'), 'size': ('single', 'wide')}
for key, allowed in choices.items():
if key in options and options[key] not in allowed:
raise ValueError('Ongeldige tegelinstelling: ' + key)
# The five-day strip and the sun path only fit a double-width card.
if options.get('display') in WIDE_ONLY:
options = {**options, 'size': 'wide'}
if options.get('tap') == 'toggle' and domain not in {'light','switch','input_boolean','fan','media_player','climate'}:
raise ValueError('Deze entiteit ondersteunt geen aan/uit-actie.')
if options.get('inline') == 'slider' and domain not in {'light','fan','cover','number','input_number','media_player'}:
raise ValueError('Deze entiteit ondersteunt geen mini-schuif.')
if 'history_hours' in options and (type(options['history_hours']) is not int or options['history_hours'] not in (1,6,24)):
raise ValueError('Geschiedenis: kies 1, 6 of 24 uur.')
if options.get('display') == 'watch' and options.get('inline') == 'slider':
raise ValueError('Kies grote waarde of mini-schuif.')
if 'controls' in options:
allowed = ('none',) + tuple(key for key, _ in CONTROLS.get(domain, ()))
if not isinstance(options['controls'], str) or options['controls'] not in allowed:
raise ValueError('Deze entiteit ondersteunt die directe bediening niet.')
item['options'] = dict(options)
clean.append(item)
result = {'title': title.strip(), 'tiles': clean}
if 'settings' in data:
result['settings'] = validate_settings(data['settings'])
return result
def local_clock(value, tz):
"""HH:MM in the home's time zone for an ISO timestamp; '' when unusable."""
try:
moment = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
except (ValueError, TypeError):
return ''
if moment.tzinfo is None:
moment = moment.replace(tzinfo=timezone.utc)
return moment.astimezone(tz or timezone.utc).strftime('%H:%M')
def forecast_number(entry, name):
value = entry.get(name)
return round(value, 1) if isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) else None
def forecast_time(entry, tz):
try:
return datetime.fromisoformat(str(entry.get('datetime')).replace('Z', '+00:00')).astimezone(tz or timezone.utc)
except (ValueError, TypeError):
return None
def epoch(value):
"""Unix time of an ISO timestamp (a scene's state, a script's last_triggered); None when unusable."""
try:
moment = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
except (ValueError, TypeError):
return None
if moment.tzinfo is None:
moment = moment.replace(tzinfo=timezone.utc)
return int(moment.timestamp())
def extras(tile, states, forecast=None, tz=None, hourly=None, now=None):
"""Small, pre-computed values the firmware cannot derive itself (time zones, forecasts)."""
domain = tile['entity'].split('.')[0]
attrs = states.get(tile['entity'], {}).get('attributes', {})
if domain == 'weather' and (forecast or hourly):
result = {}
days = []
for entry in forecast or []:
if not isinstance(entry, dict) or len(days) == 5:
continue
day = forecast_time(entry, tz)
if day is None:
continue
item = {'d': WEEKDAYS[day.weekday()], 'c': short(entry.get('condition') or '', 20)}
# h/l: high and low; p: chance of rain in %; r: rain in the entity's unit (mm).
for key, name in (('h', 'temperature'), ('l', 'templow'), ('p', 'precipitation_probability'), ('r', 'precipitation')):
value = forecast_number(entry, name)
if value is not None:
item[key] = value
days.append(item)
if days:
result['days'] = days
# The next eight hours from now, for the weather card's hourly strip.
hours = []
# The running hour still counts: an entry stays until its hour has passed.
start = (now or datetime.now(timezone.utc)) - timedelta(minutes=59)
for entry in hourly or []:
if not isinstance(entry, dict) or len(hours) == 8:
continue
moment = forecast_time(entry, tz)
if moment is None or moment < start:
continue
item = {'t': moment.strftime('%H:%M'), 'c': short(entry.get('condition') or '', 20)}
for key, name in (('h', 'temperature'), ('p', 'precipitation_probability'), ('r', 'precipitation')):
value = forecast_number(entry, name)
if value is not None:
item[key] = value
hours.append(item)
if hours:
result['hours'] = hours
return result or None
if domain in ('scene', 'script', 'button', 'input_button'):
# When it last ran: scripts report last_triggered; scenes and buttons carry the time as their state.
last = epoch(attrs.get('last_triggered') if domain == 'script' else states.get(tile['entity'], {}).get('state'))
return {'last': last} if last else None
if domain == 'sun':
rise, down = local_clock(attrs.get('next_rising'), tz), local_clock(attrs.get('next_setting'), tz)
return {'rise': rise, 'set': down} if rise or down else None
if domain == 'timer':
result = {}
try:
end = datetime.fromisoformat(str(attrs.get('finishes_at')).replace('Z', '+00:00'))
result['end'] = int(end.timestamp())
except (ValueError, TypeError):
pass
for key, name in (('dur', 'duration'), ('rem', 'remaining')):
if isinstance(attrs.get(name), str):
result[key] = short(attrs[name], 16)
return result or None
return None
def tile_icon(tile, attrs):
"""Codepoint the screen shows: the chosen icon, else HA's own mdi icon; None keeps the firmware default."""
choice = tile.get('options', {}).get('icon', 'auto')
return tile_icons.ICONS[choice][0] if choice in tile_icons.ICONS else tile_icons.ha_icon(attrs)
def screen_options(tile, attrs):
"""Stored options on the wire; `icon` travels as the resolved codepoint (firmware 0.2.18+, ignored before)
and `controls` only as the set the card really shows (firmware 0.2.19+, ignored before)."""
options = {k: v for k, v in tile.get('options', {}).items() if k not in ('icon', 'controls')}
icon = tile_icon(tile, attrs)
if icon:
options['icon'] = icon
controls = resolve_controls(tile)
if controls:
options['controls'] = controls
return options if options or 'options' in tile else None
def state_message(index, tile, states, extra=None):
if tile['entity'] in BUILTIN:
message = {'v': 1, 'op': 'state', 'i': index, 'entity': tile['entity'],
'name': short(tile['name'] or BUILTIN[tile['entity']], 80), 'state': 'ok', 'a': {}}
if 'options' in tile:
message['o'] = tile['options']
return message
state = states.get(tile['entity'], {})
attrs = state.get('attributes', {})
bounded = {}
for key in ATTRS:
value = attrs.get(key)
if isinstance(value, bool):
if key in BOOL_ATTRS:
bounded[key] = value
continue
if value is None:
continue
if isinstance(value, (int, float)):
# supported_features is a bit field (media players exceed 8 million); other numbers are display values.
if math.isfinite(value) and abs(value) <= (2**31 if key == 'supported_features' else 1000000):
bounded[key] = value
elif isinstance(value, str):
bounded[key] = short(value, 48)
elif isinstance(value, list):
# Attribute lists have bounded lengths, strings and numeric ranges.
limit = 2 if key == 'hs_color' else 4 if key == 'fan_speed_list' else 8
bounded[key] = [short(v, 48) if isinstance(v, str) else v for v in value[:limit]
if isinstance(v, str) or isinstance(v, (float, int)) and math.isfinite(v) and abs(v) <= 1000000]
options = screen_options(tile, attrs)
return {'v': 1, 'op': 'state', 'i': index, 'entity': tile['entity'],
'name': short(tile['name'] or attrs.get('friendly_name') or tile['entity'], 80),
'state': short(state.get('state', 'unavailable'), 160), 'a': bounded,
**({'o': options} if options is not None else {}), **({'x': extra} if extra else {})}
def packets(message, token=None):
raw = json.dumps(message, ensure_ascii=False, separators=(',', ':'), allow_nan=False).encode()
if len(raw) > 4096:
raise ValueError('Het schermbericht is te groot.')
encoded = base64.b64encode(raw).decode('ascii')
token = token or secrets.token_hex(8)
chunks = [encoded[i:i+200] for i in range(0, len(encoded), 200)]
return [f'{token}|{i}|{int(i == len(chunks)-1)}|{part}' for i, part in enumerate(chunks)]
def discover(registry, states, devices, areas):
device_map = {d['id']: d for d in devices}
area_map = {a['area_id']: a['name'] for a in areas}
versions = {item.get('device_id'): states.get(item['entity_id'], {}).get('state', 'onbekend') for item in registry
if item.get('platform') == 'esphome' and item.get('original_name') == 'Schermfirmware'}
boards = {item.get("device_id"): "guition" for item in registry
if item.get("platform") == "esphome" and item.get("original_name") == "Guition schermtype"}
def diagnostic(name, pattern):
found = {}
for item in registry:
if item.get('platform') == 'esphome' and item.get('original_name') == name:
value = states.get(item['entity_id'], {}).get('state', '')
if isinstance(value, str) and re.fullmatch(pattern, value):
found[item.get('device_id')] = value
return found
nodes = diagnostic('Apparaatnaam', r'[a-z0-9][a-z0-9-]{0,30}')
addresses = diagnostic('IP-adres', r'\d{1,3}(\.\d{1,3}){3}')
screens, entities = [], []
for item in registry:
eid = item['entity_id']
device = device_map.get(item.get('device_id'), {})
state = states.get(eid, {})
area = area_map.get(item.get('area_id') or device.get('area_id'), '')
if item.get('platform') == 'esphome' and eid.startswith('text.') and item.get('original_name') == 'Tegelinstellingen' and not item.get('disabled_by'):
screens.append({'id': eid, 'name': device.get('name_by_user') or device.get('name') or eid,
'firmware': versions.get(item.get('device_id'), 'onbekend'),
'board': boards.get(item.get('device_id'), 'unknown'),
'node': nodes.get(item.get('device_id')), 'ip': addresses.get(item.get('device_id')),
'device': device.get('name') or '',
'area': area, 'online': state.get('state') not in (None, 'unknown', 'unavailable'),
'status': state.get('state', 'Niet verbonden')})
if not entity_id(eid) or item.get('disabled_by'):
continue
entities.append({'id': eid, 'name': state.get('attributes', {}).get('friendly_name') or item.get('name') or item.get('original_name') or eid,
'device': device.get('name_by_user') or device.get('name') or '', 'area': area,
'state': state.get('state', 'unavailable'), 'icon': tile_icons.ha_icon(state.get('attributes'))})
# YAML entities may not have an entity-registry entry.
registered = {e['id'] for e in entities}
for eid, state in states.items():
if entity_id(eid) and eid not in registered and not any(r['entity_id'] == eid for r in registry):
entities.append({'id': eid, 'name': state.get('attributes', {}).get('friendly_name', eid), 'device': '', 'area': '', 'state': state['state'],
'icon': tile_icons.ha_icon(state.get('attributes'))})
return screens, sorted(entities, key=lambda e: e['name'].casefold())
def installation_yaml(data):
board, name, friendly = data.get('board'), data.get('name'), data.get('friendly_name')
if board not in REFS or not isinstance(name, str) or not re.fullmatch(r'[a-z][a-z0-9-]{0,29}', name):
raise ValueError('Kies een bord en een unieke naam (kleine letters, cijfers, streepjes; maximaal 30 tekens).')
if not isinstance(friendly, str) or not friendly.strip() or len(friendly) > 60:
raise ValueError('Geef het scherm een herkenbare naam (maximaal 60 tekens).')
quote = lambda s: json.dumps(s, ensure_ascii=False)
key, ota, ap = base64.b64encode(secrets.token_bytes(32)).decode(), secrets.token_urlsafe(24), secrets.token_urlsafe(12)
return f'''# Bewaar dit bestand: het bevat de unieke sleutels voor dit scherm.
# Wifi komt uit de secrets.yaml van ESPHome Device Builder.
substitutions:
DEVICE_NAME: {quote(name)}
DEVICE_FRIENDLY_NAME: {quote(friendly.strip())}
esphome:
name: {quote(name)}
friendly_name: {quote(friendly.strip())}
packages:
display:
url: {REPO}
ref: {REFS[board]}
files: [packages/{board}.yaml]
refresh: 0s
api:
encryption:
key: {quote(key)}
ota:
- platform: esphome
password: {quote(ota)}
wifi:
ssid: !secret wifi_ssid
password: !secret wifi_password
power_save_mode: none
ap:
ssid: {quote(name + ' Setup')}
password: {quote(ap)}
captive_portal:
'''