dashboard: use new moninj protocol

artiq/dashboard/moninj.py

@@ -1,13 +1,10 @@
 import asyncio
-import threading
 import logging
-import socket
-import struct
 
 from PyQt5 import QtCore, QtWidgets, QtGui
 
-from artiq.tools import TaskObject
 from artiq.protocols.sync_struct import Subscriber
+from artiq.coredevice.moninj import MonInjComm
 from artiq.gui.tools import LayoutWidget
 from artiq.gui.flowlayout import FlowLayout
 
@@ -15,20 +12,12 @@ from artiq.gui.flowlayout import FlowLayout
 logger = logging.getLogger(__name__)
 
 
-_mode_enc = {
-    "exp": 0,
-    "1": 1,
-    "0": 2,
-    "in": 3
-}
-
-
 class _TTLWidget(QtWidgets.QFrame):
-    def __init__(self, channel, send_to_device, force_out, title):
+    def __init__(self, channel, set_mode, force_out, title):
         QtWidgets.QFrame.__init__(self)
 
         self.channel = channel
-        self.send_to_device = send_to_device
+        self.set_mode = set_mode
         self.force_out = force_out
 
         self.setFrameShape(QtWidgets.QFrame.Box)
@@ -81,7 +70,10 @@ class _TTLWidget(QtWidgets.QFrame):
         self.override.clicked.connect(self.override_toggled)
         self.level.clicked.connect(self.level_toggled)
 
-        self.set_value(0, False, False)
+        self.cur_value = False
+        self.cur_oe = False
+        self.cur_override = False
+        self.refresh_display()
 
     def enterEvent(self, event):
         self.stack.setCurrentIndex(1)
@@ -97,46 +89,40 @@ class _TTLWidget(QtWidgets.QFrame):
             return
         if override:
             if self.level.isChecked():
-                self.set_mode("1")
+                self.set_mode(self.channel, "1")
             else:
-                self.set_mode("0")
+                self.set_mode(self.channel, "0")
         else:
-            self.set_mode("exp")
+            self.set_mode(self.channel, "exp")
 
     def level_toggled(self, level):
         if self.programmatic_change:
             return
         if self.override.isChecked():
             if level:
-                self.set_mode("1")
+                self.set_mode(self.channel, "1")
             else:
-                self.set_mode("0")
+                self.set_mode(self.channel, "0")
 
-    def set_mode(self, mode):
+    def refresh_display(self):
-        data = struct.pack("bbb",
+        value_s = "1" if self.cur_value else "0"
-                           2,  # MONINJ_REQ_TTLSET
+        if self.cur_override:
-                           self.channel, _mode_enc[mode])
-        self.send_to_device(data)
-
-    def set_value(self, value, oe, override):
-        value_s = "1" if value else "0"
-        if override:
             value_s = "<b>" + value_s + "</b>"
             color = " color=\"red\""
         else:
             color = ""
         self.value.setText("<font size=\"5\"{}>{}</font>".format(
                             color, value_s))
-        oe = oe or self.force_out
+        oe = self.cur_oe or self.force_out
         direction = "OUT" if oe else "IN"
         self.direction.setText("<font size=\"2\">" + direction + "</font>")
 
         self.programmatic_change = True
         try:
-            self.override.setChecked(bool(override))
+            self.override.setChecked(self.cur_override)
-            if override:
+            if self.cur_override:
                 self.stack.setCurrentIndex(1)
-                self.level.setChecked(bool(value))
+                self.level.setChecked(self.cur_value)
         finally:
             self.programmatic_change = False
 
@@ -145,12 +131,11 @@ class _TTLWidget(QtWidgets.QFrame):
 
 
 class _DDSWidget(QtWidgets.QFrame):
-    def __init__(self, bus_channel, channel, sysclk, title):
+    def __init__(self, bus_channel, channel, title):
         QtWidgets.QFrame.__init__(self)
 
         self.bus_channel = bus_channel
         self.channel = channel
-        self.sysclk = sysclk
 
         self.setFrameShape(QtWidgets.QFrame.Box)
         self.setFrameShadow(QtWidgets.QFrame.Raised)
@@ -174,85 +159,169 @@ class _DDSWidget(QtWidgets.QFrame):
         grid.setRowStretch(2, 0)
         grid.setRowStretch(3, 1)
 
-        self.set_value(0)
+        self.cur_frequency = 0
+        self.refresh_display()
 
-    def set_value(self, ftw):
+    def refresh_display(self):
-        frequency = ftw*self.sysclk()/2**32
         self.value.setText("<font size=\"4\">{:.7f}</font><font size=\"2\"> MHz</font>"
-                           .format(frequency/1e6))
+                           .format(self.cur_frequency/1e6))
 
     def sort_key(self):
         return (self.bus_channel, self.channel)
 
 
+TTL_PROBE_LEVEL = 0
+TTL_PROBE_OE = 1
+
+TTL_OVERRIDE_ENABLE = 0
+TTL_OVERRIDE_LEVEL = 1
+TTL_OVERRIDE_OE = 2
+
 class _DeviceManager:
-    def __init__(self, send_to_device, init):
+    def __init__(self, init):
+        self.core_addr = None
+        self.new_core_addr = asyncio.Event()
+        self.core_connection = None
+        self.core_connector_task = asyncio.ensure_future(self.core_connector())
+
         self.dds_sysclk = 0
-        self.send_to_device = send_to_device
-        self.ddb = dict()
         self.ttl_cb = lambda: None
         self.ttl_widgets = dict()
+        self.ttl_widgets_by_channel = dict()
         self.dds_cb = lambda: None
         self.dds_widgets = dict()
+        self.dds_widgets_by_channel = dict()
         for k, v in init.items():
             self[k] = v
 
-    def get_dds_sysclk(self):
-        return self.dds_sysclk
-
     def __setitem__(self, k, v):
         if k in self.ttl_widgets:
             del self[k]
         if k in self.dds_widgets:
             del self[k]
-        self.ddb[k] = v
         if not isinstance(v, dict):
             return
         try:
             if v["type"] == "local":
                 widget = None
-                if v["module"] == "artiq.coredevice.ttl":
+                if v["module"] == "artiq.coredevice.comm_tcp":
+                    self.core_addr = v["arguments"]["host"]
+                    self.new_core_addr.set()
+                elif v["module"] == "artiq.coredevice.ttl":
                     channel = v["arguments"]["channel"]
                     force_out = v["class"] == "TTLOut"
                     widget = _TTLWidget(
-                        channel, self.send_to_device, force_out, k)
+                        channel, self.ttl_set_mode, force_out, k)
                     self.ttl_widgets[k] = widget
+                    self.ttl_widgets_by_channel[channel] = widget
                     self.ttl_cb()
-                if (v["module"] == "artiq.coredevice.dds"
+                    self.setup_ttl_monitoring(True, channel)
+                elif (v["module"] == "artiq.coredevice.dds"
                         and v["class"] == "DDSGroupAD9914"):
                     self.dds_sysclk = v["arguments"]["sysclk"]
-                if (v["module"] == "artiq.coredevice.dds"
+                elif (v["module"] == "artiq.coredevice.dds"
-                        and v["class"] in {"DDSChannelAD9914"}):
+                        and v["class"] == "DDSChannelAD9914"):
                     bus_channel = v["arguments"]["bus_channel"]
                     channel = v["arguments"]["channel"]
-                    widget = _DDSWidget(
+                    widget = _DDSWidget(bus_channel, channel, k)
-                        bus_channel, channel, self.get_dds_sysclk, k)
+                    self.dds_widgets[k] = widget
-                    self.dds_widgets[channel] = widget
+                    self.dds_widgets_by_channel[(bus_channel, channel)] = widget
                     self.dds_cb()
+                    self.setup_dds_monitoring(True, bus_channel, channel)
                 if widget is not None and "comment" in v:
                     widget.setToolTip(v["comment"])
         except KeyError:
             pass
 
     def __delitem__(self, k):
-        del self.ddb[k]
         if k in self.ttl_widgets:
-            self.ttl_widgets[k].deleteLater()
+            widget = self.ttl_widgets[k]
+            self.setup_ttl_monitoring(False, widget.channel)
+            widget.deleteLater()
+            del self.ttl_widgets_by_channel[widget.channel]
             del self.ttl_widgets[k]
             self.ttl_cb()
         if k in self.dds_widgets:
-            self.dds_widgets[k].deleteLater()
+            widget = self.dds_widgets[k]
+            self.setup_dds_monitoring(False, widget.bus_channel, widget.channel)
+            widget.deleteLater()
+            del self.dds_widgets_by_channel[(widget.bus_channel, widget.channel)]
             del self.dds_widgets[k]
             self.dds_cb()
 
-    def get_core_addr(self):
+    def ttl_set_mode(self, channel, mode):
+        if self.core_connection is not None:
+            widget = self.ttl_widgets_by_channel[channel]
+            if mode == "0":
+                widget.cur_override = True
+                self.core_connection.inject(channel, TTL_OVERRIDE_LEVEL, 0)
+                self.core_connection.inject(channel, TTL_OVERRIDE_OE, 1)
+                self.core_connection.inject(channel, TTL_OVERRIDE_EN, 1)
+            elif mode == "1":
+                widget.cur_override = True
+                self.core_connection.inject(channel, TTL_OVERRIDE_LEVEL, 1)
+                self.core_connection.inject(channel, TTL_OVERRIDE_OE, 1)
+                self.core_connection.inject(channel, TTL_OVERRIDE_EN, 1)
+            elif mode == "exp":
+                widget.cur_override = False
+                self.core_connection.inject(channel, TTL_OVERRIDE_EN, 0)
+            else:
+                raise ValueError
+
+    def setup_ttl_monitoring(self, enable, channel):
+        if self.core_connection is not None:
+            self.core_connection.monitor(enable, channel, TTL_PROBE_LEVEL)
+            self.core_connection.monitor(enable, channel, TTL_PROBE_OE)
+
+    def setup_dds_monitoring(self, enable, bus_channel, channel):
+        if self.core_connection is not None:
+            self.core_connection.monitor(enable, bus_channel, channel)
+
+    def monitor_cb(self, channel, probe, value):
+        if channel in self.ttl_widgets_by_channel:
+            widget = self.ttl_widgets_by_channel[channel]
+            if probe == TTL_PROBE_LEVEL:
+                widget.cur_level = bool(value)
+            elif probe == TTL_PROBE_OE:
+                widget.cur_oe = bool(value)
+            widget.refresh_display()
+        if (bus_channel, channel) in self.dds_widgets_by_channel:
+            widget = self.dds_widgets_by_channel[(channel, probe)]
+            widget.cur_frequency = value*self.dds_sysclk/2**32
+            widget.refresh_display()
+
+    def injection_status_cb(self, channel, override, value):
+        if channel in self.ttl_widgets_by_channel:
+            self.ttl_widgets_by_channel[channel].cur_override = bool(value)
+
+    async def core_connector(self):
+        while True:
+            await self.new_core_addr.wait()
+            self.new_core_addr.clear()
+            if self.core_connection is not None:
+                await self.core_connection.close()
+                self.core_connection = None
+            new_core_connection = MonInjComm(self.monitor_cb, self.injection_status_cb,
+                    lambda: logger.error("lost connection to core device moninj"))
             try:
-            comm = self.ddb["comm"]
+                await new_core_connection.connect(self.core_addr, 1383)
-            while isinstance(comm, str):
+            except:
-                comm = self.ddb[comm]
+                logger.error("failed to connect to core device moninj", exc_info=True)
-            return comm["arguments"]["host"]
+            else:
-        except KeyError:
+                self.core_connection = new_core_connection
-            return None
+                for ttl_channel in self.ttl_widgets_by_channel.keys():
+                    self.setup_ttl_monitoring(True, ttl_channel)
+                for bus_channel, channel in self.dds_widgets_by_channel.keys():
+                    self.setup_dds_monitoring(True, bus_channel, channel)
+
+    async def close(self):
+        self.core_connector_task.cancel()
+        try:
+            await asyncio.wait_for(self.core_connector_task, None)
+        except asyncio.CancelledError:
+            pass
+        if self.core_connection is not None:
+            await self.core_connection.close()
 
 
 class _MonInjDock(QtWidgets.QDockWidget):
@@ -278,108 +347,24 @@ class _MonInjDock(QtWidgets.QDockWidget):
         scroll_area.setWidget(grid_widget)
 
 
-class MonInj(TaskObject):
+class MonInj:
     def __init__(self):
         self.ttl_dock = _MonInjDock("TTL")
         self.dds_dock = _MonInjDock("DDS")
 
         self.subscriber = Subscriber("devices", self.init_devices)
         self.dm = None
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        self.socket.bind(("", 0))
-        # Never ceasing to disappoint, asyncio has an issue about UDP
-        # not being supported on Windows (ProactorEventLoop) open since 2014.
-        self.loop = asyncio.get_event_loop()
-        self.thread = threading.Thread(target=self.receiver_thread,
-                                       daemon=True)
-        self.thread.start()
 
     async def start(self, server, port):
         await self.subscriber.connect(server, port)
-        try:
-            TaskObject.start(self)
-        except:
-            await self.subscriber.close()
-            raise
 
     async def stop(self):
-        await TaskObject.stop(self)
         await self.subscriber.close()
-        try:
+        if self.dm is not None:
-            # This is required to make recvfrom terminate in the thread.
+            await self.dm.close()
-            # On Linux, this raises "OSError: Transport endpoint is not
-            # connected", but still has the intended effect.
-            self.socket.shutdown(socket.SHUT_RDWR)
-        except OSError:
-            pass
-        self.socket.close()
-        self.thread.join()
-
-    def receiver_thread(self):
-        while True:
-            try:
-                data, addr = self.socket.recvfrom(2048)
-            except OSError:
-                # Windows does this when the socket is terminated
-                break
-            if addr is None:
-                # Linux does this when the socket is terminated
-                break
-            self.loop.call_soon_threadsafe(self.datagram_received, data)
-
-    def datagram_received(self, data):
-        if self.dm is None:
-            logger.debug("received datagram, but device manager "
-                         "is not present yet")
-            return
-        try:
-            hlen = 8*3+4
-            (ttl_levels, ttl_oes, ttl_overrides,
-             dds_rtio_first_channel, dds_channels_per_bus) = \
-                struct.unpack(">QQQHH", data[:hlen])
-            for w in self.dm.ttl_widgets.values():
-                channel = w.channel
-                w.set_value(ttl_levels & (1 << channel),
-                            ttl_oes & (1 << channel),
-                            ttl_overrides & (1 << channel))
-            dds_data = data[hlen:]
-            ndds = len(dds_data)//4
-            ftws = struct.unpack(">" + "I"*ndds, dds_data)
-            for w in self.dm.dds_widgets.values():
-                bus_nr = w.bus_channel - dds_rtio_first_channel
-                offset = dds_channels_per_bus*bus_nr + w.channel
-                try:
-                    ftw = ftws[offset]
-                except KeyError:
-                    pass
-                else:
-                    w.set_value(ftw)
-        except:
-            logger.warning("failed to process datagram", exc_info=True)
-
-    def send_to_device(self, data):
-        if self.dm is None:
-            logger.debug("cannot sent to device yet, no device manager")
-            return
-        ca = self.dm.get_core_addr()
-        logger.debug("core device address: %s", ca)
-        if ca is None:
-            logger.error("could not find core device address")
-        else:
-            try:
-                self.socket.sendto(data, (ca, 3250))
-            except:
-                logger.debug("could not send to device",
-                             exc_info=True)
-
-    async def _do(self):
-        while True:
-            await asyncio.sleep(0.2)
-            # MONINJ_REQ_MONITOR
-            self.send_to_device(b"\x01")
 
     def init_devices(self, d):
-        self.dm = _DeviceManager(self.send_to_device, d)
+        self.dm = _DeviceManager(d)
         self.dm.ttl_cb = lambda: self.ttl_dock.layout_widgets(
                             self.dm.ttl_widgets.items())
         self.dm.dds_cb = lambda: self.dds_dock.layout_widgets(