Xdesk/tool/nodescan2.py

# -*- coding: utf-8 -*-

"""
Valid-License-Identifier: GPL-2.0-only
SPDX-URL: https://spdx.org/licenses/GPL-2.0-only.html

(c) 2012 micha@librelight.de
"""
import time
import socket, struct
import sys
import os
import _thread as thread
import copy
import random
import traceback

sys.stdout.write("\x1b]2;Nodescan\x07")


def UDP_Socket(bind=False,ip='',port=6454):
    sock = False
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        if bind:
            sock.bind((ip, port))
    except socket.error as e:
        print("Socket 6454 ", "ERR: {0} ".format(e.args))
        sock = False

    return sock


def ArtPoll(sock=None,ip="2.255.255.255",port=6454):
    print("ArtPoll",ip,port)
    if not sock:
        sock=UDP_Socket()
    #port=6454
    #ip="2.255.255.255"
    PKG=b'Art-Net\x00\x00 \x00\x0e\x06\x00'
    print("SEND:",[PKG])
    sock.sendto(PKG,(ip,port)) # ArtPol / ping
    sock.close()


def convert_mac(MAC):
    #MAC = data[201:201+6]
    _MAC = []
    for x in MAC:
        #x = hex(ord(x))[2:]
        x = hex(x)[2:]
        x = x.rjust(2,"0")
        _MAC.append(x)
    _MAC = ":".join(_MAC)
    return _MAC

def convert_bin(d):
    return bin(d)[2:].rjust(8,"0")

def convert_ip(d):
    IP="[0,0,0,0]"
    _ip = []
    try:
        _ip.append( d[0] )
        _ip.append( d[1] )
        _ip.append( d[2] )
        _ip.append( d[3] )
        IP = str(_ip)
    except:pass

    return IP

def convert_to_hex(x,d):
    out = b""
    try:
        a = struct.unpack(x, d)[0]
        #out = hex(a)
        out = "{0:#0{1}x}".format(a,6)
    except:
        pass
    return out 

def ArtNet_decode_pollreplay(data):
    debug = 0
    node = {}
    if len(data) < 10: #min opcode
        return node

    opcode=convert_to_hex("<h",data[8:10])

    if opcode != '0x2100': #OpPollReplay
        return node

    if len(data) < 207: #Mal
        return node
    #print(data[174:174+4])
    #print("===================================================================-")
    #print("decode",data[:13])
    

    # UDP PACKAGE VALUE:INDEX:RAGE
    CONF = {}
    CONF["IP"]         = [10,14+1]
    CONF["port"]       = [14,15+1]
    CONF["version"]    = [16,17+1]
    CONF["NetSwitch"]  = [18] 
    CONF["SubSwitch"]  = [19]
    CONF["oem"]        = [20,21+1]
    CONF["ubea"]       = [22]
    CONF["status"]     = [23]
    CONF["esta"]       = [24,25+1]
    CONF["sname"]      = [26,26+17]
    CONF["lname"]      = [44,44+43]
    CONF["NodeReport"] = [108,108+20]
    CONF["NumPort"]    = [173]
    CONF["PortTypes"]  = [174,174+4]
    CONF["GoodInput"]  = [178,178+4]
    CONF["GoodOutput"] = [182,182+4]
    CONF["SwIn"]       = [186,186+4]
    CONF["SwOut"]      = [190,190+4]
    #CONF["MSG"]        = [108,108+40]
    CONF["MAC"]        = [201,201+6]
    
    cleanup = ["sname","lname","MSG","NodeReport"]
    for k,v in CONF.items():
        val = b'undefined'
        if len(v) == 2:
            val = data[v[0]:v[1]]
            if k in cleanup:
                val = val.strip(b"\x00")
                val = val.decode(errors="ignore")
        if len(v) == 1:
            val = data[v[0]]
        node[k] = val
    
    # ================================

    node["MAC"]    = convert_mac(node["MAC"])
    node["IP"]     = convert_ip(node["IP"])
    node["status"] = convert_bin(node["status"])
    node["opcode"] = opcode
    
    unpack = {"port":b"<H"}
    for k,v in node.items():
        if k in unpack:
            up = unpack[k]
            #print(k,v,up)
            node[k] = struct.unpack(up,v)[0]

    to_hex = {"version":'>H',"oem":'>H',"esta":"<H"}
    for k,u in to_hex.items():
        if k in node:
            v=node[k]
            node[k] = convert_to_hex(u,v)

    #for k,v in node.items():
    #    if type(node[k]) is bytes:
    #        node[k] = v.decode(errors="ignore")

    return node

def test():
    UDP_ArtPollReplay = b'Art-Net\x00\x00!\x02\x00\x00T6\x19\x03P\x00\x00\x11\x10\x00\x00RUAADN-01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00AVR-ArtNet DMX NODE\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00Node is ready\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x80\x00\x00\x00\x08\x00\x00\x00\x82\x00\x00\x00\x07\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00p\xb3\xd5\xfa\xff\xfa\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'

    NODE_OLD = {'IP': '[2, 0, 0, 84]', 'port': (6454,), 'version': '\x03P', 'NetSwitch': 0, 'SubSwitch': 0, 'oem': '\x11\x10', 'ubea': 0, 'status': 0, 'esta': 'RU', 'sname': 'AADN-01', 'lname': 'AVR-ArtNet DMX NODE', 'NumPort': 1, 'PortTypes': '\x00\x00\x00', 'GoodInput': '\x08\x00\x00\x00', 'GoodOutput': '\x00\x00\x00', 'SwIn': '\x07\x00\x00\x00', 'SwOut': '\x00\x00\x00\x00', 'MSG': 'Node is ready', 'MAC': '70:b3:d5:fa:ff:fa'}
    
    NODE = {'IP': '[2, 0, 0, 84]', 'port': 6454, 'version': '0x0350', 'NetSwitch': 0, 'SubSwitch': 0, 'oem': '0x1110', 'ubea': 0, 'status': '00000000', 'esta': '0x5552', 'sname': 'AADN-01', 'lname': 'AVR-ArtNet DMX NODE', 'NodeReport': 'Node is ready', 'NumPort': 1, 'PortTypes': b'\x80\x00\x00\x00', 'GoodInput': b'\x08\x00\x00\x00', 'GoodOutput': b'\x82\x00\x00\x00', 'SwIn': b'\x07\x00\x00\x00', 'SwOut': b'\x00\x00\x00\x00', 'MAC': '70:b3:d5:fa:ff:fa', 'opcode': '0x2100'}

    node = ArtNet_decode_pollreplay(UDP_ArtPollReplay)
    k_miss =[]
    if NODE != node:
        for k,v in node.items():
            if k in NODE:
                print(v==NODE[k],"- diff -",k,v, NODE[k])
            else:
                k_miss.append(k)
        print("MISSING KEY:",k_miss)
        print(node)
        print(NODE)
    assert NODE == node


def ArtPollReply(sock=None):
    print("ArtPollReply()")
    if not sock:
        sock=UDP_Socket()
    
    port = 6454
    content = []
    header = []

    # Name, 7byte + 0x00
    content.append(b"Art-Net\x00")
    # OpCode ArtPollReply -> 0x2100, Low Byte first
    content.append(struct.pack('<H', 0x2100))
    # Protocol Version 14, High Byte first
    content.append( b"\xff\xff\xff\xff")

    # Port
    content.append(struct.pack('<H', 0x1936))
    # Firmware Version
    content.append(struct.pack('>H', 200))
    # Net and subnet of this node
    net = 0
    subnet = 0
    content.append(chr(net))
    content.append(chr(subnet))
    # OEM Code (E:Cue 1x DMX Out)
    content.append(struct.pack('>H', 0x0360))
    # UBEA Version -> Nope -> 0
    content.append(chr(0))
    # Status1
    content.append(struct.pack('>H', 0b11010000))
    # Manufacture ESTA Code
    content.append(chr(0))

    hostname=os.popen("hostname").read().strip()
    # Short Name, len == 30
    content.append(hostname.ljust(18,"\x00")[:30])
    # Long Name, len == 64
    content.append(("LibreLight "+hostname).ljust(64,"\x00")[:64] )

    content2=[]
    for c in content:
        if type(c) is not bytes:
            c= bytes(c,"ascii")
        content2.append(c)

    content = b''.join(content2)
    fill = 240-len(content)-1
    print()
    if fill > 0:
        content = content + b"\x00"*fill

    def inject(i,v,content):
        content = content[:i]+ v + content[i+1:]
        return content
    def create_ip(IP,content):
        #CONF["IP"]         = [10,14+1]
        x=[]
        j=10
        for i in IP.split("."):
            i = 33
            _ip = struct.pack("B",int(i))
            # content = content[:10+j]+ _ip + content[10+j:]
            content = inject(j,_ip,content)
            j+=1
            x.append([_ip,i])
        print(IP,x)
    
    #CONF["MAC"]        = [201,201+6]
    content = inject(201,b"\xfa",content)
    content = inject(201+1,b"\xfa",content)
    content = inject(201+2,b"\xfa",content)
    content = inject(201+3,b"\xfa",content)
    content = inject(201+4,b"\xfa",content)
    content = inject(201+5,b"\xfa",content)

    #CONF["oem"]        = [20,21+1]
    content = inject(20,b"\x11",content)
    content = inject(20+1,b"\x10",content)

    content = inject(100,b"\x11",content)

    print()
    IP="2.0.0.255"
    create_ip(IP,content)
    sock.sendto(content, (IP, port))

    IP="10.10.10.255"
    content = [0]*(282-42) # new PKG
    content[0:0]   = b"Art-Net\x00"
    content[8:9+1]   = b"\x00\x21"         # Protocol 0x2100
    content[10:13+1] = b"\xaa\xaa\xaa\xaa" # IP
    content[20:21+1] = b"\x11\x10"         # oem
    content[26] = ord("A")  # SHORT NAME
    content[26+18] = ord("B") #0x42  # LONG NAME
    content[108:108] = b"HalloX" #0x42  # MSG  26+18+64
    print("send" ,[content])
    j=0
    content2 = content[:]
    for i in IP.split("."):
        content2[10+j] = int(i)
        j+=1
    content2 = bytes(content2)

    #create_ip(IP,content)
    sock.sendto(content2, (IP, port))
    
    print("send" ,[content])
    IP="192.168.2.255"
    j=0
    content2 = content[:]
    for i in IP.split("."):
        content2[10+j] = int(i)
        j+=1
    #create_ip(IP,content)
    content2 = bytes(content2)
    sock.sendto(content2, (IP, port))
    
    IP="2.0.0.255"
    j=0
    content2 = content[:]
    for i in IP.split("."):
        content2[10+j] = int(i)
        j+=1
    #create_ip(IP,content)
    content2 = bytes(content2)
    sock.sendto(content2, (IP, port))

    print("send" ,[content2])
    #print(dir(sock),sock.getsockname())

def main():
    print("start main()")
    sock = UDP_Socket(bind=True,ip='',port=6454)
    print("-- loop --")
    while sock:
        data, addr = sock.recvfrom(300)
        opcode=""
        if len(data) >= 10:
            opcode=convert_to_hex("<h",data[8:10])
        print("-",addr,len(data),opcode)
        print()

        nodes = ArtNet_decode_pollreplay(data)
        for k,v in nodes.items():
            print("-",[k,v])
    print("end main()")

if __name__ == "__main__":
    test()
    if "--ArtPoll" in sys.argv:
        ArtPoll()
    if "--ArtPollReplay" in sys.argv:
        ArtPollReply()
    else:
        main()