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
|
"""
core/reference_registry.py
Generic (device_id, channel_id) reference-rename propagation.
A Signal (physical device channel) or a Channel (derived, device_id ==
"derived") can be renamed by the user. Both kinds are referenced the same
way everywhere downstream — plot traces, control targets, filter
pipelines, other derived channels' sources — as a plain
(device_id, channel_id) pair. Renaming one without rewriting every
reference to it leaves those references dangling: they silently stop
receiving data (this is exactly the bug that motivated this module — an
Arduino analog pin remap changed a Signal's id and orphaned its plot
trace).
propagate_rename() is the single place that walks every subsystem able to
hold such a reference and rewrites it. It expects a MainWindow instance
(or anything exposing the same .engine/.processor/._chart/._ctrl/
._win_config attributes) since that's the one place all of them are
reachable from.
"""
from typing import List, Tuple
Ref = Tuple[str, str]
def propagate_rename(app, old_ref: Ref, new_ref: Ref) -> None:
if old_ref == new_ref:
return
# Physical signals live in the acquisition engine's buffers; derived
# Channels don't (their buffers live in SignalProcessor._derived_bufs).
if old_ref[0] != "derived":
app.engine.rename_channel(old_ref[0], old_ref[1], new_ref[1])
# Filter pipelines, other derived channels' sources, latest-value cache,
# and (if old_ref is itself a derived Channel) its ring buffer.
app.processor.rename_reference(old_ref, new_ref)
# Live plot — trace specs in the strip chart's current layout.
chart = getattr(app, "_chart", None)
changed = False
if chart is not None and chart._cfg is not None:
for pane in chart._cfg.panes:
for tr in pane.traces:
if (tr.device_id, tr.channel_id) == old_ref:
tr.device_id, tr.channel_id = new_ref
changed = True
if changed:
chart.refresh()
# Control panel — saved specs (profile persistence) + live bound widgets
# (actual I/O routing — ControlWidget reads its own .device_id/.channel_id
# directly in _write()/_check_stale(), not from the spec).
ctrl = getattr(app, "_ctrl", None)
if ctrl is not None:
for spec in ctrl.get_specs():
if (spec.device_id, spec.channel_id) == old_ref:
spec.device_id, spec.channel_id = new_ref
for widget in getattr(ctrl, "_widgets", []):
if (widget.device_id, widget.channel_id) == old_ref:
widget.device_id, widget.channel_id = new_ref
# Config window tabs, if currently open — refresh so pickers/tables
# don't show the stale id until next full refresh.
win = getattr(app, "_win_config", None)
if win is not None:
if chart is not None:
win.refresh_plot(chart._cfg)
if hasattr(win, "refresh_derived"):
win.refresh_derived()
if hasattr(win, "refresh_devices"):
win.refresh_devices()
def find_references(app, ref: Ref) -> List[str]:
"""Reverse lookup: human-readable names of everything currently holding
a (device_id, channel_id) reference — same 5 subsystems propagate_rename()
walks, but collecting instead of rewriting. Used to flag a signal_id an
Apply & Reconnect just dropped (still referenced elsewhere = orphaned),
without blocking the reconnect itself — see devices_window/channels_window/
control_panel "missing source" indicators, which do the live per-widget
check; this is only for a one-line summary breadcrumb."""
found: List[str] = []
for dc in app.processor.get_derived():
if ref in dc.sources:
found.append(f"Channel '{dc.name}'")
chart = getattr(app, "_chart", None)
if chart is not None and chart._cfg is not None:
for pane in chart._cfg.panes:
for tr in pane.traces:
if (tr.device_id, tr.channel_id) == ref:
found.append(f"Plot '{pane.title}' trace '{tr.label or tr.channel_id}'")
ctrl = getattr(app, "_ctrl", None)
if ctrl is not None:
for spec in ctrl.get_specs():
if (spec.device_id, spec.channel_id) == ref:
found.append(f"Control '{spec.title}'")
return found
|
