502 lines
19 KiB
Python
502 lines
19 KiB
Python
import sys, os, math, time, json
|
|
from datetime import datetime
|
|
from queue import Queue, Empty
|
|
from collections import deque
|
|
|
|
_SETTINGS_PATH = os.path.expanduser('~/.nmea_client_plots.json')
|
|
|
|
def _load_plot_settings() -> list:
|
|
try:
|
|
with open(_SETTINGS_PATH) as f:
|
|
return json.load(f).get('plots', [])
|
|
except Exception:
|
|
return []
|
|
|
|
def _save_plot_settings(settings: list):
|
|
try:
|
|
with open(_SETTINGS_PATH, 'w') as f:
|
|
json.dump({'plots': settings}, f, indent=2)
|
|
except Exception:
|
|
pass
|
|
|
|
_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
_gui = os.path.dirname(os.path.abspath(__file__))
|
|
for _p in [_root, _gui]:
|
|
if _p not in sys.path:
|
|
sys.path.insert(0, _p)
|
|
|
|
from PyQt5.QtWidgets import (
|
|
QWidget, QVBoxLayout, QHBoxLayout, QLabel, QComboBox,
|
|
QSpinBox, QListWidget, QListWidgetItem, QSizePolicy, QApplication,
|
|
)
|
|
from PyQt5.QtGui import QPainter, QColor, QPen, QBrush, QFont, QImage
|
|
from PyQt5.QtCore import Qt, QPointF, QRectF, QThread, pyqtSignal
|
|
|
|
from data_model import ReceiverManager
|
|
|
|
_Y_OPTIONS = ['C/N₀', 'Azimuth', 'Elevation', 'pDOP']
|
|
_COLORS = ['#2196F3', '#F44336', '#4CAF50', '#FF9800',
|
|
'#9C27B0', '#00BCD4', '#FFEB3B', '#795548']
|
|
|
|
# Static colors — created once in main thread, read-only everywhere else
|
|
_C_BG = QColor('#1a1a2e')
|
|
_C_PLOT_BG = QColor('#12121e')
|
|
_C_GRID = QColor('#2a2a4a')
|
|
_C_AXIS = QColor('#555577')
|
|
_C_LABEL = QColor('#aaaacc')
|
|
_C_TEXT = QColor('#ddddee')
|
|
_QCOLORS = [QColor(c) for c in _COLORS]
|
|
|
|
|
|
# ── Tick helper ───────────────────────────────────────────────────────────────
|
|
|
|
def _nice_ticks(lo: float, hi: float, n: int = 5) -> list:
|
|
if hi - lo < 1e-9:
|
|
hi = lo + 1.0
|
|
raw = (hi - lo) / n
|
|
exp = math.floor(math.log10(raw))
|
|
base = 10 ** exp
|
|
step = base * 10
|
|
for m in (1, 2, 2.5, 5, 10):
|
|
if base * m >= raw:
|
|
step = base * m
|
|
break
|
|
first = math.ceil(lo / step) * step
|
|
ticks, v = [], first
|
|
while v <= hi + step * 0.01:
|
|
ticks.append(v)
|
|
v += step
|
|
return ticks
|
|
|
|
|
|
# ── Render thread ─────────────────────────────────────────────────────────────
|
|
|
|
class PlotRenderThread(QThread):
|
|
"""Renders a time-series plot to QImage in a background thread.
|
|
|
|
snapshot format: dict[rid -> (list[timestamp_float], list[value_float])]
|
|
"""
|
|
|
|
frame_ready = pyqtSignal(object) # QImage
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._queue = Queue(maxsize=1)
|
|
self._running = True
|
|
|
|
def submit(self, snapshot: dict, y_label: str, width: int, height: int):
|
|
try:
|
|
self._queue.get_nowait() # discard stale pending frame
|
|
except Empty:
|
|
pass
|
|
self._queue.put((snapshot, y_label, width, height))
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
try:
|
|
self._queue.put_nowait(None)
|
|
except Exception:
|
|
pass
|
|
self.wait(2000)
|
|
|
|
# ── Thread body ───────────────────────────────────────────────────────────
|
|
|
|
def run(self):
|
|
while self._running:
|
|
try:
|
|
item = self._queue.get(timeout=0.2)
|
|
except Empty:
|
|
continue
|
|
if item is None:
|
|
break
|
|
snapshot, y_label, w, h = item
|
|
if w > 0 and h > 0:
|
|
self.frame_ready.emit(self._render(snapshot, y_label, w, h))
|
|
|
|
# ── Drawing (QPainter on QImage is thread-safe) ───────────────────────────
|
|
|
|
_ML, _MR, _MT, _MB = 56, 8, 10, 36 # margins: left/right/top/bottom px
|
|
|
|
def _render(self, snapshot: dict, y_label: str, w: int, h: int) -> QImage:
|
|
img = QImage(w, h, QImage.Format_RGB32)
|
|
p = QPainter(img)
|
|
p.setRenderHint(QPainter.Antialiasing)
|
|
|
|
ML, MR, MT, MB = self._ML, self._MR, self._MT, self._MB
|
|
pw = w - ML - MR
|
|
ph = h - MT - MB
|
|
|
|
p.fillRect(0, 0, w, h, _C_BG)
|
|
if pw > 0 and ph > 0:
|
|
p.fillRect(ML, MT, pw, ph, _C_PLOT_BG)
|
|
|
|
if pw <= 0 or ph <= 0:
|
|
p.end()
|
|
return img
|
|
|
|
all_ts = [t for ts, _ in snapshot.values() for t in ts]
|
|
all_v = [v for _, vs in snapshot.values() for v in vs]
|
|
|
|
if not all_ts:
|
|
self._draw_empty(p, ML, MT, pw, ph)
|
|
p.end()
|
|
return img
|
|
|
|
ts_min = min(all_ts); ts_max = max(all_ts)
|
|
v_min = min(all_v); v_max = max(all_v)
|
|
|
|
if ts_max - ts_min < 1:
|
|
ts_min = ts_max - 1.0
|
|
|
|
v_ticks = _nice_ticks(v_min, v_max)
|
|
v_min, v_max = v_ticks[0], v_ticks[-1]
|
|
if v_max - v_min < 1e-9:
|
|
v_max = v_min + 1.0
|
|
|
|
ts_span = ts_max - ts_min
|
|
v_span = v_max - v_min
|
|
|
|
def px(ts, v):
|
|
x = ML + (ts - ts_min) / ts_span * pw
|
|
y = MT + ph - (v - v_min) / v_span * ph
|
|
return x, y
|
|
|
|
self._draw_grid(p, ML, MT, pw, ph, v_ticks, v_min, v_max)
|
|
self._draw_curves(p, snapshot, px)
|
|
self._draw_legend(p, snapshot, ML, MT)
|
|
self._draw_y_labels(p, ML, MT, ph, v_ticks, v_min, v_max, y_label)
|
|
self._draw_x_labels(p, ML, MT, pw, ph, ts_min, ts_max)
|
|
|
|
p.end()
|
|
return img
|
|
|
|
def _draw_empty(self, p, ml, mt, pw, ph):
|
|
f = QFont(); f.setPointSize(8)
|
|
p.setFont(f); p.setPen(QPen(_C_LABEL))
|
|
p.drawText(QRectF(ml, mt, pw, ph), Qt.AlignCenter,
|
|
"Keine Daten\nQuelle + Satellit auswählen")
|
|
|
|
def _draw_grid(self, p, ml, mt, pw, ph, v_ticks, v_min, v_max):
|
|
p.setPen(QPen(_C_GRID, 1, Qt.DotLine))
|
|
v_span = v_max - v_min
|
|
for v in v_ticks:
|
|
if v_min <= v <= v_max:
|
|
y = mt + ph - (v - v_min) / v_span * ph
|
|
p.drawLine(QPointF(ml, y), QPointF(ml + pw, y))
|
|
for i in range(1, 5):
|
|
x = ml + i * pw / 5
|
|
p.drawLine(QPointF(x, mt), QPointF(x, mt + ph))
|
|
p.setPen(QPen(_C_AXIS, 1))
|
|
p.setBrush(Qt.NoBrush)
|
|
p.drawRect(ml, mt, pw, ph)
|
|
|
|
def _draw_curves(self, p, snapshot, px):
|
|
for i, (rid, (ts_list, v_list)) in enumerate(snapshot.items()):
|
|
if len(ts_list) < 1:
|
|
continue
|
|
pen = QPen(_QCOLORS[i % len(_QCOLORS)], 1.5)
|
|
p.setPen(pen)
|
|
pts = [QPointF(*px(t, v)) for t, v in zip(ts_list, v_list)]
|
|
for j in range(1, len(pts)):
|
|
p.drawLine(pts[j - 1], pts[j])
|
|
|
|
def _draw_legend(self, p, snapshot, ml, mt):
|
|
f = QFont(); f.setPointSize(7); p.setFont(f)
|
|
lx, ly = ml + 5, mt + 5
|
|
for i, rid in enumerate(snapshot):
|
|
color = _QCOLORS[i % len(_QCOLORS)]
|
|
p.setPen(QPen(color, 2))
|
|
p.drawLine(QPointF(lx, ly + 5), QPointF(lx + 14, ly + 5))
|
|
p.setPen(QPen(_C_TEXT))
|
|
p.drawText(QRectF(lx + 16, ly, 80, 12), Qt.AlignLeft, str(rid))
|
|
lx += max(20 + len(rid) * 6, 70)
|
|
|
|
def _draw_y_labels(self, p, ml, mt, ph, v_ticks, v_min, v_max, y_label):
|
|
f = QFont(); f.setPointSize(7); p.setFont(f)
|
|
p.setPen(QPen(_C_LABEL))
|
|
v_span = v_max - v_min
|
|
for v in v_ticks:
|
|
if v_min <= v <= v_max:
|
|
y = mt + ph - (v - v_min) / v_span * ph
|
|
lbl = f"{v:.1f}" if abs(v) < 1000 else f"{v:.0f}"
|
|
p.drawText(QRectF(2, y - 8, ml - 4, 16),
|
|
Qt.AlignRight | Qt.AlignVCenter, lbl)
|
|
# rotated axis label
|
|
p.save()
|
|
p.translate(10, mt + ph / 2)
|
|
p.rotate(-90)
|
|
f2 = QFont(); f2.setPointSize(8); p.setFont(f2)
|
|
p.drawText(QRectF(-40, -10, 80, 20), Qt.AlignCenter, y_label)
|
|
p.restore()
|
|
|
|
def _draw_x_labels(self, p, ml, mt, pw, ph, ts_min, ts_max):
|
|
f = QFont(); f.setPointSize(7); p.setFont(f)
|
|
p.setPen(QPen(_C_LABEL))
|
|
for i in range(6):
|
|
ts = ts_min + (ts_max - ts_min) * i / 5
|
|
x = ml + i * pw / 5
|
|
lbl = datetime.fromtimestamp(ts).strftime('%H:%M:%S')
|
|
p.drawText(QRectF(x - 28, mt + ph + 4, 56, 14),
|
|
Qt.AlignCenter, lbl)
|
|
|
|
|
|
# ── Canvas widget (displays QImage from render thread) ────────────────────────
|
|
|
|
class _Canvas(QWidget):
|
|
resized = pyqtSignal(int, int)
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._image = None
|
|
self.setAttribute(Qt.WA_OpaquePaintEvent)
|
|
self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding)
|
|
self.setMinimumHeight(150)
|
|
|
|
def set_image(self, img: QImage):
|
|
self._image = img
|
|
self.update()
|
|
|
|
def paintEvent(self, _event):
|
|
p = QPainter(self)
|
|
if self._image and not self._image.isNull():
|
|
p.drawImage(0, 0, self._image)
|
|
else:
|
|
p.fillRect(self.rect(), _C_BG)
|
|
p.end()
|
|
|
|
def resizeEvent(self, event):
|
|
super().resizeEvent(event)
|
|
self.resized.emit(self.width(), self.height())
|
|
|
|
|
|
# ── Single configurable plot ──────────────────────────────────────────────────
|
|
|
|
class SinglePlotWidget(QWidget):
|
|
def __init__(self, model: ReceiverManager, plot_index: int = 0):
|
|
super().__init__()
|
|
self._model = model
|
|
self._plot_index = plot_index
|
|
self._data: dict = {} # rid → [deque(ts), deque(val)]
|
|
self._known_sats: dict = {} # rid → set[sat_label] (selected only)
|
|
self._restore_sat = '' # persisted satellite, applied on first combo fill
|
|
|
|
self._build_ui()
|
|
self._load_settings()
|
|
|
|
self._thread = PlotRenderThread()
|
|
self._thread.frame_ready.connect(self._canvas.set_image)
|
|
self._thread.start()
|
|
|
|
app = QApplication.instance()
|
|
if app:
|
|
app.aboutToQuit.connect(self._thread.stop)
|
|
|
|
model.receiver_added.connect(self._on_receiver_added)
|
|
model.receiver_removed.connect(self._on_receiver_removed)
|
|
model.satellite_update.connect(self._on_sat_update)
|
|
model.pdop_update.connect(self._on_pdop_update)
|
|
|
|
# ── UI ────────────────────────────────────────────────────────────────────
|
|
|
|
def _build_ui(self):
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(2, 2, 2, 2)
|
|
layout.setSpacing(2)
|
|
|
|
bar = QHBoxLayout()
|
|
bar.setSpacing(5)
|
|
|
|
bar.addWidget(QLabel("Y-Achse:"))
|
|
self._y_cb = QComboBox(); self._y_cb.addItems(_Y_OPTIONS)
|
|
self._y_cb.setMaximumWidth(85)
|
|
self._y_cb.currentTextChanged.connect(self._on_y_changed)
|
|
bar.addWidget(self._y_cb)
|
|
|
|
bar.addWidget(QLabel("Satellit:"))
|
|
self._val_cb = QComboBox()
|
|
self._val_cb.setMinimumWidth(90); self._val_cb.setMaximumWidth(120)
|
|
self._val_cb.currentTextChanged.connect(self._on_sat_changed)
|
|
bar.addWidget(self._val_cb)
|
|
|
|
bar.addWidget(QLabel("Quelle:"))
|
|
self._src_list = QListWidget()
|
|
self._src_list.setFlow(QListWidget.LeftToRight)
|
|
self._src_list.setWrapping(True)
|
|
self._src_list.setResizeMode(QListWidget.Adjust)
|
|
self._src_list.setMaximumHeight(46); self._src_list.setMaximumWidth(200)
|
|
self._src_list.setSizePolicy(QSizePolicy.Fixed, QSizePolicy.Fixed)
|
|
self._src_list.itemChanged.connect(self._on_source_changed)
|
|
bar.addWidget(self._src_list)
|
|
|
|
bar.addWidget(QLabel("Punkte:"))
|
|
self._hist_spin = QSpinBox()
|
|
self._hist_spin.setRange(10, 10000); self._hist_spin.setValue(300)
|
|
self._hist_spin.setMaximumWidth(70)
|
|
self._hist_spin.valueChanged.connect(self._on_history_changed)
|
|
bar.addWidget(self._hist_spin)
|
|
bar.addStretch()
|
|
layout.addLayout(bar)
|
|
|
|
self._canvas = _Canvas()
|
|
self._canvas.resized.connect(lambda w, h: self._submit_render())
|
|
layout.addWidget(self._canvas)
|
|
|
|
# ── Source helpers ────────────────────────────────────────────────────────
|
|
|
|
def _selected_rids(self) -> list:
|
|
result = []
|
|
for i in range(self._src_list.count()):
|
|
it = self._src_list.item(i)
|
|
if it.checkState() == Qt.Checked:
|
|
result.append(it.data(Qt.UserRole))
|
|
return result
|
|
|
|
# ── Persistence ───────────────────────────────────────────────────────────
|
|
|
|
def _load_settings(self):
|
|
all_s = _load_plot_settings()
|
|
s = all_s[self._plot_index] if self._plot_index < len(all_s) else {}
|
|
if 'y_axis' in s:
|
|
self._y_cb.blockSignals(True)
|
|
self._y_cb.setCurrentText(s['y_axis'])
|
|
self._y_cb.blockSignals(False)
|
|
self._val_cb.setEnabled(s['y_axis'] != 'pDOP')
|
|
if 'history' in s:
|
|
self._hist_spin.blockSignals(True)
|
|
self._hist_spin.setValue(s['history'])
|
|
self._hist_spin.blockSignals(False)
|
|
if 'satellite' in s:
|
|
self._restore_sat = s['satellite']
|
|
|
|
def _save_settings(self):
|
|
all_s = _load_plot_settings()
|
|
while len(all_s) <= self._plot_index:
|
|
all_s.append({})
|
|
all_s[self._plot_index] = {
|
|
'y_axis': self._y_cb.currentText(),
|
|
'satellite': self._val_cb.currentText(),
|
|
'history': self._hist_spin.value(),
|
|
}
|
|
_save_plot_settings(all_s)
|
|
|
|
# ── Source helpers ────────────────────────────────────────────────────────
|
|
|
|
def _rebuild_val_combo(self):
|
|
selected = set(self._selected_rids())
|
|
current = self._val_cb.currentText() or self._restore_sat
|
|
self._val_cb.blockSignals(True)
|
|
self._val_cb.clear()
|
|
sats = set()
|
|
for rid in selected:
|
|
sats.update(self._known_sats.get(rid, set()))
|
|
for s in sorted(sats):
|
|
self._val_cb.addItem(s)
|
|
if current in sats:
|
|
self._val_cb.setCurrentText(current)
|
|
self._restore_sat = ''
|
|
self._val_cb.blockSignals(False)
|
|
|
|
# ── Render submission ─────────────────────────────────────────────────────
|
|
|
|
def _submit_render(self):
|
|
selected = set(self._selected_rids())
|
|
snapshot = {}
|
|
for rid in selected:
|
|
ts_dq, v_dq = self._data.get(rid, ([], []))
|
|
if ts_dq:
|
|
snapshot[rid] = (list(ts_dq), list(v_dq))
|
|
w, h = self._canvas.width(), self._canvas.height()
|
|
if w > 0 and h > 0:
|
|
self._thread.submit(snapshot, self._y_cb.currentText(), w, h)
|
|
|
|
def _push_point(self, rid: str, ts: float, val: float):
|
|
if rid not in self._data:
|
|
n = self._hist_spin.value()
|
|
self._data[rid] = [deque(maxlen=n), deque(maxlen=n)]
|
|
self._data[rid][0].append(ts)
|
|
self._data[rid][1].append(val)
|
|
self._submit_render()
|
|
|
|
# ── Config change handlers ────────────────────────────────────────────────
|
|
|
|
def _on_sat_changed(self, _text: str):
|
|
self._save_settings()
|
|
self._submit_render()
|
|
|
|
def _on_y_changed(self, _text: str):
|
|
is_pdop = (_text == 'pDOP')
|
|
self._val_cb.setEnabled(not is_pdop)
|
|
for rid in list(self._data):
|
|
n = self._hist_spin.value()
|
|
self._data[rid] = [deque(maxlen=n), deque(maxlen=n)]
|
|
self._save_settings()
|
|
self._submit_render()
|
|
|
|
def _on_source_changed(self):
|
|
self._rebuild_val_combo()
|
|
self._submit_render()
|
|
|
|
def _on_history_changed(self, n: int):
|
|
for rid in list(self._data):
|
|
old_ts, old_v = self._data[rid]
|
|
self._data[rid] = [deque(old_ts, maxlen=n), deque(old_v, maxlen=n)]
|
|
self._save_settings()
|
|
self._submit_render()
|
|
|
|
# ── Model signals ─────────────────────────────────────────────────────────
|
|
|
|
def _on_receiver_added(self, rid: str):
|
|
item = QListWidgetItem(f" {rid} ")
|
|
item.setFlags(item.flags() | Qt.ItemIsUserCheckable)
|
|
item.setCheckState(Qt.Checked)
|
|
item.setData(Qt.UserRole, rid)
|
|
self._src_list.addItem(item)
|
|
self._known_sats[rid] = set()
|
|
n = self._hist_spin.value()
|
|
self._data[rid] = [deque(maxlen=n), deque(maxlen=n)]
|
|
|
|
def _on_receiver_removed(self, rid: str):
|
|
for i in range(self._src_list.count()):
|
|
if self._src_list.item(i).data(Qt.UserRole) == rid:
|
|
self._src_list.takeItem(i)
|
|
break
|
|
self._known_sats.pop(rid, None)
|
|
self._data.pop(rid, None)
|
|
self._rebuild_val_combo()
|
|
self._submit_render()
|
|
|
|
def _on_sat_update(self, rid: str, sats: list):
|
|
selected = set(self._selected_rids())
|
|
if rid not in selected:
|
|
return
|
|
y_axis = self._y_cb.currentText()
|
|
if y_axis == 'pDOP':
|
|
return
|
|
|
|
# Update satellite combo
|
|
new_labels = {s.sat_label() for s in sats}
|
|
if not new_labels.issubset(self._known_sats.get(rid, set())):
|
|
self._known_sats[rid] = self._known_sats.get(rid, set()) | new_labels
|
|
self._rebuild_val_combo()
|
|
|
|
target = self._val_cb.currentText()
|
|
if not target:
|
|
return
|
|
|
|
ts_now = time.time()
|
|
for sat in sats:
|
|
if sat.sat_label() != target:
|
|
continue
|
|
val = {'C/N₀': sat.cno, 'Azimuth': sat.azimuth,
|
|
'Elevation': sat.elevation}.get(y_axis)
|
|
if val is not None:
|
|
self._push_point(rid, ts_now, val)
|
|
break
|
|
|
|
def _on_pdop_update(self, rid: str, pdop: float, ts: float):
|
|
if self._y_cb.currentText() != 'pDOP':
|
|
return
|
|
if rid not in set(self._selected_rids()):
|
|
return
|
|
self._push_point(rid, ts, pdop)
|