updates
This commit is contained in:
Joshua King
@@ -0,0 +1,235 @@
|
||||
#!/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',
|
||||
}
|
||||
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 = (
|
||||
0x0D131B, 0x17222E, 0x263442, 0xEDF3F7, 0x91A3B5, 0x60DE9A)
|
||||
|
||||
|
||||
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')
|
||||
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 card(pid, name, title, x, y, width=238, height=104, font=26):
|
||||
return {'obj': {'x': x, 'y': y, 'width': width, 'height': height,
|
||||
'bg_color': CARD, 'radius': 16, 'border_width': 0,
|
||||
'pad_all': 0, 'scrollable': False, 'gesture_bubble': True,
|
||||
'widgets': [
|
||||
label(title, 16, 12, width - 32, 14, text_color=MUTED),
|
||||
label('--', 16, 38, width - 32, font, id=f'{pid}_{name}_label', height=height - 42),
|
||||
]}}
|
||||
|
||||
|
||||
def page_for(printer, index, total):
|
||||
pid = printer['id']
|
||||
widgets = [
|
||||
label(printer['name'], 24, 16, 545, 28),
|
||||
label(printer['subtitle'], 24, 53, 545, 14, text_color=MUTED),
|
||||
{'obj': {'x': 586, 'y': 20, 'width': 190, 'height': 42,
|
||||
'bg_color': CARD, 'radius': 21, 'border_width': 0, 'pad_all': 0,
|
||||
'scrollable': False, 'gesture_bubble': True,
|
||||
'widgets': [label('SYNCING', 0, 11, 190, 16,
|
||||
id=f'{pid}_status_label', text_align='CENTER', text_color=MUTED)]}},
|
||||
label('Waiting for Home Assistant', 24, 88, 752, 20, id=f'{pid}_task_label'),
|
||||
{'arc': {'id': f'{pid}_progress_arc', 'x': 28, 'y': 132, 'width': 224, 'height': 224,
|
||||
'min_value': 0, 'max_value': 100, 'value': 0,
|
||||
'start_angle': 135, 'end_angle': 45, 'arc_width': 14,
|
||||
'arc_color': TRACK, 'indicator': {'arc_color': GREEN, 'arc_width': 14},
|
||||
'adjustable': False, 'gesture_bubble': True}},
|
||||
label('--', 42, 207, 196, 48, id=f'{pid}_progress_label', text_align='CENTER'),
|
||||
label('PRINT PROGRESS', 42, 266, 196, 14, text_align='CENTER', text_color=MUTED),
|
||||
card(pid, 'nozzle', 'NOZZLES L/R / TARGET' if printer.get('dual_nozzle')
|
||||
else 'NOZZLE / TARGET', 282, 132, height=118,
|
||||
font=18 if printer.get('dual_nozzle') else 26),
|
||||
card(pid, 'bed', 'BED / TARGET', 538, 132, height=118),
|
||||
card(pid, 'remaining', 'REMAINING', 282, 262, height=86),
|
||||
card(pid, 'layers', 'LAYER', 538, 262, height=86),
|
||||
label('Telemetry unavailable', 24, 358, 752, 16, id=f'{pid}_stage_label', text_color=MUTED),
|
||||
label('FILAMENT / ACTIVE SLOT OUTLINED IN GREEN', 24, 383, 752, 12, text_color=MUTED),
|
||||
]
|
||||
trays = trays_for(printer)
|
||||
if trays:
|
||||
width = (752 - 8 * (len(trays) - 1)) // len(trays)
|
||||
for i, tray in enumerate(trays):
|
||||
widgets.append({'obj': {
|
||||
'id': f'{pid}_tray_{i}_card', 'x': 24 + (width + 8) * i, 'y': 401,
|
||||
'width': width, 'height': 32, 'bg_color': CARD, 'radius': 6,
|
||||
'border_width': 1, 'border_color': TRACK, 'pad_all': 0,
|
||||
'scrollable': False, 'gesture_bubble': True,
|
||||
'widgets': [
|
||||
{'obj': {'id': f'{pid}_tray_{i}_swatch', 'x': 1, 'y': 1, 'width': width - 4,
|
||||
'height': 5, 'bg_color': TRACK, 'radius': 2, 'border_width': 0,
|
||||
'pad_all': 0, 'scrollable': False, 'gesture_bubble': True}},
|
||||
label(f'{tray["label"]} --', 6, 11, width - 12, 12,
|
||||
id=f'{pid}_tray_{i}_label'),
|
||||
]}})
|
||||
else:
|
||||
widgets.append(label('No filament slots configured', 24, 403, 752, 14, text_color=MUTED))
|
||||
for direction, x, text in (('previous', 24, '< PREV'), ('next', 688, 'NEXT >')):
|
||||
widgets.append({'button': {'x': x, 'y': 444, 'width': 88, 'height': 30,
|
||||
'bg_color': CARD, 'radius': 8, 'border_width': 0,
|
||||
'pad_all': 0, 'gesture_bubble': True,
|
||||
'widgets': [{'label': {'text': text, 'align': 'CENTER',
|
||||
'text_font': 'montserrat_12'}}],
|
||||
'on_click': [{f'lvgl.page.{direction}': {}}]}})
|
||||
widgets.append(label(f'{index + 1} / {total} SWIPE LEFT / RIGHT', 130, 451, 540, 14,
|
||||
text_align='CENTER', text_color=MUTED))
|
||||
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 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)
|
||||
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;')
|
||||
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', 'progress_arc', 'progress_label', 'nozzle_label',
|
||||
'bed_label', 'remaining_label', 'layers_label', 'stage_label')]
|
||||
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)))
|
||||
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()
|
||||
Reference in New Issue
Block a user