Commit ed7f35f1 authored by Paul Tolstoi's avatar Paul Tolstoi

first working version

parent aeeb1764
Pipeline #151 skipped
from .knechtqt import KnechtQT
import argparse
from . import KnechtQT
if __name__ == '__main__':
cli = argparse.ArgumentParser()
cli.add_argument("-p", "--port", default=60602,
type=int, help="port to listen on")
cli.add_argument("-l", "--listen", default="0.0.0.0",
help="ip to listen on")
args = cli.parse_args()
knechtqt = KnechtQT(args)
knechtqt.launch()
import asyncio
import re
class CommandReceiver(asyncio.Protocol):
LINEBUF_MAX = 1024 * 512
def __init__(self, knechtqt):
self.buf = bytearray()
self.knechtqt = knechtqt
self.peername = None
def connection_made(self, transport):
self.transport = transport
self.peername = transport.get_extra_info('peername')
def data_received(self, data):
if len(self.buf) + len(data) > self.LINEBUF_MAX:
self.transport.write(b'too much!\n')
self.close()
return
self.buf += data
npos = self.buf.find(b"\n")
if npos != -1:
self.transport.write(b"yay!\n")
self.transport.close()
self._process_command(self.buf.decode())
def _process_command(self, command):
regex = re.compile(r'^(?:([0-9]+):)?(.*)$')
match = regex.match(command)
if match:
text = match.group(2)
duration = match.group(1)
if duration and duration.isdigit():
duration = int(duration)
else:
duration = None
self.knechtqt.show_text(text, duration)
else:
print('Should never reach this point')
def connection_lost(self, exc):
pass
import sys
import signal
import asyncio
import time
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QApplication, QMainWindow, QLabel
from quamash import QEventLoop, QThreadExecutor
from .commandreceiver import CommandReceiver
class KnechtQT:
def __init__(self, args):
self.port = args.port
self.address = args.listen
self.app = None
self.window = None
self.loop = None
def _close_window(self):
if self.window:
self.window.showNormal()
self.window.hide()
self.window.close()
self.window = None
def show_text(self, text, duration=None):
if duration is None or duration <= 0:
duration = 10
if self.window:
self._close_window()
label = QLabel()
label.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
label.setTextFormat(Qt.RichText)
label.setWordWrap(True)
label.setText('<span style="font-size:100px">%s</span>' % text)
self.window = QMainWindow()
self.window.setWindowTitle("KnechtQT")
self.window.setCentralWidget(label)
self.window.showFullScreen()
self.loop.call_later(duration, self._close_window)
def launch(self):
self.app = QApplication([])
self.app.setQuitOnLastWindowClosed(False)
self.loop = QEventLoop(self.app)
asyncio.set_event_loop(self.loop)
signal.signal(signal.SIGINT, signal.SIG_DFL)
with self.loop:
coro = self.loop.create_server(lambda: CommandReceiver(self),
self.address, self.port)
server = self.loop.run_until_complete(coro)
self.loop.run_forever()
server.close()
loop.run_until_complete(server.wait_closed())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment