#!/usr/bin/env python3 """Generate the ESPHome fleet package from printers.yaml. Requires PyYAML.""" import argparse import json from pathlib import Path import re import yaml HERE = Path(__file__).resolve().parent NUMERIC = { 'progress': 'print_progress', 'nozzle': 'nozzle_temperature', 'nozzle_target': 'nozzle_target_temperature', 'bed': 'bed_temperature', 'bed_target': 'bed_target_temperature', 'time_left': 'remaining_time', 'layer': 'current_layer', 'total_layers': 'total_layer_count', 'fan': 'cooling_fan_speed', } DUAL = { 'left_nozzle': 'left_nozzle_temperature', 'left_target': 'left_nozzle_target_temperature', 'right_nozzle': 'right_nozzle_temperature', 'right_target': 'right_nozzle_target_temperature', } EXTRA_FANS = {'aux_fan': 'aux_fan_speed', 'exhaust_fan': 'chamber_fan_speed'} TEXT = {'status': 'print_status', 'task': 'task_name', 'stage': 'current_stage'} BINARY = {'online': 'online', 'print_error': 'print_error', 'hms_error': 'hms_errors'} BG, CARD, TRACK, TEXT_COLOR, MUTED, GREEN = ( 0x000000, 0x000000, 0x19232B, 0xEEEEEE, 0xAEB3B7, 0x00FF00) ORANGE, CYAN = 0xFFAA00, 0x00E5E5 class Lambda(str): """A C++ lambda serialized with ESPHome's YAML tag.""" class Dumper(yaml.SafeDumper): def ignore_aliases(self, data): return True Dumper.add_representer(Lambda, lambda dumper, text: dumper.represent_scalar('!lambda', text, style='|')) def entity(printer, key, suffix, domain='sensor'): return printer.get('entities', {}).get(key, f'{domain}.{printer["entity_prefix"]}_{suffix}') def trays_for(printer): result = [] for bank in printer.get('filament_banks', []): for slot in range(1, 5): result.append({'label': f'{bank["label"]}{slot}', 'entity': f'sensor.{bank["entity_prefix"]}_tray_{slot}'}) return result + printer.get('external_spools', []) def validate_printers(printers): if not isinstance(printers, list) or not printers: raise ValueError('printers must be a nonempty list') ids = set() for printer in printers: for key in ('id', 'name', 'subtitle', 'entity_prefix'): if not isinstance(printer.get(key), str) or not printer[key]: raise ValueError(f'Each printer needs a nonempty {key}') pid = printer['id'] if not re.fullmatch(r'[a-z][a-z0-9_]*', pid) or pid in ids: raise ValueError(f'Invalid or duplicate printer id: {pid}') ids.add(pid) if len(trays_for(printer)) > 8: raise ValueError(f'{pid}: the filament strip fits up to eight slots') if 'dual_nozzle' in printer and not isinstance(printer['dual_nozzle'], bool): raise ValueError(f'{pid}: dual_nozzle must be true or false') if any(key not in EXTRA_FANS for key in printer.get('extra_fans', [])): raise ValueError(f'{pid}: extra_fans supports aux_fan and exhaust_fan') for value in printer.get('entities', {}).values(): if not isinstance(value, str) or not re.fullmatch(r'(sensor|binary_sensor)\.[a-z0-9_]+', value): raise ValueError(f'{pid}: invalid entity override: {value}') def label(text, x, y, width, font=18, **kwargs): return {'label': {'text': text, 'x': x, 'y': y, 'width': width, 'height': font + 8, 'text_color': TEXT_COLOR, 'text_font': f'montserrat_{font}', 'long_mode': 'DOT', 'gesture_bubble': True, **kwargs}} def gauge(pid, key, title, x, y, color, maximum): return [ {'arc': {'id': f'{pid}_{key}_arc', 'x': x + 8, 'y': y, 'width': 154, 'height': 154, 'min_value': 0, 'max_value': maximum, 'value': 0, 'start_angle': 135, 'end_angle': 45, 'arc_width': 13, 'arc_color': TRACK, 'indicator': {'arc_color': color, 'arc_width': 13}, 'adjustable': False, 'gesture_bubble': True}}, label('--', x + 14, y + 47, 142, 32, id=f'{pid}_{key}_label', text_align='CENTER'), label('--', x + 14, y + 86, 142, 16, id=f'{pid}_{key}_detail', text_align='CENTER', text_color=MUTED), label(title, x, y + 137, 170, 18, text_align='CENTER', text_color=color), ] def page_for(printer, index, total): pid = printer['id'] widgets = [ {'bar': {'id': f'{pid}_top_progress', 'x': 0, 'y': 0, 'width': 800, 'height': 5, 'min_value': 0, 'max_value': 100, 'value': 0, 'bg_color': TRACK, 'indicator': {'bg_color': GREEN}, 'gesture_bubble': True}}, label(printer['name'], 16, 17, 485, 26, text_color=GREEN), label(f'{index + 1} / {total}', 520, 24, 70, 14, text_color=MUTED, text_align='CENTER'), label('SYNCING', 598, 21, 186, 16, id=f'{pid}_status_label', text_align='RIGHT', text_color=MUTED), {'obj': {'x': 578, 'y': 64, 'width': 1, 'height': 362, 'bg_color': TRACK, 'border_width': 0, 'pad_all': 0, 'scrollable': False, 'gesture_bubble': True}}, ] widgets += gauge(pid, 'progress', 'Progress', 16, 62, GREEN, 100) widgets += gauge(pid, 'nozzle', 'Nozzle L' if printer.get('dual_nozzle') else 'Nozzle', 202, 62, ORANGE, 300) widgets += gauge(pid, 'bed', 'Bed', 388, 62, CYAN, 120) widgets += gauge(pid, 'fan', 'Part', 16, 229, 0xEEEEEE, 100) widgets += gauge(pid, 'aux', 'Nozzle R' if printer.get('dual_nozzle') else 'Aux', 202, 229, ORANGE if printer.get('dual_nozzle') else 0xEEEEEE, 300 if printer.get('dual_nozzle') else 100) widgets += gauge(pid, 'exhaust', 'Exhaust', 388, 229, GREEN, 100) widgets += [ label('--', 16, 401, 544, 26, id=f'{pid}_remaining_label', text_color=GREEN, text_align='CENTER'), label('Telemetry unavailable', 16, 430, 544, 12, id=f'{pid}_stage_label', text_color=MUTED, text_align='CENTER'), label('--', 594, 397, 190, 16, id=f'{pid}_layers_label', text_color=MUTED, text_align='CENTER'), ] trays = trays_for(printer) for i, tray in enumerate(trays): bank = i // 4 slot = i % 4 if slot == 0: title = 'AMS ' + tray['label'][:-1] if tray['label'][-1:].isdigit() else 'EXTERNAL SPOOL' widgets.append(label(title, 594, 70 + bank * 145, 190, 20, text_align='CENTER', text_color=MUTED)) widgets.append({'obj': { 'id': f'{pid}_tray_{i}_card', 'x': 594 + 48 * slot, 'y': 102 + bank * 145, 'width': 44, 'height': 82, 'bg_color': BG, 'radius': 5, 'border_width': 1, 'border_color': TRACK, 'pad_all': 0, 'scrollable': False, 'gesture_bubble': True, 'widgets': [ {'obj': {'id': f'{pid}_tray_{i}_swatch', 'x': 3, 'y': 3, 'width': 36, 'height': 40, 'bg_color': TRACK, 'radius': 4, 'border_width': 0, 'pad_all': 0, 'scrollable': False, 'gesture_bubble': True}}, label(f'{tray["label"]}\n--', 1, 46, 40, 12, height=34, id=f'{pid}_tray_{i}_label', text_align='CENTER'), ]}}) if not trays: widgets.append(label('No AMS', 594, 150, 190, 20, text_color=MUTED, text_align='CENTER')) for direction, x, text in (('previous', 16, '<'), ('next', 720, '>')): widgets.append({'button': {'x': x, 'y': 446, 'width': 64, 'height': 30, 'bg_color': TRACK, 'radius': 8, 'border_width': 0, 'pad_all': 0, 'gesture_bubble': True, 'widgets': [{'label': {'text': text, 'align': 'CENTER', 'text_color': TEXT_COLOR, 'text_font': 'montserrat_18'}}], 'on_click': [{f'lvgl.page.{direction}': {}}]}}) widgets.append(label('Waiting for Home Assistant', 94, 449, 612, 14, id=f'{pid}_task_label', text_align='CENTER')) return {'id': f'{pid}_page', 'bg_color': BG, 'bg_opa': 'COVER', 'pad_all': 0, 'scrollable': False, 'scrollbar_mode': 'OFF', 'on_swipe_left': [ {'lambda': Lambda('if (auto *input = lv_indev_active()) lv_indev_wait_release(input);')}, {'lvgl.page.next': {'animation': 'MOVE_LEFT', 'time': '180ms'}}], 'on_swipe_right': [ {'lambda': Lambda('if (auto *input = lv_indev_active()) lv_indev_wait_release(input);')}, {'lvgl.page.previous': {'animation': 'MOVE_RIGHT', 'time': '180ms'}}], 'on_load': [{'script.execute': f'{pid}_refresh'}], 'widgets': widgets} def color_test_page(): widgets = [label('DISPLAY COLOR TEST', 16, 18, 768, 28), label('Each block must match its label. Swipe to return to the printers.', 16, 62, 768, 18)] colors = [('RED', 0xFF0000), ('GREEN', 0x00FF00), ('BLUE', 0x0000FF), ('WHITE', 0xFFFFFF), ('CYAN', 0x00FFFF), ('MAGENTA', 0xFF00FF)] for i, (name, color) in enumerate(colors): widgets += [ {'obj': {'x': 16 + i * 129, 'y': 118, 'width': 122, 'height': 200, 'bg_color': color, 'bg_opa': 'COVER', 'border_width': 0, 'scrollable': False, 'gesture_bubble': True}}, label(name, 16 + i * 129, 332, 122, 18, text_align='CENTER'), ] return {'id': 'color_test_page', 'skip': True, 'bg_color': BG, 'bg_opa': 'COVER', 'pad_all': 0, 'scrollable': False, 'on_swipe_left': [{'lvgl.page.next': {}}], 'on_swipe_right': [{'lvgl.page.previous': {}}], 'widgets': widgets} def generate(printers): validate_printers(printers) package = {'sensor': [], 'text_sensor': [], 'script': [], 'interval': [{'interval': '1s', 'then': []}], 'lvgl': {'pages': []}} for index, printer in enumerate(printers): pid = printer['id'] fields = dict(NUMERIC) if printer.get('dual_nozzle'): fields.update(DUAL) fields.update({key: EXTRA_FANS[key] for key in printer.get('extra_fans', [])}) expressions = {} for key, suffix in fields.items(): sid = f'{pid}_{key}' eid = entity(printer, key, suffix) package['sensor'].append({'platform': 'homeassistant', 'id': sid, 'entity_id': eid, 'internal': True}) expressions[key] = f'id({sid}).state' text_fields = {key: (entity(printer, key, suffix), None) for key, suffix in TEXT.items()} text_fields.update({key: (entity(printer, key, suffix, 'binary_sensor'), None) for key, suffix in BINARY.items()}) for key, numeric_key in [('remaining_unit', 'time_left'), ('nozzle_unit', 'nozzle'), ('bed_unit', 'bed')]: text_fields[key] = (entity(printer, numeric_key, NUMERIC[numeric_key]), 'unit_of_measurement') trays = trays_for(printer) for i, tray in enumerate(trays): for key, attr in [('type', 'type'), ('color', 'color'), ('active', 'active')]: text_fields[f'tray_{i}_{key}'] = (tray['entity'], attr) for key, (eid, attribute) in text_fields.items(): sid = f'{pid}_{key}' subscription = {'platform': 'homeassistant', 'id': sid, 'entity_id': eid, 'internal': True} if attribute: subscription['attribute'] = attribute package['text_sensor'].append(subscription) expressions[key] = f'id({sid}).state' lines = ['bambu_display::Snapshot data;', 'data.connected = id(ha_api).is_connected_with_state_subscription();'] for key, expr in expressions.items(): if not key.startswith('tray_'): lines.append(f'data.{key} = {expr};') if printer.get('dual_nozzle'): lines.append('data.dual_nozzle = true;') lines += [ 'const int seconds_left = bambu_display::duration_seconds(data.time_left, data.remaining_unit);', 'const auto clock = id(ha_time).now();', 'if (clock.is_valid() && seconds_left >= 0)', ' data.finish_time = esphome::ESPTime::from_epoch_local(clock.timestamp + seconds_left).strftime("%H:%M");', ] tray_widgets = [] for i, tray in enumerate(trays): lines.append('data.trays.push_back({' + ', '.join([ json.dumps(tray['label']), expressions[f'tray_{i}_type'], expressions[f'tray_{i}_color'], expressions[f'tray_{i}_active']]) + '});') tray_widgets.append('{' + ', '.join(f'id({pid}_tray_{i}_{part})' for part in ('card', 'swatch', 'label')) + '}') refs = [f'id({pid}_{key})' for key in ( 'status_label', 'task_label', 'top_progress', 'remaining_label', 'layers_label', 'stage_label')] for key in ('progress', 'nozzle', 'bed', 'fan', 'aux', 'exhaust'): refs.append('{' + ', '.join(f'id({pid}_{key}_{part})' for part in ('arc', 'label', 'detail')) + '}') refs.append('{' + ', '.join(tray_widgets) + '}') lines.append('bambu_display::render(data, {' + ', '.join(refs) + '});') package['script'].append({'id': f'{pid}_refresh', 'then': [{'lambda': Lambda('\n'.join(lines))}]}) package['interval'][0]['then'].append({'if': { 'condition': {'lvgl.page.is_showing': f'{pid}_page'}, 'then': [{'script.execute': f'{pid}_refresh'}]}}) package['lvgl']['pages'].append(page_for(printer, index, len(printers))) package['lvgl']['pages'].append(color_test_page()) package['button'] = [{'platform': 'template', 'name': 'Display Color Test', 'entity_category': 'diagnostic', 'on_press': [{'lvgl.page.show': 'color_test_page'}]}] return package def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--check', action='store_true', help='Fail if generated.yaml is stale') args = parser.parse_args() printers = yaml.safe_load((HERE / 'printers.yaml').read_text())['printers'] package = generate(printers) content = ('# AUTO-GENERATED by bambu-display/generate.py. Edit printers.yaml or generate.py.\n' + yaml.dump(package, Dumper=Dumper, sort_keys=False, allow_unicode=True, width=110)) destination = HERE / 'generated.yaml' if args.check: if not destination.exists() or destination.read_text() != content: raise SystemExit('generated.yaml is stale; run generate.py') print(f'Generated configuration is current ({len(printers)} printer pages).') else: destination.write_text(content) print(f'Generated {destination.name} with {len(printers)} printer pages.') if __name__ == '__main__': main()