You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

200 lines
6.3 KiB
Python

# bureau
import functools
import json
import os.path
import subprocess
import tempfile
import threading
import zmq
from mako.template import Template
def update_commands(cls):
for name, method in cls.__dict__.items():
print("name %s method %s" % (name, method))
if hasattr(method, "command"):
comstr = method.command
cls.commands[comstr] = method
return cls
def add_command(comstr, name=""):
def decorator(func):
@functools.wraps(func)
def func_wrap(*args, **kwargs):
return func(*args, **kwargs)
func_wrap.command = comstr
func_wrap.name = name
return func_wrap
return decorator
def add_api(apistr, name=""):
def decorator(func):
@functools.wraps(func)
def func_wrap(*args, **kwargs):
return func(*args, **kwargs)
func_wrap.api = apistr
func_wrap.name = name
return func_wrap
return decorator
# @update_commands
class Bureau:
name = "TEST"
prefix = "00"
version = 0
config = {}
def __init__(self):
""" set up ZeroMQ connections and register commands"""
self.commands = {}
self.api = {}
self.context = zmq.Context()
self._recv = self.context.socket(zmq.SUB)
self._recv.connect("tcp://localhost:10101")
self._recv.setsockopt_string(zmq.SUBSCRIBE, self.prefix)
print(("bureau " + self.name + " waiting for messages"))
self._send = self.context.socket(zmq.PUB)
self._send.connect("tcp://localhost:10100")
# update_commands(self)
# self.registerCommands()
print("commands: ")
print(self.commands)
def send(self, message, data=None):
if data:
message += json.dumps(data)
self._send.send_string(message)
def _publish_methods(self):
# register bureau with Inhuman Resources
bureau_json = json.dumps({"name": self.name, "prefix": self.prefix,
"desc": self.__doc__})
self.send("IRaddbureau." + bureau_json)
# find and store all published methods
for member in dir(self):
method = getattr(self, member)
# ignore anything that is not a method with command or api details
if not (callable(method) and (hasattr(method, "command") or
hasattr(method, "api"))):
continue
if hasattr(method, "command"):
self.commands[method.command] = method
cmd_json = json.dumps({"cmdname": method.name,
"prefix": self.prefix,
"cmd": method.command,
"desc": method.__doc__})
self.send("IRaddcommand." + cmd_json)
elif hasattr(method, "api"):
self.api[method.api] = method
cmd_json = json.dumps({"apiname": method.name,
"prefix": self.prefix,
"api": method.api,
"desc": method.__doc__})
self.send("IRaddapi." + cmd_json)
print("registered:")
print(self.commands)
print(self.api)
def print_full(self, template, **kwargs):
"""print a full page (A4) document """
# TODO: look up the printer LPR name
lpname = kwargs.get("printer", "default")
templ = Template(filename=template)
texfile, texfilepath = tempfile.mkstemp(".tex")
texfile.write(templ.render_unicode(
**kwargs).encode('utf-8', 'replace'))
texdir = os.path.dirname(texfilepath)
subprocess.call("cd " + texdir + "; xelatex " + texfilepath)
pdffile = texfilepath[0:-4] + ".pdf"
subprocess.call("lpr -P " + lpname + " " + pdffile)
def print_small(self, text, printer="/dev/usb/lp1"):
lp = open(printer, "w")
text += "\r\n" * 10
text += ".d0"
lp.write(text)
lp.close()
@add_command("test")
def test(self):
# stupid test to see if modules work
print(("hi! testing. " + self.name + " bureau seems to work!"))
def run_io(self):
"""process hardware or timed input
This method can be ignored for most Bureaus.
It should be overloaded for any services that need to independently
generate messages as this is just a placeholder.
It is run in a separate thread and should handle any input or
timed events. Messages are then sent to other Bureaus via the
self.send connection to the OfficeManager. Don't forget to
to consier thread safety issues when accessing data!
"""
pass
def run(self):
"""main loop for processing messages
This runs all relelvant event processing loops.
"""
# start the hardware input handler
io = threading.Thread(target=self.run_io)
io.start()
# register commands and api methods
self._publish_methods()
while True:
msg = self._recv.recv_string()
try:
dot = msg.find(".")
ref = msg[2:dot]
if (dot < len(msg) - 1) and (dot > 0):
data = json.loads(msg[dot + 1:])
else:
data = None
print("data: " + str(data))
except IndexError as e:
print("invalid message: ", e)
continue
print(("got method: " + ref))
if (ref in self.commands) or (ref in self.api):
# catch TypeErrors for case of bogus params
try:
if data:
self.api[ref](data)
else:
self.commands[ref]()
except TypeError as e:
print(e)
print("invalid data for command '{}': {}".format(ref, data))
except KeyError as e:
print(e)
print("You are calling a command as an API or vice-versa.")
else:
print("error! Command/API %s not found", ref)
if __name__ == "__main__":
test = Bureau("test", "00")
test.run()