1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
|
#!/usr/bin/env python
import socket
import time
import itertools
import sys
from contextlib import suppress
import asyncio
import threading
import config
import matelight
import bdf
import crap
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()
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 __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 _fallbackiter(it, fallback):
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() ))
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()
|
