Re-did the rendering infrastructure. Still leaking memory.

2 files changed, 85 insertions, 170 deletions

diff --git a/host/matelight.py b/host/matelight.py
index 6d0f5bf..59b6373 100644
--- a/host/matelight.py
+++ b/host/matelight.py
@@ -18,7 +18,7 @@ FRAME_SIZE = DISPLAY_WIDTH*DISPLAY_HEIGHT
 GAMMA = 2.5
 
 # Brightness of the LEDs in percent. 1.0 means 100%.
-BRIGHTNESS = 0.2
+BRIGHTNESS = 1.0
 
 ml = CDLL('./libml.so')
 ml.matelight_open.restype = c_void_p
diff --git a/host/server.py b/host/server.py
index b942af4..cdf2b1d 100755
--- a/host/server.py
+++ b/host/server.py
@@ -1,9 +1,12 @@
 #!/usr/bin/env python
 
 from socketserver import *
+import socket
+import struct
+import zlib
 from time import time, strftime, sleep
 from collections import namedtuple, deque
-from itertools import product, cycle
+import itertools
 import threading
 import random
 
@@ -53,200 +56,112 @@ def printframe(fb):
 	print('\0337\033[H', end='')
 	print('Rendering frame @{}'.format(time()))
 	bdf.console_render_buffer(fb.ctypes.data_as(POINTER(c_uint8)), w, h)
-	print('\033[0m\033[KCurrently rendering', current_entry.entrytype, 'from', current_entry.remote, ':', current_entry.text, '\0338', end='')
+	#print('\033[0m\033[KCurrently rendering', current_entry.entrytype, 'from', current_entry.remote, ':', current_entry.text, '\0338', end='')
 	printlock.release()
 
-QueueEntry = namedtuple('QueueEntry', ['entrytype', 'remote', 'timestamp', 'text'])
-defaultlines = [ QueueEntry('text', '127.0.0.1', 0, l[:-1].replace('\\x1B', '\x1B')) for l in open('default.lines').readlines() ]
-random.shuffle(defaultlines)
-defaulttexts = cycle(defaultlines)
-current_entry = next(defaulttexts)
-conns = {}
-textqueue = []
-
 def log(*args):
 	printlock.acquire()
 	print(strftime('\x1B[93m[%m-%d %H:%M:%S]\x1B[0m'), ' '.join(str(arg) for arg in args), '\x1B[0m')
 	printlock.release()
 
-class MateLightUDPServer(UDPServer):
-	"""
-	The server class for the SocketServer interface.
-	"""
-	def __init__(self, *args, **kwargs):
-		"""
-		Setup the deque for the frame
-		"""
-		super(MateLightUDPServer, self).__init__(*args, **kwargs)
-		self.frame_deque = deque(maxlen=30)
-		
-	def get_next_frame(self):
-		try:
-			return self.frame_deque.popleft()
-		except IndexError:
-			return None
-
-class MateLightTCPServer(TCPServer):
-	def __init__(self, *args, **kwargs):
-		super(MateLightTCPServer, self).__init__(*args, **kwargs)
-		# A deque for the texts, contains a tuple of the form (text, width, height)
-		self.text_deque = deque(maxlen=32)
-		self.current_text = None
-
-	def get_next_frame(self):
-		if self.current_text is None or self.i > self.current_text[1][0]:
-			self.current_text = None
+class TextRenderer:
+	def __init__(self, text):
+		self.text = text
+		self.width, _ = compute_text_bounds(text)
+
+	def __iter__(self):
+		for i in range(-DISPLAY_WIDTH, self.width):
+			yield render_text(self.text, i)
+
+class MateLightUDPServer:
+	def __init__(self, port=1337, ip=''):
+		self.current_client = None
+		self.last_timestamp = 0
+		self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+		self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+		self.socket.bind((ip, port))
+		self.thread = threading.Thread(target = self.udp_receive)
+		self.thread.daemon = True
+		self.start = self.thread.start
+		self.frame_condition = threading.Condition()
+		self.frame = None
+	
+	def frame_da(self):
+		return self.frame is not None
+
+	def __iter__(self):
+		while True:
+			with self.frame_condition:
+				if not self.frame_condition.wait_for(self.frame_da, timeout=UDP_TIMEOUT):
+					raise StopIteration()
+				frame, self.frame = self.frame, None
+			yield frame
+
+	def udp_receive(self):
+		while True:
 			try:
-				self.current_text = self.text_deque.popleft()
-				self.i = -DISPLAY_WIDTH
-			except IndexError:
-				return None
-
-		frame = render_text(self.current_text[0], self.i)
-		self.i += 1
-		return frame
-
-class MateLightUDPHandler(BaseRequestHandler):
-	"""
-	Handles one UDP connection to the matelight
-	"""
-	def handle(self):
-		try:
-			# Housekeeping - FIXME: This is *so* the wrong place for this.
-			for k,v in conns.items():
-				if time() - v.timestamp > UDP_TIMEOUT:
-					del conns[k]
-
-			global current_entry, conns
-			data = self.request[0].strip()
-			if len(data) != FRAME_SIZE*3+4:
-				#raise ValueError('Invalid frame size: Expected {}, got {}'.format(FRAME_SIZE+4, len(data)))
-				return
-			frame = data[:-4]
-			#crc1, = struct.unpack('!I', data[-4:])
-			#crc2, = zlib.crc32(frame, 0),
-			#if crc1 != crc2:
-			#	raise ValueError('Invalid frame CRC checksum: Expected {}, got {}'.format(crc2, crc1))
-			#socket.sendto(b'ACK', self.client_address)
-			a = np.array(list(frame), dtype=np.uint8)
-			timestamp = time()
-			addr = self.client_address[0]
-			conn = QueueEntry('udp', addr, timestamp, '')
-			if addr not in conns:
-				log('\x1B[91mNew UDP connection from\x1B[0m', addr)
-				current_entry = conn
-			conns[addr] = current_entry
-			if current_entry.entrytype == 'udp' and current_entry.remote == addr:
-				current_entry = conn
-				frame = a.reshape((DISPLAY_WIDTH, DISPLAY_HEIGHT, 3))
-				self.server.frame_deque.append(frame)
-				#sendframe(frame)
-				printframe(np.pad(frame, ((0,0),(0,0),(0,1)), 'constant', constant_values=(0,0)))
-		except Exception as e:
-			log('Error receiving UDP frame:', e)
+				data, (addr, sport) = self.socket.recvfrom(FRAME_SIZE*3+4)
+				timestamp = time()
+				if timestamp - self.last_timestamp > UDP_TIMEOUT:
+					self.current_client = addr
+					log('\x1B[91mAccepting UDP data from\x1B[0m', addr)
+				if addr == self.current_client:
+					if len(data) == FRAME_SIZE*3+4:
+						frame = data[:-4]
+						crc1, = struct.unpack('!I', data[-4:])
+						if crc1:
+							crc2, = zlib.crc32(frame, 0),
+							if crc1 != crc2:
+								raise ValueError('Invalid frame CRC checksum: Expected {}, got {}'.format(crc2, crc1))
+					elif len(data) == FRAME_SIZE*3:
+						frame = data
+					else:
+						raise ValueError('Invalid frame size: {}'.format(len(data)))
+					self.last_timestamp = timestamp
+					with self.frame_condition:
+						self.frame = np.frombuffer(frame, dtype=np.uint8).reshape((DISPLAY_WIDTH, DISPLAY_HEIGHT, 3))
+						self.frame_condition.notify()
+			except Exception as e:
+				log('Error receiving UDP frame:', e)
+
+renderqueue = deque()
 
 class MateLightTCPTextHandler(BaseRequestHandler):
 	def handle(self):
-		global current_entry, conns
+		global render_deque
 		data = str(self.request.recv(1024).strip(), 'UTF-8')
 		addr = self.client_address[0]
-		timestamp = time()
 		if len(data) > 140:
 			self.request.sendall('TOO MUCH INFORMATION!\n')
 			return
 		log('\x1B[95mText from\x1B[0m {}: {}\x1B[0m'.format(addr, data))
-		#textqueue.append(QueueEntry('text', addr, timestamp, data))
-		self.server.text_deque.append((data, compute_text_bounds(data)))
+		renderqueue.append(TextRenderer(data))
 		self.request.sendall(b'KTHXBYE!\n')
 
 TCPServer.allow_reuse_address = True
-tserver = MateLightTCPServer(('', 1337), MateLightTCPTextHandler)
+tserver = TCPServer(('', 1337), MateLightTCPTextHandler)
 t = threading.Thread(target=tserver.serve_forever)
 t.daemon = True
 t.start()
 
-UDPServer.allow_reuse_address = True
-userver = MateLightUDPServer(('', 1337), MateLightUDPHandler)
-t = threading.Thread(target=userver.serve_forever)
-t.daemon = True
-t.start()
-
-def udp_sub_loop():
-	"""
-	If we can get at least one frame from get_next_frame() we will wait
-	for 1000 frame cycles till we return. If we do not get a valid frame
-	in the first loop, we return immediatly.
-	"""
-	invalid_frames = 300
-	while True:
-		next_frame = userver.get_next_frame()
-		if next_frame is not None:
-			invalid_frames = 0
-			sendframe(next_frame)
-			sleep(0.01)
-		else:
-			invalid_frames += 1
-			sleep(0.01)
-		if invalid_frames > 300:
-			break
-
-def tcp_sub_loop():
-	"""
-	Send as many frames as there are texts in the buffer until the text deque
-	is empty.
-	"""
-	while True:
-		next_frame = tserver.get_next_frame()
-		if next_frame is not None:
-			sendframe(next_frame)
-		else:
-			break
-
-class MateLightDefaultScroller():
-	def __init__(self, *args, **kwargs):
-		# A deque for the texts, contains a tuple of the form (text, width, height)
-		self.text_deque = deque(maxlen=32)
-		self.current_text = None
-
-	def get_next_frame(self):
-		if self.current_text is None or self.i > self.current_text[1][0]:
-			self.current_text = None
-			text = next(defaulttexts).text
-			self.current_text = (text, compute_text_bounds(text))
-			self.i = -DISPLAY_WIDTH
-
-		frame = render_text(self.current_text[0], self.i)
-		self.i += 1
-		return frame
+userver = MateLightUDPServer()
+userver.start()
 
+defaultlines = [ TextRenderer(l[:-1].replace('\\x1B', '\x1B')) for l in open('default.lines').readlines() ]
+random.shuffle(defaultlines)
+defaulttexts = itertools.cycle(itertools.chain(*defaultlines))
 
 if __name__ == '__main__':
-	print('\033[?1049h'+'\n'*9)
-	default_scroller = MateLightDefaultScroller()
-	while True:
-		udp_sub_loop()
-		tcp_sub_loop()
-		sendframe(default_scroller.get_next_frame())
-		
-def bla():
+	#print('\033[?1049h'+'\n'*9)
 	while True:
-		if current_entry.entrytype == 'text':
-			if scroll(current_entry.text):
-				if current_entry in textqueue:
-					textqueue.remove(current_entry)
-				if textqueue:
-					current_entry = textqueue[0]
-				else:
-					if conns:
-						current_entry = random.choice(list(conns.values()))
-					else:
-						current_entry = next(defaulttexts)
-			if current_entry.entrytype != 'udp' and textqueue:
-				current_entry = textqueue[0]
-				log('\x1B[92mScrolling\x1B[0m', current_entry.text)
-		if current_entry.entrytype == 'udp':
-			if time() - current_entry.timestamp > UDP_TIMEOUT:
-				current_entry = next(defaulttexts)
-			else:
-				sleep(0.1)
+		if renderqueue:
+			renderer = renderqueue.pop_front()
+		elif userver.frame_da():
+			renderer = userver
+		else:
+			sendframe(next(defaulttexts))
+			continue
+		for frame in renderer:
+			sendframe(frame)
+			#printframe(frame)