Merge branch 'applets_pipeipc'

This commit is contained in:
Sebastien Bourdeauducq 2016-02-08 22:25:36 +01:00
commit 338e5fe3fc
10 changed files with 291 additions and 195 deletions

View File

@ -1,6 +1,6 @@
#!/usr/bin/env python3.5
from quamash import QtWidgets
from PyQt5 import QtWidgets
from artiq.applets.simple import SimpleApplet

View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3.5
import numpy as np
import PyQt5 # make sure pyqtgraph imports Qt5
import pyqtgraph
from artiq.applets.simple import SimpleApplet

View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3.5
import numpy as np
import PyQt5 # make sure pyqtgraph imports Qt5
import pyqtgraph
from artiq.applets.simple import SimpleApplet

View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3.5
import numpy as np
from quamash import QtWidgets
from PyQt5 import QtWidgets
import pyqtgraph
from artiq.applets.simple import SimpleApplet

View File

@ -1,10 +1,70 @@
import logging
import argparse
import asyncio
from quamash import QEventLoop, QtWidgets, QtGui, QtCore
from artiq.protocols.sync_struct import Subscriber
from artiq.protocols.pc_rpc import Client
from artiq.protocols.sync_struct import Subscriber, process_mod
from artiq.protocols import pyon
from artiq.protocols.pipe_ipc import AsyncioChildComm
logger = logging.getLogger(__name__)
class AppletIPCClient(AsyncioChildComm):
def set_close_cb(self, close_cb):
self.close_cb = close_cb
def write_pyon(self, obj):
self.write(pyon.encode(obj).encode() + b"\n")
async def read_pyon(self):
line = await self.readline()
return pyon.decode(line.decode())
async def embed(self, win_id):
# This function is only called when not subscribed to anything,
# so the only normal replies are embed_done and terminate.
self.write_pyon({"action": "embed",
"win_id": win_id})
reply = await self.read_pyon()
if reply["action"] == "terminate":
self.close_cb()
elif reply["action"] != "embed_done":
logger.error("unexpected action reply to embed request: %s",
action)
self.close_cb()
async def listen(self):
data = None
while True:
obj = await self.read_pyon()
try:
action = obj["action"]
if action == "terminate":
self.close_cb()
return
elif action == "mod":
mod = obj["mod"]
if mod["action"] == "init":
data = self.init_cb(mod["struct"])
else:
process_mod(data, mod)
self.mod_cb(mod)
else:
raise ValueError("unknown action in parent message")
except:
logger.error("error processing parent message",
exc_info=True)
self.close_cb()
def subscribe(self, datasets, init_cb, mod_cb):
self.write_pyon({"action": "subscribe",
"datasets": datasets})
self.init_cb = init_cb
self.mod_cb = mod_cb
asyncio.ensure_future(self.listen())
class SimpleApplet:
@ -13,27 +73,27 @@ class SimpleApplet:
self.main_widget_class = main_widget_class
self.argparser = argparse.ArgumentParser(description=cmd_description)
self.argparser.add_argument("--update-delay", type=float,
default=default_update_delay,
help="time to wait after a mod (buffering other mods) "
"before updating (default: %(default).2f)")
group = self.argparser.add_argument_group("data server")
group = self.argparser.add_argument_group("standalone mode (default)")
group.add_argument(
"--server-notify", default="::1",
help="hostname or IP to connect to for dataset notifications")
"--server", default="::1",
help="hostname or IP of the master to connect to "
"for dataset notifications "
"(ignored in embedded mode)")
group.add_argument(
"--port-notify", default=3250, type=int,
help="TCP port to connect to for dataset notifications")
group = self.argparser.add_argument_group("GUI server")
group.add_argument(
"--server-gui", default="::1",
help="hostname or IP to connect to for GUI control")
group.add_argument(
"--port-gui", default=6501, type=int,
help="TCP port to connect to for GUI control")
group.add_argument("--embed", default=None, type=int,
help="embed main widget into existing window")
"--port", default=3250, type=int,
help="TCP port to connect to")
self.argparser.add_argument("--embed", default=None,
help="embed into GUI", metavar="IPC_ADDRESS")
self._arggroup_datasets = self.argparser.add_argument_group("datasets")
self.dataset_args = set()
def add_dataset(self, name, help=None, required=True):
@ -56,6 +116,15 @@ class SimpleApplet:
self.loop = QEventLoop(app)
asyncio.set_event_loop(self.loop)
def ipc_init(self):
if self.args.embed is not None:
self.ipc = AppletIPCClient(self.args.embed)
self.loop.run_until_complete(self.ipc.connect())
def ipc_close(self):
if self.args.embed is not None:
self.ipc.close()
def create_main_widget(self):
self.main_widget = self.main_widget_class(self.args)
# Qt window embedding is ridiculously buggy, and empirical testing
@ -65,15 +134,11 @@ class SimpleApplet:
# 3. applet sends the ID to host, host embeds the widget
# 4. applet shows the widget
# Doing embedding the other way around (using QWindow.setParent in the
# applet) breaks resizing; furthermore the host needs to know our
# window ID to request graceful termination by closing the window.
# applet) breaks resizing.
if self.args.embed is not None:
self.ipc.set_close_cb(self.main_widget.close)
win_id = int(self.main_widget.winId())
remote = Client(self.args.server_gui, self.args.port_gui, "applets")
try:
remote.embed(self.args.embed, win_id)
finally:
remote.close_rpc()
self.loop.run_until_complete(self.ipc.embed(win_id))
self.main_widget.show()
def sub_init(self, data):
@ -81,6 +146,10 @@ class SimpleApplet:
return data
def filter_mod(self, mod):
if self.args.embed is not None:
# the parent already filters for us
return True
if mod["action"] == "init":
return True
if mod["path"]:
@ -108,21 +177,32 @@ class SimpleApplet:
else:
self.main_widget.data_changed(self.data, [mod])
def create_subscriber(self):
self.subscriber = Subscriber("datasets",
self.sub_init, self.sub_mod)
self.loop.run_until_complete(self.subscriber.connect(
self.args.server_notify, self.args.port_notify))
def subscribe(self):
if self.args.embed is None:
self.subscriber = Subscriber("datasets",
self.sub_init, self.sub_mod)
self.loop.run_until_complete(self.subscriber.connect(
self.args.server, self.args.port))
else:
self.ipc.subscribe(self.datasets, self.sub_init, self.sub_mod)
def unsubscribe(self):
if self.args.embed is None:
self.loop.run_until_complete(self.subscriber.close())
def run(self):
self.args_init()
self.quamash_init()
try:
self.create_main_widget()
self.create_subscriber()
self.ipc_init()
try:
self.loop.run_forever()
self.create_main_widget()
self.subscribe()
try:
self.loop.run_forever()
finally:
self.unsubscribe()
finally:
self.loop.run_until_complete(self.subscriber.close())
self.ipc_close()
finally:
self.loop.close()

View File

@ -14,7 +14,7 @@ from pyqtgraph import dockarea
from artiq import __artiq_dir__ as artiq_dir
from artiq.tools import *
from artiq.protocols.pc_rpc import AsyncioClient, Server
from artiq.protocols.pc_rpc import AsyncioClient
from artiq.gui.models import ModelSubscriber
from artiq.gui import (state, experiments, shortcuts, explorer,
moninj, datasets, applets, schedule, log, console)
@ -113,9 +113,9 @@ def main():
d_datasets = datasets.DatasetsDock(win, dock_area, sub_clients["datasets"])
appletmgr = applets.AppletManager(dock_area)
atexit_register_coroutine(appletmgr.stop)
smgr.register(appletmgr)
d_applets = applets.AppletsDock(dock_area, sub_clients["datasets"])
atexit_register_coroutine(d_applets.stop)
smgr.register(d_applets)
if os.name != "nt":
d_ttl_dds = moninj.MonInj()
@ -135,11 +135,11 @@ def main():
if os.name != "nt":
dock_area.addDock(d_ttl_dds.dds_dock, "top")
dock_area.addDock(d_ttl_dds.ttl_dock, "above", d_ttl_dds.dds_dock)
dock_area.addDock(appletmgr.main_dock, "above", d_ttl_dds.ttl_dock)
dock_area.addDock(d_datasets, "above", appletmgr.main_dock)
dock_area.addDock(d_applets, "above", d_ttl_dds.ttl_dock)
dock_area.addDock(d_datasets, "above", d_applets)
else:
dock_area.addDock(appletmgr.main_dock, "top")
dock_area.addDock(d_datasets, "above", appletmgr.main_dock)
dock_area.addDock(d_applets, "top")
dock_area.addDock(d_datasets, "above", d_applets)
dock_area.addDock(d_shortcuts, "above", d_datasets)
dock_area.addDock(d_explorer, "above", d_shortcuts)
dock_area.addDock(d_console, "bottom")
@ -155,11 +155,6 @@ def main():
if d_log0 is not None:
dock_area.addDock(d_log0, "right", d_explorer)
# start RPC server
rpc_server = Server({"applets": appletmgr.rpc})
loop.run_until_complete(rpc_server.start("::1", 6501))
atexit_register_coroutine(rpc_server.stop)
# run
win.show()
loop.run_until_complete(win.exit_request.wait())

View File

@ -7,17 +7,88 @@ from functools import partial
from quamash import QtCore, QtGui, QtWidgets
from pyqtgraph import dockarea
from artiq.protocols.pipe_ipc import AsyncioParentComm
from artiq.protocols import pyon
logger = logging.getLogger(__name__)
class AppletIPCServer(AsyncioParentComm):
def __init__(self, datasets_sub):
AsyncioParentComm.__init__(self)
self.datasets_sub = datasets_sub
self.datasets = set()
def write_pyon(self, obj):
self.write(pyon.encode(obj).encode() + b"\n")
async def read_pyon(self):
line = await self.readline()
return pyon.decode(line.decode())
def _synthesize_init(self, data):
struct = {k: v for k, v in data.items() if k in self.datasets}
return {"action": "init",
"struct": struct}
def _on_mod(self, mod):
if mod["action"] == "init":
mod = self._synthesize_init(mod["struct"])
else:
if mod["path"]:
if mod["path"][0] not in self.datasets:
return
elif mod["action"] in {"setitem", "delitem"}:
if mod["key"] not in self.datasets:
return
self.write_pyon({"action": "mod", "mod": mod})
async def serve(self, embed_cb):
self.datasets_sub.notify_cbs.append(self._on_mod)
try:
while True:
obj = await self.read_pyon()
try:
action = obj["action"]
if action == "embed":
embed_cb(obj["win_id"])
self.write_pyon({"action": "embed_done"})
elif action == "subscribe":
self.datasets = obj["datasets"]
if self.datasets_sub.model is not None:
mod = self._synthesize_init(
self.datasets_sub.model.backing_store)
self.write_pyon({"action": "mod", "mod": mod})
else:
raise ValueError("unknown action in applet message")
except:
logger.warning("error processing applet message",
exc_info=True)
self.write_pyon({"action": "error"})
except asyncio.CancelledError:
pass
except:
logger.error("error processing data from applet, "
"server stopped", exc_info=True)
finally:
self.datasets_sub.notify_cbs.remove(self._on_mod)
def start(self, embed_cb):
self.server_task = asyncio.ensure_future(self.serve(embed_cb))
async def stop(self):
self.server_task.cancel()
await asyncio.wait([self.server_task])
class AppletDock(dockarea.Dock):
def __init__(self, token, name, command):
dockarea.Dock.__init__(self, "applet" + str(token),
def __init__(self, datasets_sub, uid, name, command):
dockarea.Dock.__init__(self, "applet" + str(uid),
label="Applet: " + name,
closable=True)
self.setMinimumSize(QtCore.QSize(500, 400))
self.token = token
self.datasets_sub = datasets_sub
self.applet_name = name
self.command = command
@ -26,41 +97,38 @@ class AppletDock(dockarea.Dock):
self.label.setText("Applet: " + name)
async def start(self):
self.ipc = AppletIPCServer(self.datasets_sub)
command = self.command.format(python=sys.executable,
embed_token=self.token)
ipc_address=self.ipc.get_address())
logger.debug("starting command %s for %s", command, self.applet_name)
try:
self.process = await asyncio.create_subprocess_exec(
*shlex.split(command))
await self.ipc.create_subprocess(*shlex.split(command))
except:
logger.warning("Applet %s failed to start", self.applet_name,
exc_info=True)
self.ipc.start(self.embed)
def capture(self, win_id):
def embed(self, win_id):
logger.debug("capturing window 0x%x for %s", win_id, self.applet_name)
self.captured_window = QtGui.QWindow.fromWinId(win_id)
self.captured_widget = QtWidgets.QWidget.createWindowContainer(
self.captured_window)
self.addWidget(self.captured_widget)
embed_window = QtGui.QWindow.fromWinId(win_id)
embed_widget = QtWidgets.QWidget.createWindowContainer(embed_window)
self.addWidget(embed_widget)
async def terminate(self):
if hasattr(self, "captured_window"):
self.captured_window.close()
self.captured_widget.deleteLater()
del self.captured_window
del self.captured_widget
if hasattr(self, "process"):
if hasattr(self, "ipc"):
await self.ipc.stop()
self.ipc.write_pyon({"action": "terminate"})
try:
await asyncio.wait_for(self.process.wait(), 2.0)
await asyncio.wait_for(self.ipc.process.wait(), 2.0)
except:
logger.warning("Applet %s failed to exit, killing",
self.applet_name)
try:
self.process.kill()
self.ipc.process.kill()
except ProcessLookupError:
pass
await self.process.wait()
del self.process
await self.ipc.process.wait()
del self.ipc
async def restart(self):
await self.terminate()
@ -69,24 +137,27 @@ class AppletDock(dockarea.Dock):
_templates = [
("Big number", "{python} -m artiq.applets.big_number "
"--embed {embed_token} NUMBER_DATASET"),
"--embed {ipc_address} NUMBER_DATASET"),
("Histogram", "{python} -m artiq.applets.plot_hist "
"--embed {embed_token} COUNTS_DATASET "
"--embed {ipc_address} COUNTS_DATASET "
"--x BIN_BOUNDARIES_DATASET"),
("XY", "{python} -m artiq.applets.plot_xy "
"--embed {embed_token} Y_DATASET --x X_DATASET "
"--embed {ipc_address} Y_DATASET --x X_DATASET "
"--error ERROR_DATASET --fit FIT_DATASET"),
("XY + Histogram", "{python} -m artiq.applets.plot_xy_hist "
"--embed {embed_token} X_DATASET "
"--embed {ipc_address} X_DATASET "
"HIST_BIN_BOUNDARIES_DATASET "
"HISTS_COUNTS_DATASET"),
]
class AppletsDock(dockarea.Dock):
def __init__(self, manager):
self.manager = manager
self.token_to_checkbox = dict()
def __init__(self, dock_area, datasets_sub):
self.dock_area = dock_area
self.datasets_sub = datasets_sub
self.dock_to_checkbox = dict()
self.applet_uids = set()
self.workaround_pyqtgraph_bug = False
dockarea.Dock.__init__(self, "Applets")
self.setMinimumSize(QtCore.QSize(850, 450))
@ -129,6 +200,18 @@ class AppletsDock(dockarea.Dock):
self.table.cellChanged.connect(self.cell_changed)
def create(self, uid, name, command):
dock = AppletDock(self.datasets_sub, uid, name, command)
# If a dock is floated and then dock state is restored, pyqtgraph
# leaves a "phantom" window open.
if self.workaround_pyqtgraph_bug:
self.dock_area.addDock(dock)
else:
self.dock_area.floatDock(dock)
asyncio.ensure_future(dock.start())
dock.sigClosed.connect(partial(self.on_dock_closed, dock))
return dock
def cell_changed(self, row, column):
if column == 0:
item = self.table.item(row, column)
@ -141,30 +224,36 @@ class AppletsDock(dockarea.Dock):
name = ""
else:
name = name.text()
token = self.manager.create(name, command)
item.applet_token = token
self.token_to_checkbox[token] = item
dock = self.create(item.applet_uid, name, command)
item.applet_dock = dock
self.dock_to_checkbox[dock] = item
else:
token = getattr(item, "applet_token", None)
if token is not None:
# cell_changed is emitted at row creation
self.manager.delete(token)
dock = item.applet_dock
if dock is not None:
# This calls self.on_dock_closed
dock.close()
elif column == 1 or column == 2:
new_value = self.table.item(row, column).text()
token = getattr(self.table.item(row, 0), "applet_token", None)
if token is not None:
dock = self.table.item(row, 0).applet_dock
if dock is not None:
if column == 1:
self.manager.rename(token, new_value)
dock.rename(new_value)
else:
self.manager.set_command(token, new_value)
dock.command = new_value
def disable_token(self, token):
checkbox_item = self.token_to_checkbox[token]
checkbox_item.applet_token = None
del self.token_to_checkbox[token]
def on_dock_closed(self, dock):
asyncio.ensure_future(dock.terminate())
checkbox_item = self.dock_to_checkbox[dock]
checkbox_item.applet_dock = None
del self.dock_to_checkbox[dock]
checkbox_item.setCheckState(QtCore.Qt.Unchecked)
def new(self):
def new(self, uid=None):
if uid is None:
uid = next(iter(set(range(len(self.applet_uids) + 1))
- self.applet_uids))
self.applet_uids.add(uid)
row = self.table.rowCount()
self.table.insertRow(row)
checkbox = QtWidgets.QTableWidgetItem()
@ -172,6 +261,8 @@ class AppletsDock(dockarea.Dock):
QtCore.Qt.ItemIsUserCheckable |
QtCore.Qt.ItemIsEnabled)
checkbox.setCheckState(QtCore.Qt.Unchecked)
checkbox.applet_uid = uid
checkbox.applet_dock = None
self.table.setItem(row, 0, checkbox)
self.table.setItem(row, 1, QtWidgets.QTableWidgetItem())
self.table.setItem(row, 2, QtWidgets.QTableWidgetItem())
@ -185,31 +276,43 @@ class AppletsDock(dockarea.Dock):
selection = self.table.selectedRanges()
if selection:
row = selection[0].topRow()
token = getattr(self.table.item(row, 0), "applet_token", None)
if token is not None:
asyncio.ensure_future(self.manager.restart(token))
dock = self.table.item(row, 0).applet_dock
if dock is not None:
asyncio.ensure_future(dock.restart())
def delete(self):
selection = self.table.selectedRanges()
if selection:
row = selection[0].topRow()
token = getattr(self.table.item(row, 0), "applet_token", None)
if token is not None:
self.manager.delete(token)
item = self.table.item(row, 0)
dock = item.applet_dock
if dock is not None:
# This calls self.on_dock_closed
dock.close()
self.applet_uids.remove(item.applet_uid)
self.table.removeRow(row)
async def stop(self):
for row in range(self.table.rowCount()):
dock = self.table.item(row, 0).applet_dock
if dock is not None:
await dock.terminate()
def save_state(self):
state = []
for row in range(self.table.rowCount()):
uid = self.table.item(row, 0).applet_uid
enabled = self.table.item(row, 0).checkState() == QtCore.Qt.Checked
name = self.table.item(row, 1).text()
command = self.table.item(row, 2).text()
state.append((enabled, name, command))
state.append((uid, enabled, name, command))
return state
def restore_state(self, state):
for enabled, name, command in state:
row = self.new()
self.workaround_pyqtgraph_bug = True
for uid, enabled, name, command in state:
row = self.new(uid)
item = QtWidgets.QTableWidgetItem()
item.setText(name)
self.table.setItem(row, 1, item)
@ -218,72 +321,4 @@ class AppletsDock(dockarea.Dock):
self.table.setItem(row, 2, item)
if enabled:
self.table.item(row, 0).setCheckState(QtCore.Qt.Checked)
class AppletManagerRPC:
def __init__(self, parent):
self.parent = parent
def embed(self, token, win_id):
self.parent.embed(token, win_id)
class AppletManager:
def __init__(self, dock_area):
self.dock_area = dock_area
self.main_dock = AppletsDock(self)
self.rpc = AppletManagerRPC(self)
self.applet_docks = dict()
self.workaround_pyqtgraph_bug = False
def embed(self, token, win_id):
if token not in self.applet_docks:
logger.warning("Ignored incorrect embed token %d for winid 0x%x",
token, win_id)
return
self.applet_docks[token].capture(win_id)
def create(self, name, command):
token = next(iter(set(range(len(self.applet_docks) + 1))
- self.applet_docks.keys()))
dock = AppletDock(token, name, command)
self.applet_docks[token] = dock
# If a dock is floated and then dock state is restored, pyqtgraph
# leaves a "phantom" window open.
if self.workaround_pyqtgraph_bug:
self.dock_area.addDock(dock)
else:
self.dock_area.floatDock(dock)
asyncio.ensure_future(dock.start())
dock.sigClosed.connect(partial(self.on_dock_closed, token))
return token
def on_dock_closed(self, token):
asyncio.ensure_future(self.applet_docks[token].terminate())
self.main_dock.disable_token(token)
del self.applet_docks[token]
def delete(self, token):
# This in turns calls on_dock_closed and main_dock.disable_token
self.applet_docks[token].close()
def rename(self, token, name):
self.applet_docks[token].rename(name)
def set_command(self, token, command):
self.applet_docks[token].command = command
async def restart(self, token):
await self.applet_docks[token].restart()
async def stop(self):
for dock in self.applet_docks.values():
await dock.terminate()
def save_state(self):
return self.main_dock.save_state()
def restore_state(self, state):
self.workaround_pyqtgraph_bug = True
self.main_dock.restore_state(state)
self.workaround_pyqtgraph_bug = False

View File

@ -35,6 +35,7 @@ _encode_map = {
bytes: "bytes",
tuple: "tuple",
list: "list",
set: "set",
dict: "dict",
wrapping_int: "number",
Fraction: "fraction",
@ -98,6 +99,12 @@ class _Encoder:
r += "]"
return r
def encode_set(self, x):
r = "{"
r += ", ".join([self.encode(item) for item in x])
r += "}"
return r
def encode_dict(self, x):
r = "{"
if not self.pretty or len(x) < 2:
@ -149,9 +156,7 @@ class _Encoder:
def encode(x, pretty=False):
"""Serializes a Python object and returns the corresponding string in
Python syntax.
"""
Python syntax."""
return _Encoder(pretty).encode(x)
@ -181,9 +186,7 @@ _eval_dict = {
def decode(s):
"""Parses a string in the Python syntax, reconstructs the corresponding
object, and returns it.
"""
object, and returns it."""
return eval(s, _eval_dict, {})
@ -202,23 +205,3 @@ def load_file(filename):
"""Parses the specified file and returns the decoded Python object."""
with open(filename, "r") as f:
return decode(f.read())
class FlatFileDB:
def __init__(self, filename):
self.filename = filename
self.data = pyon.load_file(self.filename)
def save(self):
pyon.store_file(self.filename, self.data)
def get(self, key):
return self.data[key]
def set(self, key, value):
self.data[key] = value
self.save()
def delete(self, key):
del self.data[key]
self.save()

View File

@ -10,6 +10,7 @@ from artiq.protocols import pyon
_pyon_test_object = {
(1, 2): [(3, 4.2), (2, )],
Fraction(3, 4): np.linspace(5, 10, 1),
{"testing", "sets"},
"a": np.int8(9), "b": np.int16(-98), "c": np.int32(42), "d": np.int64(-5),
"e": np.uint8(8), "f": np.uint16(5), "g": np.uint32(4), "h": np.uint64(9),
"x": np.float16(9.0), "y": np.float32(9.0), "z": np.float64(9.0),

View File

@ -2,7 +2,7 @@ from time import sleep
import numpy as np
from artiq import *
from artiq.experiment import *
class Histograms(EnvExperiment):