Ich habe mal vor einiger Zeit angefangen das Nodle mit Python anzusteuern für ein Testprogramm, das wir intern verwenden wollten.
Ist aber unfertig und aus diversen Gründen bei mir momentan auf Eis - Aber vielleicht ein Ansatz für euch
Grüße, Souko
Python: nodle_example.py
import hid
import threading
import time
import sys
import random
#SUBSYSTEM=="usb", ATTRS{idVendor}=="16d0", ATTRS{idProduct}=="0830", MODE="0666"
vid = 0x16d0
pid = 0x0830
interface_device = None
# Globale Buffer für DMX-Daten
dmx_in_data = bytearray(512)
dmx_out_data = bytearray(512)
# Aktionen für verschiedene Daten definieren
nodle_protocol_actions = {
0x00: "1. Block",
0x0F: "16. Block",
0x10: "Mode Command",
0x11: "Config Command",
0x12: "Save Command",
0x20: "1. Block",
0x2F: "16. Block",
0x55: "Beat Command",
0x7A: "Bootloader Init",
0x66: "Execute CMD based on 1st CMD-byte of Message",
0x77: "Send DeviceLabel to PC",
0xFE: "DEBUG"
}
mode_mapping = {
0x00: "Standby",
0x01: "DMX in -> DMX out",
0x02: "PC out -> DMX out",
0x03: "DMX In + PC Out -> DMX Out",
0x04: "DMX in -> PC in",
0x05: "DMX in -> DMX out & DMX in -> PC in",
0x06: "PC out -> DMX out & DMX in -> PC in",
0x07: "DMX in + PC out -> DMX out & DMX in -> PC in",
0x08: "Sync-Full-Universe-Out. PC out -> DMX Out & DMX in -> PC in",
0x09: "Standalone-Static"
}
def bytes_to_uint16(data, start_index, endian='little'):
if endian == 'big':
# Big Endian: Das höherwertige Byte steht zuerst
return (data[start_index] << 8) + data[start_index + 1]
elif endian == 'little':
# Little Endian: Das niederwertige Byte steht zuerst
return (data[start_index + 1] << 8) + data[start_index]
else:
raise ValueError("Unbekanntes Endian-Format. Verwende 'big' oder 'little'.")
def bytes_to_uint32(data, start_index, endian='little'):
if endian == 'big':
# Big Endian: Das höherwertige Byte steht zuerst
return (data[start_index] << 24) + (data[start_index + 1] << 16) + (data[start_index + 2] << 8) + data[start_index + 3]
elif endian == 'little':
# Little Endian: Das niederwertige Byte steht zuerst
return (data[start_index + 3] << 24) + (data[start_index + 2] << 16) + (data[start_index + 1] << 8) + data[start_index]
else:
raise ValueError("Unbekanntes Endian-Format. Verwende 'big' oder 'little'.")
def handle_data(data):
# Wenn die Aktion den 1. bis 16. Block betrifft, schreibe die nachfolgenden 32 Byte in den DMX-Daten-Puffer
if 0x00 <= data[0] <= 0x0F:
dmx_in_data[32*(data[0]):32*(data[0]) + 32] = data[1:33]
hex_data = " ".join(format(byte, "02X") for byte in dmx_in_data[32*(data[0]):32*(data[0]) + 32])
print("DMX-Daten empfangen, Block: ", data[0] , " Data: ", hex_data)
return
# Wenn die empfangenen Daten einer bekannten Aktion entsprechen, führe die entsprechende Aktion aus
if data[0] in nodle_protocol_actions:
if data[0] == 0x55: #Heartbeat
return
if data[0] == 0x11: #Config
print("CONFIG:")
print("Break: " , bytes_to_uint16(data, 2) * 4 , " us")
print("Mark: ", bytes_to_uint16(data, 4) * 320 , " ns")
print("Interbyte: " , bytes_to_uint16(data, 6), " us")
print("Interframe: " , bytes_to_uint16(data, 8), " us")
print("Channelcount: ", bytes_to_uint16(data, 10))
print("Startbyte: ", data[12])
print("Repeatermode: ", data[13])
return
if data[0] == 0xFE: #Debug
hex_data = " ".join(format(byte, "02X") for byte in data)
print("DEBUG:", hex_data)
return
hex_data = " ".join(format(byte, "02X") for byte in data)
print("Unbekannte Aktion:", hex_data)
def interface_input_thread(device):
while not stop_flag:
# Lese Daten vom HID-Gerät
data = device.read(64)
# Hier kannst du deine Logik für die Datenverarbeitung und Aktionen einfügen
if data:
hex_data = " ".join(format(byte, "02X") for byte in data)
# print("Daten vom HID-Gerät:", hex_data)
handle_data(data)
# Schlafen, um CPU-Ressourcen zu sparen
time.sleep(0.1)
def process_input(user_input):
if user_input.startswith("SET "):
# Schneide das Präfix ab
user_input = user_input[len("SET "):]
process_channel_value(user_input)
return
if user_input.startswith("BOOT"):
set_interface_to_boot(interface_device)
return
def set_interface_mode(device, mode):
if mode in mode_mapping:
mode_data = mode_mapping[mode]
# Senden des Modus-Befehls an das HID-Gerät
device.write([0x10, mode])
print("Interface-Modus auf", mode, "gesetzt.")
else:
print("Ungültiger Modus:", mode)
def set_interface_to_boot(device):
print("Bashing that CPU into Boot!")
device.write([0x7A])
device.close()
return
def process_channel_value(input_str):
# Trenne den Kanal und den Wert anhand von "x"
channel_value = input_str.split(":")
if len(channel_value) != 2:
print("Ungültiges Eingabeformat. Bitte verwenden Sie das Format 'Kanal:Wert', z.B. '03:FF'.")
return
channel_str, value_str = channel_value
try:
# Konvertiere den Kanal in eine Zahl (Dezimal oder Hexadezimal)
if channel_str.startswith("0x"):
channel = int(channel_str, 16)
else:
channel = int(channel_str)
# Überprüfe den Bereich des Kanals
if channel < 1 or channel > 512:
print("Ungültiger Kanalwert. Der Kanal muss zwischen 1 und 512 liegen.")
return
# Konvertiere den Wert in eine Zahl (Dezimal oder Hexadezimal)
if value_str.startswith("0x"):
value = int(value_str, 16)
else:
value = int(value_str)
# Überprüfe den Bereich des Werts
if value < 0 or value > 255:
print("Ungültiger Wert. Der Wert muss zwischen 0 und 255 liegen.")
return
# Setze den Wert im globalen Buffer dmx_out_data für den entsprechenden Kanal
dmx_out_data[channel - 1] = value
# Sende den passenden 32-Byte-Block an das HID-Gerät
send_dmx_channel(channel)
except ValueError:
print("Ungültige Eingabe. Stellen Sie sicher, dass Kanal und Wert Zahlen sind.")
def send_dmx_channel(channel):
# Bestimme den Index des entsprechenden 32-Byte-Blocks im dmx_out_data-Buffer
block_index = (channel - 1) // 32
# Bestimme den Startindex des aktuellen Blocks
start_index = block_index * 32
# Bestimme den 32-Byte-Block, der gesendet werden soll
dmx_block = dmx_out_data[start_index:start_index + 32]
# Füge das Byte für den Blockindex am Anfang des Blocks ein, wenn das erste byte 0 ist,
# dann muss ein extra 0-byte davor...
# siehe: https://github.com/node-hid/node-hid/issues/187
if block_index == 0:
dmx_block_with_index = bytes([0x00]) + bytes([block_index]) + bytes(dmx_block)
else:
dmx_block_with_index = bytes([block_index]) + bytes(dmx_block)
# Sende den 32-Byte-Block mit dem Byte für den Blockindex an das HID-Gerät
interface_device.write(dmx_block_with_index)
def find_interface_by_sn(devices, serial):
for device in devices:
if 'serial_number' in device and device['serial_number'] == serial:
return device['device_path']
return None
def compare_dmx_blocks(out_data, in_data, start_index):
# Die nächsten 32 Bytes vergleichen
for i in range(start_index, start_index + 32):
if out_data[i] != in_data[i]:
return False
return True
def test_dmx_in_out(device):
#DMX-Out buffer mit random-werten fuellen und DMX-Input mit 0
dmx_out_data = bytearray(random.randint(0, 255) for _ in range(512))
dmx_in_data = bytearray()
test_ok = True
# Starte den Hintergrund-Thread fuer den Empfang in einem separaten Thread
background_thread = threading.Thread(target=interface_input_thread, args=(interface_device,))
background_thread.start()
#setze Interface auf Mode 6
set_interface_mode(interface_device, 6)
# 200ms warten, damit der DC/DC wandler stabil ist und der Ausgang und fuers gute Gefuehl
time.sleep(0.200)
for block in range(16):
channel = block * 32
# DMX-Kanal senden / immer nur einer aus dem Block, weil es werden immer 32 Kanaele gesendet
send_dmx_channel(channel)
# 20ms warten
time.sleep(0.020)
# Prüfen, ob der Block korrekt gesendet und empfangen wurde
if compare_dmx_blocks(dmx_out_data, dmx_in_data, channel):
print(f"Block {block} successfully sent and recieved. match!")
else:
sendBlock_data = " ".join(format(byte, "02X") for byte in dmx_out_data[channel:channel+32])
recievedBlock_data = " ".join(format(byte, "02X") for byte in dmx_in_data[channel:channel+32])
print(f"Fehler!!: Block {block} not sent correctly")
print(f"Sended Data: ", sendBlock_data )
print(f"Recieved Data: ", recievedBlock_data)
test_ok = False
break
# Setze die Flagge, um den Thread zu beenden
global stop_flag
stop_flag = True
# Warte auf das Ende des Hintergrund-Threads
background_thread.join()
# Schließe das HID-Gerät
interface_device.close()
if test_ok == False:
print("FEHLER!!:")
print("DMX In-Out Test war nicht erfolgreich!")
input("Enter druecken")
def main():
global interface_device
while True:
# HID-Geraete auflisten
devices = hid.enumerate()
# Nach Nodle suchen (Seriennummer 1000000000001234)
device_path = find_interface_by_sn(devices, '1000000000001234')
if device_path:
print("Nodle gefunden. Device Path:", device_path)
interface_device = hid.device()
interface_device.open_path(device_path)
test_dmx_in_out(interface_device)
if __name__ == "__main__":
# Setze die Flagge, um den Thread zu beenden
stop_flag = False
# Führe das Hauptprogramm aus
main()
Display More