Add network repl to control the camp atmo

master
Sebastian Lohff 5 years ago
parent d56ef5f64e
commit 87f6bf8e21

@ -0,0 +1,83 @@
#!/usr/bin/env python3
import socketserver
import threading
class WrongArgumentCount(Exception):
def __init__(self, command, expected, found):
if isinstance(expected, list):
expected = "{} to {}".format(**expected)
msg = "Command {} expects {} arguments, found {}".format(command, expected, found)
super(WrongArgumentCount, self).__init__(msg)
class CommandNotFound(Exception):
def __init__(self, command):
msg = "Command {} does not exist".format(command)
super(CommandNotFound, self).__init__(msg)
class AtmoClientHandler(socketserver.StreamRequestHandler):
def setup(self):
super(AtmoClientHandler, self).setup()
self.authenticated = True
self.quit = False
def handle(self):
try:
while not self.quit:
self.send(">>> ", end='')
data = self.rfile.readline().decode().strip()
tokens = data.split()
if len(tokens) > 0:
self.handle_command(tokens)
except BrokenPipeError:
print(" >> Client disconnected")
def handle_command(self, tokens):
cmd = tokens[0].lower()
args = tokens[1:]
try:
if cmd == "reload":
self._ensure_args(cmd, 0, args)
self.atmo.load_sounds()
self.send("Sounds reloaded")
elif cmd == "help":
self._ensure_args(cmd, 0, args)
self.send("The following commands are available: ")
self.send(" reload")
elif cmd == "exit":
self.quit = True
self.send("Bye")
else:
raise CommandNotFound(cmd)
except (CommandNotFound, WrongArgumentCount) as e:
self.send("Error: {}".format(e))
@staticmethod
def _ensure_args(cmd, expected, args):
if isinstance(expected, list):
if not expected[0] <= len(args) <= expected[1]:
raise WrongArgumentCount(cmd, expected, len(args))
else:
if len(args) != expected:
raise WrongArgumentCount(cmd, expected, len(args))
def send(self, line, end='\n'):
self.wfile.write("{}{}".format(line, end).encode())
class ReusableThreadingTCPServer(socketserver.ThreadingTCPServer):
daemon_threads = True
allow_reuse_address = True
class AtmoReplRunner(threading.Thread):
def __init__(self, atmo, host='0.0.0.0', port=7723):
super(AtmoReplRunner, self).__init__()
AtmoClientHandler.atmo = atmo
self._client = ReusableThreadingTCPServer((host, port), AtmoClientHandler)
self.daemon = True
def run(self):
self._client.serve_forever()

@ -5,6 +5,7 @@ import random
from glob import glob
from pygame import mixer
from atmorepl import AtmoReplRunner
VERBOSITY = 0
@ -80,6 +81,8 @@ class CampAtmo:
def main():
atmo = CampAtmo()
atmoRepl = AtmoReplRunner(atmo, port=7723)
atmoRepl.start()
try:
atmo.run_forever()
except KeyboardInterrupt:

Loading…
Cancel
Save