ebbe.bass
/
PiXelTubes
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

testing minified code 

Signed-off-by: Ebbe Baß <ebbe.bass>
main
Ebbe Baß 2024-02-18 16:58:44 +01:00
parent caa93cbabf
commit c5734959c4
1 changed files with 36 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
client/main.py
View File

@ -1,98 +1,43 @@
import subprocess _D='message'
import neopixel _C='success'
import requests _B='192.168.0.1'
import time _A=None
import paho.mqtt.client as mqtt import subprocess,neopixel,requests,time,paho.mqtt.client as mqtt
from getmac import get_mac_address from getmac import get_mac_address
import board import board
SERVER_IP=_B
SERVER_IP = '192.168.0.1' SERVER_PORT=5000
SERVER_PORT = 5000 wlan_mac_address=str(get_mac_address(interface='wlan0'))
LED_STRIP_PIN=board.D18
# Dynamically obtain the MAC address of the WLAN interface
wlan_mac_address = str(get_mac_address(interface="wlan0"))
# Replace with the GPIO pin connected to the data input of the WS2812B LED strip
LED_STRIP_PIN = board.D18
global LED_COUNT global LED_COUNT
LED_COUNT = 30 LED_COUNT=30
global LEDS_PER_PIXEL global LEDS_PER_PIXEL
LEDS_PER_PIXEL = 5 LEDS_PER_PIXEL=5
# Global variables for LED strip control
global strip global strip
strip = neopixel.NeoPixel(pin = board.D18, n = LED_COUNT, auto_write = True) strip=neopixel.NeoPixel(pin=board.D18,n=LED_COUNT,auto_write=True)
def register_tube(): def register_tube():
# Register or reauthenticate the tube with the server
try: try:
response = requests.post(f'http://{SERVER_IP}:{SERVER_PORT}/register_tube', data={'mac_address': wlan_mac_address}) response=requests.post(f"http://{SERVER_IP}:{SERVER_PORT}/register_tube",data={'mac_address':wlan_mac_address});print(response);data=response.json()
print(response) if data.get(_C):print('Tube registered successfully.')
data = response.json() else:print(f"Registration failed: {data.get(_D)}")
if data.get('success'): except requests.RequestException as e:print(f"Registration failed: {e}")
print('Tube registered successfully.')
else:
print(f'Registration failed: {data.get("message")}')
except requests.RequestException as e:
print(f'Registration failed: {e}')
def get_assigned_params(): def get_assigned_params():
try: try:
response = requests.get(f'http://{SERVER_IP}:{SERVER_PORT}/get_assigned_params/{wlan_mac_address}') response=requests.get(f"http://{SERVER_IP}:{SERVER_PORT}/get_assigned_params/{wlan_mac_address}");data=response.json()
data = response.json() if data.get(_C):return data.get('universe'),data.get('dmx_address')
if data.get('success'): else:print(f"Failed to fetch assigned parameters: {data.get(_D)}");return _A,_A
return data.get('universe'), data.get('dmx_address') except requests.RequestException as e:print(f"Failed to fetch assigned parameters: {e}");return _A,_A
else: def is_connected_to_wifi():output=subprocess.check_output(['iwgetid']).decode();return output.split('"')[1]
print(f'Failed to fetch assigned parameters: {data.get("message")}') def update_led_strip(rgb_values,pixel,strip):strip[int(pixel)]=rgb_values
return None, None def on_message(mqttc,obj,msg):global rgb_values_list;rgb_values_list=eval(msg.payload.decode())
except requests.RequestException as e: if __name__=='__main__':
print(f'Failed to fetch assigned parameters: {e}') if is_connected_to_wifi()is not _A:
return None, None register_tube();time.sleep(1);mqttc=mqtt.Client(mqtt.CallbackAPIVersion.VERSION2);mqttc.connect(_B,1883,60);mqttc.on_message=on_message;mqttc.subscribe('tube-'+str(wlan_mac_address)+'/pixel_colors',0);global rgb_values_list;rgb_values_list=eval("['[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]']");mqttc.loop_start()
def is_connected_to_wifi():
output = subprocess.check_output(['iwgetid']).decode()
return output.split('"')[1]
def update_led_strip(rgb_values, pixel, strip):
strip[int(pixel)] = rgb_values
def on_message(mqttc, obj, msg):
global rgb_values_list
rgb_values_list = eval(msg.payload.decode())
if __name__ == "__main__":
# Connect to Wi-Fi
if is_connected_to_wifi() is not None:
# Register/reauthenticate the tube
register_tube()
time.sleep(1)
mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.connect("192.168.0.1", 1883, 60)
mqttc.on_message = on_message
mqttc.subscribe("tube-"+str(wlan_mac_address)+"/pixel_colors", 0)
global rgb_values_list
rgb_values_list = eval("['[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]', '[0, 0, 0]']")
mqttc.loop_start()
while True: while True:
for pixel in range(LEDS_PER_PIXEL): for pixel in range(LEDS_PER_PIXEL):update_led_strip(tuple(eval(rgb_values_list[0])),pixel,strip)
update_led_strip(tuple(eval(rgb_values_list[0])), pixel, strip) for pixel in range(LEDS_PER_PIXEL,LEDS_PER_PIXEL*2):update_led_strip(tuple(eval(rgb_values_list[1])),pixel,strip)
for pixel in range(LEDS_PER_PIXEL*2,LEDS_PER_PIXEL*3):update_led_strip(tuple(eval(rgb_values_list[2])),pixel,strip)
for pixel in range(LEDS_PER_PIXEL, LEDS_PER_PIXEL*2): for pixel in range(LEDS_PER_PIXEL*3,LEDS_PER_PIXEL*4):update_led_strip(tuple(eval(rgb_values_list[3])),pixel,strip)
update_led_strip(tuple(eval(rgb_values_list[1])), pixel, strip) for pixel in range(LEDS_PER_PIXEL*4,LEDS_PER_PIXEL*5):update_led_strip(tuple(eval(rgb_values_list[4])),pixel,strip)
for pixel in range(LEDS_PER_PIXEL*5,LEDS_PER_PIXEL*6):update_led_strip(tuple(eval(rgb_values_list[5])),pixel,strip)
for pixel in range(LEDS_PER_PIXEL*2, LEDS_PER_PIXEL*3): time.sleep(.5)
update_led_strip(tuple(eval(rgb_values_list[2])), pixel, strip)
for pixel in range(LEDS_PER_PIXEL*3, LEDS_PER_PIXEL*4):
update_led_strip(tuple(eval(rgb_values_list[3])), pixel, strip)
for pixel in range(LEDS_PER_PIXEL*4, LEDS_PER_PIXEL*5):
update_led_strip(tuple(eval(rgb_values_list[4])), pixel, strip)
for pixel in range(LEDS_PER_PIXEL*5, LEDS_PER_PIXEL*6):
update_led_strip(tuple(eval(rgb_values_list[5])), pixel, strip)
time.sleep(0.5)