Source code for artiq.dashboard.applets_ccb

import asyncio
import logging

from PyQt5 import QtCore, QtWidgets

from artiq.gui import applets


logger = logging.getLogger(__name__)


[docs] class AppletsCCBDock(applets.AppletsDock): def __init__(self, *args, **kwargs): applets.AppletsDock.__init__(self, *args, **kwargs) sep = QtWidgets.QAction(self.table) sep.setSeparator(True) self.table.addAction(sep) ccbp_group_menu = QtWidgets.QMenu() actiongroup = QtWidgets.QActionGroup(self.table) actiongroup.setExclusive(True) self.ccbp_group_none = QtWidgets.QAction("No policy", self.table) self.ccbp_group_none.setCheckable(True) self.ccbp_group_none.triggered.connect(lambda: self.set_ccbp("")) ccbp_group_menu.addAction(self.ccbp_group_none) actiongroup.addAction(self.ccbp_group_none) self.ccbp_group_ignore = QtWidgets.QAction("Ignore requests", self.table) self.ccbp_group_ignore.setCheckable(True) self.ccbp_group_ignore.triggered.connect(lambda: self.set_ccbp("ignore")) ccbp_group_menu.addAction(self.ccbp_group_ignore) actiongroup.addAction(self.ccbp_group_ignore) self.ccbp_group_create = QtWidgets.QAction("Create applets", self.table) self.ccbp_group_create.setCheckable(True) self.ccbp_group_create.triggered.connect(lambda: self.set_ccbp("create")) ccbp_group_menu.addAction(self.ccbp_group_create) actiongroup.addAction(self.ccbp_group_create) self.ccbp_group_enable = QtWidgets.QAction("Create and enable/disable applets", self.table) self.ccbp_group_enable.setCheckable(True) self.ccbp_group_enable.triggered.connect(lambda: self.set_ccbp("enable")) ccbp_group_menu.addAction(self.ccbp_group_enable) actiongroup.addAction(self.ccbp_group_enable) self.ccbp_group_action = QtWidgets.QAction("Group CCB policy", self.table) self.ccbp_group_action.setMenu(ccbp_group_menu) self.table.addAction(self.ccbp_group_action) self.table.itemSelectionChanged.connect(self.update_group_ccbp_menu) self.update_group_ccbp_menu() ccbp_global_menu = QtWidgets.QMenu() actiongroup = QtWidgets.QActionGroup(self.table) actiongroup.setExclusive(True) self.ccbp_global_ignore = QtWidgets.QAction("Ignore requests", self.table) self.ccbp_global_ignore.setCheckable(True) ccbp_global_menu.addAction(self.ccbp_global_ignore) actiongroup.addAction(self.ccbp_global_ignore) self.ccbp_global_create = QtWidgets.QAction("Create applets", self.table) self.ccbp_global_create.setCheckable(True) self.ccbp_global_create.setChecked(True) ccbp_global_menu.addAction(self.ccbp_global_create) actiongroup.addAction(self.ccbp_global_create) self.ccbp_global_enable = QtWidgets.QAction("Create and enable/disable applets", self.table) self.ccbp_global_enable.setCheckable(True) ccbp_global_menu.addAction(self.ccbp_global_enable) actiongroup.addAction(self.ccbp_global_enable) ccbp_global_action = QtWidgets.QAction("Global CCB policy", self.table) ccbp_global_action.setMenu(ccbp_global_menu) self.table.addAction(ccbp_global_action) def update_group_ccbp_menu(self): selection = self.table.selectedItems() if selection: item = selection[0] if item.ty == "applet": item = item.parent() if item is None: self.ccbp_group_action.setEnabled(False) else: self.ccbp_group_action.setEnabled(True) ccbp = item.text(1) if ccbp == "": self.ccbp_group_none.setChecked(True) else: getattr(self, "ccbp_group_" + ccbp).setChecked(True) else: self.ccbp_group_action.setEnabled(False) def set_ccbp(self, ccbp): item = self.table.selectedItems()[0] if item.ty == "applet": item = item.parent() item.setText(1, ccbp) def get_ccpb_global(self): if self.ccbp_global_ignore.isChecked(): return "ignore" if self.ccbp_global_create.isChecked(): return "create" if self.ccbp_global_enable.isChecked(): return "enable" def get_ccpb(self, group): ccbp = self.get_ccpb_global() parent = self.table.invisibleRootItem() for g in group: new_parent = None for i in range(parent.childCount()): child = parent.child(i) if child.ty == "group" and child.text(0) == g: c_ccbp = child.text(1) if c_ccbp: ccbp = c_ccbp new_parent = child break if new_parent is None: return ccbp else: parent = new_parent return ccbp def locate_applet(self, name, group, create_groups): parent = self.table.invisibleRootItem() for g in group: new_parent = None for i in range(parent.childCount()): child = parent.child(i) if child.ty == "group" and child.text(0) == g: new_parent = child break if new_parent is None: if create_groups: new_parent = self.new_group(g, parent=parent) else: return None, None parent = new_parent applet = None for i in range(parent.childCount()): child = parent.child(i) if child.ty == "applet" and child.text(0) == name: applet = child break return parent, applet
[docs] def ccb_create_applet(self, name, command, group=None, code=None): """Requests the creation of a new applet. An applet is identified by its name and an optional list of groups that represent a path (nested groups). If ``group`` is a string, it corresponds to a single group. If ``group`` is ``None`` or an empty list, it corresponds to the root. ``command`` gives the command line used to run the applet, as if it was started from a shell. The dashboard substitutes variables such as ``$python`` that gives the complete file name of the Python interpreter running the dashboard. If the name already exists (after following any specified groups), the command or code of the existing applet with that name is replaced, and the applet is restarted and shown at its previous position. If not, a new applet entry is created and the applet is shown at any position on the screen. If the group(s) do not exist, they are created. If ``code`` is not ``None``, it should be a string that contains the full source code of the applet. In this case, ``command`` is used to specify (optional) command-line arguments to the applet. This function is called when a CCB ``create_applet`` is issued. """ if group is None: group = [] elif isinstance(group, str): group = [group] ccbp = self.get_ccpb(group) if ccbp == "ignore": return parent, applet = self.locate_applet(name, group, True) if code is None: spec = {"ty": "command", "command": command} else: spec = {"ty": "code", "code": code, "command": command} if applet is None: logger.debug("Applet %s does not exist: creating", name) applet = self.new(name=name, spec=spec, parent=parent) else: if spec != self.get_spec(applet): logger.debug("Applet %s already exists: updating existing spec", name) self.set_spec(applet, spec) if applet.applet_dock: asyncio.ensure_future(applet.applet_dock.restart()) else: logger.debug("Applet %s already exists and no update required", name) if ccbp == "enable": applet.setCheckState(0, QtCore.Qt.Checked)
[docs] def ccb_disable_applet(self, name, group=None): """Disables an applet. The applet is identified by its name, after following any specified groups. This function is called when a CCB ``disable_applet`` is issued. """ if group is None: group = [] elif isinstance(group, str): group = [group] ccbp = self.get_ccpb(group) if ccbp != "enable": return parent, applet = self.locate_applet(name, group, False) if applet is not None: applet.setCheckState(0, QtCore.Qt.Unchecked)
[docs] def ccb_disable_applet_group(self, group): """Disables all the applets in a group. If the group is nested, ``group`` should be a list, with the names of the parents preceding the name of the group to disable. This function is called when a CCB ``disable_applet_group`` is issued. """ if isinstance(group, str): group = [group] ccbp = self.get_ccpb(group) if ccbp != "enable": return if not group: return wi = self.table.invisibleRootItem() for g in group: nwi = None for i in range(wi.childCount()): child = wi.child(i) if child.ty == "group" and child.text(0) == g: nwi = child break if nwi is None: return else: wi = nwi wi.setCheckState(0, QtCore.Qt.Unchecked)
def ccb_notify(self, message): try: service = message["service"] args = message["args"] kwargs = message["kwargs"] if service == "create_applet": self.ccb_create_applet(*args, **kwargs) elif service == "disable_applet": self.ccb_disable_applet(*args, **kwargs) elif service == "disable_applet_group": self.ccb_disable_applet_group(*args, **kwargs) except: logger.error("failed to process CCB", exc_info=True) def save_state(self): return { "applets": applets.AppletsDock.save_state(self), "ccbp_global": self.get_ccpb_global() } def restore_state(self, state): applets.AppletsDock.restore_state(self, state["applets"]) getattr(self, "ccbp_global_" + state["ccbp_global"]).setChecked(True)