diff options
Diffstat (limited to 'host')
| -rw-r--r-- | host/bdf.py | 4 | ||||
| -rw-r--r-- | host/config.py | 4 | ||||
| -rwxr-xr-x | host/server.py | 159 |
3 files changed, 94 insertions, 73 deletions
diff --git a/host/bdf.py b/host/bdf.py index a8fc8f3..fb4c928 100644 --- a/host/bdf.py +++ b/host/bdf.py @@ -40,7 +40,7 @@ class Font: textw, texth = c_size_t(0), c_size_t(0) res = lib.framebuffer_get_text_bounds(textbytes, self.font, byref(textw), byref(texth)) if res: - raise ValueError('Invalid text') + raise RuntimeError('Invalid text') return textw.value, texth.value def render_text(self, text, offset): @@ -49,7 +49,7 @@ class Font: res = lib.framebuffer_render_text(textbytes, self.font, self.cbuf, config.display_width, config.display_height, offset) if res: - raise ValueError('Invalid text') + raise RuntimeError('Invalid text') return self.cbuf unifont = Font() diff --git a/host/config.py b/host/config.py index 02a047c..b76e2aa 100644 --- a/host/config.py +++ b/host/config.py @@ -44,3 +44,7 @@ crap_fw_addr, crap_fw_port = '127.0.0.1', 1338 # USB Serial number of matelight to control as byte string (None for first matelight connected) ml_usb_serial_match = None +# Maximum width allowed for marquee texts in px. For reference: Using GNU unifont, a normal (half-width) char such as ∀ +# is 8px wide, a full-width char such as 水 is 16px wide. +max_marquee_width = 140*8 + diff --git a/host/server.py b/host/server.py index f83e2cd..e622d9b 100755 --- a/host/server.py +++ b/host/server.py @@ -4,9 +4,12 @@ import socket import time import itertools import sys +import re from contextlib import suppress import asyncio import threading +import functools +import operator import config @@ -14,86 +17,100 @@ import matelight import bdf import crap +import secret_sauce + def log(*args): - print(time.strftime('\x1B[93m[%m-%d %H:%M:%S]\x1B[0m'), ' '.join(str(arg) for arg in args), '\x1B[0m') - sys.stdout.flush() + print(time.strftime('\x1B[93m[%m-%d %H:%M:%S]\x1B[0m'), ' '.join(str(arg) for arg in args), '\x1B[0m') + sys.stdout.flush() -class TextRenderer: - def __init__(self, text, title='default', font=bdf.unifont): - self.text = text - self.font = font - (self.width, _), _testrender = font.compute_text_bounds(text), font.render_text(text, 0) - self.title = title +def addrcolor(addr): + col = 16 + (functools.reduce(operator.xor, (int(e or '0') for e in re.split('[.:]', addr))) % 216) + return '\x1B[38;5;{}m{}\x1B[0m'.format(col, addr) - def __iter__(self): - for i in range(-config.display_width, self.width): - yield self.title, self.font.render_text(self.text, i) - time.sleep(0.05) + +class TextRenderer: + def __init__(self, text, title='default', checkwidth=False, font=bdf.unifont): + self.text = text + self.font = font + (self.width, _), _testrender = font.compute_text_bounds(text), font.render_text(text, 0) + self.title = title + if self.width > config.max_marquee_width: + raise ValueError() + + def __iter__(self): + for i in range(-config.display_width, self.width): + yield self.title, self.font.render_text(self.text, i) + time.sleep(0.05) class MatelightTCPServer: - def __init__(self, ip, port, loop): - coro = asyncio.start_server(self.handle_conn, ip, port, loop=loop) - server = loop.run_until_complete(coro) - self.renderqueue = [] - self.close = server.close - - def __iter__(self): - q, self.renderqueue = self.renderqueue, [] - for title, frame in itertools.chain(*q): - yield title, frame - - async def handle_conn(self, reader, writer): - line = (await reader.read(1024)).decode('UTF-8').strip() - if len(line) > 140: # Unicode string length, *not* byte length of encoded UTF-8 - writer.write(b'TOO MUCH INFORMATION!\n') - else: - addr,*rest = writer.get_extra_info('peername') - log('\x1B[95mText from\x1B[0m {}: {}\x1B[0m'.format(addr, line)) - try: - self.renderqueue.append(TextRenderer(line, title='tcp:'+addr)) - except: - writer.write(b'STAHPTROLLINK?\n') - else: - writer.write(b'KTHXBYE!\n') - await writer.drain() - writer.close() + def __init__(self, ip, port, loop): + coro = asyncio.start_server(self.handle_conn, ip, port, loop=loop) + server = loop.run_until_complete(coro) + self.renderqueue = [] + self.close = server.close + + def __iter__(self): + q, self.renderqueue = self.renderqueue, [] + for title, frame in itertools.chain(*q): + yield title, frame + + async def handle_conn(self, reader, writer): + line = (await reader.read(1024)).decode('UTF-8').strip() + addr,*rest = writer.get_extra_info('peername') + + log('\x1B[95mText from\x1B[0m {}: {}\x1B[0m'.format(addrcolor(addr), line)) + try: + secret_sauce.check_spam(str(addr), line) + renderer = TextRenderer(line, title='tcp:'+addr, checkwidth=True) + except secret_sauce.SpamError as err: + log('\x1B[91mMessage rejected from {}: {}'.format(addrcolor(addr), err)) + writer.write(b'BLERGH!\n') + except ValueError: # Text too long + writer.write(b'TOO MUCH INFORMATION!\n') + except RuntimeError: # Invalid escape etc. + writer.write(b'STAHPTROLLINK?\n') + else: + self.renderqueue.append(renderer) + writer.write(b'KTHXBYE!\n') + await writer.drain() + writer.close() def _fallbackiter(it, fallback): - for fel in fallback: - for el in it: - yield el - yield fel + for fel in fallback: + for el in it: + yield el + yield fel def defaulttexts(filename='default.lines'): - with open(filename) as f: - return itertools.chain.from_iterable(( TextRenderer(l[:-1].replace('\\x1B', '\x1B')) for l in f.readlines() )) + with open(filename) as f: + return itertools.chain.from_iterable(( TextRenderer(l[:-1].replace('\\x1B', '\x1B')) for l in f.readlines() )) if __name__ == '__main__': - try: - ml = matelight.Matelight(config.ml_usb_serial_match) - except ValueError as e: - print(e, 'Starting in headless mode.', file=sys.stderr) - ml = None - - loop = asyncio.get_event_loop() - - tcp_server = MatelightTCPServer(config.tcp_addr, config.tcp_port, loop) - udp_server = crap.CRAPServer(config.udp_addr, config.udp_port) - forwarder = crap.CRAPClient(config.crap_fw_addr, config.crap_fw_port) if config.crap_fw_addr is not None else None - - async_thr = threading.Thread(target=loop.run_forever) - async_thr.daemon = True - async_thr.start() - - with suppress(KeyboardInterrupt): - while True: - for title, frame in _fallbackiter(udp_server, _fallbackiter(tcp_server, defaulttexts())): - if ml: - ml.sendframe(frame) - if forwarder: - forwarder.sendframe(frame) - - tcp_server.close() - udp_server.close() - forwarder.close() + try: + ml = matelight.Matelight(config.ml_usb_serial_match) + except ValueError as e: + print(e, 'Starting in headless mode.', file=sys.stderr) + ml = None + + loop = asyncio.get_event_loop() + + tcp_server = MatelightTCPServer(config.tcp_addr, config.tcp_port, loop) + udp_server = crap.CRAPServer(config.udp_addr, config.udp_port) + forwarder = crap.CRAPClient(config.crap_fw_addr, config.crap_fw_port) if config.crap_fw_addr is not None else None + + async_thr = threading.Thread(target=loop.run_forever) + async_thr.daemon = True + async_thr.start() + + with suppress(KeyboardInterrupt): + while True: + for title, frame in _fallbackiter(udp_server, _fallbackiter(tcp_server, defaulttexts())): + if ml: + ml.sendframe(frame) + if forwarder: + forwarder.sendframe(frame) + + tcp_server.close() + udp_server.close() + forwarder.close() |