Add dash based device monitor

docs/device_object.rst

@@ -194,7 +194,7 @@ This stores all parameter values (of parameters that can be set). With
 
 you can reset it to the stored
 configuration. Alternatively you can use "device.save_state(name)" and "device.set_state(name)" to store and set multiple working points with
-custom names. They can also be accessed via "device.states" in case you forgot the name. To store a state in json-file, use device.save_to_file(name, path), you can load it 
+custom names. They can also be accessed via "device.states" in case you forgot the name. To store a state in json-file, use device.save_to_file(name, path), you can load it
 via "device.load_state_from_file(name, path)" to load it (use "device.set_state(name)" afterwards to set it) or with "device.set_state_from_file(name, path)" to directly apply the configuration to the device.
 For all of those methods the parameters are ramped to the final state by default (with the default QuMada ramp rate), again depending on the "device.ramp" setting.
 

pyproject.toml

@@ -24,12 +24,14 @@ dependencies = [
     "jsonschema",
     "zhinst-toolkit >= 0.3.3",
     "pyqt5",
+    "shapely",
     "versioningit ~= 2.2.0",
 ]
 
 [project.optional-dependencies]
 gui = ["plottr"]
 spyder = ["spyder"]
+monitor = ["dash", "ezdxf", "dash-extensions"]
 
 [project.urls]
 Repository = "https://github.com/qutech/qumada"

src/qumada-monitor/README.md

@@ -0,0 +1,30 @@
+# QuMADA Device Monitor
+
+This is a dash app that is intended to monitor a qumada device. To use it you need to install the `monitor` dependencies for example via
+`pip install qumada[monitor]` or with a local editable install `pip install -e .[monitor]`.
+
+The feature that separates it from the qcodes monitor is the ability to display a proper layout.
+
+```python
+from qumada.utils.device_server import start_monitor_socket
+
+device: 'QumadaDevice' = ...
+my_dxf_file = ...
+
+start_monitor_socket(device, my_dxf_file)
+```
+
+This does the following steps under the hood:
+
+1. Create a list of `qumada.utils.geometry.Gate` objects with the correct labels
+    - Load a dxf file with `doc = ezdxf.readfile('my_layout.dxf')`
+    - Crop it to the desired region with `raw_gates = qumada.utils.dxf.get_gates_from_cropped_region`
+    - Merge gates from the same layer that overlap (necessary for some reason) with `merged_gates = qumada.utils.dxf.auto_merge`
+    - Give the gates the proper labels with a helper GUI: `gates = qumada.utils.dxf.label_gates(merged_gates)`
+    - Save the gate list to json with `qumada.utils.geometry.store_to_file(gates, 'my_gates.json')`
+2. Start the device monitor server and set its gate_geometry
+
+To view the monitor you need to start the webserver by running `python -m qumada-monitor` in a shell
+
+The webserver will try to match parameters to gates based on the gate labels being present in the parameter labels.
+Un-matched gates are nan.

src/qumada-monitor/__main__.py

@@ -0,0 +1,13 @@
+import argparse
+
+from .app import make_app
+
+p = argparse.ArgumentParser(description="Dash front-end for your measurement WebSocket")
+p.add_argument("--ws", default="ws://127.200.200.9:6789", help="WebSocket URL, e.g. ws://localhost:8765")
+p.add_argument("--host", default="127.0.0.1", help="Dash host to bind")
+p.add_argument("--port", default=8050, type=int, help="Dash port (default: 8050)")
+p.add_argument("--debug", default=False, action="store_true", help="Enable debug mode")
+args = p.parse_args()
+
+app = make_app(args.ws)
+app.run(host=args.host, port=args.port, debug=args.debug)

src/qumada-monitor/app.py

@@ -0,0 +1,228 @@
+#!/usr/bin/env python3
+"""
+Live quantum‑device monitor: geometry + voltages ➜ interactive SVG plot.
+
+Start with:
+    python dash_device_monitor.py --ws ws://localhost:8765 --colorscale Cividis
+"""
+import argparse
+import itertools
+import json
+import os
+import warnings
+from collections import defaultdict
+
+import dash.exceptions
+import plotly.express as px
+import plotly.graph_objects as go
+from dash import Dash, Input, Output, State, dash_table, dcc, html, no_update
+from dash_extensions import WebSocket  # pip install dash-extensions
+from plotly.colors import sample_colorscale
+from shapely.geometry import Polygon
+
+# ---------------- helpers -------------------------------------------------- #
+from qumada.utils.geometry import Gate, string_to_gate_list
+
+
+def layer_palette(layers):
+    """Return dict layer -> rgba border colour (qualitative palette, repeats if needed)."""
+    palette = itertools.cycle(px.colors.qualitative.Plotly)
+    return {layer: next(palette) for layer in layers}
+
+
+def voltage_colour(v, vabs_max, colorscale):
+    """Map voltage to rgba string using the chosen Plotly colourscale."""  # flat value
+    ratio = 0.5 + v / vabs_max / 2
+    return sample_colorscale(colorscale, [ratio])[0]  # :contentReference[oaicite:9]{index=9}
+
+
+def _set_alpha(color: str, value) -> str:
+    if color.startswith("rgb("):
+        result = color.replace("rgb(", "rgba(")
+        result = result.replace(")", f",{value})")
+    elif color.startswith("rgba("):
+        result = color.rsplit(",", 1)[0]
+        result = f"{result},{value})"
+    else:
+        raise NotImplementedError(color)
+    return result
+
+
+def gates_to_figure(gates: list[Gate], voltages: list[dict], colorscale):
+    """Build a Plotly figure from gate list + latest voltages dict."""
+    if not gates:
+        return go.Figure()
+
+    gate_runtime_info = []
+    for gate in gates:
+        candidates = []
+        for data in voltages:
+            if gate.label in data["label"]:
+                candidates.append(data)
+        if not candidates:
+            gate_runtime_info.append(None)
+        else:
+            if len(candidates) > 1:
+                data = min(candidates, key=lambda d: d["label"])
+            else:
+                (data,) = candidates
+            gate_runtime_info.append(data)
+
+    # 1) derive colour mapping
+    v_values = [info["value"] for info in gate_runtime_info if info is not None]
+    vabs_max = max(map(abs, v_values + [1.0]))
+    layers = sorted({g.layer for g in gates})
+    border_col = layer_palette(layers)  # :contentReference[oaicite:10]{index=10}
+
+    fig = go.Figure()
+    for gate, info in zip(gates, gate_runtime_info):
+        poly: Polygon = gate.polygon
+        x, y = poly.exterior.xy
+        x = list(x)
+        y = list(y)
+        if info:
+            v = info["value"]
+            unit = info["unit"]
+            text = f"{gate.label}:<br>{v:.3f} {unit}"
+            fill_col = voltage_colour(v, vabs_max, colorscale)
+        else:
+            text = f"{gate.label}:<br>NaN"
+            fill_col = None
+        line_col = border_col[gate.layer]
+
+        if fill_col is None:
+            fill_col = "rgba(0,0,0,0)"
+        else:
+            fill_col = _set_alpha(fill_col, 0.3)
+
+        fig.add_trace(
+            go.Scatter(
+                x=x,
+                y=y,
+                fill="toself",  # polygon fill trick :contentReference[oaicite:11]{index=11}
+                fillcolor=fill_col,
+                line=dict(color=line_col, width=1),
+                hoverinfo="text",
+                text=text,
+                showlegend=False,
+                mode="lines",
+            )
+        )
+        # add static label at predefined position
+        if getattr(gate, "label_position", None):
+            lx, ly = gate.label_position
+            fig.add_annotation(x=lx, y=ly, text=text, showarrow=False, font=dict(size=10, color="black"))
+
+    fig.update_layout(
+        xaxis=dict(scaleanchor="y", visible=False),
+        yaxis=dict(visible=False),
+        margin=dict(l=0, r=0, t=0, b=0),
+        plot_bgcolor="white",
+    )
+    return fig
+
+
+# ---------------- Dash app -------------------------------------------------- #
+def make_app(ws_url: str) -> Dash:
+    app = Dash(__name__, title="Quantum‑dot voltage monitor")
+
+    app.layout = html.Div(
+        [
+            html.H3("Live device layout"),
+            WebSocket(id="ws", url=ws_url),
+            dcc.Store(id="gate-geometry"),
+            dcc.Store(id="voltages"),
+            dcc.Store(id="parameters"),
+            dcc.Graph(id="layout-graph", style={"height": "700px"}),
+            dcc.Dropdown(
+                id="colorscale-dropdown",
+                value="rdbu",
+                options=[{"label": cs, "value": cs} for cs in px.colors.named_colorscales()],
+                clearable=False,
+                style={"width": "250px"},
+            ),
+            html.Hr(),
+            dash_table.DataTable(
+                id="parameter-table",
+                columns=[
+                    {"name": "Label", "id": "label"},
+                    {"name": "Value", "id": "value"},
+                    {"name": "Unit", "id": "unit"},
+                    {"name": "Timestamp", "id": "timestamp"},
+                    {"name": "Name", "id": "name"},
+                ],
+                style_cell={"fontFamily": "monospace", "padding": "2px 6px"},
+                style_table={"max-height": "400px", "overflowY": "auto"},
+            ),
+        ],
+        style={"font-family": "Source Sans Pro, sans-serif", "margin": "0 20px"},
+    )
+
+    # ---------- 1) unpack every WebSocket message -------------------------- #
+    @app.callback(
+        Output("gate-geometry", "data"),
+        Output("parameters", "data"),
+        Input("ws", "message"),
+        prevent_initial_call=True,
+    )
+    def _unpack_ws(msg):
+        if msg is None:
+            return no_update, no_update
+
+        data = json.loads(msg["data"])
+
+        parameters = gate_geometry = no_update
+        if "parameters" in data:
+            parameters = data["parameters"]
+        if "gate_geometry" in data:
+            gate_geometry = data["gate_geometry"]
+        return gate_geometry, parameters
+
+    @app.callback(
+        Output("voltages", "data"),
+        Input("parameters", "data"),
+        prevent_initial_call=True,
+    )
+    def _select_voltages(parameters):
+        voltages = []
+        for parameter in parameters:
+            assert "unit" in parameter, f"{parameter!r}"
+            if "V" in parameter["unit"]:
+                value = {attr: parameter[attr] for attr in ["name", "value", "label", "unit"]}
+                voltages.append(value)
+        return voltages
+
+    @app.callback(
+        Output("parameter-table", "data"),
+        Input("parameters", "data"),
+    )
+    def _parameter_table(parameters):
+        if not parameters:
+            return []
+
+        cols = ["label", "value", "unit", "timestamp", "name"]
+        return [{col: parameter[col] for col in cols} for parameter in parameters]
+
+    @app.callback(
+        Output("layout-graph", "figure"),
+        Input("gate-geometry", "data"),
+        Input("voltages", "data"),
+        Input("colorscale-dropdown", "value"),
+    )
+    def _draw_layout(geom_str, volts, colorscale_selected):
+        if geom_str is None or volts is None:
+            fig = go.Figure()
+        else:
+            gates = string_to_gate_list(geom_str)
+            fig = gates_to_figure(gates, volts, colorscale_selected)
+
+        fig.update_layout(
+            xaxis=dict(scaleanchor="y", visible=False),
+            yaxis=dict(visible=False),
+            margin=dict(l=0, r=0, t=0, b=0),
+            plot_bgcolor="white",
+            uirevision=hash(geom_str),
+        )
+        return fig
+
+    return app

src/qumada/instrument/custom_drivers/ZI/MFLI.py

@@ -41,11 +41,13 @@ class MFLI(Instrument):
     """
 
     def __init__(
-        self, name: str, device: str,
+        self,
+        name: str,
+        device: str,
         serverhost: str = "localhost",
-        existing_session: Session = None, 
-        allow_version_mismatch = False,
-        **kwargs
+        existing_session: Session = None,
+        allow_version_mismatch=False,
+        **kwargs,
     ):
         super().__init__(name, **kwargs)
         if isinstance(existing_session, Session):

src/qumada/instrument/mapping/Dummies/DummyDac.py

@@ -60,8 +60,8 @@ def ramp(
 
         if not start_values:
             start_values = [param.get() for param in parameters]
-            
-        print("Ramping...")    
+
+        print("Ramping...")
         instrument._triggered_ramp_channels(
             [param._instrument for param in parameters], start_values, end_values, ramp_time, num_points
         )
@@ -96,7 +96,7 @@ def pulse(
 
     def setup_trigger_in():
         pass
-    
+
     def force_trigger(self):
         self._instrument.force_trigger()
-        self._instrument._is_triggered.clear()
+        self._instrument._is_triggered.clear()

src/qumada/instrument/mapping/Harvard/Decadac.py

@@ -19,12 +19,14 @@
 # - Till Huckeman
 
 
+import logging
+
 from qcodes.instrument.parameter import Parameter
 
 from qumada.instrument.custom_drivers.Harvard.Decadac import Decadac
 from qumada.instrument.mapping import DECADAC_MAPPING
 from qumada.instrument.mapping.base import InstrumentMapping
-import logging
+
 logger = logging.getLogger(__name__)
 
 
@@ -159,6 +161,6 @@ def trigger_in(self, trigger: str | None) -> None:
     def force_trigger(self):
         string = ""
         for b in range(0, 6):
-            for c in range(0,6):
-                string+=f"B{b};C{c};G0;"
-        self._instrument.write(string)
+            for c in range(0, 6):
+                string += f"B{b};C{c};G0;"
+        self._instrument.write(string)

src/qumada/instrument/mapping/QDevil/qdac.py

@@ -30,6 +30,7 @@ class QDacMapping(InstrumentMapping):
     def __init__(self):
         super().__init__(QDAC_MAPPING)
         self.max_ramp_channels = 8
+
     def ramp(
         self,
         parameters: list[Parameter],
@@ -72,7 +73,7 @@ def ramp(
 
     def setup_trigger_in(self):
         raise Exception("QDac does not have a trigger input!")
-        
+
     def force_trigger(self):
         pass
         # Not required as QDac has no trigger input and starts ramps instantly.

src/qumada/instrument/mapping/QDevil/qdac2.py

@@ -182,7 +182,7 @@ def setup_trigger_in(self):
             "QDac2 does not have a trigger input \
             not yet supported!"
         )
-            
+
     def force_trigger(self):
         pass
         # Currently no trigger inputs are supported, thus all ramps are started