From dc05be751bf163ba679f20093e36b1ecce13c9bb Mon Sep 17 00:00:00 2001 From: Jens Ahrensfeld Date: Sun, 17 May 2026 17:00:48 +0200 Subject: [PATCH] refactored heavily --- connection_panel.py | 211 ++++++++++++++ defines.py | 22 ++ gnss_sky_plot.py | 638 ------------------------------------------- gnss_thread.py | 51 ++++ lru_store.py | 33 +++ sat_table.py | 70 +++++ sky_plot.py | 107 ++++++++ sky_plot_widget.py | 127 +++++++++ transport.py | 13 + transport_network.py | 30 ++ transport_serial.py | 21 ++ 11 files changed, 685 insertions(+), 638 deletions(-) create mode 100644 connection_panel.py create mode 100644 defines.py delete mode 100644 gnss_sky_plot.py create mode 100644 gnss_thread.py create mode 100644 lru_store.py create mode 100644 sat_table.py create mode 100644 sky_plot.py create mode 100644 sky_plot_widget.py create mode 100644 transport.py create mode 100644 transport_network.py create mode 100644 transport_serial.py diff --git a/connection_panel.py b/connection_panel.py new file mode 100644 index 0000000..df71797 --- /dev/null +++ b/connection_panel.py @@ -0,0 +1,211 @@ +# --------------------------------------------------------------------------- +# Verbindungs-Panel +# --------------------------------------------------------------------------- +import serial +from serial.tools.list_ports import comports +from PyQt5.QtCore import pyqtSignal +from PyQt5.QtWidgets import QGroupBox, QVBoxLayout, QTabWidget, QWidget, QFormLayout, QComboBox, QPushButton, \ + QHBoxLayout, QLineEdit, QSpinBox + +from lru_store import LruStore +from transport_network import TcpTransport +from transport_serial import SerialTransport + +# --------------------------------------------------------------------------- +# Stil-Konstante +# --------------------------------------------------------------------------- +FIELD_STYLE = """ + QLineEdit, QComboBox, QSpinBox { + background-color: #0d1b2a; + color: #cccccc; + border: 1px solid #2a4a6a; + padding: 3px 6px; + border-radius: 3px; + } + QComboBox QAbstractItemView { + background-color: #0d1b2a; + color: #cccccc; + selection-background-color: #1a3a5c; + } +""" + +class ConnectionPanel(QGroupBox): + connect_requested = pyqtSignal(object, str) + disconnect_requested = pyqtSignal() + + def __init__(self, parent=None): + super().__init__("Verbindung", parent) + self.setStyleSheet(""" + QGroupBox { color: #7ab7d4; border: 1px solid #2a4a6a; + border-radius: 4px; margin-top: 8px; padding: 6px; } + QGroupBox::title { subcontrol-origin: margin; left: 8px; } + """ + FIELD_STYLE) + self._lru = LruStore(max_size=5) + self._build_ui() + + def _build_ui(self): + root = QVBoxLayout(self) + + self._tabs = QTabWidget() + self._tabs.setStyleSheet(""" + QTabBar::tab { background: #1a3a5c; color: #cccccc; + padding: 4px 12px; border-radius: 3px; } + QTabBar::tab:selected { background: #2a5a8c; color: #ffffff; } + """) + + # ── Tab: Seriell ───────────────────────────────────────────────────── + serial_widget = QWidget() + serial_form = QFormLayout(serial_widget) + serial_form.setContentsMargins(6, 6, 6, 6) + + self._port_combo = QComboBox() + self._port_combo.setMinimumWidth(200) + refresh_btn = QPushButton("⟳") + refresh_btn.setFixedWidth(28) + refresh_btn.clicked.connect(self._refresh_ports) + port_row = QHBoxLayout() + port_row.addWidget(self._port_combo) + port_row.addWidget(refresh_btn) + serial_form.addRow("Port:", port_row) + + self._baud_combo = QComboBox() + for b in ("9600", "19200", "38400", "57600", "115200", "230400", "460800"): + self._baud_combo.addItem(b) + self._baud_combo.setCurrentText("9600") + serial_form.addRow("Baudrate:", self._baud_combo) + + self._tabs.addTab(serial_widget, "🔌 Seriell") + + # ── Tab: TCP/IP ─────────────────────────────────────────────────────── + tcp_widget = QWidget() + tcp_layout = QVBoxLayout(tcp_widget) + tcp_layout.setContentsMargins(6, 6, 6, 6) + tcp_layout.setSpacing(6) + + # LRU-Gruppe + lru_group = QGroupBox("Letzte Verbindungen") + lru_group.setStyleSheet(""" + QGroupBox { color: #7ab7d4; border: 1px solid #2a4a6a; + border-radius: 3px; margin-top: 6px; padding: 4px; } + QGroupBox::title { subcontrol-origin: margin; left: 6px; } + """ + FIELD_STYLE) + lru_layout = QHBoxLayout(lru_group) + lru_layout.setContentsMargins(4, 10, 4, 4) + + self._lru_combo = QComboBox() + self._lru_combo.setMinimumWidth(220) + self._lru_combo.setPlaceholderText("– Frühere Verbindung wählen –") + self._lru_combo.currentIndexChanged.connect(self._on_lru_selected) + lru_layout.addWidget(self._lru_combo, stretch=1) + + clear_btn = QPushButton("🗑") + clear_btn.setFixedWidth(28) + clear_btn.setToolTip("Liste leeren") + clear_btn.clicked.connect(self._clear_lru) + lru_layout.addWidget(clear_btn) + + tcp_layout.addWidget(lru_group) + + # Manuelle Eingabe + tcp_form = QFormLayout() + tcp_form.setSpacing(6) + + self._tcp_host = QLineEdit() + self._tcp_host.setPlaceholderText("IP-Adresse oder Hostname") + tcp_form.addRow("Host:", self._tcp_host) + + self._tcp_port = QSpinBox() + self._tcp_port.setRange(1, 65535) + self._tcp_port.setValue(4001) + tcp_form.addRow("Port:", self._tcp_port) + + self._tcp_timeout = QSpinBox() + self._tcp_timeout.setRange(1, 30) + self._tcp_timeout.setValue(5) + self._tcp_timeout.setSuffix(" s") + tcp_form.addRow("Timeout:", self._tcp_timeout) + + tcp_layout.addLayout(tcp_form) + tcp_layout.addStretch() + + self._tabs.addTab(tcp_widget, "🌐 TCP/IP") + + root.addWidget(self._tabs) + + # Verbinden-Knopf + btn_row = QHBoxLayout() + self._connect_btn = QPushButton("▶ Verbinden") + self._connect_btn.clicked.connect(self._on_connect_clicked) + btn_row.addStretch() + btn_row.addWidget(self._connect_btn) + root.addLayout(btn_row) + + self._refresh_ports() + self._refresh_lru_combo() + + # ── LRU ────────────────────────────────────────────────────────────────── + + def _refresh_lru_combo(self): + self._lru_combo.blockSignals(True) + self._lru_combo.clear() + for entry in self._lru.entries(): + self._lru_combo.addItem(entry) + self._lru_combo.setCurrentIndex(-1) + self._lru_combo.blockSignals(False) + + def _on_lru_selected(self, index: int): + if index < 0: + return + entry = self._lru_combo.currentText() + if not entry: + return + try: + host, port = self._lru.parse(entry) + self._tcp_host.setText(host) + self._tcp_port.setValue(port) + except (ValueError, IndexError): + pass + + def _clear_lru(self): + self._lru.clear() + self._refresh_lru_combo() + + # ── Seriell ─────────────────────────────────────────────────────────────── + + def _refresh_ports(self): + self._port_combo.clear() + for p in comports(): + self._port_combo.addItem(f"{p.device} – {p.description}", p.device) + if self._port_combo.count() == 0: + self._port_combo.addItem("Kein Port gefunden", None) + + # ── Verbinden ───────────────────────────────────────────────────────────── + + def _on_connect_clicked(self): + if self._connect_btn.text().startswith("■"): + self.disconnect_requested.emit() + return + + if self._tabs.currentIndex() == 0: + port = self._port_combo.currentData() + baud = int(self._baud_combo.currentText()) + if not port: + return + factory = lambda p=port, b=baud: SerialTransport(p, b) + label = f"{port} @ {baud} Baud" + else: + host = self._tcp_host.text().strip() + port = self._tcp_port.value() + timeout = self._tcp_timeout.value() + if not host: + return + self._lru.add(host, port) + self._refresh_lru_combo() + factory = lambda h=host, p=port, t=timeout: TcpTransport(h, p, t) + label = f"{host}:{port}" + + self.connect_requested.emit(factory, label) + + def set_connected(self, connected: bool): + self._connect_btn.setText("■ Trennen" if connected else "▶ Verbinden") + self._tabs.setEnabled(not connected) diff --git a/defines.py b/defines.py new file mode 100644 index 0000000..9c913c2 --- /dev/null +++ b/defines.py @@ -0,0 +1,22 @@ +from PyQt5.QtGui import QColor + +GNSS_NAMES = { + 0: "GPS", + 1: "SBAS", + 2: "Galileo", + 3: "BeiDou", + 4: "IMES", + 5: "QZSS", + 6: "GLONASS", + 7: "NavIC" +} +GNSS_COLORS = { + 0: QColor("#1E90FF"), + 1: QColor("#FFA500"), + 2: QColor("#00C853"), + 3: QColor("#E53935"), + 4: QColor("#9C27B0"), + 5: QColor("#00BCD4"), + 6: QColor("#FF4081"), + 7: QColor("#808080") +} diff --git a/gnss_sky_plot.py b/gnss_sky_plot.py deleted file mode 100644 index 68864ae..0000000 --- a/gnss_sky_plot.py +++ /dev/null @@ -1,638 +0,0 @@ -#!/usr/bin/env python3 -""" -GNSS Sky Plot Viewer -Unterstützt UBX über serielle Schnittstelle UND TCP/IP (Ser2net). -TCP/IP-Verbindungen werden in einer persistenten LRU-Liste (max. 5) gespeichert. -""" - -import sys -import socket -import math -import serial -import serial.tools.list_ports -from PyQt5.QtWidgets import ( - QApplication, QMainWindow, QWidget, QVBoxLayout, - QHBoxLayout, QLabel, QComboBox, QPushButton, - QStatusBar, QTableWidget, QTableWidgetItem, - QHeaderView, QSplitter, QTabWidget, QLineEdit, - QSpinBox, QFormLayout, QGroupBox, -) -from PyQt5.QtCore import Qt, QThread, pyqtSignal, QSettings -from PyQt5.QtGui import ( - QPainter, QPen, QBrush, QColor, QFont, - QRadialGradient, -) -from ubx import UbxPacketizer, build_poll, extract_satellites, UBX_CLASS_NAV, UBX_ID_NAV_SAT, UBX_ID_NAV_SVINFO -ubx_packetizer = UbxPacketizer() - -# --------------------------------------------------------------------------- -# UBX Protokoll-Konstanten -# --------------------------------------------------------------------------- -GNSS_NAMES = { - 0: "GPS", - 1: "SBAS", - 2: "Galileo", - 3: "BeiDou", - 4: "IMES", - 5: "QZSS", - 6: "GLONASS", - 7: "NavIC" -} - -GNSS_COLORS = { - 0: QColor("#1E90FF"), - 1: QColor("#FFA500"), - 2: QColor("#00C853"), - 3: QColor("#E53935"), - 4: QColor("#9C27B0"), - 5: QColor("#00BCD4"), - 6: QColor("#FF4081"), - 7: QColor("#808080") -} - -# --------------------------------------------------------------------------- -# Transport-Abstraktionen -# --------------------------------------------------------------------------- -class GnssTransport: - def read(self, n: int) -> bytes: - raise NotImplementedError - - def write(self, data: bytes) -> None: - raise NotImplementedError - - def close(self) -> None: - raise NotImplementedError - - -class SerialTransport(GnssTransport): - def __init__(self, port: str, baudrate: int): - self._ser = serial.Serial(port, baudrate, timeout=1) - - def read(self, n: int) -> bytes: - return self._ser.read(n) - - def write(self, data: bytes) -> None: - self._ser.write(data) - - def close(self) -> None: - if self._ser.is_open: - self._ser.close() - - -class TcpTransport(GnssTransport): - def __init__(self, host: str, port: int, timeout: float = 5.0): - self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - self._sock.settimeout(timeout) - self._sock.connect((host, port)) - self._sock.settimeout(1.0) - - def read(self, n: int) -> bytes: - try: - return self._sock.recv(n) - except socket.timeout: - return b"" - - def write(self, data: bytes) -> None: - self._sock.sendall(data) - - def close(self) -> None: - try: - self._sock.shutdown(socket.SHUT_RDWR) - except OSError: - pass - self._sock.close() - - -# --------------------------------------------------------------------------- -# GNSS Lese-Thread -# --------------------------------------------------------------------------- -class GnssReader(QThread): - satellites_updated = pyqtSignal(list) - error_occurred = pyqtSignal(str) - connected = pyqtSignal(str) - - def __init__(self, transport_factory): - super().__init__() - self._factory = transport_factory - self._running = False - self._transport = None - - def run(self): - self._running = True - buf = b"" - try: - self._transport = self._factory() - self.connected.emit("Verbunden") - poll_msg = build_poll(UBX_CLASS_NAV, UBX_ID_NAV_SAT) - self._transport.write(poll_msg) - - while self._running: - chunk = self._transport.read(512) - frames = ubx_packetizer.on_recv(chunk) - sats = extract_satellites(frames) - if sats: - self.satellites_updated.emit(sats) - else: - # Kein Datum – erneut pollen - self._transport.write(poll_msg) - - except (serial.SerialException, OSError, ConnectionRefusedError, - socket.timeout, socket.gaierror) as exc: - self.error_occurred.emit(str(exc)) - finally: - if self._transport: - self._transport.close() - - def stop(self): - self._running = False - self.wait(3000) - - -# --------------------------------------------------------------------------- -# LRU-Speicher (persistent via QSettings) -# --------------------------------------------------------------------------- -class LruStore: - """Speichert bis zu max_size TCP-Verbindungen persistent (host:port).""" - - def __init__(self, max_size: int = 5): - self._max = max_size - self._cfg = QSettings("gnss_skyplot", "connections") - - def entries(self) -> list[str]: - val = self._cfg.value("tcp_lru", []) - if isinstance(val, str): - val = [val] - return val or [] - - def add(self, host: str, port: int) -> None: - entry = f"{host}:{port}" - entries = [e for e in self.entries() if e != entry] - entries.insert(0, entry) - self._cfg.setValue("tcp_lru", entries[: self._max]) - self._cfg.sync() - - def clear(self) -> None: - self._cfg.remove("tcp_lru") - self._cfg.sync() - - def parse(self, entry: str) -> tuple[str, int]: - host, port_str = entry.rsplit(":", 1) - return host.strip(), int(port_str.strip()) - - -# --------------------------------------------------------------------------- -# Sky-Plot Widget -# --------------------------------------------------------------------------- -class SkyPlotWidget(QWidget): - def __init__(self, parent=None): - super().__init__(parent) - self._satellites = [] - self.setMinimumSize(400, 400) - self.setStyleSheet("background-color: #1a1a2e;") - - def update_satellites(self, sats: list): - self._satellites = sats - self.update() - - def paintEvent(self, event): - painter = QPainter(self) - painter.setRenderHint(QPainter.Antialiasing) - w, h = self.width(), self.height() - cx, cy = w / 2, h / 2 - radius = min(w, h) / 2 - 30 - self._draw_background(painter, cx, cy, radius) - self._draw_grid(painter, cx, cy, radius) - self._draw_satellites(painter, cx, cy, radius) - self._draw_legend(painter) - - def _draw_background(self, painter, cx, cy, radius): - grad = QRadialGradient(cx, cy, radius) - grad.setColorAt(0.0, QColor("#0d1b2a")) - grad.setColorAt(1.0, QColor("#1a1a2e")) - painter.setBrush(QBrush(grad)) - painter.setPen(Qt.NoPen) - painter.drawEllipse( - int(cx - radius), int(cy - radius), - int(radius * 2), int(radius * 2), - ) - - def _draw_grid(self, painter, cx, cy, radius): - pen = QPen(QColor("#2a4a6a"), 1, Qt.DashLine) - painter.setPen(pen) - painter.setFont(QFont("Monospace", 8)) - - for elev in (0, 30, 60): - r = radius * (1 - elev / 90) - painter.drawEllipse(int(cx - r), int(cy - r), int(r * 2), int(r * 2)) - painter.setPen(QColor("#4a7a9b")) - painter.drawText(int(cx + 4), int(cy - r), f"{elev}°") - painter.setPen(pen) - - for azim in range(0, 360, 45): - rad = math.radians(azim) - painter.drawLine( - int(cx + math.sin(rad) * radius * 0.05), - int(cy - math.cos(rad) * radius * 0.05), - int(cx + math.sin(rad) * radius), - int(cy - math.cos(rad) * radius), - ) - - painter.setPen(QColor("#7ab7d4")) - painter.setFont(QFont("Monospace", 10, QFont.Bold)) - for label, azim in (("N", 0), ("O", 90), ("S", 180), ("W", 270)): - rad = math.radians(azim) - painter.drawText( - int(cx + math.sin(rad) * (radius + 15) - 6), - int(cy - math.cos(rad) * (radius + 15) + 5), - label, - ) - - @staticmethod - def _az_el_to_xy(azim, elev, cx, cy, radius): - r = radius * (1 - elev / 90) - rad = math.radians(azim) - return cx + math.sin(rad) * r, cy - math.cos(rad) * r - - def _draw_satellites(self, painter, cx, cy, radius): - for sat in self._satellites: - if sat["elev"] < 0: - continue - color = GNSS_COLORS.get(sat["gnssId"], QColor("#ffffff")) - x, y = self._az_el_to_xy(sat["azim"], sat["elev"], cx, cy, radius) - cno = max(0, min(50, sat["cno"])) - dot_r = 4 + int(cno / 50 * 8) - - glow = QColor(color) - glow.setAlpha(60) - painter.setBrush(QBrush(glow)) - painter.setPen(Qt.NoPen) - painter.drawEllipse( - int(x - dot_r * 1.8), int(y - dot_r * 1.8), - int(dot_r * 3.6), int(dot_r * 3.6), - ) - - if sat["used"]: - painter.setBrush(QBrush(color)) - painter.setPen(QPen(QColor("#ffffff"), 1)) - else: - faded = QColor(color) - faded.setAlpha(120) - painter.setBrush(QBrush(faded)) - painter.setPen(QPen(color, 1, Qt.DashLine)) - - painter.drawEllipse( - int(x - dot_r), int(y - dot_r), dot_r * 2, dot_r * 2 - ) - painter.setPen(QColor("#ffffff")) - painter.setFont(QFont("Monospace", 7)) - painter.drawText(int(x + dot_r + 2), int(y + 4), str(sat["svId"])) - - def _draw_legend(self, painter): - painter.setFont(QFont("Monospace", 8)) - x, y = 10, 20 - for gnss_id, name in GNSS_NAMES.items(): - color = GNSS_COLORS[gnss_id] - painter.setBrush(QBrush(color)) - painter.setPen(Qt.NoPen) - painter.drawEllipse(x, y - 8, 10, 10) - painter.setPen(QColor("#cccccc")) - painter.drawText(x + 14, y, name) - y += 16 - - -# --------------------------------------------------------------------------- -# Satelliten-Tabelle -# --------------------------------------------------------------------------- -class SatelliteTable(QTableWidget): - HEADERS = ["GNSS", "SV", "Elevation", "Azimut", "C/N₀ (dB)", "Genutzt"] - - def __init__(self, parent=None): - super().__init__(0, len(self.HEADERS), parent) - self.setHorizontalHeaderLabels(self.HEADERS) - self.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) - self.setStyleSheet(""" - QTableWidget { - background-color: #0d1b2a; color: #cccccc; - gridline-color: #2a4a6a; font-family: monospace; font-size: 11px; - } - QHeaderView::section { - background-color: #1a3a5c; color: #7ab7d4; padding: 4px; - } - """) - self.setEditTriggers(QTableWidget.NoEditTriggers) - self.setSelectionBehavior(QTableWidget.SelectRows) - - def update_satellites(self, sats: list): - self.setRowCount(0) - for sat in sorted(sats, key=lambda s: (s["gnssId"], s["svId"])): - row = self.rowCount() - self.insertRow(row) - color = GNSS_COLORS.get(sat["gnssId"], QColor("#ffffff")) - vals = [ - GNSS_NAMES.get(sat["gnssId"], f"?({sat['gnssId']})"), - str(sat["svId"]), - f"{sat['elev']}°", - f"{sat['azim']}°", - str(sat["cno"]), - "✓" if sat["used"] else "–", - ] - for col, val in enumerate(vals): - item = QTableWidgetItem(val) - item.setForeground(color if col == 0 else QColor("#cccccc")) - if sat["used"]: - item.setBackground(QColor("#0d2a1a")) - self.setItem(row, col, item) - - -# --------------------------------------------------------------------------- -# Stil-Konstante -# --------------------------------------------------------------------------- -FIELD_STYLE = """ - QLineEdit, QComboBox, QSpinBox { - background-color: #0d1b2a; - color: #cccccc; - border: 1px solid #2a4a6a; - padding: 3px 6px; - border-radius: 3px; - } - QComboBox QAbstractItemView { - background-color: #0d1b2a; - color: #cccccc; - selection-background-color: #1a3a5c; - } -""" - - -# --------------------------------------------------------------------------- -# Verbindungs-Panel -# --------------------------------------------------------------------------- -class ConnectionPanel(QGroupBox): - connect_requested = pyqtSignal(object, str) - disconnect_requested = pyqtSignal() - - def __init__(self, parent=None): - super().__init__("Verbindung", parent) - self.setStyleSheet(""" - QGroupBox { color: #7ab7d4; border: 1px solid #2a4a6a; - border-radius: 4px; margin-top: 8px; padding: 6px; } - QGroupBox::title { subcontrol-origin: margin; left: 8px; } - """ + FIELD_STYLE) - self._lru = LruStore(max_size=5) - self._build_ui() - - def _build_ui(self): - root = QVBoxLayout(self) - - self._tabs = QTabWidget() - self._tabs.setStyleSheet(""" - QTabBar::tab { background: #1a3a5c; color: #cccccc; - padding: 4px 12px; border-radius: 3px; } - QTabBar::tab:selected { background: #2a5a8c; color: #ffffff; } - """) - - # ── Tab: Seriell ───────────────────────────────────────────────────── - serial_widget = QWidget() - serial_form = QFormLayout(serial_widget) - serial_form.setContentsMargins(6, 6, 6, 6) - - self._port_combo = QComboBox() - self._port_combo.setMinimumWidth(200) - refresh_btn = QPushButton("⟳") - refresh_btn.setFixedWidth(28) - refresh_btn.clicked.connect(self._refresh_ports) - port_row = QHBoxLayout() - port_row.addWidget(self._port_combo) - port_row.addWidget(refresh_btn) - serial_form.addRow("Port:", port_row) - - self._baud_combo = QComboBox() - for b in ("9600", "19200", "38400", "57600", "115200", "230400", "460800"): - self._baud_combo.addItem(b) - self._baud_combo.setCurrentText("9600") - serial_form.addRow("Baudrate:", self._baud_combo) - - self._tabs.addTab(serial_widget, "🔌 Seriell") - - # ── Tab: TCP/IP ─────────────────────────────────────────────────────── - tcp_widget = QWidget() - tcp_layout = QVBoxLayout(tcp_widget) - tcp_layout.setContentsMargins(6, 6, 6, 6) - tcp_layout.setSpacing(6) - - # LRU-Gruppe - lru_group = QGroupBox("Letzte Verbindungen") - lru_group.setStyleSheet(""" - QGroupBox { color: #7ab7d4; border: 1px solid #2a4a6a; - border-radius: 3px; margin-top: 6px; padding: 4px; } - QGroupBox::title { subcontrol-origin: margin; left: 6px; } - """ + FIELD_STYLE) - lru_layout = QHBoxLayout(lru_group) - lru_layout.setContentsMargins(4, 10, 4, 4) - - self._lru_combo = QComboBox() - self._lru_combo.setMinimumWidth(220) - self._lru_combo.setPlaceholderText("– Frühere Verbindung wählen –") - self._lru_combo.currentIndexChanged.connect(self._on_lru_selected) - lru_layout.addWidget(self._lru_combo, stretch=1) - - clear_btn = QPushButton("🗑") - clear_btn.setFixedWidth(28) - clear_btn.setToolTip("Liste leeren") - clear_btn.clicked.connect(self._clear_lru) - lru_layout.addWidget(clear_btn) - - tcp_layout.addWidget(lru_group) - - # Manuelle Eingabe - tcp_form = QFormLayout() - tcp_form.setSpacing(6) - - self._tcp_host = QLineEdit() - self._tcp_host.setPlaceholderText("IP-Adresse oder Hostname") - tcp_form.addRow("Host:", self._tcp_host) - - self._tcp_port = QSpinBox() - self._tcp_port.setRange(1, 65535) - self._tcp_port.setValue(4001) - tcp_form.addRow("Port:", self._tcp_port) - - self._tcp_timeout = QSpinBox() - self._tcp_timeout.setRange(1, 30) - self._tcp_timeout.setValue(5) - self._tcp_timeout.setSuffix(" s") - tcp_form.addRow("Timeout:", self._tcp_timeout) - - tcp_layout.addLayout(tcp_form) - tcp_layout.addStretch() - - self._tabs.addTab(tcp_widget, "🌐 TCP/IP") - - root.addWidget(self._tabs) - - # Verbinden-Knopf - btn_row = QHBoxLayout() - self._connect_btn = QPushButton("▶ Verbinden") - self._connect_btn.clicked.connect(self._on_connect_clicked) - btn_row.addStretch() - btn_row.addWidget(self._connect_btn) - root.addLayout(btn_row) - - self._refresh_ports() - self._refresh_lru_combo() - - # ── LRU ────────────────────────────────────────────────────────────────── - - def _refresh_lru_combo(self): - self._lru_combo.blockSignals(True) - self._lru_combo.clear() - for entry in self._lru.entries(): - self._lru_combo.addItem(entry) - self._lru_combo.setCurrentIndex(-1) - self._lru_combo.blockSignals(False) - - def _on_lru_selected(self, index: int): - if index < 0: - return - entry = self._lru_combo.currentText() - if not entry: - return - try: - host, port = self._lru.parse(entry) - self._tcp_host.setText(host) - self._tcp_port.setValue(port) - except (ValueError, IndexError): - pass - - def _clear_lru(self): - self._lru.clear() - self._refresh_lru_combo() - - # ── Seriell ─────────────────────────────────────────────────────────────── - - def _refresh_ports(self): - self._port_combo.clear() - for p in serial.tools.list_ports.comports(): - self._port_combo.addItem(f"{p.device} – {p.description}", p.device) - if self._port_combo.count() == 0: - self._port_combo.addItem("Kein Port gefunden", None) - - # ── Verbinden ───────────────────────────────────────────────────────────── - - def _on_connect_clicked(self): - if self._connect_btn.text().startswith("■"): - self.disconnect_requested.emit() - return - - if self._tabs.currentIndex() == 0: - port = self._port_combo.currentData() - baud = int(self._baud_combo.currentText()) - if not port: - return - factory = lambda p=port, b=baud: SerialTransport(p, b) - label = f"{port} @ {baud} Baud" - else: - host = self._tcp_host.text().strip() - port = self._tcp_port.value() - timeout = self._tcp_timeout.value() - if not host: - return - self._lru.add(host, port) - self._refresh_lru_combo() - factory = lambda h=host, p=port, t=timeout: TcpTransport(h, p, t) - label = f"{host}:{port}" - - self.connect_requested.emit(factory, label) - - def set_connected(self, connected: bool): - self._connect_btn.setText("■ Trennen" if connected else "▶ Verbinden") - self._tabs.setEnabled(not connected) - - -# --------------------------------------------------------------------------- -# Hauptfenster -# --------------------------------------------------------------------------- -class MainWindow(QMainWindow): - def __init__(self): - super().__init__() - self.setWindowTitle("GNSS Sky Plot – UBX | Seriell + TCP/IP") - self.resize(980, 680) - self.setStyleSheet("background-color: #12192b; color: #cccccc;") - self._reader = None - self._build_ui() - - def _build_ui(self): - central = QWidget() - self.setCentralWidget(central) - root = QVBoxLayout(central) - root.setContentsMargins(8, 8, 8, 8) - - # Obere Leiste - top = QHBoxLayout() - - self._conn_panel = ConnectionPanel() - self._conn_panel.connect_requested.connect(self._connect) - self._conn_panel.disconnect_requested.connect(self._disconnect) - top.addWidget(self._conn_panel) - - self._sat_label = QLabel("Satelliten: –") - self._sat_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) - self._sat_label.setStyleSheet("font-size: 13px; color: #7ab7d4;") - top.addStretch() - top.addWidget(self._sat_label) - - root.addLayout(top) - - # Splitter: Sky Plot | Tabelle - splitter = QSplitter(Qt.Horizontal) - self._sky_plot = SkyPlotWidget() - self._table = SatelliteTable() - splitter.addWidget(self._sky_plot) - splitter.addWidget(self._table) - splitter.setSizes([540, 380]) - root.addWidget(splitter) - - self._status = QStatusBar() - self.setStatusBar(self._status) - self._status.showMessage("Nicht verbunden.") - - def _connect(self, factory, label: str): - self._reader = GnssReader(factory) - self._reader.satellites_updated.connect(self._on_satellites) - self._reader.error_occurred.connect(self._on_error) - self._reader.connected.connect( - lambda msg: self._status.showMessage(f"{msg}: {label}") - ) - self._reader.start() - self._conn_panel.set_connected(True) - self._status.showMessage(f"Verbinde mit {label} …") - - def _disconnect(self): - if self._reader: - self._reader.stop() - self._reader = None - self._conn_panel.set_connected(False) - self._status.showMessage("Verbindung getrennt.") - - def _on_satellites(self, sats: list): - visible = [s for s in sats if s["elev"] >= 0] - used = sum(1 for s in sats if s["used"]) - self._sat_label.setText(f"Satelliten: {len(visible)} sichtbar, {used} genutzt") - self._sky_plot.update_satellites(visible) - self._table.update_satellites(visible) - - def _on_error(self, msg: str): - self._status.showMessage(f"Fehler: {msg}") - self._disconnect() - - def closeEvent(self, event): - self._disconnect() - super().closeEvent(event) - - -# --------------------------------------------------------------------------- -if __name__ == "__main__": - app = QApplication(sys.argv) - app.setStyle("Fusion") - win = MainWindow() - win.show() - sys.exit(app.exec_()) diff --git a/gnss_thread.py b/gnss_thread.py new file mode 100644 index 0000000..4f561b5 --- /dev/null +++ b/gnss_thread.py @@ -0,0 +1,51 @@ +# --------------------------------------------------------------------------- +# GNSS Lese-Thread +# --------------------------------------------------------------------------- +import socket + +import serial +from PyQt5.QtCore import QThread, pyqtSignal + +from ubx import UbxPacketizer, build_poll, UBX_CLASS_NAV, UBX_ID_NAV_SAT, extract_satellites + +class GnssReader(QThread): + satellites_updated = pyqtSignal(list) + error_occurred = pyqtSignal(str) + connected = pyqtSignal(str) + + def __init__(self, transport_factory): + super().__init__() + self._factory = transport_factory + self._running = False + self._transport = None + self.packetizer = UbxPacketizer() + + def run(self): + self._running = True + buf = b"" + try: + self._transport = self._factory() + self.connected.emit("Verbunden") + poll_msg = build_poll(UBX_CLASS_NAV, UBX_ID_NAV_SAT) + self._transport.write(poll_msg) + + while self._running: + chunk = self._transport.read(512) + frames = self.packetizer.on_recv(chunk) + sats = extract_satellites(frames) + if sats: + self.satellites_updated.emit(sats) + else: + # Kein Datum – erneut pollen + self._transport.write(poll_msg) + + except (serial.SerialException, OSError, ConnectionRefusedError, + socket.timeout, socket.gaierror) as exc: + self.error_occurred.emit(str(exc)) + finally: + if self._transport: + self._transport.close() + + def stop(self): + self._running = False + self.wait(3000) diff --git a/lru_store.py b/lru_store.py new file mode 100644 index 0000000..9fefe8a --- /dev/null +++ b/lru_store.py @@ -0,0 +1,33 @@ +# --------------------------------------------------------------------------- +# LRU-Speicher (persistent via QSettings) +# --------------------------------------------------------------------------- +from PyQt5.QtCore import QSettings + + +class LruStore: + """Speichert bis zu max_size TCP-Verbindungen persistent (host:port).""" + + def __init__(self, max_size: int = 5): + self._max = max_size + self._cfg = QSettings("gnss_skyplot", "connections") + + def entries(self) -> list[str]: + val = self._cfg.value("tcp_lru", []) + if isinstance(val, str): + val = [val] + return val or [] + + def add(self, host: str, port: int) -> None: + entry = f"{host}:{port}" + entries = [e for e in self.entries() if e != entry] + entries.insert(0, entry) + self._cfg.setValue("tcp_lru", entries[: self._max]) + self._cfg.sync() + + def clear(self) -> None: + self._cfg.remove("tcp_lru") + self._cfg.sync() + + def parse(self, entry: str) -> tuple[str, int]: + host, port_str = entry.rsplit(":", 1) + return host.strip(), int(port_str.strip()) diff --git a/sat_table.py b/sat_table.py new file mode 100644 index 0000000..2cbc761 --- /dev/null +++ b/sat_table.py @@ -0,0 +1,70 @@ +# --------------------------------------------------------------------------- +# Satelliten-Tabelle +# --------------------------------------------------------------------------- +from PyQt5.QtGui import QColor +from PyQt5.QtWidgets import QTableWidget, QHeaderView, QTableWidgetItem + +# --------------------------------------------------------------------------- +# UBX Protokoll-Konstanten +# --------------------------------------------------------------------------- +GNSS_NAMES = { + 0: "GPS", + 1: "SBAS", + 2: "Galileo", + 3: "BeiDou", + 4: "IMES", + 5: "QZSS", + 6: "GLONASS", + 7: "NavIC" +} + +GNSS_COLORS = { + 0: QColor("#1E90FF"), + 1: QColor("#FFA500"), + 2: QColor("#00C853"), + 3: QColor("#E53935"), + 4: QColor("#9C27B0"), + 5: QColor("#00BCD4"), + 6: QColor("#FF4081"), + 7: QColor("#808080") +} + +class SatelliteTable(QTableWidget): + HEADERS = ["GNSS", "SV", "Elevation", "Azimut", "C/N₀ (dB)", "Genutzt"] + + def __init__(self, parent=None): + super().__init__(0, len(self.HEADERS), parent) + self.setHorizontalHeaderLabels(self.HEADERS) + self.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) + self.setStyleSheet(""" + QTableWidget { + background-color: #0d1b2a; color: #cccccc; + gridline-color: #2a4a6a; font-family: monospace; font-size: 11px; + } + QHeaderView::section { + background-color: #1a3a5c; color: #7ab7d4; padding: 4px; + } + """) + self.setEditTriggers(QTableWidget.NoEditTriggers) + self.setSelectionBehavior(QTableWidget.SelectRows) + + def update_satellites(self, sats: list): + self.setRowCount(0) + for sat in sorted(sats, key=lambda s: (s["gnssId"], s["svId"])): + row = self.rowCount() + self.insertRow(row) + color = GNSS_COLORS.get(sat["gnssId"], QColor("#ffffff")) + vals = [ + GNSS_NAMES.get(sat["gnssId"], f"?({sat['gnssId']})"), + str(sat["svId"]), + f"{sat['elev']}°", + f"{sat['azim']}°", + str(sat["cno"]), + "✓" if sat["used"] else "–", + ] + for col, val in enumerate(vals): + item = QTableWidgetItem(val) + item.setForeground(color if col == 0 else QColor("#cccccc")) + if sat["used"]: + item.setBackground(QColor("#0d2a1a")) + self.setItem(row, col, item) diff --git a/sky_plot.py b/sky_plot.py new file mode 100644 index 0000000..53b7404 --- /dev/null +++ b/sky_plot.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +""" +GNSS Sky Plot Viewer +Unterstützt UBX über serielle Schnittstelle UND TCP/IP (Ser2net). +TCP/IP-Verbindungen werden in einer persistenten LRU-Liste (max. 5) gespeichert. +""" + +import sys +from PyQt5.QtWidgets import ( + QApplication, QMainWindow, QWidget, QVBoxLayout, + QHBoxLayout, QLabel, QStatusBar, QSplitter, ) +from PyQt5.QtCore import Qt + +from connection_panel import ConnectionPanel +from gnss_thread import GnssReader +from sat_table import SatelliteTable +from sky_plot_widget import SkyPlotWidget + + +# --------------------------------------------------------------------------- +# Hauptfenster +# --------------------------------------------------------------------------- +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle("GNSS Sky Plot – UBX | Seriell + TCP/IP") + self.resize(980, 680) + self.setStyleSheet("background-color: #12192b; color: #cccccc;") + self._reader = None + self._build_ui() + + def _build_ui(self): + central = QWidget() + self.setCentralWidget(central) + root = QVBoxLayout(central) + root.setContentsMargins(8, 8, 8, 8) + + # Obere Leiste + top = QHBoxLayout() + + self._conn_panel = ConnectionPanel() + self._conn_panel.connect_requested.connect(self._connect) + self._conn_panel.disconnect_requested.connect(self._disconnect) + top.addWidget(self._conn_panel) + + self._sat_label = QLabel("Satelliten: –") + self._sat_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) + self._sat_label.setStyleSheet("font-size: 13px; color: #7ab7d4;") + top.addStretch() + top.addWidget(self._sat_label) + + root.addLayout(top) + + # Splitter: Sky Plot | Tabelle + splitter = QSplitter(Qt.Horizontal) + self._sky_plot = SkyPlotWidget() + self._table = SatelliteTable() + splitter.addWidget(self._sky_plot) + splitter.addWidget(self._table) + splitter.setSizes([540, 380]) + root.addWidget(splitter) + + self._status = QStatusBar() + self.setStatusBar(self._status) + self._status.showMessage("Nicht verbunden.") + + def _connect(self, factory, label: str): + self._reader = GnssReader(factory) + self._reader.satellites_updated.connect(self._on_satellites) + self._reader.error_occurred.connect(self._on_error) + self._reader.connected.connect( + lambda msg: self._status.showMessage(f"{msg}: {label}") + ) + self._reader.start() + self._conn_panel.set_connected(True) + self._status.showMessage(f"Verbinde mit {label} …") + + def _disconnect(self): + if self._reader: + self._reader.stop() + self._reader = None + self._conn_panel.set_connected(False) + self._status.showMessage("Verbindung getrennt.") + + def _on_satellites(self, sats: list): + visible = [s for s in sats if s["elev"] >= 0] + used = sum(1 for s in sats if s["used"]) + self._sat_label.setText(f"Satelliten: {len(visible)} sichtbar, {used} genutzt") + self._sky_plot.update_satellites(visible) + self._table.update_satellites(visible) + + def _on_error(self, msg: str): + self._status.showMessage(f"Fehler: {msg}") + self._disconnect() + + def closeEvent(self, event): + self._disconnect() + super().closeEvent(event) + + +# --------------------------------------------------------------------------- +if __name__ == "__main__": + app = QApplication(sys.argv) + app.setStyle("Fusion") + win = MainWindow() + win.show() + sys.exit(app.exec_()) diff --git a/sky_plot_widget.py b/sky_plot_widget.py new file mode 100644 index 0000000..a53bbd7 --- /dev/null +++ b/sky_plot_widget.py @@ -0,0 +1,127 @@ +# --------------------------------------------------------------------------- +# Sky-Plot Widget +# --------------------------------------------------------------------------- +import math + +from PyQt5.QtCore import Qt +from PyQt5.QtGui import QPainter, QRadialGradient, QColor, QBrush, QPen, QFont +from PyQt5.QtWidgets import QWidget + +from defines import GNSS_NAMES, GNSS_COLORS + + +class SkyPlotWidget(QWidget): + def __init__(self, parent=None): + super().__init__(parent) + self._satellites = [] + self.setMinimumSize(400, 400) + self.setStyleSheet("background-color: #1a1a2e;") + + def update_satellites(self, sats: list): + self._satellites = sats + self.update() + + def paintEvent(self, event): + painter = QPainter(self) + painter.setRenderHint(QPainter.Antialiasing) + w, h = self.width(), self.height() + cx, cy = w / 2, h / 2 + radius = min(w, h) / 2 - 30 + self._draw_background(painter, cx, cy, radius) + self._draw_grid(painter, cx, cy, radius) + self._draw_satellites(painter, cx, cy, radius) + self._draw_legend(painter) + + def _draw_background(self, painter, cx, cy, radius): + grad = QRadialGradient(cx, cy, radius) + grad.setColorAt(0.0, QColor("#0d1b2a")) + grad.setColorAt(1.0, QColor("#1a1a2e")) + painter.setBrush(QBrush(grad)) + painter.setPen(Qt.NoPen) + painter.drawEllipse( + int(cx - radius), int(cy - radius), + int(radius * 2), int(radius * 2), + ) + + def _draw_grid(self, painter, cx, cy, radius): + pen = QPen(QColor("#2a4a6a"), 1, Qt.DashLine) + painter.setPen(pen) + painter.setFont(QFont("Monospace", 8)) + + for elev in (0, 30, 60): + r = radius * (1 - elev / 90) + painter.drawEllipse(int(cx - r), int(cy - r), int(r * 2), int(r * 2)) + painter.setPen(QColor("#4a7a9b")) + painter.drawText(int(cx + 4), int(cy - r), f"{elev}°") + painter.setPen(pen) + + for azim in range(0, 360, 45): + rad = math.radians(azim) + painter.drawLine( + int(cx + math.sin(rad) * radius * 0.05), + int(cy - math.cos(rad) * radius * 0.05), + int(cx + math.sin(rad) * radius), + int(cy - math.cos(rad) * radius), + ) + + painter.setPen(QColor("#7ab7d4")) + painter.setFont(QFont("Monospace", 10, QFont.Bold)) + for label, azim in (("N", 0), ("O", 90), ("S", 180), ("W", 270)): + rad = math.radians(azim) + painter.drawText( + int(cx + math.sin(rad) * (radius + 15) - 6), + int(cy - math.cos(rad) * (radius + 15) + 5), + label, + ) + + @staticmethod + def _az_el_to_xy(azim, elev, cx, cy, radius): + r = radius * (1 - elev / 90) + rad = math.radians(azim) + return cx + math.sin(rad) * r, cy - math.cos(rad) * r + + def _draw_satellites(self, painter, cx, cy, radius): + for sat in self._satellites: + if sat["elev"] < 0: + continue + color = GNSS_COLORS.get(sat["gnssId"], QColor("#ffffff")) + x, y = self._az_el_to_xy(sat["azim"], sat["elev"], cx, cy, radius) + cno = max(0, min(50, sat["cno"])) + dot_r = 4 + int(cno / 50 * 8) + + glow = QColor(color) + glow.setAlpha(60) + painter.setBrush(QBrush(glow)) + painter.setPen(Qt.NoPen) + painter.drawEllipse( + int(x - dot_r * 1.8), int(y - dot_r * 1.8), + int(dot_r * 3.6), int(dot_r * 3.6), + ) + + if sat["used"]: + painter.setBrush(QBrush(color)) + painter.setPen(QPen(QColor("#ffffff"), 1)) + else: + faded = QColor(color) + faded.setAlpha(120) + painter.setBrush(QBrush(faded)) + painter.setPen(QPen(color, 1, Qt.DashLine)) + + painter.drawEllipse( + int(x - dot_r), int(y - dot_r), dot_r * 2, dot_r * 2 + ) + painter.setPen(QColor("#ffffff")) + painter.setFont(QFont("Monospace", 7)) + painter.drawText(int(x + dot_r + 2), int(y + 4), str(sat["svId"])) + + def _draw_legend(self, painter): + painter.setFont(QFont("Monospace", 8)) + x, y = 10, 20 + for gnss_id, name in GNSS_NAMES.items(): + color = GNSS_COLORS[gnss_id] + painter.setBrush(QBrush(color)) + painter.setPen(Qt.NoPen) + painter.drawEllipse(x, y - 8, 10, 10) + painter.setPen(QColor("#cccccc")) + painter.drawText(x + 14, y, name) + y += 16 diff --git a/transport.py b/transport.py new file mode 100644 index 0000000..5eb2fcd --- /dev/null +++ b/transport.py @@ -0,0 +1,13 @@ +# --------------------------------------------------------------------------- +# Transport-Abstraktionen +# --------------------------------------------------------------------------- + +class GnssTransport: + def read(self, n: int) -> bytes: + raise NotImplementedError + + def write(self, data: bytes) -> None: + raise NotImplementedError + + def close(self) -> None: + raise NotImplementedError diff --git a/transport_network.py b/transport_network.py new file mode 100644 index 0000000..2005306 --- /dev/null +++ b/transport_network.py @@ -0,0 +1,30 @@ +# --------------------------------------------------------------------------- +# Transport over Network +# --------------------------------------------------------------------------- +import socket + +from transport import GnssTransport + + +class TcpTransport(GnssTransport): + def __init__(self, host: str, port: int, timeout: float = 5.0): + self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._sock.settimeout(timeout) + self._sock.connect((host, port)) + self._sock.settimeout(1.0) + + def read(self, n: int) -> bytes: + try: + return self._sock.recv(n) + except socket.timeout: + return b"" + + def write(self, data: bytes) -> None: + self._sock.sendall(data) + + def close(self) -> None: + try: + self._sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self._sock.close() diff --git a/transport_serial.py b/transport_serial.py new file mode 100644 index 0000000..472c173 --- /dev/null +++ b/transport_serial.py @@ -0,0 +1,21 @@ +# --------------------------------------------------------------------------- +# Transport over Serial +# --------------------------------------------------------------------------- +import serial + +from transport import GnssTransport + + +class SerialTransport(GnssTransport): + def __init__(self, port: str, baudrate: int): + self._ser = serial.Serial(port, baudrate, timeout=1) + + def read(self, n: int) -> bytes: + return self._ser.read(n) + + def write(self, data: bytes) -> None: + self._ser.write(data) + + def close(self) -> None: + if self._ser.is_open: + self._ser.close()