"""Panels to interact with AbiPy flows."""
from __future__ import annotations
import panel as pn
import panel.widgets as pnw
import param
# import bokeh.models.widgets as bkw
from panel.viewable import Viewer
from abipy.flowtk.flows import Flow
from abipy.flowtk.tasks import AbinitTask
from abipy.panels.core import depends_on_btn_click, dfc
from abipy.panels.nodes import NodeParameterized
[docs]
class WorkTaskSelector(Viewer):
"""
Panel widget to select a Work and a Task within a Flow.
"""
task = param.ClassSelector(class_=AbinitTask, doc="Task object")
def __init__(self, flow: Flow, **params):
"""
Args:
flow: |Flow| object.
params: Parameters passed to the parent class.
"""
self._wstr2work = {
f"w{i} ({work.__class__.__name__}, len: {len(work)})": work for (i, work) in enumerate(flow.works)
}
options = list(self._wstr2work.keys())
self.work_select = pnw.Select(name="Select a Work", value=options[0], options=options)
options = [f"t{i} ({task.__class__.__name__}, {task.status!s})" for (i, task) in enumerate(flow[0])]
self.task_select = pnw.Select(name="Select a Task in the Work", value=options[0], options=options)
super().__init__(**params)
self.layout = pn.Column(self.work_select, self.task_select)
self.sync_widgets()
def __panel__(self):
return self.layout
[docs]
@pn.depends("work_select.value", watch=True)
def update_work(self) -> None:
"""Update the selected Work."""
self.work = self._wstr2work[self.work_select.value]
self.task_select.options = [
f"t{i} ({task.__class__.__name__}, {task.status!s})" for (i, task) in enumerate(self.work)
]
self.task = self.work[0]
[docs]
class FlowPanel(NodeParameterized):
"""Provides widgets and callbacks to interact with an AbiPy Flow."""
def __init__(self, flow: Flow, **params):
"""
Args:
flow: |Flow| object.
params: Parameters passed to the parent class.
"""
NodeParameterized.__init__(self, node=flow, **params)
self.structures_btn = pnw.Button(name="Show Structures", button_type="primary")
self.structures_io_checkbox = pnw.CheckBoxGroup(
name="Input/Output Structure", value=["output"], options=["input", "output"], inline=True
)
self.wt_selector = WorkTaskSelector(flow)
self.task_btn = pnw.Button(name="Analyze Task", button_type="primary")
[docs]
def get_task_view(self) -> pn.Column:
"""Return the task view panel."""
wbox = pn.WidgetBox
return pn.Column(
wbox(
"## Select Work and Task",
self.wt_selector,
self.task_btn,
),
pn.layout.Divider(),
self.on_task_btn,
sizing_mode="stretch_width",
)
[docs]
@depends_on_btn_click("task_btn")
def on_task_btn(self) -> pn.Column:
"""Return panel associated to the selected task."""
task = self.wt_selector.task
return pn.Column(
f"## {task!r}",
task.get_panel(),
sizing_mode="stretch_width",
)
[docs]
@depends_on_btn_click("structures_btn")
def on_structures_btn(self) -> pn.Row:
"""Callback for structures button click."""
what = ""
if "input" in self.structures_io_checkbox.value:
what += "i"
if "output" in self.structures_io_checkbox.value:
what += "o"
dfs = self.flow.compare_structures(
nids=None, # select_nids(flow, options),
what=what,
verbose=self.verbose,
with_spglib=False,
printout=False,
with_colors=False,
)
return pn.Row(dfc(dfs.lattice), sizing_mode="scale_width")
[docs]
def get_panel(self, as_dict=False, **kwargs):
"""Return tabs with widgets to interact with the flow."""
d = super().get_panel(as_dict=True)
# row = pn.Row(bkw.PreText(text=self.ddb.to_string(verbose=self.verbose), sizing_mode="scale_both"))
d["Task"] = self.get_task_view()
# d["Work"] = self.get_work_view()
# d["Structures"] = pn.Row(pn.Column(self.structures_io_checkbox, self.structures_btn), self.on_structures_btn)
###ws = pn.Column(self.ebands_plotter_mode, self.ebands_ksamp_checkbox, self.ebands_df_checkbox, self.ebands_plotter_btn)
###d["Ebands"] = pn.Row(ws, self.on_ebands_btn)
# d["Browse"] = self.get_workdir_view()
if as_dict:
return d
return self.get_template_from_tabs(d, template=kwargs.get("template"), closable=False)
[docs]
class JsPane(pn.pane.HTML):
"""Based on: https://discourse.holoviz.org/t/how-to-make-a-dynamic-link-in-panel/2137"""
def __init__(self):
"""Initialize the object."""
super().__init__(width=0, height=0, margin=0, sizing_mode="fixed")
[docs]
def execute_js(self, script: str) -> None:
"""Execute javascript script."""
script = f'<script type="text/javascript">{script}</script>'
self.object = script
self.object = ""
[docs]
class FlowMultiPageApp:
"""
Multi-page application for interacting with an AbiPy Flow.
"""
def __init__(self, flow: Flow, template, spectator_mode=True, **kwargs):
"""
Args:
flow: |Flow| object.
template: Panel template.
spectator_mode: True if flow is in spectator mode.
kwargs: Extra parameters.
"""
if spectator_mode:
# We are in read-only mode so we have to disable signals to avoid side effects and callbacks
flow.set_spectator_mode()
self.flow = flow
self.template = template
self.wt_selector = WorkTaskSelector(flow)
goto_work_btn = pnw.Button(name="Go to Work", button_type="primary")
goto_work_btn.on_click(self.on_goto_work_bnt)
goto_task_btn = pnw.Button(name="Go to Task", button_type="primary")
goto_task_btn.on_click(self.on_goto_task_bnt)
self.new_tab = pnw.Checkbox(value=True, name="Open in new Tab")
self.js_panel = JsPane()
self.sidebar = pn.WidgetBox(
self.wt_selector,
self.new_tab,
pn.layout.Divider(),
goto_work_btn,
goto_task_btn,
self.js_panel,
)
def handle_home():
app = FlowPanel(self.flow).get_panel(template=template)
if hasattr(app, "sidebar"):
app.sidebar.append(self.sidebar)
# app.header.append(self.sidebar)
return app
# url --> handler
self.routes = {
"/": handle_home,
r"/w\d+/?": self.handle_wt,
r"/w\d+/t\d+": self.handle_wt,
}
[docs]
def on_goto_work_bnt(self, event) -> None:
"""Callback for 'Go to Work' button click."""
work = self.wt_selector.work
url = "/w%d" % work.pos
code = f"window.open('{url}')" if self.new_tab.value else f"window.location.href='{url}'"
self.js_panel.execute_js(code)
[docs]
def on_goto_task_bnt(self, event) -> None:
"""Callback for 'Go to Task' button click."""
task = self.wt_selector.task
url = "/w%d/t%d" % (task.pos[0], task.pos[1])
code = f"window.open('{url}')" if self.new_tab.value else f"window.location.href='{url}'"
self.js_panel.execute_js(code)
[docs]
def handle_wt(self):
"""Handle Work/Task route."""
# URL example: /w1/t5/w\d+/t\d+
# print("in handle_wt with pn.state.app_url:", pn.state.app_url)
tokens = pn.state.app_url.split("/")
work_idx = int(tokens[1][1:])
task_idx = None
if tokens[2].startswith("t"):
task_idx = int(tokens[2][1:])
# print("got request with work_idx:", work_idx, "task_idx:", task_idx)
# from abipy.panels.core import abipanel
# abipanel()
if task_idx is None:
work = self.flow[work_idx]
app = work.get_panel(template=self.template, title=repr(work))
else:
task = self.flow[work_idx][task_idx]
app = task.get_panel(template=self.template, title=repr(task))
if hasattr(app, "sidebar"):
app.sidebar.append(self.sidebar)
# app.header.append(self.sidebar)
# app.title = title
return app
[docs]
def serve(self, **serve_kwargs):
"""Serve the application."""
return pn.serve(self.routes, **serve_kwargs)