jens/skyplot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
jens
2026-05-16 17:10:24 +02:00
commit 6b37db6c51
3 changed files with 636 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -0,0 +1 @@
.idea/
gnss_sky_plot.py
+635
View File
@@ -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("<H", len(payload))
raw = bytes([cls, msg_id]) + length + payload
ck_a, ck_b = ubx_checksum(raw)
return header + length + payload + bytes([ck_a, ck_b])
def build_poll(cls: int, msg_id: int) -> 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("<H", data, i + 4)[0]
end = i + 6 + length + 2
if end > 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(
"<BBBbhiI", payload, off
)
sats.append({
"gnssId": gnss_id, "svId": sv_id, "cno": cno,
"elev": elev, "azim": azim, "used": bool(flags & 0x08),
})
return sats
def parse_nav_svinfo(payload: bytes) -> 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(
"<BBBBBbhI", payload, off
)
sats.append({
"gnssId": 0, "svId": svid, "cno": cno,
"elev": elev, "azim": azim, "used": bool(flags & 0x01),
})
return sats
def extract_satellites(frames: list[dict]) -> 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_())
readme.md
View File