import sys, os
from collections import deque
_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, QHBoxLayout, QVBoxLayout, QLabel, QListWidget,
QListWidgetItem, QTableWidget, QTableWidgetItem, QHeaderView,
QSplitter, QGroupBox, QFrame, QCheckBox, QSpinBox,
)
from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtGui import QColor, QBrush, QIcon, QPixmap
from data_model import ReceiverManager, GNSS_NAMES, GNSS_COLORS
from sky_widget import SkyPlotWidget
_COL_HEADERS = ['GNSS', 'SV', 'Elev °', 'Az °', 'C/N₀', 'Fix', 'Receiver', 'Msg']
_COL_MSG = len(_COL_HEADERS) - 1
_TRAIL_MAXLEN = 86400 # 24 hours at 1 Hz
_TRAIL_MAX_HOURS = 24
_MSG_HIGHLIGHT_MS = 10_000
class SkyTab(QWidget):
def __init__(self, model: ReceiverManager):
super().__init__()
self._model = model
self._store: dict = {} # (rid, gnss_id, sv_id) -> SatelliteData
self._table_keys: list = [] # current row order as (rid, gnss_id, sv_id)
self._pdop: dict = {} # rid -> float
self._info_cards: dict = {} # rid -> QFrame
self._info_labels: dict = {} # rid -> (lbl_pdop, lbl_visible, lbl_used)
self._name_labels: dict = {} # rid -> QLabel (receiver display name)
self._trail: dict = {} # (rid, gnss_id, sv_id) -> deque[(elev, azim)]
self._hi_timers: dict = {} # (rid, gnss_id, sv_id) -> QTimer
self._gnss_checks: dict = {} # gnss_id -> QCheckBox
self._enabled_gnss: set = set(GNSS_NAMES.keys())
outer = QVBoxLayout(self)
outer.setContentsMargins(4, 4, 4, 4)
outer.setSpacing(4)
# ── Source selector + trail toggle ────────────────────────────────────
src_box = QGroupBox("Receiver Sources (check to include)")
src_box.setMaximumHeight(80)
src_bl = QHBoxLayout(src_box)
src_bl.setContentsMargins(4, 2, 4, 2)
src_bl.setSpacing(8)
self._src_list = QListWidget()
self._src_list.setFlow(QListWidget.LeftToRight)
self._src_list.setWrapping(True)
self._src_list.setResizeMode(QListWidget.Adjust)
self._src_list.itemChanged.connect(self._refresh_immediate)
src_bl.addWidget(self._src_list)
self._trail_cb = QCheckBox("Show Trails")
self._trail_cb.setChecked(False)
self._trail_cb.toggled.connect(self._on_trail_toggled)
src_bl.addWidget(self._trail_cb)
self._trail_spin = QSpinBox()
self._trail_spin.setRange(0, _TRAIL_MAX_HOURS)
self._trail_spin.setValue(1)
self._trail_spin.setSuffix(" h")
self._trail_spin.setFixedWidth(58)
self._trail_spin.setEnabled(False)
self._trail_spin.valueChanged.connect(self._on_trail_toggled)
src_bl.addWidget(self._trail_spin)
# ── Constellation legend / filter ─────────────────────────────────────
const_box = QGroupBox("Constellations (check to show)")
const_box.setMaximumHeight(80)
const_bl = QHBoxLayout(const_box)
const_bl.setContentsMargins(4, 2, 4, 2)
const_bl.setSpacing(8)
for gid in sorted(GNSS_NAMES):
cb = QCheckBox(GNSS_NAMES[gid])
cb.setChecked(True)
pix = QPixmap(10, 10)
pix.fill(QColor(GNSS_COLORS.get(gid, '#9E9E9E')))
cb.setIcon(QIcon(pix))
cb.toggled.connect(lambda checked, g=gid: self._on_gnss_toggled(g, checked))
self._gnss_checks[gid] = cb
const_bl.addWidget(cb)
const_bl.addStretch()
filter_row = QHBoxLayout()
filter_row.addWidget(src_box)
filter_row.addWidget(const_box)
outer.addLayout(filter_row)
# ── Info bar (one card per receiver) ──────────────────────────────────
info_frame = QFrame()
info_frame.setFrameShape(QFrame.StyledPanel)
info_frame.setMaximumHeight(42)
self._info_layout = QHBoxLayout(info_frame)
self._info_layout.setContentsMargins(6, 2, 6, 2)
self._info_layout.setSpacing(12)
self._info_layout.addStretch()
outer.addWidget(info_frame)
# ── Sky plot + satellite table ─────────────────────────────────────────
splitter = QSplitter(Qt.Horizontal)
self._sky = SkyPlotWidget()
splitter.addWidget(self._sky)
self._table = QTableWidget()
self._table.setColumnCount(len(_COL_HEADERS))
self._table.setHorizontalHeaderLabels(_COL_HEADERS)
hdr = self._table.horizontalHeader()
hdr.setSectionResizeMode(QHeaderView.ResizeToContents)
hdr.setSectionResizeMode(_COL_MSG, QHeaderView.Fixed)
hdr.setStretchLastSection(False)
self._table.setColumnWidth(_COL_MSG, 36)
self._table.setEditTriggers(QTableWidget.NoEditTriggers)
self._table.setSelectionBehavior(QTableWidget.SelectRows)
self._table.setAlternatingRowColors(False)
splitter.addWidget(self._table)
splitter.setStretchFactor(0, 2)
splitter.setStretchFactor(1, 1)
outer.addWidget(splitter)
# ── Table throttle timer (max ~2 Hz rebuilds) ─────────────────────────
self._table_timer = QTimer(self)
self._table_timer.setSingleShot(True)
self._table_timer.setInterval(500)
self._table_timer.timeout.connect(self._rebuild_table_from_store)
# ── Model signals ─────────────────────────────────────────────────────
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)
model.version_update.connect(self._on_version_update)
# ── Source list management ────────────────────────────────────────────────
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)
card = QFrame()
card.setFrameShape(QFrame.Box)
cl = QHBoxLayout(card)
cl.setContentsMargins(8, 1, 8, 1)
cl.setSpacing(10)
name_lbl = QLabel(f"{rid}")
lbl_pdop = QLabel("pDOP: —")
lbl_visible = QLabel("Visible: —")
lbl_used = QLabel("Used: —")
for w in (name_lbl, lbl_pdop, lbl_visible, lbl_used):
cl.addWidget(w)
self._info_cards[rid] = card
self._info_labels[rid] = (lbl_pdop, lbl_visible, lbl_used)
self._name_labels[rid] = name_lbl
self._info_layout.insertWidget(self._info_layout.count() - 1, card)
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
for k in [k for k in self._store if k[0] == rid]: del self._store[k]
for k in [k for k in self._trail if k[0] == rid]: del self._trail[k]
self._pdop.pop(rid, None)
card = self._info_cards.pop(rid, None)
if card:
self._info_layout.removeWidget(card)
card.deleteLater()
self._info_labels.pop(rid, None)
self._name_labels.pop(rid, None)
self._refresh_immediate()
def _on_version_update(self, rid: str, device_id: str, sw_version: str):
label = self._model.display_name(rid)
for i in range(self._src_list.count()):
item = self._src_list.item(i)
if item.data(Qt.UserRole) == rid:
item.setText(f" {label} ")
break
lbl = self._name_labels.get(rid)
if lbl:
lbl.setText(f"{label}")
self._table_keys = [] # force a full rebuild so the Receiver column updates
self._refresh_immediate()
def _selected_rids(self) -> set:
result = set()
for i in range(self._src_list.count()):
item = self._src_list.item(i)
if item.checkState() == Qt.Checked:
result.add(item.data(Qt.UserRole))
return result
def _visible_sats(self) -> list:
selected = self._selected_rids()
return [(k[0], self._store[k])
for k in self._store if k[0] in selected and k[1] in self._enabled_gnss]
def _active_trails(self) -> dict:
if not self._trail_cb.isChecked():
return {}
n_points = self._trail_spin.value() * 3600
if n_points == 0:
return {}
selected = self._selected_rids()
return {key: list(dq)[-n_points:]
for key, dq in self._trail.items()
if key[0] in selected and key[1] in self._enabled_gnss}
def _on_gnss_toggled(self, gid: int, checked: bool):
if checked:
self._enabled_gnss.add(gid)
else:
self._enabled_gnss.discard(gid)
self._refresh_immediate()
# ── Trail toggle ──────────────────────────────────────────────────────────
def _on_trail_toggled(self, _value=None):
self._trail_spin.setEnabled(self._trail_cb.isChecked())
self._sky.update_data(self._visible_sats(), self._active_trails())
# ── Info bar ──────────────────────────────────────────────────────────────
def _update_info(self):
for rid, (lbl_pdop, lbl_visible, lbl_used) in self._info_labels.items():
sats = [self._store[k] for k in self._store if k[0] == rid]
n_visible = len(sats)
n_used = sum(1 for s in sats if s.used_in_fix)
pdop = self._pdop.get(rid)
lbl_pdop.setText( f"pDOP: {pdop:.1f}" if pdop is not None else "pDOP: —")
lbl_visible.setText(f"Visible: {n_visible}")
lbl_used.setText( f"Used: {n_used}")
# ── Data update ───────────────────────────────────────────────────────────
def _on_sat_update(self, rid: str, sats: list):
selected = self._selected_rids()
for sat in sats:
key = (rid, sat.gnss_id, sat.sv_id)
self._store[key] = sat
if key not in self._trail:
self._trail[key] = deque(maxlen=_TRAIL_MAXLEN)
self._trail[key].append((sat.elevation, sat.azimuth))
if rid in selected:
self._highlight_row(key)
self._sky.update_data(self._visible_sats(), self._active_trails())
self._update_info()
if not self._table_timer.isActive():
self._table_timer.start()
def _on_pdop_update(self, rid: str, pdop: float, _ts: float):
self._pdop[rid] = pdop
self._update_info()
def _refresh_immediate(self):
self._table_timer.stop()
visible = self._visible_sats()
self._sky.update_data(visible, self._active_trails())
self._update_info()
self._rebuild_table(visible)
def _rebuild_table_from_store(self):
self._rebuild_table(self._visible_sats())
# ── Row highlight on message receive ─────────────────────────────────────
def _highlight_row(self, key):
try:
row = self._table_keys.index(key)
except ValueError:
return # row not in table yet (full rebuild pending)
item = self._table.item(row, _COL_MSG)
if item:
item.setText("✓")
t = self._hi_timers.get(key)
if t:
t.start(_MSG_HIGHLIGHT_MS)
else:
t = QTimer(self)
t.setSingleShot(True)
t.timeout.connect(lambda k=key, r=row: self._clear_highlight(k, r))
t.start(_MSG_HIGHLIGHT_MS)
self._hi_timers[key] = t
def _clear_highlight(self, key, row):
self._hi_timers.pop(key, None)
if row >= self._table.rowCount() or row >= len(self._table_keys):
return
if self._table_keys[row] != key:
return # row was reassigned during a full rebuild
item = self._table.item(row, _COL_MSG)
if item:
item.setText("")
# ── Table ─────────────────────────────────────────────────────────────────
def _rebuild_table(self, sats: list):
sorted_sats = sorted(sats, key=lambda x: (x[1].gnss_id, x[1].sv_id))
new_keys = [(rid, sat.gnss_id, sat.sv_id) for rid, sat in sorted_sats]
if new_keys == self._table_keys:
# Only values changed — update in place, scroll position untouched
for row, (_, sat) in enumerate(sorted_sats):
self._table.item(row, 2).setText(f"{sat.elevation:.0f}")
self._table.item(row, 3).setText(f"{sat.azimuth:.0f}")
self._table.item(row, 4).setText(f"{sat.cno:.1f}")
self._table.item(row, 5).setText("✓" if sat.used_in_fix else "")
return
# Satellite set changed — full rebuild
self._table_keys = new_keys
vp = self._table.viewport()
vp.setUpdatesEnabled(False)
self._table.setSortingEnabled(False)
self._table.setRowCount(len(sorted_sats))
for row, (rid, sat) in enumerate(sorted_sats):
gnss_name = GNSS_NAMES.get(sat.gnss_id, f'SYS{sat.gnss_id}')
bg = QColor(GNSS_COLORS.get(sat.gnss_id, '#9E9E9E'))
bg.setAlpha(55)
brush = QBrush(bg)
def cell(text, br=brush):
it = QTableWidgetItem(str(text))
it.setTextAlignment(Qt.AlignCenter)
it.setBackground(br)
return it
self._table.setItem(row, 0, cell(gnss_name))
self._table.setItem(row, 1, cell(sat.sv_id))
self._table.setItem(row, 2, cell(f"{sat.elevation:.0f}"))
self._table.setItem(row, 3, cell(f"{sat.azimuth:.0f}"))
self._table.setItem(row, 4, cell(f"{sat.cno:.1f}"))
self._table.setItem(row, 5, cell("✓" if sat.used_in_fix else ""))
self._table.setItem(row, 6, cell(self._model.display_name(rid)))
msg_item = QTableWidgetItem("")
msg_item.setTextAlignment(Qt.AlignCenter)
msg_item.setBackground(brush)
self._table.setItem(row, _COL_MSG, msg_item)
self._table.setSortingEnabled(True)
vp.setUpdatesEnabled(True)