import param import panel as pn import toml from .util import get_plans, get_manifest from .composition import Composition from .runner import TestRunner STAGE_WELCOME = 'Welcome' STAGE_CONFIG_COMPOSITION = 'Configure' STAGE_RUN_TEST = 'Run' class Welcome(param.Parameterized): composition = param.Parameter() composition_picker = pn.widgets.FileInput(accept='.toml') plan_picker = param.Selector() ready = param.Boolean() def __init__(self, **params): super().__init__(**params) self.composition_picker.param.watch(self._composition_updated, 'value') self.param.watch(self._plan_selected, 'plan_picker') self.param['plan_picker'].objects = ['Select a Plan'] + get_plans() def panel(self): tabs = pn.Tabs( ('New Compostion', self.param['plan_picker']), ('Existing Composition', self.composition_picker), ) return pn.Column( "Either choose an existing composition or select a plan to create a new composition:", tabs, ) def _composition_updated(self, *args): print('composition updated') content = self.composition_picker.value.decode('utf8') comp_toml = toml.loads(content) manifest = get_manifest(comp_toml['global']['plan']) self.composition = Composition.from_dict(comp_toml, manifest=manifest) print('existing composition: {}'.format(self.composition)) self.ready = True def _plan_selected(self, evt): if evt.new == 'Select a Plan': return print('plan selected: {}'.format(evt.new)) manifest = get_manifest(evt.new) self.composition = Composition(manifest=manifest, add_default_group=True) print('new composition: ', self.composition) self.ready = True class ConfigureComposition(param.Parameterized): composition = param.Parameter() @param.depends('composition') def panel(self): if self.composition is None: return pn.Pane("no composition :(") print('composition: ', self.composition) return self.composition.panel() class WorkflowPipeline(object): def __init__(self): stages = [ (STAGE_WELCOME, Welcome(), dict(ready_parameter='ready')), (STAGE_CONFIG_COMPOSITION, ConfigureComposition()), (STAGE_RUN_TEST, TestRunner()), ] self.pipeline = pn.pipeline.Pipeline(debug=True, stages=stages) def panel(self): return pn.Column( pn.Row( self.pipeline.title, self.pipeline.network, self.pipeline.prev_button, self.pipeline.next_button, ), self.pipeline.stage, sizing_mode='stretch_width', ) class App(object): def __init__(self): self.workflow = WorkflowPipeline() def ui(self): return self.workflow.panel().servable("Testground Composer")