#!/usr/bin/env python3
"""
WeConnect GUI client.
Connects to the collect.py TCP push server and displays live data.
Tabs
----
Connector — server address, connect / disconnect, status
Dashboard — live tree view of the latest snapshot
Plot — time-series charts of all physical {value, unit} fields
"""
import json
import socket
import sys
import threading
from collections import deque
from datetime import datetime
from pathlib import Path
from PyQt5.QtCore import QObject, Qt, pyqtSignal
from PyQt5.QtGui import QColor, QFont
from PyQt5.QtWidgets import (
QApplication, QComboBox, QFormLayout, QGridLayout, QGroupBox,
QHBoxLayout, QLabel, QLineEdit, QMainWindow, QPushButton, QSizePolicy,
QSpinBox, QStatusBar, QTabWidget, QTreeWidget, QTreeWidgetItem,
QVBoxLayout, QWidget,
)
import pyqtgraph as pg
pg.setConfigOption("background", "w")
pg.setConfigOption("foreground", "k")
MAX_PLOT_POINTS = 200
_SETTINGS_PATH = Path.home() / ".config" / "we_monitor" / "plot_settings.json"
# ── TCP reader ────────────────────────────────────────────────────────────────
class TcpReader(QObject):
"""Background thread that emits decoded JSON messages via Qt signals."""
message = pyqtSignal(dict)
status_changed = pyqtSignal(str) # emitted only on unexpected loss
def __init__(self):
super().__init__()
self._sock = None
self._running = False
def connect_to(self, host: str, port: int) -> str | None:
"""Open connection; returns error string or None on success."""
try:
self._sock = socket.create_connection((host, port), timeout=5)
self._sock.settimeout(None) # blocking reads after connect
except OSError as exc:
return str(exc)
self._running = True
threading.Thread(target=self._read_loop, daemon=True).start()
return None
def disconnect(self):
self._running = False
if self._sock:
try:
self._sock.close()
except OSError:
pass
self._sock = None
def _read_loop(self):
buf = ""
try:
while self._running:
chunk = self._sock.recv(4096)
if not chunk:
break
buf += chunk.decode(errors="replace")
while "\n" in buf:
line, buf = buf.split("\n", 1)
line = line.strip()
if line:
try:
self.message.emit(json.loads(line))
except json.JSONDecodeError:
pass
except OSError:
pass
if self._running:
self._running = False
self.status_changed.emit("Connection lost")
# ── helpers ───────────────────────────────────────────────────────────────────
def _is_phys(v) -> bool:
return isinstance(v, dict) and "value" in v and "unit" in v
def _extract_phys(data: dict, prefix: str = "") -> dict[str, dict]:
"""Return a flat dict of path → {value, unit} for every physical leaf."""
result = {}
for k, v in data.items():
if k == "ts":
continue
path = f"{prefix}.{k}" if prefix else k
if _is_phys(v):
result[path] = v
elif isinstance(v, dict):
result.update(_extract_phys(v, path))
return result
# ── Connector tab ─────────────────────────────────────────────────────────────
class ConnectorTab(QWidget):
connect_requested = pyqtSignal(str, int)
disconnect_requested = pyqtSignal()
def __init__(self):
super().__init__()
layout = QVBoxLayout(self)
layout.setAlignment(Qt.AlignTop)
grp = QGroupBox("Server")
form = QFormLayout(grp)
self._host = QLineEdit("127.0.0.1")
self._port = QSpinBox()
self._port.setRange(1, 65535)
self._port.setValue(9999)
form.addRow("Host:", self._host)
form.addRow("Port:", self._port)
btn_row = QHBoxLayout()
self._btn = QPushButton("Connect")
self._btn.setFixedWidth(110)
self._btn.clicked.connect(self._toggle)
btn_row.addWidget(self._btn)
btn_row.addStretch()
form.addRow(btn_row)
self._status = QLabel("Disconnected")
self._status.setStyleSheet("color: gray;")
form.addRow("Status:", self._status)
layout.addWidget(grp)
def _toggle(self):
if self._btn.text() == "Connect":
self.connect_requested.emit(self._host.text(), self._port.value())
else:
self.disconnect_requested.emit()
def set_connected(self, host: str, port: int):
self._btn.setText("Disconnect")
self._host.setEnabled(False)
self._port.setEnabled(False)
self._status.setText(f"Connected → {host}:{port}")
self._status.setStyleSheet("color: green; font-weight: bold;")
def set_disconnected(self, reason: str = ""):
self._btn.setText("Connect")
self._host.setEnabled(True)
self._port.setEnabled(True)
if reason:
self._status.setText(f"Disconnected ({reason})")
self._status.setStyleSheet("color: red;")
else:
self._status.setText("Disconnected")
self._status.setStyleSheet("color: gray;")
# ── Dashboard tab ─────────────────────────────────────────────────────────────
class DashboardTab(QWidget):
def __init__(self):
super().__init__()
layout = QVBoxLayout(self)
self._ts = QLabel("Last update: —")
self._ts.setAlignment(Qt.AlignRight)
self._ts.setStyleSheet("color: gray; font-size: 11px;")
layout.addWidget(self._ts)
self._tree = QTreeWidget()
self._tree.setColumnCount(3)
self._tree.setHeaderLabels(["Field", "Value", "Unit"])
self._tree.setColumnWidth(0, 320)
self._tree.setColumnWidth(1, 180)
self._tree.setColumnWidth(2, 80)
self._tree.setAlternatingRowColors(True)
self._tree.setRootIsDecorated(True)
layout.addWidget(self._tree)
def update(self, snapshot: dict):
self._ts.setText(f"Last update: {snapshot.get('ts', '—')}")
expanded = self._expanded_paths()
self._tree.clear()
bold = QFont()
bold.setBold(True)
domain_color = QColor("#1a4d99")
for domain, domain_data in snapshot.items():
if domain == "ts" or not isinstance(domain_data, dict):
continue
d_item = QTreeWidgetItem([domain.upper()])
d_item.setFont(0, bold)
d_item.setForeground(0, domain_color)
for obj_name, fields in domain_data.items():
if not isinstance(fields, dict):
continue
obj_item = QTreeWidgetItem([obj_name])
obj_item.setFont(0, bold)
self._add_fields(obj_item, fields, f"{domain}/{obj_name}")
d_item.addChild(obj_item)
self._tree.addTopLevelItem(d_item)
self._tree.expandAll()
self._restore_expanded(expanded)
def _add_fields(self, parent: QTreeWidgetItem, data: dict, path: str):
for k, v in data.items():
child_path = f"{path}/{k}"
if _is_phys(v):
item = QTreeWidgetItem([k, str(v["value"]), v["unit"]])
item.setTextAlignment(1, Qt.AlignRight | Qt.AlignVCenter)
parent.addChild(item)
elif isinstance(v, dict):
node = QTreeWidgetItem([k])
self._add_fields(node, v, child_path)
parent.addChild(node)
else:
item = QTreeWidgetItem([k, str(v), ""])
item.setTextAlignment(1, Qt.AlignRight | Qt.AlignVCenter)
parent.addChild(item)
# ── preserve expand state across refreshes ──────────────────────────────
def _expanded_paths(self) -> set[str]:
paths = set()
root = self._tree.invisibleRootItem()
self._collect_expanded(root, "", paths)
return paths
def _collect_expanded(self, item, prefix, paths):
for i in range(item.childCount()):
child = item.child(i)
path = f"{prefix}/{child.text(0)}"
if child.isExpanded():
paths.add(path)
self._collect_expanded(child, path, paths)
def _restore_expanded(self, paths: set[str]):
root = self._tree.invisibleRootItem()
self._apply_expanded(root, "", paths)
def _apply_expanded(self, item, prefix, paths):
for i in range(item.childCount()):
child = item.child(i)
path = f"{prefix}/{child.text(0)}"
if path in paths:
child.setExpanded(True)
self._apply_expanded(child, path, paths)
# ── Plot tab ──────────────────────────────────────────────────────────────────
class PlotTab(QWidget):
NUM_PLOTS = 4
MAX_TRACES = 4
COLORS = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728"]
def __init__(self):
super().__init__()
main_layout = QVBoxLayout(self)
main_layout.setContentsMargins(4, 4, 4, 4)
main_layout.setSpacing(4)
# ── toolbar ───────────────────────────────────────────────────────────
tb = QWidget()
tb_layout = QHBoxLayout(tb)
tb_layout.setContentsMargins(0, 0, 0, 0)
reset_btn = QPushButton("Reset zoom")
reset_btn.setFixedWidth(100)
reset_btn.clicked.connect(self._reset_zoom)
tb_layout.addWidget(reset_btn)
self._cursor_label = QLabel("")
self._cursor_label.setTextFormat(Qt.RichText)
self._cursor_label.setStyleSheet("font-size: 11px; color: #333;")
tb_layout.addWidget(self._cursor_label)
tb_layout.addStretch()
main_layout.addWidget(tb)
# ── data ──────────────────────────────────────────────────────────────
self._traces: list[list[str]] = [[] for _ in range(self.NUM_PLOTS)]
self._saved_traces: list[list[str]] = self._load_settings()
self._series: dict[str, deque] = {}
self._units: dict[str, str] = {}
# ── 2×2 plot grid (each cell: control bar + plot) ─────────────────────
plot_container = QWidget()
plot_grid = QGridLayout(plot_container)
plot_grid.setContentsMargins(0, 0, 0, 0)
plot_grid.setSpacing(4)
self._plot_widgets: list[pg.PlotWidget] = []
self._vlines: list[pg.InfiniteLine] = []
self._proxies: list = []
self._add_combos: list[QComboBox] = []
self._chip_bars: list[QHBoxLayout] = []
for i in range(self.NUM_PLOTS):
cell = QWidget()
cell_vbox = QVBoxLayout(cell)
cell_vbox.setContentsMargins(0, 0, 0, 0)
cell_vbox.setSpacing(2)
# control bar
ctrl = QWidget()
ctrl.setMaximumHeight(30)
ctrl_row = QHBoxLayout(ctrl)
ctrl_row.setContentsMargins(0, 0, 0, 0)
ctrl_row.setSpacing(4)
add_combo = QComboBox()
add_combo.setMinimumWidth(180)
add_combo.addItem("— add trace —")
add_combo.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self._add_combos.append(add_combo)
ctrl_row.addWidget(add_combo)
add_btn = QPushButton("+")
add_btn.setFixedSize(26, 22)
add_btn.clicked.connect(lambda checked, idx=i: self._add_trace(idx))
ctrl_row.addWidget(add_btn)
ctrl_row.addSpacing(8)
chip_bar = QHBoxLayout()
chip_bar.setSpacing(4)
chip_bar.setContentsMargins(0, 0, 0, 0)
self._chip_bars.append(chip_bar)
ctrl_row.addLayout(chip_bar)
ctrl_row.addStretch()
cell_vbox.addWidget(ctrl)
# plot
pw = pg.PlotWidget(
background="white",
axisItems={"bottom": pg.DateAxisItem(orientation="bottom")},
)
pw.showGrid(x=True, y=True, alpha=0.3)
pw.addLegend(offset=(0, 0), labelTextSize="8pt")
pw.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
vl = pg.InfiniteLine(
angle=90, movable=False,
pen=pg.mkPen("gray", width=1, style=Qt.DashLine),
)
vl.setVisible(False)
pw.addItem(vl, ignoreBounds=True)
self._vlines.append(vl)
proxy = pg.SignalProxy(
pw.scene().sigMouseMoved,
rateLimit=30,
slot=lambda evt, idx=i: self._on_mouse_move(evt, idx),
)
self._proxies.append(proxy)
cell_vbox.addWidget(pw)
self._plot_widgets.append(pw)
plot_grid.addWidget(cell, i // 2, i % 2)
main_layout.addWidget(plot_container)
# ── trace management ──────────────────────────────────────────────────────
def _add_trace(self, plot_idx: int):
key = self._add_combos[plot_idx].currentText()
traces = self._traces[plot_idx]
if key.startswith("—") or key not in self._series:
return
if key in traces or len(traces) >= self.MAX_TRACES:
return
traces.append(key)
self._add_chip(plot_idx, key, len(traces) - 1)
self._save_settings()
self._redraw()
def _add_chip(self, plot_idx: int, key: str, color_idx: int):
short = key.split(".")[-1]
color = self.COLORS[color_idx]
btn = QPushButton(f"{short} ×")
btn.setFixedHeight(22)
btn.setStyleSheet(
f"QPushButton {{ background:{color}; color:white; border-radius:3px;"
f" padding:1px 6px; font-size:11px; }}"
f"QPushButton:hover {{ background:#888; }}"
)
btn.clicked.connect(lambda: self._remove_trace(plot_idx, key))
self._chip_bars[plot_idx].addWidget(btn)
def _remove_trace(self, plot_idx: int, key: str):
if key not in self._traces[plot_idx]:
return
self._traces[plot_idx].remove(key)
self._rebuild_chips(plot_idx)
self._save_settings()
self._redraw()
def _rebuild_chips(self, plot_idx: int):
bar = self._chip_bars[plot_idx]
while bar.count():
item = bar.takeAt(0)
if item.widget():
item.widget().deleteLater()
for j, key in enumerate(self._traces[plot_idx]):
self._add_chip(plot_idx, key, j)
# ── public update called on every new snapshot ────────────────────────────
def update(self, snapshot: dict):
ts_raw = snapshot.get("ts", "")
try:
ts = datetime.fromisoformat(ts_raw).timestamp()
except (ValueError, TypeError):
ts = datetime.utcnow().timestamp()
phys = _extract_phys(snapshot)
if not phys:
return
new_keys = [k for k in phys if k not in self._series]
for k in new_keys:
self._series[k] = deque(maxlen=MAX_PLOT_POINTS)
for k, pv in phys.items():
self._units[k] = pv.get("unit", "")
try:
self._series[k].append((ts, float(pv["value"])))
except (TypeError, ValueError):
pass
if new_keys:
self._refresh_combos()
self._redraw()
# ── combo refresh + saved-trace restore ──────────────────────────────────
def _refresh_combos(self):
signals = sorted(self._series.keys())
for combo in self._add_combos:
current = combo.currentText()
combo.blockSignals(True)
combo.clear()
combo.addItem("— add trace —")
for s in signals:
combo.addItem(s)
idx = combo.findText(current)
combo.setCurrentIndex(idx if idx >= 0 else 0)
combo.blockSignals(False)
for i, saved in enumerate(self._saved_traces):
for key in list(saved):
if key not in self._series:
continue
traces = self._traces[i]
if key not in traces and len(traces) < self.MAX_TRACES:
traces.append(key)
self._add_chip(i, key, len(traces) - 1)
saved.remove(key)
# ── settings persistence ──────────────────────────────────────────────────
def _save_settings(self):
try:
_SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
_SETTINGS_PATH.write_text(
json.dumps({"traces": [list(t) for t in self._traces]}, indent=2)
)
except OSError:
pass
def _load_settings(self) -> list[list[str]]:
try:
data = json.loads(_SETTINGS_PATH.read_text())
saved = data.get("traces", [])
while len(saved) < self.NUM_PLOTS:
saved.append([])
return [list(t) for t in saved[: self.NUM_PLOTS]]
except (OSError, json.JSONDecodeError, KeyError):
return [[] for _ in range(self.NUM_PLOTS)]
# ── drawing ───────────────────────────────────────────────────────────────
def _redraw(self):
for i, pw in enumerate(self._plot_widgets):
pi = pw.getPlotItem()
pi.clear()
pi.addItem(self._vlines[i], ignoreBounds=True)
if pi.legend:
pi.legend.clear()
for j, key in enumerate(self._traces[i]):
if key not in self._series:
continue
series = self._series[key]
if not series:
continue
times, values = zip(*series)
label = f"{key.split('.')[-1]} [{self._units.get(key, '')}]"
pw.plot(
list(times), list(values),
pen=pg.mkPen(self.COLORS[j], width=1.5),
name=label,
)
# ── zoom reset ────────────────────────────────────────────────────────────
def _reset_zoom(self):
for pw in self._plot_widgets:
pw.getViewBox().autoRange()
# ── cursor tracing ────────────────────────────────────────────────────────
def _on_mouse_move(self, evt, plot_idx: int):
pos = evt[0]
pw = self._plot_widgets[plot_idx]
vl = self._vlines[plot_idx]
if not pw.sceneBoundingRect().contains(pos):
vl.setVisible(False)
self._cursor_label.setText("")
return
x = pw.getPlotItem().vb.mapSceneToView(pos).x()
vl.setPos(x)
vl.setVisible(True)
parts = []
for j, key in enumerate(self._traces[plot_idx]):
if key not in self._series:
continue
series = list(self._series[key])
if not series:
continue
_, nearest_v = min(series, key=lambda p: abs(p[0] - x))
label = key.split(".")[-1]
unit = self._units.get(key, "")
color = self.COLORS[j]
parts.append(
f'■'
f' {label}: {nearest_v:.2f} {unit}'
)
self._cursor_label.setText(" ".join(parts))
# ── Main window ───────────────────────────────────────────────────────────────
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.setWindowTitle("WeConnect Monitor")
self.resize(960, 700)
self._reader = TcpReader()
self._reader.message.connect(self._on_message)
self._reader.status_changed.connect(self._on_connection_lost)
self._connector = ConnectorTab()
self._dashboard = DashboardTab()
self._plot = PlotTab()
tabs = QTabWidget()
tabs.addTab(self._connector, "Connector")
tabs.addTab(self._dashboard, "Dashboard")
tabs.addTab(self._plot, "Plot")
self.setCentralWidget(tabs)
self._statusbar = QStatusBar()
self.setStatusBar(self._statusbar)
self._statusbar.showMessage("Disconnected")
self._connector.connect_requested.connect(self._connect)
self._connector.disconnect_requested.connect(self._disconnect)
self._record_count = 0
def _connect(self, host: str, port: int):
err = self._reader.connect_to(host, port)
if err:
self._connector.set_disconnected(err)
self._statusbar.showMessage(f"Connection failed: {err}")
else:
self._record_count = 0
self._connector.set_connected(host, port)
self._statusbar.showMessage(f"Connected to {host}:{port} | Records: 0")
def _disconnect(self):
self._reader.disconnect()
self._connector.set_disconnected()
self._statusbar.showMessage("Disconnected")
def _on_message(self, data: dict):
self._dashboard.update(data)
self._plot.update(data)
self._record_count += 1
self._statusbar.showMessage(
f"Last snapshot: {data.get('ts', '?')} | Records: {self._record_count}"
)
def _on_connection_lost(self, reason: str):
self._connector.set_disconnected(reason)
self._statusbar.showMessage(reason)
# ── entry point ───────────────────────────────────────────────────────────────
def main():
app = QApplication(sys.argv)
app.setStyle("Fusion")
win = MainWindow()
win.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()