1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
|
"""
ui/namespace_panel.py
NAMESPACE panel — dockable, live browser of everything referenceable in a
script: Signals (raw device endpoints), Channels (processed/derived
values), and vars (the shared dict bridging Control Panel On Action
Scripts and Channel scripts). Read-only reference, not an editor — editing
still happens in Configuration > Channels.
Docked opposite the Control Panel (ui/main_window.py), built the same way:
a QDockWidget you can drag, float, redock, or close.
"""
from __future__ import annotations
from typing import Dict, Tuple
from PyQt6.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QLabel,
QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView,
QApplication,
)
from PyQt6.QtCore import QTimer, pyqtSlot
from PyQt6.QtGui import QColor
from devices.device_registry import DeviceRegistry
from core.signal_processor import SignalProcessor
_VARS_POLL_MS = 400
_C_KIND = 0
_C_ID = 1
_C_NAME = 2
_C_DIR = 3
_C_VALUE = 4
_C_SOURCE = 5
def _fmt_value(v) -> str:
if v is None:
return "—"
if isinstance(v, float):
return f"{v:.4g}"
return str(v)
def _fmt_ref(dev_id: str, ch_id: str) -> str:
"""Absolute reference for a source — device_id/channel_id for a Signal
(device_id is stable across machines even when the COM/tty port isn't,
see ProfileManager.apply()'s port re-detection), or the bare channel_id
for another Channel (derived channels have no device to qualify with)."""
if dev_id == "derived":
return ch_id
return f"{dev_id}/{ch_id}"
class NamespacePanel(QWidget):
def __init__(self, registry: DeviceRegistry, processor: SignalProcessor):
super().__init__()
self.registry = registry
self.processor = processor
# row key -> row index. Keys:
# ("signal", dev_id, ch_id) | ("channel", channel_id) | ("var", name)
self._row_index: Dict[Tuple, int] = {}
self._build()
self._rebuild_rows()
self._poll = QTimer(self)
self._poll.setInterval(_VARS_POLL_MS)
self._poll.timeout.connect(self._on_poll)
self._poll.start()
def _build(self):
lay = QVBoxLayout(self)
lay.setContentsMargins(0, 0, 0, 0)
lay.setSpacing(0)
hint = QLabel(
"Double-click a row to copy reference to clipboard"
)
hint.setObjectName("traceSource")
hint.setWordWrap(True)
hint.setContentsMargins(8, 6, 8, 6)
lay.addWidget(hint)
self._table = QTableWidget()
self._table.setObjectName("channelTable")
self._table.setColumnCount(6)
self._table.setHorizontalHeaderLabels(
["Kind", "ID", "Name", "Dir", "Value", "Source"]
)
hdr = self._table.horizontalHeader()
hdr.setSectionResizeMode(_C_SOURCE, QHeaderView.ResizeMode.Stretch)
hdr.setSectionResizeMode(_C_KIND, QHeaderView.ResizeMode.ResizeToContents)
hdr.setSectionResizeMode(_C_ID, QHeaderView.ResizeMode.ResizeToContents)
hdr.setSectionResizeMode(_C_NAME, QHeaderView.ResizeMode.ResizeToContents)
hdr.setSectionResizeMode(_C_DIR, QHeaderView.ResizeMode.ResizeToContents)
hdr.setSectionResizeMode(_C_VALUE, QHeaderView.ResizeMode.ResizeToContents)
self._table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self._table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self._table.setAlternatingRowColors(True)
self._table.verticalHeader().setVisible(False)
self._table.cellDoubleClicked.connect(self._on_double_clicked)
lay.addWidget(self._table)
# ── Row-set (structural) rebuild — devices/channels/vars added or removed ──
def _desired_keys(self):
keys = []
for dev in self.registry.all_instances():
for ch in dev.info.signals:
if ch.enabled:
keys.append(("signal", dev.info.device_id, ch.signal_id))
for dc in self.processor.get_derived():
keys.append(("channel", dc.channel_id))
for name in self.processor.list_vars():
keys.append(("var", name))
return keys
def _rebuild_rows(self):
keys = self._desired_keys()
if list(self._row_index.keys()) == keys:
return # no structural change — values refresh independently
self._row_index.clear()
self._table.setRowCount(len(keys))
for row, key in enumerate(keys):
self._row_index[key] = row
self._paint_row(row, key)
def _paint_row(self, row: int, key: Tuple):
kind = key[0]
if kind == "signal":
_, dev_id, ch_id = key
dev = self.registry.get_instance(dev_id)
ch = dev.get_signal(ch_id) if dev else None
name = ch.name if ch else ch_id
is_wr = bool(ch and ch.writable)
self._set_row(row, "Signal", ch_id, name, "WRITE" if is_wr else "READ",
None, _fmt_ref(dev_id, ch_id))
elif kind == "channel":
_, channel_id = key
dc = next((d for d in self.processor.get_derived()
if d.channel_id == channel_id), None)
name = dc.name if dc else channel_id
if dc and dc.sources:
# Absolute reference(s) this Channel is chained from — same
# format as another row's ID column, so a chain (Channel 1 ->
# Channel 2 -> Channel 3) can be traced by eye: Channel 3's
# Source literally matches Channel 2's ID.
src = ", ".join(_fmt_ref(d, c) for d, c in dc.sources)
else:
src = dc.kind if dc else ""
self._set_row(row, "Channel", channel_id, name, "READ", None, src)
else: # var
_, name = key
val = self.processor.get_var(name, None)
self._set_row(row, "var", name, name, "R/W", val, "shared (vars)")
def _set_row(self, row: int, kind: str, id_ref: str, name: str,
direction: str, value, source: str):
self._table.setItem(row, _C_KIND, QTableWidgetItem(kind))
id_item = QTableWidgetItem(id_ref)
id_item.setForeground(QColor("#94a3b8"))
self._table.setItem(row, _C_ID, id_item)
self._table.setItem(row, _C_NAME, QTableWidgetItem(name))
dir_item = QTableWidgetItem(direction)
dir_item.setForeground(_dir_color(direction))
self._table.setItem(row, _C_DIR, dir_item)
self._table.setItem(row, _C_VALUE, QTableWidgetItem(_fmt_value(value)))
self._table.setItem(row, _C_SOURCE, QTableWidgetItem(source))
# ── Live value updates ──────────────────────────────────────────────────
@pyqtSlot(str, str, float, float)
def on_processed_data(self, device_id: str, channel_id: str, ts: float, value: float):
key = ("channel", channel_id) if device_id == "derived" else ("signal", device_id, channel_id)
row = self._row_index.get(key)
if row is None:
return
item = self._table.item(row, _C_VALUE)
if item is not None:
item.setText(_fmt_value(value))
def _on_poll(self):
self._rebuild_rows()
for name in self.processor.list_vars():
row = self._row_index.get(("var", name))
if row is None:
continue
item = self._table.item(row, _C_VALUE)
if item is not None:
item.setText(_fmt_value(self.processor.get_var(name, None)))
# ── Copy-to-clipboard reference ─────────────────────────────────────────
def _on_double_clicked(self, row: int, _col: int):
key = next((k for k, r in self._row_index.items() if r == row), None)
if key is None:
return
if key[0] == "var":
snippet = f'vars.get("{key[1]}", 0)'
elif key[0] == "signal":
snippet = f'channels.get("{key[2]}", 0)'
else: # channel
snippet = f'channels.get("{key[1]}", 0)'
QApplication.clipboard().setText(snippet)
def _dir_color(direction: str) -> QColor:
if direction == "WRITE":
return QColor("#f59e0b")
if direction == "R/W":
return QColor("#818cf8")
return QColor("#22c55e")
|
