lotus/composer/app/runner.py
2020-06-23 10:02:05 -04:00

112 lines
3.5 KiB
Python

import os
import panel as pn
import param
from panel.io.server import unlocked
from tornado.ioloop import IOLoop, PeriodicCallback
from tornado.process import Subprocess
from subprocess import STDOUT
from bokeh.models.widgets import Div
from ansi2html import Ansi2HTMLConverter
from .composition import Composition
TESTGROUND = 'testground'
class AnsiColorText(pn.widgets.Widget):
style = param.Dict(default=None, doc="""
Dictionary of CSS property:value pairs to apply to this Div.""")
value = param.Parameter(default=None)
_format = '<div>{value}</div>'
_rename = {'name': None, 'value': 'text'}
# _target_transforms = {'value': 'target.text.split(": ")[0]+": "+value'}
#
# _source_transforms = {'value': 'value.split(": ")[1]'}
_widget_type = Div
_converter = Ansi2HTMLConverter(inline=True)
def _process_param_change(self, msg):
msg = super(AnsiColorText, self)._process_property_change(msg)
if 'value' in msg:
text = str(msg.pop('value'))
text = self._converter.convert(text)
msg['text'] = text
return msg
def scroll_down(self):
# TODO: figure out how to automatically scroll down as text is added
pass
class CommandRunner(param.Parameterized):
command_output = param.String()
def __init__(self, **params):
super().__init__(**params)
self._output_lines = []
self.proc = None
self._updater = PeriodicCallback(self._refresh_output, callback_time=1000)
@pn.depends('command_output')
def panel(self):
return pn.Param(self.param, show_name=False, sizing_mode='stretch_width', widgets={
'command_output': dict(
type=AnsiColorText,
sizing_mode='stretch_width',
height=800)
})
def run(self, *cmd):
self.command_output = ''
self._output_lines = []
self.proc = Subprocess(cmd, stdout=Subprocess.STREAM, stderr=STDOUT)
self._get_next_line()
self._updater.start()
def _get_next_line(self):
if self.proc is None:
return
loop = IOLoop.current()
loop.add_future(self.proc.stdout.read_until(bytes('\n', encoding='utf8')), self._append_output)
def _append_output(self, future):
self._output_lines.append(future.result().decode('utf8'))
self._get_next_line()
def _refresh_output(self):
text = ''.join(self._output_lines)
if len(text) != len(self.command_output):
with unlocked():
self.command_output = text
class TestRunner(param.Parameterized):
composition = param.ClassSelector(class_=Composition, precedence=-1)
testground_daemon_endpoint = param.String(default="{}:8042".format(os.environ.get('TESTGROUND_DAEMON_HOST', 'localhost')))
run_test = param.Action(lambda self: self.run())
runner = CommandRunner()
def __init__(self, **params):
super().__init__(**params)
def run(self):
# TODO: temp file management - maybe we should mount a volume and save there?
filename = '/tmp/composition.toml'
self.composition.write_to_file(filename)
self.runner.run(TESTGROUND, '--endpoint', self.testground_daemon_endpoint, 'run', 'composition', '-f', filename)
def panel(self):
return pn.Column(
self.param['testground_daemon_endpoint'],
self.param['run_test'],
self.runner.panel(),
sizing_mode='stretch_width',
)