Max Gramser
f60eed3ff6
App 0.2.39 / firmware 0.2.33, items 1, 2, 4, 5 and 6 of the 30-screen scaling check. - Only the tiles (and top bar) of changed entities are rebuilt; the screen list is cached until the registry or a screen diagnostic changes; the SSE stream follows. - Sensor history comes from one recorder/statistics_during_period call per window in a background task, with the REST history as fallback per sensor without statistics. - Firmware 0.2.33 answers a keepalive ping with its layout revision; a mismatch or a restart asks for the layout again, everything is repeated hourly as a safety net. - The screen_message API action takes one whole message; the text inbox chunks stay as fallback for older firmware. A repeated unchanged layout keeps the inbox state. - Quiet diagnostics: Opgestart (uptime as timestamp) replaces Uptime, heap every 5 min, setting numbers publish only on change. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
618 lines
33 KiB
Python
618 lines
33 KiB
Python
"""Pure validation, firmware generation and bounded display protocol."""
|
||
import base64
|
||
from datetime import datetime, timedelta, timezone
|
||
import hashlib
|
||
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.33'
|
||
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')
|
||
|
||
# Grid positions: two columns, three rows per page, at most eight pages. A tile's
|
||
# `slot` is its absolute cell (page * 6 + row * 2 + column); a wide tile starts in
|
||
# the left column and also covers the cell to its right. Empty cells are allowed.
|
||
SLOTS_PER_PAGE = 6
|
||
MAX_PAGES = 8
|
||
MAX_SLOTS = MAX_PAGES * SLOTS_PER_PAGE
|
||
|
||
def is_wide(tile):
|
||
return tile.get('options', {}).get('size') == 'wide'
|
||
|
||
def footprint(slot, wide):
|
||
return (slot, slot + 1) if wide else (slot,)
|
||
|
||
def pack_slots(tiles):
|
||
"""In-order packing, the rule before explicit positions and what firmware without
|
||
`slots` still does: fill left to right, a wide card starts a new row."""
|
||
position, slots = 0, []
|
||
for tile in tiles:
|
||
wide = is_wide(tile)
|
||
if wide and position % 2:
|
||
position += 1
|
||
slots.append(position)
|
||
position += 2 if wide else 1
|
||
return slots
|
||
|
||
def has_gaps(tiles):
|
||
"""True when the stored positions differ from the in-order packing, so firmware
|
||
before 0.2.26 (which ignores `slots`) would show another arrangement."""
|
||
return [t.get('slot') for t in tiles] != pack_slots(tiles)
|
||
|
||
# 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
|
||
|
||
# Diagnostic entities every ESP Screens firmware exposes; the manager watches them for screens.
|
||
SCREEN_ENTITY_NAMES = frozenset({'Tegelinstellingen', 'Schermfirmware', 'Guition schermtype', 'Apparaatnaam', 'IP-adres'})
|
||
|
||
# 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
|
||
|
||
# ----- Top bar (firmware 0.2.32+): the screen name on the left, up to six items on the right.
|
||
# Without a `header` the screen keeps its name and the clock of `show_clock`. -----
|
||
HEADER_MIN_FIRMWARE = (0, 2, 32)
|
||
# Firmware 0.2.33+ takes a whole message in one API action (esphome.<node>_screen_message)
|
||
# and answers a keepalive ping with its layout revision; older firmware gets base64
|
||
# chunks in the text inbox and a full repeat every keepalive.
|
||
TRANSPORT_MIN_FIRMWARE = (0, 2, 33)
|
||
MESSAGE_ACTION = 'screen_message'
|
||
HEADER_MAX_ITEMS = 6
|
||
# Items the screen draws on its own clock, without Home Assistant.
|
||
HEADER_BUILTIN = {'clock': 'Tijd', 'analog': 'Analoge klok', 'date': 'Datum'}
|
||
# Only shown, never controlled: the top bar takes these besides every tile domain.
|
||
HEADER_ONLY_DOMAINS = frozenset('device_tracker zone lock alarm_control_panel counter event input_datetime input_text water_heater humidifier'.split())
|
||
HEADER_CONTENTS = ('state', 'last_changed')
|
||
HEADER_SHOWS = ('always', 'active')
|
||
|
||
def header_entity(value):
|
||
return (isinstance(value, str) and len(value) <= 120 and re.fullmatch(r'[a-z0-9_]+\.[a-z0-9_]+', value) is not None
|
||
and value.split('.')[0] in (DOMAINS - {'screen'}) | HEADER_ONLY_DOMAINS)
|
||
|
||
def header_items(layout):
|
||
"""Items the top bar shows: the stored ones, else what firmware before the top bar drew (the clock)."""
|
||
if 'header' in layout:
|
||
return layout['header']['items']
|
||
return [{'type': 'clock'}] if layout.get('settings', {}).get('show_clock', True) else []
|
||
|
||
def validate_header(data):
|
||
if not isinstance(data, dict) or set(data) - {'items'} or not isinstance(data.get('items'), list):
|
||
raise ValueError('Ongeldige bovenbalk; vernieuw de beheerpagina.')
|
||
if len(data['items']) > HEADER_MAX_ITEMS:
|
||
raise ValueError(f'De bovenbalk heeft plaats voor maximaal {HEADER_MAX_ITEMS} onderdelen.')
|
||
items, seen = [], set()
|
||
for item in data['items']:
|
||
kind = item.get('type') if isinstance(item, dict) else None
|
||
if kind in HEADER_BUILTIN:
|
||
if set(item) != {'type'}:
|
||
raise ValueError('Ongeldige instelling in de bovenbalk.')
|
||
clean = {'type': kind}
|
||
elif kind == 'entity':
|
||
if set(item) - {'type', 'entity', 'content', 'icon', 'show'}:
|
||
raise ValueError('Onbekende instelling in de bovenbalk; vernieuw de beheerpagina.')
|
||
if not header_entity(item.get('entity')):
|
||
raise ValueError('Deze entiteit kan niet in de bovenbalk.')
|
||
clean = {'type': 'entity', 'entity': item['entity'], 'content': item.get('content', 'state'),
|
||
'icon': item.get('icon', 'auto'), 'show': item.get('show', 'always')}
|
||
if clean['content'] not in HEADER_CONTENTS or clean['show'] not in HEADER_SHOWS:
|
||
raise ValueError('Ongeldige instelling in de bovenbalk.')
|
||
if not (clean['icon'] in ('auto', 'none') or isinstance(clean['icon'], str) and clean['icon'] in tile_icons.ICONS):
|
||
raise ValueError('Kies een icoon uit de lijst.')
|
||
else:
|
||
raise ValueError('Onbekend onderdeel in de bovenbalk; vernieuw de beheerpagina.')
|
||
key = json.dumps(clean, sort_keys=True)
|
||
if key in seen:
|
||
raise ValueError('Dit onderdeel staat al in de bovenbalk.')
|
||
seen.add(key)
|
||
items.append(clean)
|
||
return {'items': items}
|
||
|
||
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)
|
||
# Positions: every tile or none (an older editor sends none and keeps its order).
|
||
given = [tile.get('slot') for tile in tiles]
|
||
if any(slot is not None for slot in given):
|
||
occupied = set()
|
||
for item, slot in zip(clean, given):
|
||
if type(slot) is not int or not 0 <= slot < MAX_SLOTS:
|
||
raise ValueError('Ongeldige tegelpositie; vernieuw de beheerpagina.')
|
||
if is_wide(item) and slot % 2:
|
||
raise ValueError('Een dubbelbrede tegel begint in de linkerkolom.')
|
||
for cell in footprint(slot, is_wide(item)):
|
||
if cell in occupied:
|
||
raise ValueError('Twee tegels staan op dezelfde plek.')
|
||
occupied.add(cell)
|
||
item['slot'] = slot
|
||
clean.sort(key=lambda item: item['slot'])
|
||
else:
|
||
for item, slot in zip(clean, pack_slots(clean)):
|
||
item['slot'] = slot
|
||
result = {'title': title.strip(), 'tiles': clean}
|
||
# Pages kept on purpose, empty ones included; the screen shows at least what the tiles need.
|
||
if 'pages' in data:
|
||
if type(data['pages']) is not int or not 1 <= data['pages'] <= MAX_PAGES:
|
||
raise ValueError(f'Een scherm heeft 1 tot {MAX_PAGES} pagina\'s.')
|
||
result['pages'] = data['pages']
|
||
if 'settings' in data:
|
||
result['settings'] = validate_settings(data['settings'])
|
||
if 'header' in data:
|
||
result['header'] = validate_header(data['header'])
|
||
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 encode(message):
|
||
"""The message as the firmware parses it: compact JSON, at most 4096 bytes."""
|
||
raw = json.dumps(message, ensure_ascii=False, separators=(',', ':'), allow_nan=False)
|
||
if len(raw.encode()) > 4096:
|
||
raise ValueError('Het schermbericht is te groot.')
|
||
return raw
|
||
|
||
def revision(message):
|
||
"""Short fingerprint of a layout message; the screen echoes it back on every ping."""
|
||
return hashlib.sha256(json.dumps(message, sort_keys=True, separators=(',', ':')).encode()).hexdigest()[:12]
|
||
|
||
def packets(message, token=None):
|
||
"""The message as base64 chunks for the 255-character text inbox (firmware before 0.2.33)."""
|
||
encoded = base64.b64encode(encode(message).encode()).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 message_action(node):
|
||
"""Home Assistant action that hands a screen one whole message, or None without a node name."""
|
||
return alert_service(node, MESSAGE_ACTION)
|
||
|
||
# ----- Alerts (firmware 0.2.31+): the reference the cheatsheet shows. tests/test_alerts_reference.py
|
||
# keeps every value here equal to what the board profiles compile. -----
|
||
ALERT_MIN_FIRMWARE = '0.2.31'
|
||
ALERT_EVENT = 'esphome.screen_alert'
|
||
ALERT_ENDINGS = (('ok', 'De knop is ingedrukt'), ('timeout', 'De timeout is verstreken'),
|
||
('replaced', 'Een nieuwe alert kwam eroverheen'), ('remote', 'dismiss_alert vanuit Home Assistant'))
|
||
ALERT_FALLBACK_ICON = 'alert-outline'
|
||
# (field, ESPHome type, label, explanation, example) in the order Home Assistant shows them.
|
||
ALERT_FIELDS = (
|
||
('title', 'string', 'Titel', 'Eén regel bovenaan de kaart; wat niet past krijgt puntjes. Leeg wordt "Melding".', 'Iemand belt aan'),
|
||
('subtitle', 'string', 'Subtitle', 'Toelichting onder de titel, over meerdere regels; een regeleinde mag. Leeg mag.', 'Deur 3, achterkant'),
|
||
('icon', 'string', 'Icoon', 'Een naam uit de lijst, ook als mdi:naam of als hex-codepoint (F12E6). Onbekend of leeg geeft de waarschuwingsdriehoek.', 'doorbell'),
|
||
('color', 'string', 'Kleur', 'Een van de negen pastelkleuren van de tegels. Leeg geeft de witte kaart.', 'orange'),
|
||
('button_text', 'string', 'Knoptekst', 'De tekst op de knop. Leeg is "Oké".', 'Ik kom'),
|
||
('timeout', 'int', 'Timeout', 'Seconden waarna de kaart vanzelf verdwijnt. 0 wacht op de knop, hoe lang dat ook duurt. De knop sluit altijd direct, ook met een timeout.', 0),
|
||
('flash', 'bool', 'Knipperen', 'Aan laat de backlight vier keer knipperen als de alert binnenkomt; daarna blijft het scherm gewoon aan.', True),
|
||
)
|
||
# Bytes per field the firmware keeps (the profiles' ALERT_*_MAX); a Dutch letter with accent takes two.
|
||
ALERT_LIMITS = {'cyd': {'title': 48, 'subtitle': 160, 'button_text': 12}, 'guition': {'title': 64, 'subtitle': 240, 'button_text': 16}}
|
||
ALERT_SUGGESTED_ICONS = ('doorbell', 'bell', 'bell-ring', 'alert-outline', 'alarm-light', 'lock', 'lock-open-variant', 'door-open',
|
||
'window-closed-variant', 'motion-sensor', 'cctv', 'smoke-detector', 'water-alert', 'fire', 'mailbox', 'car',
|
||
'account', 'account-group', 'washing-machine', 'robot-vacuum', 'timer-outline', 'check')
|
||
|
||
def alert_service(node, action='show_alert'):
|
||
"""Home Assistant registers a device's actions as esphome.<node>_<action>, dashes as underscores."""
|
||
return f"esphome.{node.replace('-', '_')}_{action}" if isinstance(node, str) and node else None
|
||
|
||
def alert_reference():
|
||
"""Everything the Alerts cheatsheet shows besides the screens themselves."""
|
||
return {'min_firmware': ALERT_MIN_FIRMWARE, 'event': ALERT_EVENT,
|
||
'endings': [{'action': action, 'label': label} for action, label in ALERT_ENDINGS],
|
||
'fallback_icon': ALERT_FALLBACK_ICON, 'fallback_cp': tile_icons.GLYPHS[ALERT_FALLBACK_ICON],
|
||
'fields': [{'name': name, 'type': kind, 'label': label, 'help': help_, 'example': example} for name, kind, label, help_, example in ALERT_FIELDS],
|
||
'limits': ALERT_LIMITS,
|
||
'colors': [{'name': name, 'label': item['label'], 'color': item['color']} for name, item in TILE_BACKGROUNDS.items() if item['color']],
|
||
'suggested_icons': [{'name': name, 'cp': tile_icons.GLYPHS[name]} for name in ALERT_SUGGESTED_ICONS],
|
||
'extra_icons': [{'name': name, 'cp': cp} for name, cp in tile_icons.FIXED]}
|
||
|
||
def screen_items(registry):
|
||
"""The ESPHome entities that describe a screen; the subset `discover_screens` needs."""
|
||
return [item for item in registry if item.get('platform') == 'esphome' and item.get('original_name') in SCREEN_ENTITY_NAMES]
|
||
|
||
def discover_screens(registry, states, devices, areas):
|
||
"""Paired screens: every enabled ESPHome inbox with the diagnostics of its device."""
|
||
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 = []
|
||
for item in registry:
|
||
eid = item['entity_id']
|
||
if not (item.get('platform') == 'esphome' and eid.startswith('text.') and item.get('original_name') == 'Tegelinstellingen' and not item.get('disabled_by')):
|
||
continue
|
||
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'), '')
|
||
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')})
|
||
return screens
|
||
|
||
def discover(registry, states, devices, areas):
|
||
"""(screens, entities): the paired screens and every entity a tile or the top bar can show."""
|
||
device_map = {d['id']: d for d in devices}
|
||
area_map = {a['area_id']: a['name'] for a in areas}
|
||
screens = discover_screens(registry, states, devices, areas)
|
||
entities = []
|
||
for item in registry:
|
||
eid = item['entity_id']
|
||
tile = entity_id(eid)
|
||
if not (tile or header_entity(eid)) or item.get('disabled_by'):
|
||
continue
|
||
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'), '')
|
||
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')),
|
||
# Top-bar-only domains (a phone's tracker, a lock) stay out of the tile picker.
|
||
**({} if tile else {'tile': False})})
|
||
# YAML entities may not have an entity-registry entry.
|
||
registered = {e['id'] for e in entities}
|
||
in_registry = {r['entity_id'] for r in registry}
|
||
for eid, state in states.items():
|
||
tile = entity_id(eid)
|
||
if (tile or header_entity(eid)) and eid not in registered and eid not in 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')), **({} if tile else {'tile': False})})
|
||
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:
|
||
'''
|