homeassistant_espscreen/screen_manager/app/core.py

"""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': 'Clock'}
NEW_DOMAINS = frozenset('sun timer person screen'.split())
WEEKDAYS = ['Mo', 'Tu', 'We', 'Th', 'Fr', 'Sa', 'Su']
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.39'
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': 'Default', 'color': None},
    # No card behind the tile: contents keep their size and place on the screen background.
    'none': {'label': 'None', 'color': None},
    'red': {'label': 'Red', 'color': '#FADADD'},
    'orange': {'label': 'Orange', 'color': '#FFE1C6'},
    'yellow': {'label': 'Yellow', 'color': '#FFF0C2'},
    'green': {'label': 'Green', 'color': '#D9EEDC'},
    'mint': {'label': 'Mint', 'color': '#D5F0EA'},
    'blue': {'label': 'Blue', 'color': '#D9EAFB'},
    'purple': {'label': 'Purple', 'color': '#E9DDF5'},
    'pink': {'label': 'Pink', 'color': '#F7DDEC'},
    'gray': {'label': 'Gray', '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', 'Temperature − / +'), ('mode', 'Off, heat, cool')),
    'switch': (('toggle', 'On/off switch'),),
    'input_boolean': (('toggle', 'On/off switch'),),
    'light': (('toggle', 'On/off switch'), ('brightness', 'Brightness slider')),
    'fan': (('toggle', 'On/off switch'), ('speed', 'Speed slider')),
    'vacuum': (('buttons', 'Start, stop, dock'),),
    'cover': (('buttons', 'Open, stop, close'), ('position', 'Position slider')),
    'media_player': (('volume', 'Volume and mute'), ('playback', 'Previous, play/pause, next')),
    'number': (('stepper', 'Value − / +'), ('slider', 'Slider')),
    'input_number': (('stepper', 'Value − / +'), ('slider', 'Slider')),
    'select': (('stepper', 'Previous / next choice'),),
    'input_select': (('stepper', 'Previous / next choice'),),
    'timer': (('buttons', 'Start/pause and cancel'),),
    'scene': (('run', 'Activate button'),),
    'script': (('run', 'Run button'),),
    'button': (('run', 'Press button'),),
    'input_button': (('run', 'Press button'),),
}

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': 'None'}]}
            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.
# Firmware built before the English translation still registers the Dutch originals, so both
# forms are recognised until every board has been reflashed with an English `name:`.
NAME_TILE_SETTINGS = ('Tile settings', 'Tegelinstellingen')
NAME_SCREEN_FIRMWARE = ('Screen firmware', 'Schermfirmware')
NAME_GUITION_TYPE = ('Guition screen type', 'Guition schermtype')
NAME_DEVICE_NAME = ('Device name', 'Apparaatnaam')
NAME_IP_ADDRESS = ('IP address', 'IP-adres')
SCREEN_ENTITY_NAMES = frozenset(NAME_TILE_SETTINGS + NAME_SCREEN_FIRMWARE + NAME_GUITION_TYPE + NAME_DEVICE_NAME + NAME_IP_ADDRESS)

def entity_slug(name):
    """The end of an entity id Home Assistant derives from an entity name (ASCII names)."""
    return re.sub(r'[^a-z0-9]+', '_', str(name).lower()).strip('_')

# An inbox entity id is text.<device>_tile_settings, or text.<device>_tegelinstellingen before firmware 0.2.34.
INBOX_SUFFIXES = tuple('_' + entity_slug(name) for name in NAME_TILE_SETTINGS)

def inbox_prefix(entity):
    """The device part of an inbox entity id ('office_1' for text.office_1_tile_settings), or None."""
    if isinstance(entity, str) and entity.startswith('text.'):
        for suffix in INBOX_SUFFIXES:
            if entity.endswith(suffix) and len(entity) > len('text.') + len(suffix):
                return entity[len('text.'):-len(suffix)]
    return None

def device_prefixes(registry, device_id):
    """Entity id prefixes Home Assistant gave the ESPHome entities of one device: the device name when each
    entity was created. Entities whose name never changed keep the prefix an older inbox id carried."""
    prefixes = set()
    for item in registry:
        name = item.get('original_name')
        if item.get('device_id') != device_id or item.get('platform') != 'esphome' or not isinstance(name, str):
            continue
        object_id, suffix = item['entity_id'].split('.', 1)[-1], '_' + entity_slug(name)
        if len(suffix) > 1 and object_id.endswith(suffix) and len(object_id) > len(suffix):
            prefixes.add(object_id[:-len(suffix)])
    return prefixes

# 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('Unknown screen settings; refresh the management page.')
    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'Invalid value for {key}.')
        clean[key] = value
    if max(clean['standby_brightness'], clean['night_brightness']) > clean['brightness']:
        raise ValueError('Standby and night brightness may not be higher than normal.')
    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': 'Time', 'analog': 'Analog clock', 'date': 'Date'}
# 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('Invalid top bar; refresh the management page.')
    if len(data['items']) > HEADER_MAX_ITEMS:
        raise ValueError(f'The top bar has room for at most {HEADER_MAX_ITEMS} items.')
    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('Invalid setting in the top bar.')
            clean = {'type': kind}
        elif kind == 'entity':
            if set(item) - {'type', 'entity', 'content', 'icon', 'show'}:
                raise ValueError('Unknown setting in the top bar; refresh the management page.')
            if not header_entity(item.get('entity')):
                raise ValueError("This entity can't go in the top bar.")
            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('Invalid setting in the top bar.')
            if not (clean['icon'] in ('auto', 'none') or isinstance(clean['icon'], str) and clean['icon'] in tile_icons.ICONS):
                raise ValueError('Choose an icon from the list.')
        else:
            raise ValueError('Unknown item in the top bar; refresh the management page.')
        key = json.dumps(clean, sort_keys=True)
        if key in seen:
            raise ValueError('This item is already in the top bar.')
        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('Invalid layout.')
    title, tiles = data.get('title'), data.get('tiles')
    if not isinstance(title, str) or not title.strip() or len(title.encode()) > 96:
        raise ValueError('Give the screen a title of at most 96 bytes.')
    if not isinstance(tiles, list) or len(tiles) > 20:
        raise ValueError('Choose at most 20 tiles.')
    clean, seen = [], set()
    for tile in tiles:
        if not isinstance(tile, dict) or not entity_id(tile.get('entity')):
            raise ValueError("This entity isn't supported.")
        if tile['entity'] in seen:
            raise ValueError('An entity can only appear once on a screen.')
        name = tile.get('name', '')
        if not isinstance(name, str) or len(name.encode()) > 80:
            raise ValueError('A tile name may contain at most 80 bytes.')
        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('Unknown tile settings.')
            if 'background' in options and (not isinstance(options['background'],str) or options['background'] not in TILE_BACKGROUNDS):
                raise ValueError('Choose a pastel background color from the palette.')
            if 'icon' in options and not (options['icon'] == 'auto' or isinstance(options['icon'], str) and options['icon'] in tile_icons.ICONS):
                raise ValueError('Choose an icon from the list.')
            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('Invalid tile setting: ' + 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("This entity doesn't support an on/off action.")
            if options.get('inline') == 'slider' and domain not in {'light','fan','cover','number','input_number','media_player'}:
                raise ValueError("This entity doesn't support a mini-slider.")
            if 'history_hours' in options and (type(options['history_hours']) is not int or options['history_hours'] not in (1,6,24)):
                raise ValueError('History: choose 1, 6, or 24 hours.')
            if options.get('display') == 'watch' and options.get('inline') == 'slider':
                raise ValueError('Choose large value or mini-slider.')
            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("This entity doesn't support that direct control.")
            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('Invalid tile position; refresh the management page.')
            if is_wide(item) and slot % 2:
                raise ValueError('A double-width tile starts in the left column.')
            for cell in footprint(slot, is_wide(item)):
                if cell in occupied:
                    raise ValueError('Two tiles are in the same spot.')
                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'A screen has 1 to {MAX_PAGES} pages.')
        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())

# Vacuum cards (app 0.2.46, firmware 0.2.39+): how a robot cleans is often a select on its own device,
# not an attribute. Roborock has "Cleaning mode" (vacuum, vac_and_mop, mop, custom, smart_mode) and
# "Mop intensity" since HA 2026.8; Ecovacs calls its mode work_mode. They are found by translation key
# or by the end of their entity id. HA 2026.8 also dropped battery_level from vacuum entities, so the
# battery comes from the device's battery sensor.
VACUUM_MODE_KEYS = ('cleaning_mode', 'work_mode', 'clean_mode')
VACUUM_WATER_KEYS = ('mop_intensity', 'water_flow', 'water_amount', 'water_flow_level', 'water_volume', 'mop_water_level')
# Short chip labels; anything else shows its own words.
VACUUM_LABELS = {
    'vacuum': 'Vacuum', 'sweeping': 'Vacuum', 'vac_and_mop': 'Vac & mop', 'vacuum_and_mop': 'Vac & mop',
    'sweeping_and_mopping': 'Vac & mop', 'mop_after_vacuum': 'Vac, then mop', 'mopping_after_sweeping': 'Vac, then mop',
    'mop': 'Mop', 'mopping': 'Mop', 'custom': 'Custom', 'smart_mode': 'Smart',
    'off': 'Off', 'min': 'Min', 'slight': 'Slight', 'low': 'Low', 'mild': 'Mild', 'medium': 'Medium', 'moderate': 'Moderate',
    'standard': 'Standard', 'high': 'High', 'intense': 'Intense', 'extreme': 'Extreme', 'ultrahigh': 'Ultra high',
    'quiet': 'Quiet', 'silent': 'Silent', 'gentle': 'Gentle', 'balanced': 'Normal', 'turbo': 'Turbo', 'strong': 'Strong',
    'max': 'Max', 'max_plus': 'Max+', 'auto': 'Auto',
}
# What a cleaning mode does, one letter per option for the firmware: v vacuum only, m mop only, b both,
# a automatic (the robot or its app picks suction and water). Unknown modes count as both.
VACUUM_ROLES = {'vacuum': 'v', 'sweeping': 'v', 'mop': 'm', 'mopping': 'm', 'custom': 'a', 'smart_mode': 'a'}
# With a cleaning mode select these speeds and water levels belong to a mode: suction off is mop only,
# water off is vacuum only, and custom or smart settings are the Custom and Smart modes.
MODE_COVERED = frozenset(('off', 'off_raise_main_brush', 'custom', 'custom_water_flow', 'smart_mode', 'vac_followed_by_mop'))
VACUUM_CHOICES = 6

def vacuum_label(value):
    return short(VACUUM_LABELS.get(value) or str(value).replace('_', ' ').capitalize(), 24)

def vacuum_related(entity, device, states):
    """Entities on the vacuum's device that its card reads, by role: the 'mode' and 'water' selects, the
    'battery' and 'room' sensors and the 'charging' binary sensor; absent roles are left out.

    `device` holds the registry entries of the vacuum's device. Entity ids follow Home Assistant's
    language (select.s8_intensiteit_van_dweilen), so translation keys come first."""
    found = {}
    for item in device or ():
        eid = item.get('entity_id') or ''
        if eid == entity or item.get('disabled_by'):
            continue
        key = item.get('translation_key') or ''
        attrs = states.get(eid, {}).get('attributes', {})
        role = None
        if eid.startswith('select.'):
            if key in VACUUM_MODE_KEYS or eid.endswith(tuple('_' + k for k in VACUUM_MODE_KEYS)):
                role = 'mode'
            elif key in VACUUM_WATER_KEYS or eid.endswith(tuple('_' + k for k in VACUUM_WATER_KEYS)):
                role = 'water'
        elif eid.startswith('sensor.'):
            if attrs.get('device_class') == 'battery':
                role = 'battery'
            elif key == 'current_room' or eid.endswith('_current_room'):
                role = 'room'
        elif eid.startswith('binary_sensor.') and attrs.get('device_class') == 'battery_charging':
            role = 'charging'
        if role and role not in found:
            found[role] = eid
    return found

def vacuum_extras(tile, states, device):
    """The vacuum card's rows and details: the mode and water selects (`e` entity, `s` state, `o` options,
    `l` labels, `r` a role per mode), the suction speeds to offer (`fan`), the battery (`bat`), charging
    (`chg`) and the room the robot is in (`room`)."""
    entity = tile['entity']
    attrs = states.get(entity, {}).get('attributes', {})
    related = vacuum_related(entity, device, states)
    result = {}
    def row(values):
        values = [short(v, 48) for v in values][:VACUUM_CHOICES]
        return {'o': values, 'l': [vacuum_label(v) for v in values]} if values else None
    def select(eid, keep):
        state = states.get(eid) or {}
        options = state.get('attributes', {}).get('options')
        current = state.get('state') if isinstance(state.get('state'), str) else ''
        found = row([o for o in options if isinstance(o, str) and keep(o, current)]) if isinstance(options, list) else None
        return {'e': eid, 's': short(current, 48), **found} if found else None
    # Custom runs the per-room settings made in the robot's app: listed only while it is in use.
    mode = select(related['mode'], lambda o, current: o != 'custom' or o == current) if 'mode' in related else None
    covered = MODE_COVERED if mode else frozenset()
    if mode:
        mode['r'] = ''.join(VACUUM_ROLES.get(v, 'b') for v in mode['o'])
        result['mode'] = mode
    water = select(related['water'], lambda o, current: o not in covered) if 'water' in related else None
    if water:
        result['water'] = water
    speeds = attrs.get('fan_speed_list')
    fan = row([s for s in speeds if isinstance(s, str) and s not in covered]) if isinstance(speeds, list) else None
    if fan:
        result['fan'] = fan
    if not isinstance(attrs.get('battery_level'), (int, float)) and 'battery' in related:
        try:
            level = float(states[related['battery']].get('state'))
        except (TypeError, ValueError, KeyError):
            level = math.nan
        if math.isfinite(level):
            result['bat'] = max(0, min(100, round(level)))
    if states.get(related.get('charging'), {}).get('state') == 'on':
        result['chg'] = 1
    room = states.get(related.get('room'), {}).get('state')
    if isinstance(room, str) and room not in ('', 'unknown', 'unavailable', 'none'):
        result['room'] = short(room, 32)
    return result or None

def extras(tile, states, forecast=None, tz=None, hourly=None, now=None, device=None):
    """Small, pre-computed values the firmware cannot derive itself (time zones, forecasts, a vacuum's device)."""
    domain = tile['entity'].split('.')[0]
    attrs = states.get(tile['entity'], {}).get('attributes', {})
    if domain == 'vacuum':
        return vacuum_extras(tile, states, device)
    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('The screen message is too large.')
    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', 'The button was pressed'), ('timeout', 'The timeout ran out'),
                 ('replaced', 'A new alert came over it'), ('remote', 'dismiss_alert from Home Assistant'))
ALERT_FALLBACK_ICON = 'alert-outline'
# (field, ESPHome type, label, explanation, example) in the order Home Assistant shows them.
ALERT_FIELDS = (
    ('title', 'string', 'Title', 'A single line at the top of the card; what doesn\'t fit gets an ellipsis. Empty becomes "Notification".', 'Someone is at the door'),
    ('subtitle', 'string', 'Subtitle', 'Explanation under the title, across multiple lines; a line break is fine. Empty is fine.', 'Door 3, back'),
    ('icon', 'string', 'Icon', 'A name from the list, also as mdi:name or as a hex codepoint (F12E6). Unknown or empty gives the warning triangle.', 'doorbell'),
    ('color', 'string', 'Color', 'One of the nine pastel colors of the tiles. Empty gives the white card.', 'orange'),
    ('button_text', 'string', 'Button text', 'The text on the button. Empty is "OK".', 'Coming'),
    ('timeout', 'int', 'Timeout', 'Seconds after which the card disappears on its own. 0 waits for the button, however long that takes. The button always closes it immediately, even with a timeout.', 0),
    ('flash', 'bool', 'Blinking', 'On makes the backlight blink four times when the alert arrives; the screen then just stays on.', True),
)
# Bytes per field the firmware keeps (the profiles' ALERT_*_MAX); an accented letter 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')
# The firmware's MAX_TIMEOUT_SECONDS.
ALERT_MAX_TIMEOUT = 86400
# One alert for every screen (app 0.2.45): an automation fires one of these Home Assistant events and the
# app calls the matching action on each screen that can show it, screens added later included.
BROADCAST_SHOW, BROADCAST_DISMISS = 'esp_screens_show_alert', 'esp_screens_dismiss_alert'
BROADCAST_EVENTS = {BROADCAST_SHOW: 'show_alert', BROADCAST_DISMISS: 'dismiss_alert'}

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,
            'broadcast': {'show': BROADCAST_SHOW, 'dismiss': BROADCAST_DISMISS},
            '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 _whole(value):
    """A whole number from an int, a finite float or a numeric string; None for anything else."""
    if isinstance(value, bool):
        return None
    if isinstance(value, str):
        try:
            value = float(value.strip())
        except ValueError:
            return None
    if isinstance(value, float):
        return int(value) if math.isfinite(value) else None
    return value if isinstance(value, int) else None

def _flag(value):
    """True/False from a bool, 0/1 or the words Home Assistant accepts; None for anything else."""
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)) and value in (0, 1):
        return bool(value)
    text = value.strip().lower() if isinstance(value, str) else None
    return True if text in ('true', 'on', 'yes', '1') else False if text in ('false', 'off', 'no', '0') else None

def alert_data(data):
    """(service data, unusable field names): the seven show_alert fields from an event's data, typed the way
    Home Assistant validates ESPHome actions. A missing field is empty; so is an unusable one (YAML turns a bare
    `Yes` into a boolean), so one bad value never loses the whole alert. The firmware clips texts itself."""
    data = data if isinstance(data, dict) else {}
    service, unusable = {}, []
    for name, kind, *_ in ALERT_FIELDS:
        value = data.get(name)
        missing = value is None or value == ''
        if kind == 'string':
            usable = isinstance(value, (str, int, float)) and not isinstance(value, bool)
            service[name] = str(value) if usable and not missing else ''
        elif kind == 'int':
            number = _whole(value)
            usable = number is not None
            service[name] = min(max(number, 0), ALERT_MAX_TIMEOUT) if usable else 0
        else:
            flag = _flag(value)
            usable = flag is not None
            service[name] = bool(flag)
        if not usable and not missing:
            unusable.append(name)
    return service, unusable

def alert_targets(screens):
    """(ready, skipped): the paired screens that can show an alert now, and the others with the reason.

    One call per device: a screen that shows up twice (an old inbox next to a renamed one) counts once."""
    minimum = tuple(int(part) for part in ALERT_MIN_FIRMWARE.split('.'))
    ready, skipped, nodes = [], [], set()
    for screen in screens:
        try:
            version = tuple(int(part) for part in str(screen.get('firmware') or '').split('.'))
        except ValueError:
            version = ()
        node = screen.get('node')
        if node in nodes:
            continue
        if not node:
            skipped.append((screen, 'device name unknown'))
        elif not screen.get('online'):
            skipped.append((screen, 'offline'))
        elif len(version) != 3 or version < minimum:
            skipped.append((screen, f"firmware {screen.get('firmware') or 'unknown'}"))
        else:
            ready.append(screen)
            nodes.add(node)
    return ready, skipped

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', 'unknown') for item in registry
                if item.get('platform') == 'esphome' and item.get('original_name') in NAME_SCREEN_FIRMWARE}
    boards = {item.get("device_id"): "guition" for item in registry
              if item.get("platform") == "esphome" and item.get("original_name") in NAME_GUITION_TYPE}
    def diagnostic(names, pattern):
        found = {}
        for item in registry:
            if item.get('platform') == 'esphome' and item.get('original_name') in names:
                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(NAME_DEVICE_NAME, r'[a-z0-9][a-z0-9-]{0,30}')
    addresses = diagnostic(NAME_IP_ADDRESS, 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') in NAME_TILE_SETTINGS 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'), 'unknown'),
                        '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', 'Not connected')})
    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('Choose a board and a unique name (lowercase letters, digits, dashes; 30 characters max).')
    if not isinstance(friendly, str) or not friendly.strip() or len(friendly) > 60:
        raise ValueError('Give the screen a recognizable name (60 characters max).')
    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'''# Keep this file safe: it contains the unique keys for this screen.
# Wi-Fi comes from the secrets.yaml of 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:
'''