homeassistant_espscreen/screen_manager/app/core.py

"""Pure validation, firmware generation and bounded display protocol."""
import base64
from datetime import datetime, 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.18'
ATTRS = frozenset('brightness percentage current_position current_temperature temperature current_humidity min_temp max_temp target_temp_step supported_color_modes hvac_modes 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 media_title options min max step temperature_unit supported_features next_rising next_setting finishes_at duration remaining'.split())


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')

# 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'}:
                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'}:
                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.')
            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 extras(tile, states, forecast=None, tz=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:
        days = []
        for entry in forecast[:5]:
            if not isinstance(entry, dict):
                continue
            try:
                day = datetime.fromisoformat(str(entry.get('datetime')).replace('Z', '+00:00')).astimezone(tz or timezone.utc)
            except (ValueError, TypeError):
                continue
            item = {'d': WEEKDAYS[day.weekday()], 'c': short(entry.get('condition') or '', 20)}
            for key, name in (('h', 'temperature'), ('l', 'templow')):
                value = entry.get(name)
                if isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value):
                    item[key] = round(value, 1)
            days.append(item)
        return {'days': days} if days 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)."""
    options = {k: v for k, v in tile.get('options', {}).items() if k != 'icon'}
    icon = tile_icon(tile, attrs)
    if icon:
        options['icon'] = icon
    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) or value is None:
            continue
        if isinstance(value, (int, float)):
            if math.isfinite(value) and abs(value) <= 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
  ap:
    ssid: {quote(name + ' Setup')}
    password: {quote(ap)}
captive_portal:
'''