85 lines
3.0 KiB
Python
Executable File
85 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Schickt sACN/E1.31-Pakete an eine ninlight-Lampe - fuer die Abnahme von Hand.
|
|
|
|
Aufruf:
|
|
tools/sacn_send.py --universe 1 --start-slot 1 --values 255,255,0,0,0,0 --seconds 5
|
|
tools/sacn_send.py --host 192.168.2.122 --values 255,0,255,0,0,0 --seconds 5
|
|
tools/sacn_send.py --values 255,255,255,255,255,255 --seconds 3 --terminate
|
|
|
|
Ohne --host geht es per Multicast an 239.255.<hi>.<lo>, mit --host per Unicast direkt an
|
|
die Lampe. Die Werte sind der Slot-Satz ab --start-slot: Dimmer, R, G, B, CW, WW (Boards
|
|
ohne RGB: Dimmer, CW, WW).
|
|
"""
|
|
import argparse
|
|
import socket
|
|
import struct
|
|
import time
|
|
|
|
E131_PORT = 5568
|
|
ACN_PID = b"ASC-E1.17\0\0\0"
|
|
|
|
|
|
def packet(universe, sequence, values, start_slot, priority, terminated):
|
|
slots = bytearray(512)
|
|
for i, v in enumerate(values):
|
|
slots[start_slot - 1 + i] = v
|
|
count = start_slot - 1 + len(values)
|
|
data = bytes(slots[:count])
|
|
total = 126 + count
|
|
|
|
p = bytearray(total)
|
|
struct.pack_into(">HH", p, 0, 0x0010, 0x0000)
|
|
p[4:16] = ACN_PID
|
|
struct.pack_into(">H", p, 16, 0x7000 | (total - 16))
|
|
struct.pack_into(">I", p, 18, 0x00000004)
|
|
p[22:38] = b"ninlight-abnahme" # CID, 16 Byte
|
|
struct.pack_into(">H", p, 38, 0x7000 | (total - 38))
|
|
struct.pack_into(">I", p, 40, 0x00000002)
|
|
p[44:44 + 13] = b"ninlight test"
|
|
p[108] = priority
|
|
p[111] = sequence & 0xFF
|
|
p[112] = 0x40 if terminated else 0x00
|
|
struct.pack_into(">H", p, 113, universe)
|
|
struct.pack_into(">H", p, 115, 0x7000 | (total - 115))
|
|
p[117] = 0x02
|
|
p[118] = 0xA1
|
|
struct.pack_into(">HH", p, 119, 0x0000, 0x0001)
|
|
struct.pack_into(">H", p, 123, count + 1)
|
|
p[125] = 0x00
|
|
p[126:] = data
|
|
return bytes(p)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--universe", type=int, default=1)
|
|
ap.add_argument("--start-slot", type=int, default=1)
|
|
ap.add_argument("--values", default="255,255,255,0,0,0")
|
|
ap.add_argument("--priority", type=int, default=100)
|
|
ap.add_argument("--seconds", type=float, default=5.0)
|
|
ap.add_argument("--fps", type=float, default=30.0)
|
|
ap.add_argument("--host", help="Unicast statt Multicast")
|
|
ap.add_argument("--terminate", action="store_true",
|
|
help="zum Schluss ein Paket mit gesetztem stream_terminated-Bit")
|
|
a = ap.parse_args()
|
|
|
|
values = [int(v) for v in a.values.split(",")]
|
|
dest = a.host or "239.255.%d.%d" % (a.universe >> 8, a.universe & 0xFF)
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 8)
|
|
|
|
seq, deadline, step = 0, time.time() + a.seconds, 1.0 / a.fps
|
|
while time.time() < deadline:
|
|
s.sendto(packet(a.universe, seq, values, a.start_slot, a.priority, False),
|
|
(dest, E131_PORT))
|
|
seq += 1
|
|
time.sleep(step)
|
|
if a.terminate:
|
|
s.sendto(packet(a.universe, seq, values, a.start_slot, a.priority, True),
|
|
(dest, E131_PORT))
|
|
print("%d Pakete an %s:%d, Universe %d" % (seq, dest, E131_PORT, a.universe))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|