restructed & updated mqtt control

.gitignore

@@ -0,0 +1 @@
+/venv

control.py

@@ -1,72 +0,0 @@
-from neopixel import NeoPixel
-from random import randrange
-import time
-import board
-from gpiozero import LED
-
-# ------------------------------------------------------------------------------
-# Control script for 3D printer fan & LED control
-#
-#
-# Raspi Zero W Pin assignment, in order from top to bottom on PCB
-#
-#  On PCB  |  Cable color    |  Raspberry PIN
-# --------------------------------------------
-#  5V      |  not connected  |  ------
-#  LED     |  grey           |  GPIO 19 [PWM]
-#  GND     |  black          |  ------
-#  DHT22   |  white          |  GPIO 26
-#  FAN1    |  purple         |  GPIO 13
-#  FAN2    |  blue           |  GPIO 6
-#
-
-SETTINGS = {
-    'fan1': board.D13,
-    'fan2': board.D6,
-    'dht': board.D26,
-    'led': board.D19,
-}
-
-fan1 = LED(SETTINGS['fan1'])
-fan2 = LED(SETTINGS['fan2'])
-
-
-def test_leds(sleep_time=0.1, num_leds_on_at_same_time=10, max_brightness=30):
-    num_leds = 120
-    strip = NeoPixel(SETTINGS['led'], n=num_leds, auto_write=False)
-
-    # switch all leds off
-    for i in range(num_leds):
-        strip[i] = (0, 0, 0)
-    strip.show()
-
-    for i in range(num_leds):
-        strip[i] = tuple(randrange(0, max_brightness) for j in range(3))
-        if i - num_leds_on_at_same_time >= 0:
-            strip[i-num_leds_on_at_same_time] = (0, 0, 0)
-        strip.show()
-        time.sleep(sleep_time)
-
-    # switch all leds off again
-    time.sleep(2)
-    for i in range(num_leds):
-        strip[i] = (0, 0, 0)
-    strip.show()
-
-def board_test(fan_on_secs=4):
-  print("Testing fan1")
-  fan1.on()
-  time.sleep(fan_on_secs)
-  fan1.off()
-
-
-  print("Testing fan2")
-  time.sleep(fan_on_secs)
-  fan2.off()
-
-  print("Testing leds")
-  test_leds()
-
-
-if __name__ == "__main__":
-  board_test()

mqttcontrol/printerbox_hardware.py

@@ -0,0 +1,230 @@
+import random
+from gpiozero import PWMLED
+from rpi_ws281x import Color, PixelStrip, ws
+import time
+import threading
+
+# ------------------------------------------------------------------------------
+# Control script for 3D printer fan & LED control
+#
+#
+# Raspi Zero W Pin assignment, in order from top to bottom on PCB
+#
+#  On PCB  |  Cable color    |  Raspberry PIN
+# --------------------------------------------
+#  5V      |  not connected  |  ------
+#  LED     |  grey           |  GPIO 19 [PWM]
+#  GND     |  black          |  ------
+#  DHT22   |  white          |  GPIO 26
+#  FAN1    |  purple         |  GPIO 13
+#  FAN2    |  blue           |  GPIO 6
+#
+
+
+# LED strip configuration:
+LED_COUNT = 120        # Number of LED pixels.
+LED_PIN = 19           # GPIO pin connected to the pixels (must support PWM!).
+LED_FREQ_HZ = 800000   # LED signal frequency in hertz (usually 800khz)
+LED_DMA = 10           # DMA channel to use for generating signal (try 10)
+LED_BRIGHTNESS = 255   # Set to 0 for darkest and 255 for brightest
+LED_INVERT = False     # True to invert the signal (for NPN transistor)
+LED_CHANNEL = 1
+LED_STRIP = ws.SK6812_STRIP
+LED_SUBSETS = {
+    'top_back': list(range(24)),
+    'top_mid': list(range(24, 48)),
+    'left': list(range(48, 72)),
+    'top_front': list(range(72, 96)),
+    'right': list(reversed(range(96, 120))),
+}
+LED_SUBSETS['top'] = LED_SUBSETS['top_back'] + \
+    LED_SUBSETS['top_mid'] + LED_SUBSETS['top_front']
+LED_SUBSETS['sides'] = LED_SUBSETS['left'] + LED_SUBSETS['right']
+LED_SUBSETS['all'] = LED_SUBSETS['top'] + LED_SUBSETS['sides']
+
+LED_SUBSET_NAMES = {
+    'top_back': "Oben hinten",
+    'top_mid': "Oben mitte",
+    'left': "Seite links",
+    'top_front': "Oben vorne",
+    'right': "Seite rechts",
+    'top': "Oben",
+    'sides': "Seiten",
+    'all': "Alle"
+}
+
+leds = PixelStrip(LED_COUNT, LED_PIN, LED_FREQ_HZ, LED_DMA,
+                  LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL, LED_STRIP)
+leds.begin()
+
+
+# Fan configuration
+PWM_FREQ = 600
+
+venting_fan = PWMLED(13, frequency=PWM_FREQ)
+filter_fan = PWMLED(6, frequency=PWM_FREQ)
+
+#
+led_thread_running = True
+led_thread = None
+
+
+def normalize_subset(subset):
+    if type(subset) is str:
+        return LED_SUBSETS[subset]
+    else:
+        return subset
+
+
+def next_color_func(r, g, b):
+    if type(r) is tuple or type(g) is tuple or type(b) is tuple:
+        def next_color():
+            return Color(*(random.randint(e[0], e[1]) if isinstance(e, tuple) else e for e in (r, g, b)))
+    else:
+        def next_color():
+            return Color(r, g, b)
+    return next_color
+
+
+def _thread_set_leds_color(r, g, b, subset='all'):
+    global led_thread_running
+    led_thread_running = True
+    next_color = next_color_func(r, g, b)
+
+    for i in normalize_subset(subset):
+        leds.setPixelColor(i, next_color())
+    leds.show()
+    led_thread_running = False
+
+
+def _thread_set_leds_brightness(new_brightness):
+    global led_thread_running
+    led_thread_running = True
+
+    leds.setBrightness(new_brightness)
+    leds.show()
+
+    led_thread_running = False
+
+
+def _thread_set_leds_on(subset='all'):
+    global led_thread_running
+    led_thread_running = True
+
+    changed = False
+    for i in normalize_subset(subset):
+        if leds.getPixelColor(i) == 0:
+            leds.setPixelColor(i, Color(180, 180, 180))
+            changed = True
+    if changed:
+        leds.show()
+
+    led_thread_running = False
+
+
+def _thread_set_leds_off(subset='all'):
+    _thread_set_leds_color(0, 0, 0, subset=subset)
+
+
+def _thread_effect_color_wipe(r, g, b, wait_ms=25, subset='all'):
+    """Wipe color across display a pixel at a time."""
+    global led_thread_running
+
+    next_color = next_color_func(r, g, b)
+
+    _thread_set_leds_off(subset)
+    led_thread_running = True
+    while led_thread_running:
+        for i in normalize_subset(subset):
+            leds.setPixelColor(i, next_color())
+            leds.show()
+            if not led_thread_running:
+                break
+            time.sleep(wait_ms / 1000.0)
+        for i in reversed(normalize_subset(subset)):
+            leds.setPixelColor(i, Color(0, 0, 0))
+            leds.show()
+            if not led_thread_running:
+                break
+            time.sleep(wait_ms / 1000.0)
+
+
+def _thread_effect_blink_on_off(r, g, b, wait_ms=1000, subset='all'):
+    global led_thread_running
+    next_color = next_color_func(r, g, b)
+
+    _thread_set_leds_off(subset)
+    led_thread_running = True
+
+    while led_thread_running:
+        for i in normalize_subset(subset):
+            leds.setPixelColor(i, next_color())
+        leds.show()
+        time.sleep(wait_ms / 1000.0)
+
+        if not led_thread_running:
+            break
+
+        for i in normalize_subset(subset):
+            leds.setPixelColor(i, Color(0, 0, 0))
+        leds.show()
+        time.sleep(wait_ms / 1000.0)
+
+
+def _thread_effect_move_blink(r, g, b, wait_ms=500, subset='all'):
+    global led_thread_running
+    next_color = next_color_func(r, g, b)
+
+    _thread_set_leds_off(subset)
+    led_thread_running = True
+
+    ctr = 0
+    while led_thread_running:
+        for nr, led_id in enumerate(normalize_subset(subset)):
+            color = next_color() if nr % 2 == ctr % 2 else Color(0, 0, 0)
+            leds.setPixelColor(led_id, color)
+        leds.show()
+        ctr += 1
+        time.sleep(wait_ms / 1000.0)
+
+
+def _start_in_led_thread(func, *args, **kwargs):
+    global led_thread_running
+    global led_thread
+    led_thread_running = False
+    if led_thread is not None:
+        led_thread.join()
+    led_thread = threading.Thread(target=func, args=args, kwargs=kwargs)
+    led_thread.start()
+
+
+# ---------------------- Settings Lights ---------------------------------
+
+
+def set_leds_color(r, g, b, subset='all'):
+    _start_in_led_thread(_thread_set_leds_color, r, g, b, subset)
+
+
+def set_leds_on(subset='all'):
+    if not led_thread_running:
+        _start_in_led_thread(_thread_set_leds_on, subset)
+
+
+def set_leds_off(subset='all'):
+    _start_in_led_thread(_thread_set_leds_off, subset)
+
+
+def set_leds_brightness(new_brightness):
+    _start_in_led_thread(_thread_set_leds_brightness, new_brightness)
+
+
+def effect_color_wipe(r, g, b, wait_ms=25, subset='all'):
+    _start_in_led_thread(_thread_effect_color_wipe, r, g, b, wait_ms, subset)
+
+
+def effect_move_blink(r, g, b, wait_ms=500, subset='all'):
+    _start_in_led_thread(_thread_effect_move_blink, r, g, b, wait_ms, subset)
+
+
+def effect_blink_on_off(r, g, b, wait_ms=1000, subset='all'):
+    _start_in_led_thread(_thread_effect_blink_on_off, r, g, b, wait_ms, subset)

mqttcontrol/printerbox_mqtt.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import paho.mqtt.client as mqtt
+import json
+from pprint import pprint
+import printerbox_hardware as hw
+
+
+def light_cfg(id, name):
+    return {f'light_{id}': {
+        'entity_type': 'light',
+        'name': f"PrusaBox Licht {name}",
+        'unique_id': f'prusa_caselights_{id}',
+        'command_topic': f'prusa_printer/lights_{id}/switch',
+        'state_topic': f'prusa_printer/lights_{id}/switch_state',
+        'brightness_command_topic': f'prusa_printer/lights_{id}/brightness',
+        'brightness_state_topic': f'prusa_printer/lights_{id}/brightness_state',
+        'rgb_command_topic': f'prusa_printer/lights_{id}/rgb',
+        'rgb_state_topic': f'prusa_printer/lights_{id}/rgb_state',
+        'effect_command_topic': f'prusa_printer/lights_{id}/effect',
+        'effect_state_topic': f'prusa_printer/lights_{id}/effect_state',
+        'effect_list': ['color_wipe', 'blink', 'blink_move'],
+    }}
+
+
+# ----------------------------------- Config ------------------------------------------------------
+
+
+CONFIG = {
+    'mqtt_server': 'homeassistant',
+    'mqtt_user': 'prusaprinter',
+    'mqtt_password': 'ANchibUdeClEnegue',
+    'mqtt_discovery_prefix': 'homeassistant',
+
+    'devices': {
+        'venting_fan': {
+            'entity_type': 'fan',
+            'unique_id': 'prusa_venting_fan',
+            'name': 'PrusaBox Lüfter draussen',
+            'state_topic': 'prusa_printer/venting_fan',
+            'command_topic': 'prusa_printer/venting_fan/set',
+            'percentage_command_topic': 'prusa_printer/venting_fan/percentage',
+            'percentage_state_topic': 'prusa_printer/venting_fan/percentageState',
+            'min_voltage_percentage': 0.35,
+            'icon': 'mdi:fan',
+        },
+        'filter_fan': {
+            'entity_type': 'fan',
+            'unique_id': 'prusa_filter_fan',
+            'name': 'PrusaBox Filter Lüfter',
+            'state_topic': 'prusa_printer/filter_fan',
+            'command_topic': 'prusa_printer/filter_fan/set',
+            'percentage_command_topic': 'prusa_printer/filter_fan/percentage',
+            'percentage_state_topic': 'prusa_printer/filter_fan/percentageState',
+            'min_voltage_percentage': 0.5,
+            'icon': 'mdi:fan',
+        },
+    }
+}
+for id, name in hw.LED_SUBSET_NAMES.items():
+    CONFIG['devices'].update(light_cfg(id, name))
+
+# -----------------------------------------------------------------------------------------------------
+
+
+def create_discovery_msg(config_entry):
+    cfg = config_entry.copy()
+    topic = "{}/{}/{}/config".format(
+        CONFIG['mqtt_discovery_prefix'], cfg['entity_type'], cfg['unique_id'])
+    del cfg['entity_type']
+    return {'topic': topic, 'payload': json.dumps(cfg), 'retain': False}
+
+
+def send_autodiscovery_messages(client):
+    client.publish(**create_discovery_msg(CONFIG['devices']['venting_fan']))
+    client.publish(**create_discovery_msg(CONFIG['devices']['filter_fan']))
+    for id in hw.LED_SUBSET_NAMES.keys():
+        client.publish(
+            **create_discovery_msg(CONFIG['devices']['light_' + id]))
+
+
+def handle_light_message(client, msg):
+    prefix = 'prusa_printer/lights_'
+    if not msg.topic.startswith(prefix):
+        return False
+    id, cmd = msg.topic[len(prefix):].split("/")
+    payload = msg.payload.decode()
+    cfg = CONFIG['devices']['light_' + id]
+
+    if cmd == "rgb":
+        r, g, b = tuple(int(i) for i in payload.split(","))
+        hw.set_leds_color(r, g, b, subset=id)
+        client.publish(cfg['rgb_state_topic'], payload)
+    elif cmd == "switch":
+        if payload == "ON":
+            hw.set_leds_on(subset=id)
+        elif payload == "OFF":
+            hw.set_leds_off(subset=id)
+        client.publish(cfg['state_topic'], payload)
+    elif cmd == "brightness":
+        hw.set_leds_brightness(int(payload))
+        client.publish(cfg['brightness_state_topic'], payload)
+    elif cmd == "effect":
+        if payload == "color_wipe":
+            hw.effect_color_wipe(0, 0, 255)
+        elif payload == "blink":
+            print("starting blink effect")
+            hw.effect_blink_on_off(255, 0, 0)
+            print("finished starting blink effect")
+        elif payload == "blink_move":
+            hw.effect_move_blink(255, 0, 0)
+        client.publish(cfg['effect_state_topic'], payload)
+    
+def handle_fan_message(client, msg):
+    topic = msg.topic
+    payload = msg.payload.decode()
+
+    for cfg, fan in [(CONFIG['devices']['venting_fan'], hw.venting_fan),
+                     (CONFIG['devices']['filter_fan'], hw.filter_fan)]:
+        if topic == cfg['command_topic']:
+            fan.value = 1 if payload == b"ON" else 0
+            client.publish(cfg['state_topic'], payload)
+            return True
+        elif topic == cfg['percentage_command_topic']:
+            #print("setting percentage, payload = ", payload)
+            fan_percentage = float(payload) / 100
+            min_perc = cfg['min_voltage_percentage']
+            voltage_percentage = min_perc + (1 - min_perc) * fan_percentage
+            if fan_percentage == 0:
+                voltage_percentage = 0
+            #print("setting voltage percentage ", voltage_percentage)
+            fan.value = voltage_percentage
+            client.publish(cfg['percentage_state_topic'], payload)
+            return True
+
+    return False
+
+
+def on_mqtt_message(client, userdata, msg):
+    print(msg.topic, msg.payload)
+    handle_fan_message(client, msg)
+    handle_light_message(client, msg)
+
+
+def on_mqtt_connect(client, *args, **kwargs):
+    subscriptions = []
+    for device in CONFIG['devices'].values():
+        for key, value in device.items():
+            if key.endswith('_topic'):
+                subscriptions.append((value, 2))
+    client.subscribe(subscriptions)
+
+
+def run():
+    client = mqtt.Client("prusa-printer-client")
+    client.username_pw_set(CONFIG['mqtt_user'], CONFIG['mqtt_password'])
+    client.on_connect = on_mqtt_connect
+    client.on_message = on_mqtt_message
+    client.connect(CONFIG['mqtt_server'])
+    send_autodiscovery_messages(client)
+
+    client.loop_forever()
+
+
+if __name__ == "__main__":
+    run()