From 6b37db6c5132dd2ed77213a96a2aa6f7c1a0f885 Mon Sep 17 00:00:00 2001 From: Jens Ahrensfeld Date: Sat, 16 May 2026 17:10:24 +0200 Subject: [PATCH] Initial commit --- .gitignore | 1 + gnss_sky_plot.py | 635 +++++++++++++++++++++++++++++++++++++++++++++++ readme.md | 0 3 files changed, 636 insertions(+) create mode 100644 .gitignore create mode 100644 gnss_sky_plot.py create mode 100644 readme.md diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9f11b75 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.idea/ diff --git a/gnss_sky_plot.py b/gnss_sky_plot.py new file mode 100644 index 0000000..8f7912d --- /dev/null +++ b/gnss_sky_plot.py @@ -0,0 +1,635 @@ +#!/usr/bin/env python3 +""" +GNSS Sky Plot Viewer +Unterstützt UBX über serielle Schnittstelle UND TCP/IP (Ser2net). +""" + +import sys +import struct +import socket +import serial +import serial.tools.list_ports +import math +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 +from PyQt5.QtGui import ( + QPainter, QPen, QBrush, QColor, QFont, + QRadialGradient, +) + + +# --------------------------------------------------------------------------- +# UBX Protokoll-Konstanten +# --------------------------------------------------------------------------- +UBX_SYNC1 = 0xB5 +UBX_SYNC2 = 0x62 +UBX_CLASS_NAV = 0x01 +UBX_ID_NAV_SAT = 0x35 +UBX_ID_NAV_SVINFO = 0x30 + +GNSS_NAMES = { + 0: "GPS", + 1: "SBAS", + 2: "Galileo", + 3: "BeiDou", + 4: "IMES", + 5: "QZSS", + 6: "GLONASS", +} + +GNSS_COLORS = { + 0: QColor("#1E90FF"), + 1: QColor("#FFA500"), + 2: QColor("#00C853"), + 3: QColor("#E53935"), + 4: QColor("#9C27B0"), + 5: QColor("#00BCD4"), + 6: QColor("#FF4081"), +} + + +# --------------------------------------------------------------------------- +# UBX Hilfsfunktionen +# --------------------------------------------------------------------------- +def ubx_checksum(payload: bytes) -> tuple[int, int]: + ck_a = ck_b = 0 + for b in payload: + ck_a = (ck_a + b) & 0xFF + ck_b = (ck_b + ck_a) & 0xFF + return ck_a, ck_b + + +def build_ubx(cls: int, msg_id: int, payload: bytes = b"") -> bytes: + header = bytes([UBX_SYNC1, UBX_SYNC2, cls, msg_id]) + length = struct.pack(" bytes: + return build_ubx(cls, msg_id, b"") + + +def parse_ubx_frames(data: bytes) -> list[dict]: + frames, i = [], 0 + while i < len(data) - 7: + if data[i] == UBX_SYNC1 and data[i + 1] == UBX_SYNC2: + if i + 6 > len(data): + break + cls = data[i + 2] + msg_id = data[i + 3] + length = struct.unpack_from(" len(data): + break + payload = data[i + 6: i + 6 + length] + raw_chk = data[i + 2: i + 6 + length] + ck_a, ck_b = ubx_checksum(raw_chk) + if ck_a == data[end - 2] and ck_b == data[end - 1]: + frames.append({"cls": cls, "id": msg_id, "payload": payload}) + i = end + else: + i += 1 + return frames + + +def parse_nav_sat(payload: bytes) -> list[dict]: + if len(payload) < 8: + return [] + num_svs = payload[5] + sats = [] + for n in range(num_svs): + off = 8 + n * 12 + if off + 12 > len(payload): + break + gnss_id, sv_id, cno, elev, azim, _, flags = struct.unpack_from( + " list[dict]: + if len(payload) < 8: + return [] + num_ch = payload[4] + sats = [] + for n in range(num_ch): + off = 8 + n * 12 + if off + 12 > len(payload): + break + _, svid, flags, _, cno, elev, azim, _ = struct.unpack_from( + " list[dict]: + for f in frames: + if f["cls"] == UBX_CLASS_NAV and f["id"] == UBX_ID_NAV_SAT: + return parse_nav_sat(f["payload"]) + if f["cls"] == UBX_CLASS_NAV and f["id"] == UBX_ID_NAV_SVINFO: + return parse_nav_svinfo(f["payload"]) + return [] + + +# --------------------------------------------------------------------------- +# Abstrakte Transport-Basisklasse +# --------------------------------------------------------------------------- +class GnssTransport: + """Gemeinsame Lese-/Schreibschnittstelle für seriell und TCP.""" + + 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) # blockierendes Lesen mit 1 s Timeout + + 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() + + +# --------------------------------------------------------------------------- +# Einheitlicher GNSS-Lese-Thread +# --------------------------------------------------------------------------- +class GnssReader(QThread): + satellites_updated = pyqtSignal(list) + error_occurred = pyqtSignal(str) + connected = pyqtSignal(str) + + def __init__(self, transport_factory): + """ + transport_factory: callable ohne Argumente, das ein GnssTransport + zurückgibt oder eine Exception wirft. + """ + 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) + if chunk: + buf += chunk + if len(buf) > 8192: + buf = buf[-1024:] + frames = parse_ubx_frames(buf) + 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) + + +# --------------------------------------------------------------------------- +# 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) + + +# --------------------------------------------------------------------------- +# Verbindungs-Panel (Tabs: Seriell | TCP/IP) +# --------------------------------------------------------------------------- +FIELD_STYLE = """ + QLineEdit, QComboBox, QSpinBox { + background-color: #0d1b2a; + color: #cccccc; + border: 1px solid #2a4a6a; + padding: 3px 6px; + border-radius: 3px; + } +""" + +class ConnectionPanel(QGroupBox): + """Gibt über connect_requested ein transport_factory-Callable zurück.""" + + connect_requested = pyqtSignal(object, str) # (factory, label) + 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._build_ui() + + def _build_ui(self): + root = QVBoxLayout(self) + + # Tabs + 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_form = QFormLayout(tcp_widget) + tcp_form.setContentsMargins(6, 6, 6, 6) + + self._tcp_host = QLineEdit("192.168.1.100") + 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(2947) # gpsd-Standard; Ser2net oft 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) + + 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() + + 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) + + def _on_connect_clicked(self): + if self._connect_btn.text().startswith("■"): + self.disconnect_requested.emit() + return + + if self._tabs.currentIndex() == 0: + # Seriell + 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: + # TCP/IP + host = self._tcp_host.text().strip() + port = self._tcp_port.value() + timeout = self._tcp_timeout.value() + if not host: + return + 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/readme.md b/readme.md new file mode 100644 index 0000000..e69de29