gui: centralize subscribers

artiq/frontend/artiq_gui.py

@@ -12,13 +12,8 @@ from pyqtgraph import dockarea
 
 from artiq.tools import verbosity_args, init_logger, artiq_dir
 from artiq.protocols.pc_rpc import AsyncioClient
-from artiq.gui.state import StateManager
+from artiq.gui.models import ModelSubscriber
-from artiq.gui.explorer import ExplorerDock
+from artiq.gui import state, explorer, moninj, datasets, schedule, log, console
-from artiq.gui.moninj import MonInj
-from artiq.gui.datasets import DatasetsDock
-from artiq.gui.schedule import ScheduleDock
-from artiq.gui.log import LogDock
-from artiq.gui.console import ConsoleDock
 
 
 def get_argparser():
@@ -57,6 +52,12 @@ class MainWindow(QtGui.QMainWindow):
         self.restoreGeometry(QtCore.QByteArray(state))
 
 
+def atexit_register_coroutine(coroutine, loop=None):
+    if loop is None:
+        loop = asyncio.get_event_loop()
+    atexit.register(lambda: loop.run_until_complete(coroutine()))
+
+
 def main():
     args = get_argparser().parse_args()
     init_logger(args)
@@ -74,7 +75,18 @@ def main():
         atexit.register(client.close_rpc)
         rpc_clients[target] = client
 
-    smgr = StateManager(args.db_file)
+    sub_clients = dict()
+    for notifier_name, module in (("explist", explorer),
+                                  ("datasets", datasets),
+                                  ("schedule", schedule),
+                                  ("log", log)):
+        subscriber = ModelSubscriber(notifier_name, module.Model)
+        loop.run_until_complete(subscriber.connect(
+            args.server, args.port_notify))
+        atexit_register_coroutine(subscriber.close)
+        sub_clients[notifier_name] = subscriber
+
+    smgr = state.StateManager(args.db_file)
 
     win = MainWindow(app, args.server)
     area = dockarea.DockArea()
@@ -85,24 +97,20 @@ def main():
     status_bar.showMessage("Connected to {}".format(args.server))
     win.setStatusBar(status_bar)
 
-    d_explorer = ExplorerDock(win, status_bar,
+    d_explorer = explorer.ExplorerDock(win, status_bar,
-                              rpc_clients["schedule"],
+                                       sub_clients["explist"],
-                              rpc_clients["repository"])
+                                       sub_clients["schedule"],
+                                       rpc_clients["schedule"],
+                                       rpc_clients["repository"])
     smgr.register(d_explorer)
-    loop.run_until_complete(d_explorer.sub_connect(
-        args.server, args.port_notify))
-    atexit.register(lambda: loop.run_until_complete(d_explorer.sub_close()))
 
-    d_datasets = DatasetsDock(win, area)
+    d_datasets = datasets.DatasetsDock(win, area, sub_clients["datasets"])
     smgr.register(d_datasets)
-    loop.run_until_complete(d_datasets.sub_connect(
-        args.server, args.port_notify))
-    atexit.register(lambda: loop.run_until_complete(d_datasets.sub_close()))
 
     if os.name != "nt":
-        d_ttl_dds = MonInj()
+        d_ttl_dds = moninj.MonInj()
         loop.run_until_complete(d_ttl_dds.start(args.server, args.port_notify))
-        atexit.register(lambda: loop.run_until_complete(d_ttl_dds.stop()))
+        atexit_register_coroutine(d_ttl_dds.stop)
 
     if os.name != "nt":
         area.addDock(d_ttl_dds.dds_dock, "top")
@@ -112,23 +120,17 @@ def main():
         area.addDock(d_datasets, "top")
     area.addDock(d_explorer, "above", d_datasets)
 
-    d_schedule = ScheduleDock(status_bar, rpc_clients["schedule"])
+    d_schedule = schedule.ScheduleDock(
-    loop.run_until_complete(d_schedule.sub_connect(
+        status_bar, rpc_clients["schedule"], sub_clients["schedule"])
-        args.server, args.port_notify))
-    atexit.register(lambda: loop.run_until_complete(d_schedule.sub_close()))
-    d_explorer.get_current_schedule = d_schedule.get_current_schedule
 
-    d_log = LogDock()
+    d_log = log.LogDock(sub_clients["log"])
     smgr.register(d_log)
-    loop.run_until_complete(d_log.sub_connect(
-        args.server, args.port_notify))
-    atexit.register(lambda: loop.run_until_complete(d_log.sub_close()))
 
     def _set_dataset(k, v):
         asyncio.ensure_future(rpc_clients["dataset_db"].set(k, v))
     def _del_dataset(k):
         asyncio.ensure_future(rpc_clients["dataset_db"].delete(k))
-    d_console = ConsoleDock(
+    d_console = console.ConsoleDock(
         d_datasets.get_dataset,
         _set_dataset,
         _del_dataset)
@@ -139,7 +141,7 @@ def main():
 
     smgr.load()
     smgr.start()
-    atexit.register(lambda: loop.run_until_complete(smgr.stop()))
+    atexit_register_coroutine(smgr.stop)
     win.show()
     loop.run_until_complete(win.exit_request.wait())
 

artiq/gui/datasets.py

@@ -7,9 +7,8 @@ from quamash import QtGui, QtCore
 from pyqtgraph import dockarea
 from pyqtgraph import LayoutWidget
 
-from artiq.protocols.sync_struct import Subscriber
 from artiq.tools import short_format
-from artiq.gui.tools import DictSyncModel
+from artiq.gui.models import DictSyncModel
 from artiq.gui.displays import *
 
 try:
@@ -21,10 +20,9 @@ except AttributeError:
 logger = logging.getLogger(__name__)
 
 
-class DatasetsModel(DictSyncModel):
+class Model(DictSyncModel):
-    def __init__(self, parent, init):
+    def __init__(self,  init):
-        DictSyncModel.__init__(self, ["Dataset", "Persistent", "Value"],
+        DictSyncModel.__init__(self, ["Dataset", "Persistent", "Value"], init)
-                               parent, init)
 
     def sort_key(self, k, v):
         return k
@@ -47,7 +45,7 @@ def _get_display_type_name(display_cls):
 
 
 class DatasetsDock(dockarea.Dock):
-    def __init__(self, dialog_parent, dock_area):
+    def __init__(self, dialog_parent, dock_area, datasets_sub):
         dockarea.Dock.__init__(self, "Datasets", size=(1500, 500))
         self.dialog_parent = dialog_parent
         self.dock_area = dock_area
@@ -66,6 +64,9 @@ class DatasetsDock(dockarea.Dock):
             QtGui.QHeaderView.ResizeToContents)
         grid.addWidget(self.table, 1, 0)
 
+        self.table_model = Model(dict())
+        datasets_sub.add_setmodel_callback(self.set_model)
+
         add_display_box = QtGui.QGroupBox("Add display")
         grid.addWidget(add_display_box, 1, 1)
         display_grid = QtGui.QGridLayout()
@@ -79,25 +80,18 @@ class DatasetsDock(dockarea.Dock):
         self.displays = dict()
 
     def _search_datasets(self):
-        self.table_model_filter.setFilterFixedString(self.search.displayText())
+        if hasattr(self, "table_model_filter"):
+            self.table_model_filter.setFilterFixedString(
+                self.search.displayText())
 
     def get_dataset(self, key):
         return self.table_model.backing_store[key][1]
 
-    async def sub_connect(self, host, port):
+    def set_model(self, model):
-        self.subscriber = Subscriber("datasets", self.init_datasets_model,
+        self.table_model = model
-                                     self.on_mod)
-        await self.subscriber.connect(host, port)
-
-    async def sub_close(self):
-        await self.subscriber.close()
-
-    def init_datasets_model(self, init):
-        self.table_model = DatasetsModel(self.table, init)
         self.table_model_filter = QSortFilterProxyModel()
         self.table_model_filter.setSourceModel(self.table_model)
         self.table.setModel(self.table_model_filter)
-        return self.table_model
 
     def update_display_data(self, dsp):
         filtered_data = {k: self.table_model.backing_store[k][1]

artiq/gui/explorer.py

@@ -5,19 +5,18 @@ from quamash import QtGui, QtCore
 from pyqtgraph import dockarea
 from pyqtgraph import LayoutWidget
 
-from artiq.protocols.sync_struct import Subscriber
 from artiq.protocols import pyon
-from artiq.gui.tools import DictSyncModel
+from artiq.gui.models import DictSyncModel
 from artiq.gui.scan import ScanController
 from artiq.gui.shortcuts import ShortcutManager
 
 
-class _ExplistModel(DictSyncModel):
+class Model(DictSyncModel):
-    def __init__(self, explorer, parent, init):
+    def __init__(self, init):
-        self.explorer = explorer
         DictSyncModel.__init__(self,
             ["Experiment"],
-            parent, init)
+            init)
+        self.explorer = None
 
     def sort_key(self, k, v):
         return k
@@ -27,8 +26,9 @@ class _ExplistModel(DictSyncModel):
 
     def __setitem__(self, k, v):
         DictSyncModel.__setitem__(self, k, v)
-        if k == self.explorer.selected_key:
+        if self.explorer is not None:
-            self.explorer.update_selection(k, k)
+            if k == self.explorer.selected_key:
+                self.explorer.update_selection(k, k)
 
 
 class _FreeValueEntry(QtGui.QLineEdit):
@@ -216,11 +216,14 @@ class _ArgumentEditor(QtGui.QTreeWidget):
 
 
 class ExplorerDock(dockarea.Dock):
-    def __init__(self, main_window, status_bar, schedule_ctl, repository_ctl):
+    def __init__(self, main_window, status_bar,
+                 explist_sub, schedule_sub,
+                 schedule_ctl, repository_ctl):
         dockarea.Dock.__init__(self, "Explorer", size=(1500, 500))
 
         self.main_window = main_window
         self.status_bar = status_bar
+        self.schedule_sub = schedule_sub
         self.schedule_ctl = schedule_ctl
 
         self.splitter = QtGui.QSplitter(QtCore.Qt.Horizontal)
@@ -271,6 +274,8 @@ class ExplorerDock(dockarea.Dock):
         self.splitter.addWidget(self.argeditor)
         self.splitter.setSizes([grid.minimumSizeHint().width(), 1000])
         self.argeditor_states = dict()
+        self.explist_model = Model(dict())
+        explist_sub.add_setmodel_callback(self.set_model)
 
         self.shortcuts = ShortcutManager(self.main_window, self)
 
@@ -299,6 +304,11 @@ class ExplorerDock(dockarea.Dock):
         scan_repository_action.triggered.connect(scan_repository)
         self.el.addAction(scan_repository_action)
 
+    def set_model(self, model):
+        model.explorer = self
+        self.explist_model = model
+        self.el.setModel(model)
+
     def update_selection(self, selected, deselected):
         if deselected:
             self.argeditor_states[deselected] = self.argeditor.save_state()
@@ -346,19 +356,6 @@ class ExplorerDock(dockarea.Dock):
     def enable_duedate(self):
         self.datetime_en.setChecked(True)
 
-    async def sub_connect(self, host, port):
-        self.explist_subscriber = Subscriber("explist",
-                                             self.init_explist_model)
-        await self.explist_subscriber.connect(host, port)
-
-    async def sub_close(self):
-        await self.explist_subscriber.close()
-
-    def init_explist_model(self, init):
-        self.explist_model = _ExplistModel(self, self.el, init)
-        self.el.setModel(self.explist_model)
-        return self.explist_model
-
     async def submit_task(self, pipeline_name, file, class_name, arguments,
                           priority, due_date, flush):
         expid = {
@@ -414,17 +411,16 @@ class ExplorerDock(dockarea.Dock):
     def request_inst_term(self):
         if self.selected_key is not None:
             expinfo = self.explist_model.backing_store[self.selected_key]
-            # attribute get_current_schedule must be set externally after
+            if self.schedule_sub.model is not None:
-            # instance creation
+                current_schedule = self.schedule_sub.model.backing_store
-            current_schedule = self.get_current_schedule()
+                rids = []
-            rids = []
+                for rid, desc in current_schedule.items():
-            for rid, desc in current_schedule.items():
+                    expid = desc["expid"]
-                expid = desc["expid"]
+                    if ("repo_rev" in expid  # only consider runs from repository
-                if ("repo_rev" in expid  # only consider runs from repository
+                            and expid["file"] == expinfo["file"]
-                        and expid["file"] == expinfo["file"]
+                            and expid["class_name"] == expinfo["class_name"]):
-                        and expid["class_name"] == expinfo["class_name"]):
+                        rids.append(rid)
-                    rids.append(rid)
+                asyncio.ensure_future(self.request_term_multiple(rids))
-            asyncio.ensure_future(self.request_term_multiple(rids))
 
     def edit_shortcuts(self):
         experiments = sorted(self.explist_model.backing_store.keys())

artiq/gui/log.py

@@ -5,8 +5,6 @@ import time
 from quamash import QtGui, QtCore
 from pyqtgraph import dockarea, LayoutWidget
 
-from artiq.protocols.sync_struct import Subscriber
-
 try:
     QSortFilterProxyModel = QtCore.QSortFilterProxyModel
 except AttributeError:
@@ -25,9 +23,9 @@ def _level_to_name(level):
     return "DEBUG"
 
 
-class _LogModel(QtCore.QAbstractTableModel):
+class Model(QtCore.QAbstractTableModel):
-    def __init__(self, parent, init):
+    def __init__(self, init):
-        QtCore.QAbstractTableModel.__init__(self, parent)
+        QtCore.QAbstractTableModel.__init__(self)
 
         self.headers = ["Level", "Source", "Time", "Message"]
 
@@ -153,7 +151,7 @@ class _LogFilterProxyModel(QSortFilterProxyModel):
 
 
 class LogDock(dockarea.Dock):
-    def __init__(self):
+    def __init__(self, log_sub):
         dockarea.Dock.__init__(self, "Log", size=(1000, 300))
 
         grid = LayoutWidget()
@@ -183,12 +181,7 @@ class LogDock(dockarea.Dock):
         grid.addWidget(self.log, 1, 0, colspan=4)
         self.scroll_at_bottom = False
 
-    async def sub_connect(self, host, port):
+        log_sub.add_setmodel_callback(self.set_model)
-        self.subscriber = Subscriber("log", self.init_log_model)
-        await self.subscriber.connect(host, port)
-
-    async def sub_close(self):
-        await self.subscriber.close()
 
     def filter_level_changed(self):
         if not hasattr(self, "table_model_filter"):
@@ -223,8 +216,8 @@ class LogDock(dockarea.Dock):
             scrollbar = self.log.verticalScrollBar()
             scrollbar.setValue(self.scroll_value)
 
-    def init_log_model(self, init):
+    def set_model(self, model):
-        self.table_model = _LogModel(self.log, init)
+        self.table_model = model
         self.table_model_filter = _LogFilterProxyModel(
             getattr(logging, self.filter_level.currentText()),
             self.filter_freetext.text())
@@ -233,7 +226,6 @@ class LogDock(dockarea.Dock):
         self.table_model_filter.rowsAboutToBeInserted.connect(self.rows_inserted_before)
         self.table_model_filter.rowsInserted.connect(self.rows_inserted_after)
         self.table_model_filter.rowsRemoved.connect(self.rows_removed)
-        return self.table_model
 
     def save_state(self):
         return {"min_level_idx": self.filter_level.currentIndex()}

artiq/gui/models.py

@@ -1,5 +1,26 @@
 from quamash import QtCore
 
+from artiq.protocols.sync_struct import Subscriber
+
+
+class ModelSubscriber(Subscriber):
+    def __init__(self, notifier_name, model_factory):
+        Subscriber.__init__(self, notifier_name, self._create_model)
+        self.model = None
+        self._model_factory = model_factory
+        self._setmodel_callbacks = []
+
+    def _create_model(self, init):
+        self.model = self._model_factory(init)
+        for cb in self._setmodel_callbacks:
+            cb(self.model)
+        return self.model
+
+    def add_setmodel_callback(self, cb):
+        self._setmodel_callbacks.append(cb)
+        if self.model is not None:
+            cb(self.model)
+
 
 class _SyncSubstruct:
     def __init__(self, update_cb, ref):
@@ -31,12 +52,12 @@ class _SyncSubstruct:
 
 
 class DictSyncModel(QtCore.QAbstractTableModel):
-    def __init__(self, headers, parent, init):
+    def __init__(self, headers, init):
         self.headers = headers
         self.backing_store = init
         self.row_to_key = sorted(self.backing_store.keys(),
                                  key=lambda k: self.sort_key(k, self.backing_store[k]))
-        QtCore.QAbstractTableModel.__init__(self, parent)
+        QtCore.QAbstractTableModel.__init__(self)
 
     def rowCount(self, parent):
         return len(self.backing_store)
@@ -45,12 +66,11 @@ class DictSyncModel(QtCore.QAbstractTableModel):
         return len(self.headers)
 
     def data(self, index, role):
-        if not index.isValid():
+        if not index.isValid() or role != QtCore.Qt.DisplayRole:
             return None
-        elif role != QtCore.Qt.DisplayRole:
+        else:
-            return None
+            k = self.row_to_key[index.row()]
-        k = self.row_to_key[index.row()]
+            return self.convert(k, self.backing_store[k], index.column())
-        return self.convert(k, self.backing_store[k], index.column())
 
     def headerData(self, col, orientation, role):
         if (orientation == QtCore.Qt.Horizontal
@@ -113,10 +133,10 @@ class DictSyncModel(QtCore.QAbstractTableModel):
 
 
 class ListSyncModel(QtCore.QAbstractTableModel):
-    def __init__(self, headers, parent, init):
+    def __init__(self, headers, init):
         self.headers = headers
         self.backing_store = init
-        QtCore.QAbstractTableModel.__init__(self, parent)
+        QtCore.QAbstractTableModel.__init__(self)
 
     def rowCount(self, parent):
         return len(self.backing_store)
@@ -125,11 +145,11 @@ class ListSyncModel(QtCore.QAbstractTableModel):
         return len(self.headers)
 
     def data(self, index, role):
-        if not index.isValid():
+        if not index.isValid() or role != QtCore.Qt.DisplayRole:
             return None
-        elif role != QtCore.Qt.DisplayRole:
+        else:
-            return None
+            return self.convert(self.backing_store[index.row()],
-        return self.convert(self.backing_store[index.row()], index.column())
+                                index.column())
 
     def headerData(self, col, orientation, role):
         if (orientation == QtCore.Qt.Horizontal

artiq/gui/schedule.py

@@ -5,17 +5,16 @@ from functools import partial
 from quamash import QtGui, QtCore
 from pyqtgraph import dockarea
 
-from artiq.protocols.sync_struct import Subscriber
+from artiq.gui.models import DictSyncModel
-from artiq.gui.tools import DictSyncModel
 from artiq.tools import elide
 
 
-class _ScheduleModel(DictSyncModel):
+class Model(DictSyncModel):
-    def __init__(self, parent, init):
+    def __init__(self, init):
         DictSyncModel.__init__(self,
             ["RID", "Pipeline", "Status", "Prio", "Due date",
              "Revision", "File", "Class name"],
-            parent, init)
+            init)
 
     def sort_key(self, k, v):
         # order by priority, and then by due date and RID
@@ -57,7 +56,7 @@ class _ScheduleModel(DictSyncModel):
 
 
 class ScheduleDock(dockarea.Dock):
-    def __init__(self, status_bar, schedule_ctl):
+    def __init__(self, status_bar, schedule_ctl, schedule_sub):
         dockarea.Dock.__init__(self, "Schedule", size=(1000, 300))
 
         self.status_bar = status_bar
@@ -82,24 +81,12 @@ class ScheduleDock(dockarea.Dock):
         delete_action.setShortcut("SHIFT+DELETE")
         self.table.addAction(delete_action)
 
-    async def sub_connect(self, host, port):
+        self.table_model = Model(dict())
-        self.subscriber = Subscriber("schedule", self.init_schedule_model)
+        schedule_sub.add_setmodel_callback(self.set_model)
-        await self.subscriber.connect(host, port)
 
-    async def sub_close(self):
+    def set_model(self, model):
-        await self.subscriber.close()
+        self.table_model = model
-
-    def get_current_schedule(self):
-        try:
-            table_model = self.table_model
-        except AttributeError:
-            return dict()
-        return table_model.backing_store
-
-    def init_schedule_model(self, init):
-        self.table_model = _ScheduleModel(self.table, init)
         self.table.setModel(self.table_model)
-        return self.table_model
 
     async def delete(self, rid, graceful):
         if graceful: