import sys, os, math from queue import Queue, Empty _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, 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 GNSS_NAMES, GNSS_COLORS, GNSS_SHORT # ── Static colors (created once in main thread, read-only in render thread) ── _C_BG = QColor('#1a1a2e') _C_RING = QColor('#3a3a5e') _C_SPOKE = QColor('#2e2e4e') _C_LABEL = QColor('#aaaacc') _C_SAT_LBL = QColor('#ddddee') _C_ZENITH = QColor('#444466') # Pre-build GNSS QColor objects _GNSS_QCOLORS: dict = {gid: QColor(hex_) for gid, hex_ in GNSS_COLORS.items()} def _polar_xy(cx, cy, max_r, elev, azim): r = (90.0 - elev) / 90.0 * max_r rad = math.radians(azim) return cx + r * math.sin(rad), cy - r * math.cos(rad) # ── Render thread ───────────────────────────────────────────────────────────── class SkyRenderThread(QThread): """Renders the sky plot to a QImage in a background thread.""" frame_ready = pyqtSignal(object) # QImage — use object to avoid type-reg issues def __init__(self, parent=None): super().__init__(parent) # maxsize=1: submit() always replaces pending work with latest data self._queue: Queue = Queue(maxsize=1) self._running = True def submit(self, sats: list, trails: dict, width: int, height: int): # Discard any pending (not yet rendered) frame so we always render latest try: self._queue.get_nowait() except Empty: pass self._queue.put((list(sats), trails, width, height)) def stop(self): self._running = False try: self._queue.put_nowait(None) # unblock get() 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 sats, trails, w, h = item if w > 0 and h > 0: img = self._render(sats, trails, w, h) self.frame_ready.emit(img) # ── Drawing (runs in render thread — QPainter on QImage is thread-safe) ── def _render(self, sats: list, trails: dict, w: int, h: int) -> QImage: img = QImage(w, h, QImage.Format_RGB32) p = QPainter(img) p.setRenderHint(QPainter.Antialiasing) cx, cy = w / 2.0, h / 2.0 max_r = min(w, h) / 2.0 - 32 self._draw_bg(p, w, h) self._draw_grid(p, cx, cy, max_r) if trails: self._draw_trails(p, cx, cy, max_r, trails) self._draw_satellites(p, cx, cy, max_r, sats) self._draw_legend(p, sats) p.end() return img def _draw_bg(self, p: QPainter, w: int, h: int): p.fillRect(0, 0, w, h, _C_BG) def _draw_grid(self, p: QPainter, cx, cy, max_r): # Elevation rings 0°, 30°, 60° p.setPen(QPen(_C_RING, 1, Qt.DashLine)) p.setBrush(Qt.NoBrush) for elev in (0, 30, 60): r = (90 - elev) / 90.0 * max_r p.drawEllipse(QPointF(cx, cy), r, r) # Azimuth spokes every 45° p.setPen(QPen(_C_SPOKE, 1)) for az in range(0, 360, 45): rad = math.radians(az) p.drawLine(QPointF(cx, cy), QPointF(cx + max_r * math.sin(rad), cy - max_r * math.cos(rad))) # Cardinal labels f = QFont(); f.setPointSize(9); f.setBold(True) p.setFont(f) p.setPen(QPen(_C_LABEL)) off = max_r + 16 for lbl, az in (('N', 0), ('E', 90), ('S', 180), ('W', 270)): rad = math.radians(az) x = cx + off * math.sin(rad) y = cy - off * math.cos(rad) p.drawText(QRectF(x - 10, y - 10, 20, 20), Qt.AlignCenter, lbl) # Elevation ring labels f2 = QFont(); f2.setPointSize(7) p.setFont(f2) p.setPen(QPen(_C_LABEL)) for elev, lbl in ((0, '0°'), (30, '30°'), (60, '60°')): r = (90 - elev) / 90.0 * max_r p.drawText(QRectF(cx + 3, cy - r - 10, 28, 12), Qt.AlignLeft, lbl) # Zenith dot p.setBrush(QBrush(_C_ZENITH)) p.setPen(Qt.NoPen) p.drawEllipse(QPointF(cx, cy), 3, 3) def _draw_trails(self, p: QPainter, cx, cy, max_r, trails: dict): for (_, gnss_id, _), points in trails.items(): n = len(points) if n < 2: continue base_color = _GNSS_QCOLORS.get(gnss_id, _C_LABEL) r, g, b = base_color.red(), base_color.green(), base_color.blue() # Subsample to at most 120 points so rendering stays fast step = max(1, n // 120) pts = points[::step] m = len(pts) for i in range(1, m): # Alpha fades from 15 (oldest) to 200 (newest) alpha = int(15 + 185 * i / m) p.setPen(QPen(QColor(r, g, b, alpha), 1.5)) x1, y1 = _polar_xy(cx, cy, max_r, pts[i - 1][0], pts[i - 1][1]) x2, y2 = _polar_xy(cx, cy, max_r, pts[i][0], pts[i][1]) p.drawLine(QPointF(x1, y1), QPointF(x2, y2)) def _draw_satellites(self, p: QPainter, cx, cy, max_r, sats): f = QFont(); f.setPointSize(7) p.setFont(f) for _rid, sat in sats: color = _GNSS_QCOLORS.get(sat.gnss_id, _C_LABEL) x, y = _polar_xy(cx, cy, max_r, sat.elevation, sat.azimuth) size = max(4.0, min(14.0, 4.0 + (sat.cno - 20.0) * 0.33)) if sat.used_in_fix: p.setBrush(QBrush(color)) p.setPen(QPen(color.darker(160), 1.0)) else: p.setBrush(Qt.NoBrush) p.setPen(QPen(color, 1.5, Qt.DashLine)) p.drawEllipse(QPointF(x, y), size, size) short = GNSS_SHORT.get(sat.gnss_id, '?') p.setPen(QPen(_C_SAT_LBL)) p.drawText(QRectF(x + size + 1, y - 6, 28, 12), Qt.AlignLeft, f"{short}{sat.sv_id}") def _draw_legend(self, p: QPainter, sats): used = {sat.gnss_id for _, sat in sats} if not used: return f = QFont(); f.setPointSize(8) p.setFont(f) lx, ly = 8, 8 for gid in sorted(used): color = _GNSS_QCOLORS.get(gid) if color is None: continue p.setBrush(QBrush(color)) p.setPen(Qt.NoPen) p.drawEllipse(QPointF(lx + 5, ly + 6), 5, 5) p.setPen(QPen(_C_SAT_LBL)) p.drawText(lx + 14, ly + 11, GNSS_NAMES.get(gid, f'SYS{gid}')) ly += 18 # ── Widget ──────────────────────────────────────────────────────────────────── class SkyPlotWidget(QWidget): """Displays the pre-rendered QImage from SkyRenderThread. paintEvent only does a single drawImage — never blocks the main thread.""" def __init__(self): super().__init__() self._sats: list = [] self._trails: dict = {} self._image: QImage = None self.setMinimumSize(400, 400) self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) self.setAttribute(Qt.WA_OpaquePaintEvent) self._thread = SkyRenderThread() self._thread.frame_ready.connect(self._on_frame) self._thread.start() # Stop thread when application quits app = QApplication.instance() if app: app.aboutToQuit.connect(self._thread.stop) # ── Public API ──────────────────────────────────────────────────────────── def update_data(self, sats: list, trails: dict = None): self._sats = sats self._trails = trails or {} self._thread.submit(sats, self._trails, self.width(), self.height()) # ── Qt events ───────────────────────────────────────────────────────────── def resizeEvent(self, event): super().resizeEvent(event) if self._sats: self._thread.submit(self._sats, self._trails, self.width(), self.height()) 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() # ── Render thread callback (main thread via queued signal) ──────────────── def _on_frame(self, img): self._image = img self.update() # schedules repaint — returns immediately