# -*- coding: UTF-8 -*- import os import fcntl import time import socket import struct import random from optparse import OptionParser parser = OptionParser() parser.add_option("-r", "--recive", dest="recive", help="set recive ip like --recive 10.") parser.add_option("-s", "--sendto", dest="sendto", help="set sender ip like --sendto 2.255.255.255") parser.add_option("-t", "--test", dest="testuniv", help="set test univers like --test [0-16]") parser.add_option("", "--inmap", dest="inmap", help="set test univers like --test [0-16]") #parser.add_option("-q", "--quiet", # action="store_false", dest="verbose", default=True, # help="don't print status messages to stdout") (options, args) = parser.parse_args() print("option",options) print(options.sendto) def unpack_art_dmx(data): dmx = [] for i in range(len(data[18:]) ): x=data[18+i] #print("x",x) #print( "data",b'!B', data[18+i]) #x=struct.unpack( b'!B',data[18+i]) #print( "data",b'!B', data[18+i],x) #x=x[0] dmx += [x] return dmx class Socket(): def __init__(self,bind='',port=6454): self.__port =port self.__bind =bind self.__poll = 0 self.__data = [] self.__addr = "NONE" #self.__hosts = {"host":{"9":[0]*512}} self.__hosts = {} self.hosts = self.__hosts self.open() self._poll_clean_time = time.time() self._poll_clean_count = 0 def open(self): try: print("connecting to ArtNet bind:",self.__bind,"Port",self.__port) self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind((self.__bind, self.__port)) fcntl.fcntl(self.sock, fcntl.F_SETFL, os.O_NONBLOCK) #self.sock.setblocking(0) except socket.error as e: print("Socket ",self.__bind,self.__port, "ERR: {0} ".format(e.args)) #raw_input() #sys.exit() def poll_clean(self): if self._poll_clean_time+(1/25.) <= time.time(): self._poll_clean_time = time.time() self._poll_clean() x = self._poll_clean_count self._poll_clean_count = 0 return x def _poll_clean(self): while 1: try: self.__data, self.__addr = self.sock.recvfrom(self.__port) self._poll_clean_count += 1 #return 1 except socket.timeout as e: err = e.args[0] if err == 'timed out': time.sleep(1) print('recv timed out, retry later') else: print(e) break except socket.error as e: break def poll(self): if not self.__poll: try: self.__data, self.__addr = self.sock.recvfrom(self.__port) data, addr = (self.__data,self.__addr) self.host = addr[0] head = data[:18] rawdmx = data[18:] #print([head],addr) self.univ = -1 try: self.head = struct.unpack("!8sHBBBBHBB" , head ) except Exception as e: pass#print( "======E09823" , e) univ = self.head[6]/255 # /512 # * 512 self.univ = int(univ) if self.host.startswith("127."): #allways recive localhost on port self.__poll = 1 return 1 elif not options.recive: self.__poll = 1 return 1 elif self.host.startswith(options.recive): self.__poll = 1 return 1 else: self.__poll = 0 addr = str(addr) univ = str(univ) if self.__poll: if addr not in self.__hosts: self.__hosts[addr] = {} if univ not in self.__hosts[addr]: self.__hosts[addr][univ] = {} self.__hosts[addr][univ] = {"head":head,"addr":addr,"univ":univ,"dmx":rawdmx} self.hosts = self.__hosts except socket.timeout as e: err = e.args[0] if err == 'timed out': time.sleep(1) print('recv timed out, retry later') else: print(e) except socket.error as e: pass def recive(self): if self.__poll: self.__poll = 0 data, addr = (self.__data,self.__addr) #print( self.univ,self.head) self.dmx = unpack_art_dmx(data) return { "host":self.host,"dmx":self.dmx,"univ":self.univ,"head":self.head,"data":data,"addr":addr} x=0 y=0 running = True xsocket = Socket() import pygame import pygame.gfxdraw pygame.init() screen = pygame.display.set_mode((780, 610)) pygame.display.set_caption("pygame: DMX OSZI") pygame.mouse.set_visible(1) pygame.key.set_repeat(1, 30) clock = pygame.time.Clock() pygame.init() class Trans(): def __init__(self,y=150): self.y = 150 self.s = 0.25 def get_y(self,y): return int((y*self.s)+(self.y*self.s))*-1 import copy import _thread as thread class E(): def __init__(self): self.sdata = {} self.lock = thread.allocate_lock() def loop(self): sdata = {} print("loop") while 1: flag = 0 while xsocket.poll(): xx = xsocket.recive() k = xx["host"] +":"+ str(xx["head"][6]) sdata[k] = xx flag = 1 if flag: try: self.lock.acquire() self.sdata = copy.deepcopy(sdata) finally: self.lock.release() time.sleep(0.001) def get(self): try: self.lock.acquire() x = self.sdata #= copy.deepcopy(asdata) self.sdata = {} return x finally: self.lock.release() e = E() thread.start_new_thread(e.loop,()) T = Trans() import sys font = pygame.font.SysFont("FreeSans", 12) #,color=(255,0,0)) def get_value(sdata=[] ,univ=0,dmx=[1,121]): data=[] for k in sdata: xx = sdata[k] _univ = int(xx["head"][6] /256) if xx["host"].startswith('2.0.0.') and _univ == univ: for d in dmx: y = xx["dmx"][d-1] data.append(y) return data class OSZI(): def __init__(self): self._x=0 self.sdata={} self.lz = time.time() self.c=0 def draw(self,screen,data):#,c,x,y,T): x=int(self._x) if time.time() > self.lz: self.lz = time.time()+6 self._x=0 x=0 rec = pygame.Rect(x+1,T.get_y(10),30,245) # clear balken pygame.draw.rect(screen,(10,10,0),rec) T.y=-260 c=0 for d in data: y=d pygame.gfxdraw.pixel(screen,x,T.get_y(255),(255,0,0)) pygame.gfxdraw.pixel(screen,x,T.get_y(0),(10,10,25)) rec = pygame.Rect(x+4,T.get_y(0),20,-80) # clear balken pygame.draw.rect(screen,(c,210,110),rec) rec = pygame.Rect(x+2,T.get_y(0),30,-80) # clear balken pygame.draw.rect(screen,(c,10,110),rec) text = font.render( str(y), True, (0,0,0)) pygame.draw.circle(screen,(255,155,0),(x,T.get_y(y)),2) c+=50 if c >255: c=255 T.y-=275 #_x+=3.5*2 self._x+=1.5*2 def read_event(): running = True for event in pygame.event.get(): if event.type == pygame.QUIT: running = False if event.type == pygame.KEYDOWN: print(event.type) if event.key == pygame.K_ESCAPE: pygame.event.post(pygame.event.Event(pygame.QUIT)) return running class GRID(): def __init__(self): self.grid_timer = time.time() def draw(self,xsdata): grid_timer = self.grid_timer if grid_timer > time.time(): return grid_timer=time.time()+.0215 for d in xsdata: xx=sdata[d] univ = xx["head"][6] //256 #/ 255 if xx["host"].startswith('2.0.0.'): if univ == 0: rx=308 ry=10 rec = pygame.Rect(rx,ry,600,600) # clear balken pygame.draw.rect(screen,(20,20,20),rec) text = font.render( str(univ).rjust(3," "), True, (255,255,255)) screen.blit(text, ( rx-10, ry ) ) line = [] for i,dmx in enumerate(xx["dmx"]): text = font.render( str(dmx).rjust(3," "), True, (255,255,255)) screen.blit(text, ( rx+10, ry+10 ) ) rx+=29 if (i+1) % 20 == 0: rx=308 ry+=11 elif univ == 1: rx=8 ry=310 rec = pygame.Rect(rx,ry,600,600) # clear balken pygame.draw.rect(screen,(10,30,10),rec) text = font.render( str(univ).rjust(3," "), True, (255,255,255)) screen.blit(text, ( rx-10, ry ) ) line = [] for i,dmx in enumerate(xx["dmx"]): text = font.render( str(dmx).rjust(3," "), True, (255,200,255)) screen.blit(text, ( rx+10, ry+10 ) ) rx+=29 if (i+1) % 20 == 0: rx=8 ry+=11 _ips = {} def print_ips(sdata): if int(time.time()*10) % 20 == 0: for k in sdata: if k in _ips: _ips[k] += 1 else: _ips[k] = 0 print(k.ljust(15," "),_ips[k]) print() grid_a = GRID() oszi_a = OSZI() while running: clock.tick(15) #clock.tick(225) sdata = e.get() xsdata = copy.deepcopy(sdata) print_ips(sdata) data = get_value(sdata,univ=1,dmx=[21,142,261,263]) running = read_event() oszi_a.draw(screen,data) #,c,x,y,T) grid_a.draw(xsdata) pygame.display.flip()