from base64 import b16decode import time from datetime import datetime from random import * from ctypes.wintypes import VARIANT_BOOL from multiprocessing.spawn import old_main_modules import os from unicodedata import decimal import requests current_path = os.path.dirname(os.path.abspath(__file__)) try: from rpi_ws281x import * import RPi.GPIO as GPIO import paho.mqtt.client as mqtt import spotipy import urllib.request import colorgram import numpy as np from PIL import Image from dotenv import load_dotenv except ImportError: print('Some modules are missing. Try to install them with "pip3 install -r requirements.txt"') exit() # LED strip configuration: LED_COUNT = 60 # Number of LED pixels. LED_PIN = 18 # GPIO pin connected to the pixels (18 uses PWM!). #LED_PIN = 10 # GPIO pin connected to the pixels (10 uses SPI /dev/spidev0.0). 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 (when using NPN transistor level shift) LED_CHANNEL = 0 # set to '1' for GPIOs 13, 19, 41, 45 or 53 global animation_r animation_r = 255 global animation_g animation_g = 0 global animation_b animation_b = 0 def debug_print(message: str): now = datetime.now() current_time = now.strftime("%H:%M:%S") print('[DEBUG]['+current_time+'] '+message) def convertColor(hexCode): R = int(hexCode[:2],16) G = int(hexCode[2:4],16) B = int(hexCode[4:6],16) total = R + G + B if R == 0: firstPos = 0 else: firstPos = R / total if G == 0: secondPos = 0 else: secondPos = G / total return [firstPos, secondPos] def getRandomHex(): return "%06x" % random.randint(0, 0xFFFFFF) def extract_colors(url, n=4): urllib.request.urlretrieve(url, current_path+r"/cover.jpg") img = current_path+r"/cover.jpg" return colorgram.extract(img, n) def wheel(pos): if pos < 85: return Color(pos * 3, 255 - pos * 3, 0) elif pos < 170: pos -= 85 return Color(255 - pos * 3, 0, pos * 3) else: pos -= 170 return Color(0, pos * 3, 255 - pos * 3) def rainbowCycle(strip, wait_ms=5, iterations=10): for j in range(256*iterations): if mode > 0: break elif power == "False": break for i in range(strip.numPixels()): strip.setPixelColor(i, wheel((int(i * 256 / strip.numPixels()) + j) & 255)) strip.show() time.sleep(wait_ms/1000.0) def setColor(strip, color, wait_ms=10): for i in range(strip.numPixels()): strip.setPixelColor(i, color) strip.show() time.sleep(float(0.01)) time.sleep(wait_ms/1000.0) def strobe(strip, color): for i in range(strip.numPixels()): strip.setPixelColor(i, color) strip.show() time.sleep(float(0.05)) for i in range(strip.numPixels()): strip.setPixelColor(i, Color(0, 0, 0)) strip.show() time.sleep(float(0.05)) def randomColorPerLED(strip): for i in range(strip.numPixels()): strip.setPixelColor(i, Color(randint(0, 255), randint(0, 255), randint(0, 255))) strip.show() time.sleep(1) def randomColor(strip): color = Color(randint(0, 255), randint(0, 255), randint(0, 255)) for i in range(strip.numPixels()): strip.setPixelColor(i, color) strip.show() time.sleep(1) def mqtt_on_connect(client, userdata, flags, rc): client.subscribe("server-rack-led/power") client.subscribe("server-rack-led/mode") client.subscribe("server-rack-led/rgb") if __name__ == '__main__': def mqtt_on_message(client, userdata, msg): print("topic: " ,str(msg.topic)) print("payload: " ,str(msg.payload.decode("utf-8"))) topic = str(msg.topic) payload = str(msg.payload.decode("utf-8")) if topic == "server-rack-led/power": global power power = payload elif topic == "server-rack-led/rgb": splitted_payload = payload.split(",") global r r = str(splitted_payload[0]) global g g = str(splitted_payload[1]) global b b = str(splitted_payload[2]) elif topic == "server-rack-led/mode": if payload == "Rainbow": global mode mode = 0 elif payload == "Custom Color": mode = 1 elif payload == "Strobe": mode = 2 elif payload == "Random Color": mode = 3 elif payload == "Random Color per LED": mode = 4 if os.path.isfile(current_path+r'/.env') == False: # If not create it and add content with open(current_path+r'/.env', 'w') as f: var_lines = ["SPOTIPY_CLIENT_ID = ", "SPOTIPY_CLIENT_SECRET = ", "SPOTIPY_USER_NAME = ", "SPOTIPY_REDIRECT_URI = "] for var_line in var_lines: f.write(var_line+'\n') print('Please fill out the variables in the .env file ( '+current_path+r'/.env )') exit() else: load_dotenv() strip = Adafruit_NeoPixel(LED_COUNT, LED_PIN, LED_FREQ_HZ, LED_DMA, LED_INVERT, LED_BRIGHTNESS, LED_CHANNEL) strip.begin() mode = 0 power = "True" global r r = 255 global g g = 255 global b b = 255 mqttBroker ="homeassistant.ping-mee.local" client = mqtt.Client("server_rack_led") client.username_pw_set("mqtt", "pmMQTT_11!") debug_print("Connecting to MQTT Broker "+str(mqttBroker)) client.connect(mqttBroker) client.on_connect = mqtt_on_connect client.on_message = mqtt_on_message client.publish("server-rack-led/power", "True") client.publish("server-rack-led/mode", "Rainbow") try: client.loop_start() while True: if power == "True": if mode == 0: rainbowCycle(strip) elif mode == 1: setColor(strip, Color(int(r), int(g), int(b))) elif mode == 2: strobe(strip, Color(int(r), int(g), int(b))) elif mode == 3: randomColor(strip) elif mode == 4: randomColorPerLED(strip) else: setColor(strip, Color(0,0,0)) except KeyboardInterrupt: client.publish("server-rack-led/power", "False") client.loop_stop() setColor(strip, Color(0,0,0)) GPIO.cleanup()