diff --git a/LinuxMintQT/MeshtasticClient.code-workspace b/LinuxMintQT/MeshtasticClient.code-workspace new file mode 100644 index 0000000..990ca49 --- /dev/null +++ b/LinuxMintQT/MeshtasticClient.code-workspace @@ -0,0 +1,17 @@ +{ + "folders": [ + { + "path": "." + } + ], + "settings": { + "python.defaultInterpreterPath": "/home/knud/GIT/Meshtastic/Meshtastic_client/.venv/bin/python", + "python.interpreterPath": "/home/knud/GIT/Meshtastic/Meshtastic_client/.venv/bin/python", + "python.pythonPath": "/home/knud/GIT/Meshtastic/Meshtastic_client/.venv/bin/python", + "python.terminal.activateEnvironment": true, + "terminal.integrated.env.linux": { + "VIRTUAL_ENV": "${workspaceFolder}/.venv", + "PATH": "${workspaceFolder}/.venv/bin:${env:PATH}" + } + } +} diff --git a/LinuxMintQT/meshtastic.png b/LinuxMintQT/meshtastic.png new file mode 100644 index 0000000..668347f Binary files /dev/null and b/LinuxMintQT/meshtastic.png differ diff --git a/LinuxMintQT/meshtastic_client.py b/LinuxMintQT/meshtastic_client.py new file mode 100644 index 0000000..fadfb81 --- /dev/null +++ b/LinuxMintQT/meshtastic_client.py @@ -0,0 +1,1934 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Meshtastic Client - PySide6/Qt full GUI + +Qt port of the original Tkinter-based Meshtastic client. + +Implemented features (port of Tk version): +- TCP / Serial / BLE connect & disconnect +- Set IP/Port dialog +- Messages view with auto "pong" on 'ping' +- Channel selector: + * Public (broadcast, ch0) + * To selected node + * Extra radio channels (Ch 1..N) with names from the device +- Node list with: + * Search box + * Distance, speed, altitude, hops, "since" + * Right-click menu: Chat with node, Node info, Map, Traceroute, Delete node + * Double-click: sets "To selected node" as send target +- Per‑node chat windows (direct chat) +- Traceroute (via python-meshtastic if available, else meshtastic CLI) +- Tools menu: + * Clear messages + * Radio config (view, read‑only) +- Links menu: client / org / flasher / web / docker / FB DK / FB Nordjylland +- Light/Dark themes for the main UI + +Requirements (on target system): + pip install meshtastic PySide6 pubsub google-api-python-client +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import threading +import time +import datetime +import webbrowser +from typing import Any, Dict, Optional, Tuple, List + +from pathlib import Path + +from PySide6.QtCore import Qt, QTimer, QPoint +from PySide6.QtGui import QIcon, QAction +from PySide6.QtWidgets import ( + QApplication, + QMainWindow, + QWidget, + QVBoxLayout, + QHBoxLayout, + QSplitter, + QTextEdit, + QLineEdit, + QPushButton, + QLabel, + QComboBox, + QTableWidget, + QTableWidgetItem, + QMenu, + QMessageBox, + QInputDialog, + QStatusBar, + QHeaderView, + QDialog, + QPlainTextEdit, + QListWidget, +) + +# pubsub + meshtastic imports (same as Tk version) +try: + from pubsub import pub +except Exception: + pub = None + +try: + from meshtastic.tcp_interface import TCPInterface +except Exception: + TCPInterface = None + +try: + from meshtastic.serial_interface import SerialInterface +except Exception: + SerialInterface = None + +try: + from meshtastic.ble_interface import BLEInterface +except Exception: + BLEInterface = None + +try: + from meshtastic.protobuf import mesh_pb2, portnums_pb2 + import google.protobuf.json_format as _json_format +except Exception: + mesh_pb2 = None + portnums_pb2 = None + _json_format = None + +HOST_DEFAULT = "192.168.0.156" +PORT_DEFAULT = 4403 +PROJECT_PATH = Path(__file__).resolve().parent +ICON_PATH = PROJECT_PATH / "meshtastic.png" # PNG icon works better with Qt + + +def _prefer_chrome(url: str) -> None: + """Try to open URL in a browser, preferring Chrome where possible.""" + try: + webbrowser.open(url) + return + except Exception: + pass + + candidates = [ + "xdg-open", + "gio open", + "sensible-browser", + "google-chrome-stable", + "chromium", + r"C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe", + r"C:\\Program Files (x86)\\Google\\Chrome\\Application\\chrome.exe", + ] + for cmd in candidates: + try: + if os.path.isabs(cmd) and os.path.exists(cmd): + subprocess.Popen([cmd, url]) + return + else: + subprocess.Popen(cmd.split() + [url]) + return + except Exception: + continue + + +def _fmt_ago(epoch_seconds: Optional[float]) -> str: + """Format 'time since' value similar to Tk app.""" + if not epoch_seconds: + return "N/A" + try: + delta = time.time() - float(epoch_seconds) + except Exception: + return "N/A" + if delta < 0: + delta = 0 + mins = int(delta // 60) + hours = int(delta // 3600) + days = int(delta // 86400) + if delta < 60: + return "%ds" % int(delta) + if mins < 60: + return "%dm" % mins + if hours < 24: + return "%dh" % hours + if days < 7: + return "%dd" % days + dt = datetime.datetime.fromtimestamp(epoch_seconds) + return dt.strftime("%Y-%m-%d %H:%M") + + +class MeshtasticMainWindow(QMainWindow): + def __init__(self) -> None: + super().__init__() + self.setWindowTitle("Meshtastic Client (v1)") + + if ICON_PATH.exists(): + try: + self.setWindowIcon(QIcon(str(ICON_PATH))) + except Exception: + pass + + # core state + self.host: str = HOST_DEFAULT + self.port: int = PORT_DEFAULT + self.iface: Optional[object] = None + self.connected_evt = threading.Event() + self.current_theme: str = "dark" + self._last_seen_overrides: Dict[str, float] = {} + + # channel map: label -> {mode: ..., channelIndex: int} + self._channel_map: Dict[str, Dict[str, Any]] = {} + + # node sorting state + self._last_sort_col_index: int = 2 # "since" + self._last_sort_order_desc: bool = True + + # per-node chat windows + self._per_node_chats: Dict[str, "NodeChatDialog"] = {} + + self._build_ui() + self._connect_signals() + + # subscribe to Meshtastic pubsub topics + if pub is not None: + try: + pub.subscribe(self.on_connection_established, "meshtastic.connection.established") + pub.subscribe(self.on_connection_lost, "meshtastic.connection.lost") + pub.subscribe(self.on_receive, "meshtastic.receive") + pub.subscribe(self.on_node_updated, "meshtastic.node.updated") + except Exception as e: + print("pubsub subscribe failed:", e) + + # periodic refresh of node table (backup to pub events) + self.refresh_timer = QTimer(self) + self.refresh_timer.setInterval(2000) + self.refresh_timer.timeout.connect(self.refresh_nodes) + self.refresh_timer.start() + + self.apply_theme("dark") + self._append("Ready. Connection -> Connect (TCP/USB/BLE)") + self._update_title_with_host() + + # ------------------------------------------------------------------ + # UI construction + # ------------------------------------------------------------------ + def _build_ui(self) -> None: + central = QWidget(self) + self.setCentralWidget(central) + main_layout = QVBoxLayout(central) + + splitter = QSplitter(Qt.Horizontal, self) + + # --- left: messages ------------------------------------------- + left_widget = QWidget(self) + left_layout = QVBoxLayout(left_widget) + lbl_msgs = QLabel("Messages", left_widget) + left_layout.addWidget(lbl_msgs) + + self.txt_messages = QTextEdit(left_widget) + self.txt_messages.setReadOnly(True) + left_layout.addWidget(self.txt_messages) + + send_row = QHBoxLayout() + self.ent_message = QLineEdit(left_widget) + self.ent_message.setPlaceholderText("Type a message and press Enter…") + send_row.addWidget(self.ent_message, 1) + + self.btn_send = QPushButton("Send", left_widget) + send_row.addWidget(self.btn_send) + + # channel selector + self.cbo_channel = QComboBox(left_widget) + send_row.addWidget(self.cbo_channel) + left_layout.addLayout(send_row) + + splitter.addWidget(left_widget) + + # --- right: nodes --------------------------------------------- + right_widget = QWidget(self) + right_layout = QVBoxLayout(right_widget) + + header_row = QHBoxLayout() + self.lbl_nodes_header = QLabel("Nodes (0)", right_widget) + header_row.addWidget(self.lbl_nodes_header) + header_row.addStretch(1) + right_layout.addLayout(header_row) + + self.ent_search = QLineEdit(right_widget) + self.ent_search.setPlaceholderText("Search in nodes…") + right_layout.addWidget(self.ent_search) + + # node table + self.cols_all: List[str] = [ + "shortname", + "longname", + "since", + "hops", + "distkm", + "speed", + "alt", + "lastheard", + "hwmodel", + "role", + "macaddr", + "publickey", + "isunmessagable", + "id", + ] + headings = { + "shortname": "Short", + "longname": "Long", + "since": "Since", + "hops": "Hops", + "distkm": "Dist (km)", + "speed": "Speed", + "alt": "Alt (m)", + "hwmodel": "HW", + "role": "Role", + "lastheard": "", + "macaddr": "MAC", + "publickey": "Public key", + "isunmessagable": "Unmsg?", + "id": "ID", + } + self.nodes_table = QTableWidget(0, len(self.cols_all), right_widget) + for idx, name in enumerate(self.cols_all): + self.nodes_table.setHorizontalHeaderItem(idx, QTableWidgetItem(headings.get(name, name))) + + # set widths like Tk version for visible columns + widths = { + "shortname": 90, + "longname": 200, + "since": 90, + "hops": 50, + "distkm": 70, + "speed": 70, + "alt": 70, + "hwmodel": 90, + "role": 110, + } + for idx, name in enumerate(self.cols_all): + if name in widths: + self.nodes_table.setColumnWidth(idx, widths[name]) + + # hide technical columns + for name in ("lastheard", "macaddr", "publickey", "isunmessagable", "id"): + idx = self.cols_all.index(name) + self.nodes_table.setColumnHidden(idx, True) + + self.nodes_table.setSelectionBehavior(QTableWidget.SelectRows) + self.nodes_table.setEditTriggers(QTableWidget.NoEditTriggers) + self.nodes_table.horizontalHeader().setStretchLastSection(False) + self.nodes_table.horizontalHeader().setSectionResizeMode(QHeaderView.Interactive) + self.nodes_table.setContextMenuPolicy(Qt.CustomContextMenu) + + right_layout.addWidget(self.nodes_table) + splitter.addWidget(right_widget) + + splitter.setStretchFactor(0, 3) + splitter.setStretchFactor(1, 4) + main_layout.addWidget(splitter) + + # --- Menubar -------------------------------------------------- + menubar = self.menuBar() + + m_conn = menubar.addMenu("Connection") + self.act_connect_tcp = QAction("Connect (TCP)", self) + self.act_connect_serial = QAction("Connect via USB/Serial…", self) + self.act_connect_ble = QAction("Connect via Bluetooth…", self) + self.act_disconnect = QAction("Disconnect", self) + self.act_set_ipport = QAction("Set IP/Port…", self) + m_conn.addAction(self.act_connect_tcp) + m_conn.addAction(self.act_connect_serial) + m_conn.addAction(self.act_connect_ble) + m_conn.addAction(self.act_disconnect) + m_conn.addSeparator() + m_conn.addAction(self.act_set_ipport) + + m_tools = menubar.addMenu("Tools") + self.act_clear_messages = QAction("Clear messages", self) + self.act_radio_config = QAction("Radio config (view)", self) + self.act_neighbors = QAction("Neighbor table", self) + self.act_channel_editor = QAction("Channel editor", self) + m_tools.addAction(self.act_clear_messages) + m_tools.addAction(self.act_radio_config) + m_tools.addAction(self.act_neighbors) + m_tools.addAction(self.act_channel_editor) + + m_view = menubar.addMenu("View") + self.act_light_theme = QAction("Light theme", self) + self.act_dark_theme = QAction("Dark theme", self) + m_view.addAction(self.act_light_theme) + m_view.addAction(self.act_dark_theme) + + m_links = menubar.addMenu("Links") + self.act_link_client = QAction("Meshtastic client (GitHub)", self) + self.act_link_org = QAction("Meshtastic org", self) + self.act_link_flasher = QAction("Meshtastic flasher (Chrome)", self) + self.act_link_web = QAction("Meshtastic Web Client", self) + self.act_link_docker = QAction("Meshtastic docker client", self) + self.act_link_fb_dk = QAction("Meshtastic Facebook Danmark", self) + self.act_link_fb_nj = QAction("Meshtastic Facebook Nordjylland", self) + m_links.addAction(self.act_link_client) + m_links.addAction(self.act_link_org) + m_links.addAction(self.act_link_flasher) + m_links.addAction(self.act_link_web) + m_links.addAction(self.act_link_docker) + m_links.addSeparator() + m_links.addAction(self.act_link_fb_dk) + m_links.addAction(self.act_link_fb_nj) + + m_help = menubar.addMenu("Help") + self.act_about = QAction("About", self) + m_help.addAction(self.act_about) + + # --- Statusbar ------------------------------------------------ + status = QStatusBar(self) + self.setStatusBar(status) + self.lbl_status_nodes = QLabel("Nodes: 0", self) + status.addPermanentWidget(self.lbl_status_nodes) + + # Node context menu + self.node_menu = QMenu(self) + self.act_node_chat = QAction("Chat with node", self) + self.act_node_info = QAction("Node info", self) + self.act_node_map = QAction("Map", self) + self.act_node_traceroute = QAction("Traceroute", self) + self.act_node_delete = QAction("Delete node", self) + self.node_menu.addAction(self.act_node_chat) + self.node_menu.addAction(self.act_node_info) + self.node_menu.addAction(self.act_node_map) + self.node_menu.addAction(self.act_node_traceroute) + self.node_menu.addSeparator() + self.node_menu.addAction(self.act_node_delete) + + # initialize channel selector + self._reset_channel_choices() + + # ------------------------------------------------------------------ + # Signal wiring + # ------------------------------------------------------------------ + def _connect_signals(self) -> None: + self.btn_send.clicked.connect(self.send_message) + self.ent_message.returnPressed.connect(self.send_message) + self.ent_search.textChanged.connect(self.refresh_nodes) + + self.nodes_table.customContextMenuRequested.connect(self._popup_node_menu) + self.nodes_table.cellDoubleClicked.connect(self._on_node_double_clicked) + self.nodes_table.horizontalHeader().sectionClicked.connect(self._on_header_clicked) + + self.act_connect_tcp.triggered.connect(self.connect_tcp) + self.act_connect_serial.triggered.connect(self.connect_serial_dialog) + self.act_connect_ble.triggered.connect(self.connect_ble_dialog) + self.act_disconnect.triggered.connect(self.disconnect) + self.act_set_ipport.triggered.connect(self.set_ip_port) + + self.act_clear_messages.triggered.connect(self.txt_messages.clear) + self.act_radio_config.triggered.connect(self.show_radio_config_window) + self.act_neighbors.triggered.connect(self.show_neighbors_window) + self.act_channel_editor.triggered.connect(self.show_channel_editor_window) + + self.act_light_theme.triggered.connect(lambda: self.apply_theme("light")) + self.act_dark_theme.triggered.connect(lambda: self.apply_theme("dark")) + + self.act_link_client.triggered.connect( + lambda: self._open_browser_url("https://github.com/dk98174003/Meshtastic-Client") + ) + self.act_link_org.triggered.connect(lambda: self._open_browser_url("https://meshtastic.org/")) + self.act_link_flasher.triggered.connect( + lambda: self._open_browser_url("https://flasher.meshtastic.org/") + ) + self.act_link_web.triggered.connect( + lambda: self._open_browser_url("https://client.meshtastic.org") + ) + self.act_link_docker.triggered.connect( + lambda: self._open_browser_url( + "https://meshtastic.org/docs/software/linux/usage/#usage-with-docker" + ) + ) + self.act_link_fb_dk.triggered.connect( + lambda: self._open_browser_url( + "https://www.facebook.com/groups/1553839535376876/" + ) + ) + self.act_link_fb_nj.triggered.connect( + lambda: self._open_browser_url( + "https://www.facebook.com/groups/1265866668302201/" + ) + ) + + self.act_about.triggered.connect(self._show_about) + + self.act_node_chat.triggered.connect(self._cm_open_chat) + self.act_node_info.triggered.connect(self._cm_show_node_details) + self.act_node_map.triggered.connect(self._cm_open_map) + self.act_node_traceroute.triggered.connect(self._cm_traceroute) + self.act_node_delete.triggered.connect(self._cm_delete_node) + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + def _open_browser_url(self, url: str) -> None: + _prefer_chrome(url) + + def _update_title_with_host(self) -> None: + self.setWindowTitle("Meshtastic Client (v1) - %s:%s" % (self.host, self.port)) + + def _append(self, text: str) -> None: + self.txt_messages.append(text) + + # ------------------------------------------------------------------ + # Meshtastic-related helpers + # ------------------------------------------------------------------ + def _node_label(self, node_id: str) -> str: + iface = self.iface + if not iface or not getattr(iface, "nodes", None): + return node_id + node = getattr(iface, "nodes", {}).get(node_id, {}) + user = (node or {}).get("user") or {} + shortname = user.get("shortName") or "" + longname = user.get("longName") or "" + label = ("%s %s" % (shortname, longname)).strip() + if label: + return label + return node_id + + def _get_lastheard_epoch(self, node_id: str, node: Dict[str, Any]) -> Optional[float]: + raw = (node or {}).get("lastHeard") + ts_iface: Optional[float] = None + if raw is not None: + try: + val = float(raw) + ts_iface = val / 1000.0 if val > 10000000000 else val + except Exception: + ts_iface = None + ts_local = self._last_seen_overrides.get(str(node_id)) + if ts_iface and ts_local: + return max(ts_iface, ts_local) + return ts_iface or ts_local + + def _extract_latlon(self, node: dict) -> Tuple[Optional[float], Optional[float]]: + pos = (node or {}).get("position") or {} + lat = pos.get("latitude") or pos.get("latitudeI") or pos.get("latitude_i") + lon = pos.get("longitude") or pos.get("longitudeI") or pos.get("longitude_i") + try: + if lat is not None: + lat = float(lat) + if abs(lat) > 90: + lat *= 1e-7 + if lon is not None: + lon = float(lon) + if abs(lon) > 180: + lon *= 1e-7 + except Exception: + lat = lon = None + return lat, lon + + def _extract_speed_alt(self, node: dict) -> Tuple[Optional[float], Optional[float]]: + pos = (node or {}).get("position") or {} + speed = ( + pos.get("groundSpeedKmh") + or pos.get("groundSpeedKmhI") + or pos.get("groundSpeed") + or pos.get("ground_speed") + ) + alt = ( + pos.get("altitude") + or pos.get("altitudeM") + or pos.get("altitude_i") + or pos.get("altitudeI") + ) + try: + if speed is not None: + speed = float(speed) + if alt is not None: + alt = float(alt) + except Exception: + speed, alt = None, None + return speed, alt + + def _get_local_latlon(self) -> Tuple[Optional[float], Optional[float]]: + iface = self.iface + if not iface: + return (None, None) + try: + mi = getattr(iface, "myInfo", None) + nbn = getattr(iface, "nodesByNum", None) + if mi is not None and hasattr(mi, "my_node_num") and nbn: + n = nbn.get(mi.my_node_num) or {} + lat, lon = self._extract_latlon(n) + if lat is not None and lon is not None: + return lat, lon + except Exception: + pass + return (None, None) + + def _haversine_km(self, lat1, lon1, lat2, lon2) -> float: + import math + + R = 6371.0088 + phi1 = math.radians(lat1) + phi2 = math.radians(lat2) + dphi = math.radians(lat2 - lat1) + dlmb = math.radians(lon2 - lon1) + a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlmb / 2) ** 2 + c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) + return R * c + + # ------------------------------------------------------------------ + # Pubsub callbacks (run in background threads → use QTimer.singleShot) + # ------------------------------------------------------------------ + def on_connection_established(self, interface=None, **kwargs) -> None: + self.connected_evt.set() + QTimer.singleShot(0, self._on_connection_established_ui) + + def _on_connection_established_ui(self) -> None: + self._append("[+] Connected") + self.refresh_nodes() + try: + self._update_channels_from_iface() + except Exception: + pass + + def on_connection_lost(self, interface=None, **kwargs) -> None: + self.connected_evt.clear() + QTimer.singleShot(0, lambda: self._append("[-] Connection lost")) + + def on_node_updated(self, node=None, interface=None, **kwargs) -> None: + QTimer.singleShot(0, self.refresh_nodes) + + def on_receive(self, packet=None, interface=None, **kwargs) -> None: + pkt = packet or {} + QTimer.singleShot(0, lambda p=pkt: self._handle_receive(p)) + + # ------------------------------------------------------------------ + # Receive handling + # ------------------------------------------------------------------ + def _handle_receive(self, packet: dict) -> None: + decoded = packet.get("decoded", {}) if isinstance(packet, dict) else {} + app_name = decoded.get("app", "") + portnum = decoded.get("portnum") or app_name + from_id = packet.get("fromId") or packet.get("from") or "UNKNOWN" + label = self._node_label(from_id) + + tag_map = { + "TEXT_MESSAGE_APP": "MSG", + "TEXT_MESSAGE_COMPRESSED_APP": "MSG", + "POSITION_APP": "POS", + "TELEMETRY_APP": "TEL", + "NODEINFO_APP": "INFO", + "ROUTING_APP": "ROUT", + "MAP_REPORT_APP": "MAP", + "ADMIN_APP": "ADM", + "NEIGHBORINFO_APP": "NEI", + "STORE_FORWARD_APP": "SFWD", + "REMOTE_HARDWARE_APP": "RHW", + "PRIVATE_APP": "PRIV", + } + tag = tag_map.get(app_name or portnum, "INFO") + + if app_name in ("TEXT_MESSAGE_APP", "TEXT_MESSAGE_COMPRESSED_APP"): + text = decoded.get("text") or decoded.get("payload", {}).get("text", "") + if text: + self._append(f"[{tag}] {label}: {text}") + # also append to per-node chat if open + self._append_to_node_chat(str(from_id), text) + else: + self._append(f"[{tag}] {label}") + + decoded2 = packet.get("decoded", {}) if isinstance(packet, dict) else {} + portnum2 = decoded2.get("portnum") + sender = packet.get("fromId") or packet.get("from") or packet.get("fromIdShort") + if sender: + self._last_seen_overrides[str(sender)] = time.time() + + user = {} + iface = self.iface + if iface and getattr(iface, "nodes", None) and sender: + user = getattr(iface, "nodes", {}).get(sender, {}).get("user", {}) + shortname = user.get("shortName") or "" + longname = user.get("longName") or "" + label2 = str(shortname or longname or sender or "Unknown").strip() + + text2 = "" + p = decoded2.get("payload", "") + if isinstance(p, (bytes, bytearray)): + try: + text2 = p.decode("utf-8", errors="ignore") + except Exception: + text2 = repr(p) + elif isinstance(p, str): + text2 = p + else: + t = decoded2.get("text") + if isinstance(t, bytes): + text2 = t.decode("utf-8", errors="ignore") + elif isinstance(t, str): + text2 = t + + rssi = packet.get("rxRssi") + + if portnum2 == "TEXT_MESSAGE_APP": + self._append(f"[MSG] {label2}: {text2} (RSSI={rssi})") + self._append_to_node_chat(str(sender), text2) + if isinstance(text2, str) and text2.strip().lower() == "ping": + self._send_pong(sender, label2) + + self.refresh_nodes() + + def _send_pong(self, dest_id: Optional[str], label: str) -> None: + iface = self.iface + if not iface or not dest_id: + return + try: + iface.sendText("pong", destinationId=dest_id, wantAck=False) + self._append(f"[auto] pong -> {label}") + self._append_to_node_chat(str(dest_id), "[auto] pong") + except Exception as e: + self._append(f"[auto] pong failed: {e}") + + # ------------------------------------------------------------------ + # Connection actions + # ------------------------------------------------------------------ + def set_ip_port(self) -> None: + new_host, ok = QInputDialog.getText( + self, + "Set IP/Port", + "Host/IP:", + text=self.host, + ) + if not ok or not new_host.strip(): + return + + new_port, ok = QInputDialog.getInt( + self, + "Set IP/Port", + "Port:", + value=self.port, + min=1, + max=65535, + ) + if not ok: + return + + self.host = new_host.strip() + self.port = int(new_port) + self._update_title_with_host() + + def connect_tcp(self) -> None: + if self.iface: + return + host = self.host.strip() + port = int(self.port) + self._append(f"Connecting TCP {host}:{port} ...") + + def run(): + try: + if TCPInterface is None: + raise RuntimeError("meshtastic.tcp_interface not installed") + self.iface = TCPInterface(hostname=host, portNumber=port) + self.connected_evt.wait(timeout=5) + except Exception as e: + def _err(): + QMessageBox.critical(self, "TCP connect failed", str(e)) + QTimer.singleShot(0, _err) + + threading.Thread(target=run, daemon=True).start() + + def connect_serial_dialog(self) -> None: + if SerialInterface is None: + QMessageBox.critical( + self, "Unavailable", "meshtastic.serial_interface not installed." + ) + return + + port, ok = QInputDialog.getText( + self, + "Serial", + "Serial port (or type '(auto)' for automatic):", + text="(auto)", + ) + if not ok: + return + port = port.strip() + dev = None if port.lower() == "(auto)" or not port else port + self._append(f"Connecting Serial {dev or '(auto)'} ...") + + def run(): + try: + if dev: + self.iface = SerialInterface(devPath=dev) + else: + self.iface = SerialInterface() + self.connected_evt.wait(timeout=5) + except Exception as e: + def _err(): + QMessageBox.critical(self, "Serial connect failed", str(e)) + QTimer.singleShot(0, _err) + + threading.Thread(target=run, daemon=True).start() + + def connect_ble_dialog(self) -> None: + if BLEInterface is None: + QMessageBox.critical( + self, + "Unavailable", + "meshtastic.ble_interface not installed (needs bleak).", + ) + return + self._append("Scanning BLE for Meshtastic devices ...") + + def run_scan(): + try: + devices = BLEInterface.scan() + except Exception as e: + def _err(): + QMessageBox.critical(self, "BLE scan failed", str(e)) + QTimer.singleShot(0, _err) + return + + if not devices: + def _info(): + QMessageBox.information(self, "BLE", "No devices found.") + QTimer.singleShot(0, _info) + return + + options = [ + f"{i+1}. {getattr(d, 'name', '') or '(unnamed)'} [{getattr(d, 'address', '?')}]" + for i, d in enumerate(devices) + ] + + def _select(): + text, ok = QInputDialog.getItem( + self, + "Select BLE device", + "Select device:", + options, + editable=False, + ) + if not ok: + return + idx = options.index(text) + addr = getattr(devices[idx], "address", None) + if not addr: + QMessageBox.critical(self, "BLE", "Selected device has no address.") + return + self._append(f"Connecting BLE {addr} ...") + + def _run_connect(): + try: + self.iface = BLEInterface(address=addr) + self.connected_evt.wait(timeout=8) + except Exception as e: + def _err2(): + QMessageBox.critical(self, "BLE connect failed", str(e)) + QTimer.singleShot(0, _err2) + + threading.Thread(target=_run_connect, daemon=True).start() + + QTimer.singleShot(0, _select) + + threading.Thread(target=run_scan, daemon=True).start() + + def disconnect(self) -> None: + try: + if self.iface: + self.iface.close() + except Exception: + pass + self.iface = None + self.connected_evt.clear() + self._append("[*] Disconnected") + + # ------------------------------------------------------------------ + # Channel selector helpers + # ------------------------------------------------------------------ + def _reset_channel_choices(self) -> None: + """Reset channel combo to defaults (Public + To selected node).""" + self._channel_map = {} + options: List[str] = [] + + label_pub = "Public (broadcast)" + self._channel_map[label_pub] = {"mode": "broadcast", "channelIndex": 0} + options.append(label_pub) + + # Removed "To selected node" option from channel selector; + # direct messages are available via the node right-click menu. + self.cbo_channel.clear() + self.cbo_channel.addItems(options) + # Default to Public on a hard reset; user choice will be preserved + # later by _update_channels_from_iface when channels are refreshed. + self.cbo_channel.setCurrentIndex(0) + + def _set_channel_choice(self, label: str) -> None: + idx = self.cbo_channel.findText(label) + if idx >= 0: + self.cbo_channel.setCurrentIndex(idx) + + def _update_channels_from_iface(self) -> None: + iface = self.iface + if not iface: + return + local_node = getattr(iface, "localNode", None) + if not local_node: + return + + # Remember current selection so we can keep it after refreshing + prev_label = self.cbo_channel.currentText().strip() + + chans = getattr(local_node, "channels", None) + try: + if (not chans) and hasattr(local_node, "requestChannels"): + local_node.requestChannels() + time.sleep(1.5) + chans = getattr(local_node, "channels", None) + except Exception: + chans = getattr(local_node, "channels", None) + + options = [self.cbo_channel.itemText(i) for i in range(self.cbo_channel.count())] + + # update channel 0 name + try: + if chans and len(chans) > 0: + ch0 = chans[0] + try: + name0 = (getattr(ch0, "settings", None).name or "").strip() + except Exception: + try: + name0 = (ch0.settings.name or "").strip() + except Exception: + name0 = "" + if name0: + old_label = None + for lbl, meta in list(self._channel_map.items()): + if meta.get("mode") == "broadcast" and int(meta.get("channelIndex", 0) or 0) == 0: + old_label = lbl + break + if old_label: + new_label = f"Public (ch0: {name0})" + self._channel_map[new_label] = self._channel_map.pop(old_label) + options = [new_label if v == old_label else v for v in options] + if self.cbo_channel.currentText() == old_label: + self._set_channel_choice(new_label) + except Exception: + pass + + # Add remaining channels (1..N) as broadcast options. + for idx, ch in enumerate(chans or []): + if idx == 0: + continue + try: + name = (getattr(ch, "settings", None).name or "").strip() + except Exception: + try: + name = (ch.settings.name or "").strip() + except Exception: + name = "" + if not name: + label = f"Ch {idx}" + else: + label = f"Ch {idx}: {name}" + if label in self._channel_map: + continue + self._channel_map[label] = {"mode": "broadcast_channel", "channelIndex": idx} + options.append(label) + + self.cbo_channel.clear() + self.cbo_channel.addItems(options) + + # Restore previous selection if it still exists + if prev_label and prev_label in options: + self._set_channel_choice(prev_label) + + def _send_text_to_node(self, dest_id: str, msg: str) -> bool: + iface = self.iface + if not iface: + QMessageBox.warning(self, "Send", "Connect first.") + return False + msg = (msg or "").strip() + if not msg: + return False + try: + choice = self.cbo_channel.currentText() + info = self._channel_map.get(choice) or {"mode": "broadcast", "channelIndex": 0} + ch_index = int(info.get("channelIndex", 0) or 0) + except Exception: + ch_index = 0 + try: + iface.sendText(msg, destinationId=dest_id, wantAck=False, channelIndex=ch_index) + self._append("[ME -> %s] %s" % (self._node_label(dest_id), msg)) + self._append_to_node_chat(str(dest_id), "[ME] " + msg) + return True + except Exception as e: + QMessageBox.critical(self, "Send failed", str(e)) + return False + + def send_message(self) -> None: + msg = self.ent_message.text().strip() + if not msg: + return + iface = self.iface + if not iface: + QMessageBox.warning(self, "Not connected", "Connect first.") + return + + try: + choice = self.cbo_channel.currentText() + info = self._channel_map.get(choice) or {"mode": "broadcast", "channelIndex": 0} + mode = info.get("mode", "broadcast") + ch_index = int(info.get("channelIndex", 0) or 0) + except Exception: + mode = "broadcast" + ch_index = 0 + + try: + if mode == "selected": + nid = self._get_selected_node_id() + if not nid: + QMessageBox.information(self, "No selection", "Select a node first.") + return + dest = self._resolve_node_dest_id(nid) + if not dest: + QMessageBox.critical( + self, + "Send failed", + "Cannot resolve destination for selected node.", + ) + return + iface.sendText(msg, destinationId=dest, wantAck=False, channelIndex=ch_index) + self._append(f"[ME -> {self._node_label(nid)}] {msg}") + self._append_to_node_chat(str(dest), "[ME] " + msg) + else: + iface.sendText(msg, wantAck=False, channelIndex=ch_index) + if mode == "broadcast_channel" and ch_index: + self._append(f"[ME ch{ch_index}] {msg}") + else: + self._append(f"[ME] {msg}") + + self.ent_message.clear() + except Exception as e: + QMessageBox.critical(self, "Send failed", str(e)) + + # ------------------------------------------------------------------ + # Nodes table + # ------------------------------------------------------------------ + def _get_selected_node_id(self) -> Optional[str]: + rows = self.nodes_table.selectionModel().selectedRows() + if not rows: + return None + row = rows[0].row() + item = self.nodes_table.item(row, 0) # shortname column + if not item: + return None + return item.data(Qt.UserRole) + + def refresh_nodes(self) -> None: + iface = self.iface + # Also refresh available channels when we have a local node, + # in case they appear after initial connection. + if iface and getattr(iface, "localNode", None): + try: + self._update_channels_from_iface() + except Exception: + pass + + if not iface or not getattr(iface, "nodes", None): + self.nodes_table.setRowCount(0) + self.lbl_nodes_header.setText("Nodes (0)") + self.lbl_status_nodes.setText("Nodes: 0") + return + + q = self.ent_search.text().strip().lower() + + try: + nodes_snapshot = dict(getattr(iface, "nodes", {}) or {}) + except Exception: + nodes_snapshot = {} + + base_lat, base_lon = self._get_local_latlon() + + self.nodes_table.setRowCount(0) + for node_id, node in nodes_snapshot.items(): + user = (node or {}).get("user") or {} + shortname = user.get("shortName") or "" + longname = user.get("longName") or "" + hwmodel = user.get("hwModel") or "" + role = user.get("role") + if not role: + # Many nodes don't explicitly report a role; treat missing as default CLIENT + role = "CLIENT" + macaddr = user.get("macaddr") or "" + publickey = user.get("publicKey") or "" + unmsg = user.get("isUnmessagable") or user.get("isUnmessageable") or False + + lastheard_epoch = self._get_lastheard_epoch(node_id, node) + since_str = _fmt_ago(lastheard_epoch) + hops = node.get("hopsAway") + + lat, lon = self._extract_latlon(node) + if base_lat is not None and base_lon is not None and lat is not None and lon is not None: + try: + dist = self._haversine_km(base_lat, base_lon, lat, lon) + except Exception: + dist = None + else: + dist = None + dist_str = "%.1f" % dist if isinstance(dist, (int, float)) else "-" + + speed, alt = self._extract_speed_alt(node) + speed_str = "%.1f" % speed if isinstance(speed, (int, float)) else "-" + alt_str = "%.0f" % alt if isinstance(alt, (int, float)) else "-" + + lastheard_val = f"{int(lastheard_epoch or 0):015d}" + + values = { + "shortname": shortname, + "longname": longname, + "since": since_str, + "hops": str(hops) if hops is not None else "-", + "distkm": dist_str, + "speed": speed_str, + "alt": alt_str, + "lastheard": lastheard_val, + "hwmodel": hwmodel, + "role": role, + "macaddr": macaddr, + "publickey": publickey, + "isunmessagable": str(bool(unmsg)), + "id": node_id, + } + + if q and not any(q in str(v).lower() for v in values.values()): + continue + + row = self.nodes_table.rowCount() + self.nodes_table.insertRow(row) + for col_idx, col_name in enumerate(self.cols_all): + item = QTableWidgetItem(str(values[col_name])) + if col_name in {"lastheard", "distkm", "hops", "speed", "alt"}: + try: + item.setData(Qt.UserRole, float(values[col_name]) if values[col_name] not in {"-", ""} else 0.0) + except Exception: + item.setData(Qt.UserRole, 0.0) + if col_name == "shortname": + item.setData(Qt.UserRole, node_id) # store node id + self.nodes_table.setItem(row, col_idx, item) + + count = self.nodes_table.rowCount() + self.lbl_nodes_header.setText(f"Nodes ({count})") + self.lbl_status_nodes.setText(f"Nodes: {count}") + + if count > 0: + order = Qt.DescendingOrder if self._last_sort_order_desc else Qt.AscendingOrder + + sort_col = self._last_sort_col_index + try: + # When 'Since' is the active sort column, actually sort by hidden 'lastheard' + if self.cols_all[self._last_sort_col_index] == "since": + sort_col = self.cols_all.index("lastheard") + except Exception: + sort_col = self._last_sort_col_index + + self.nodes_table.sortItems(sort_col, order) + + def _on_header_clicked(self, logical_index: int) -> None: + """Handle clicks on the node-list header. + + Special case: when clicking the visible 'Since' column, we actually sort + by the hidden 'lastheard' epoch column, so the newest nodes appear first + in descending order. + """ + if logical_index == self._last_sort_col_index: + self._last_sort_order_desc = not self._last_sort_order_desc + else: + self._last_sort_col_index = logical_index + self._last_sort_order_desc = True + + order = Qt.DescendingOrder if self._last_sort_order_desc else Qt.AscendingOrder + + sort_col = logical_index + try: + # If user clicked "Since", use the hidden "lastheard" column for sorting + if self.cols_all[logical_index] == "since": + sort_col = self.cols_all.index("lastheard") + except Exception: + sort_col = logical_index + + self.nodes_table.sortItems(sort_col, order) + + # ------------------------------------------------------------------ + # Node context menu + actions + # ------------------------------------------------------------------ + def _popup_node_menu(self, pos: QPoint) -> None: + index = self.nodes_table.indexAt(pos) + if index.isValid(): + self.nodes_table.selectRow(index.row()) + # Pause auto-refresh while the context menu is open, + # otherwise the selection can disappear while clicking. + restarted = False + try: + if hasattr(self, "refresh_timer") and self.refresh_timer.isActive(): + self.refresh_timer.stop() + restarted = True + self.node_menu.exec(self.nodes_table.viewport().mapToGlobal(pos)) + finally: + if restarted: + self.refresh_timer.start() + + def _on_node_double_clicked(self, row: int, column: int) -> None: + """Double‑click on a node: mark it as target and open chat window.""" + nid = self._get_selected_node_id() + if not nid: + return + # Log in the main message area which node is now the active chat target + self._append(f"[target] will send to {self._node_label(nid)}") + # Same behaviour as right‑click -> "Chat with node" + self._open_node_chat(nid) + + def _cm_open_chat(self) -> None: + nid = self._get_selected_node_id() + if not nid: + QMessageBox.information(self, "Chat", "Select a node first.") + return + self._open_node_chat(nid) + + def _open_node_chat(self, nid: str) -> None: + key = str(nid) + if key in self._per_node_chats: + win = self._per_node_chats[key] + try: + win.show() + win.raise_() + win.activateWindow() + except Exception: + pass + return + label = self._node_label(nid) + chat = NodeChatDialog(self, key, label) + self._per_node_chats[key] = chat + chat.show() + + def _append_to_node_chat(self, node_id: str, line: str) -> None: + key = str(node_id) + win = self._per_node_chats.get(key) + if not win: + return + win.append_line(line) + + def _cm_traceroute(self) -> None: + nid = self._get_selected_node_id() + if not nid: + QMessageBox.information(self, "Traceroute", "Select a node first.") + return + iface = self.iface + if not iface or not getattr(iface, "localNode", None): + QMessageBox.warning(self, "Traceroute", "Connect first.") + return + dest = self._resolve_node_dest_id(nid) + if not dest: + QMessageBox.critical(self, "Traceroute", "Cannot determine node ID for traceroute.") + return + # Log and run traceroute synchronously on the GUI thread so + # that result dialogs are guaranteed to show. + self._append(f"[trace] Requesting traceroute to {self._node_label(nid)} ({dest})") + self._do_traceroute(dest) + + def _cm_delete_node(self) -> None: + nid = self._get_selected_node_id() + iface = self.iface + if not nid: + QMessageBox.information(self, "Delete node", "Select a node first.") + return + if not iface or not getattr(iface, "localNode", None): + QMessageBox.warning(self, "Delete node", "Connect first.") + return + dest = self._resolve_node_dest_id(nid) + if not dest: + QMessageBox.critical(self, "Delete node", "Cannot determine node ID.") + return + label = self._node_label(nid) + ret = QMessageBox.question( + self, + "Delete node", + f"Remove node {label} ({dest}) from the NodeDB on the connected radio?\n\n" + "The device might reboot after this.", + QMessageBox.Yes | QMessageBox.No, + ) + if ret != QMessageBox.Yes: + return + try: + getattr(iface, "localNode").removeNode(dest) + self._append(f"[admin] Requested delete of node {label} ({dest})") + except Exception as e: + QMessageBox.critical(self, "Delete node", f"Failed to delete node: {e}") + return + + rows = self.nodes_table.selectionModel().selectedRows() + if rows: + self.nodes_table.removeRow(rows[0].row()) + count = self.nodes_table.rowCount() + self.lbl_nodes_header.setText(f"Nodes ({count})") + self.lbl_status_nodes.setText(f"Nodes: {count}") + + def _resolve_node_dest_id(self, nid: str) -> Optional[str]: + if nid.startswith("!") or nid.isdigit(): + return nid + iface = self.iface + try: + if iface and getattr(iface, "nodes", None): + node = getattr(iface, "nodes", {}).get(nid) or {} + user = (node or {}).get("user") or {} + node_id = user.get("id") or "" + if node_id: + return node_id + except Exception: + pass + if nid: + return "!" + nid if not nid.startswith("!") else nid + return None + + # ------------------------------------------------------------------ + # Traceroute + # ------------------------------------------------------------------ + def _do_traceroute(self, dest: str, hop_limit: int = 10, channel_index: int = 0) -> None: + """Run traceroute using the meshtastic CLI only (same behaviour as Tk client). + + This avoids protobuf / RouteDiscovery compatibility issues and relies on the + already working `meshtastic --traceroute` command. + """ + # Simply delegate to the CLI helper; it will log and open the traceroute window. + self._do_traceroute_via_cli(dest) + def _do_traceroute_via_interface(self, dest: str, hop_limit: int, channel_index: int) -> None: + iface = self.iface + if not iface or mesh_pb2 is None or portnums_pb2 is None or _json_format is None: + QTimer.singleShot( + 0, lambda: QMessageBox.information(self, "Traceroute", "Python traceroute not available.") + ) + return + + evt = threading.Event() + result: Dict[str, Any] = {} + + def _num_to_label(num: int) -> str: + try: + nbn = getattr(iface, "nodesByNum", None) + if nbn and num in nbn: + n = nbn[num] + user = (n or {}).get("user") or {} + sid = user.get("id") or f"!{num:08x}" + sn = user.get("shortName") or "" + ln = user.get("longName") or "" + label = (sn or ln or sid).strip() + return f"{label} ({sid})" if sid else label + except Exception: + pass + return f"!{int(num):08x}" + + def _on_response(p: dict): + try: + rd = mesh_pb2.RouteDiscovery() + rd.ParseFromString(p["decoded"]["payload"]) + as_dict = _json_format.MessageToDict(rd) + result["packet"] = p + result["data"] = as_dict + except Exception as e: + result["error"] = str(e) + finally: + evt.set() + + try: + r = mesh_pb2.RouteDiscovery() + iface.sendData( + r, + destinationId=dest, + portNum=portnums_pb2.PortNum.TRACEROUTE_APP, + wantResponse=True, + onResponse=_on_response, + channelIndex=channel_index, + hopLimit=hop_limit, + ) + except Exception as e: + QTimer.singleShot( + 0, + lambda: QMessageBox.critical(self, "Traceroute", f"Failed to send traceroute: {e}"), + ) + return + + if not evt.wait(10.0): + QTimer.singleShot( + 0, + lambda: QMessageBox.information( + self, "Traceroute", "No traceroute response (timeout or unsupported)." + ), + ) + return + + if "error" in result: + QTimer.singleShot( + 0, + lambda: QMessageBox.critical( + self, "Traceroute", f"Failed to decode traceroute: {result['error']}" + ), + ) + return + + p = result.get("packet") or {} + data = result.get("data") or {} + UNK = -128 + + try: + origin_num = int(p.get("to")) + dest_num = int(p.get("from")) + except Exception: + origin_num = None + dest_num = None + + def _build_path(title: str, start_num: Optional[int], route_key: str, snr_key: str, end_num: Optional[int]) -> Optional[str]: + route_nums = [] + for v in data.get(route_key, []): + try: + route_nums.append(int(v)) + except Exception: + pass + snrs = [] + for v in data.get(snr_key, []): + try: + snrs.append(int(v)) + except Exception: + pass + if not start_num or not end_num: + return None + nodes = [start_num] + route_nums + [end_num] + if len(nodes) <= 1: + return None + parts = [] + for idx, num in enumerate(nodes): + label = _num_to_label(num) + if idx == 0: + parts.append(label) + else: + snr_txt = "? dB" + if (idx - 1) < len(snrs): + v = snrs[idx - 1] + if v != UNK: + snr_txt = f"{v / 4.0:.2f} dB" + parts.append(f"{label} ({snr_txt})") + return title + "\n" + " -> ".join(parts) + + lines: List[str] = [] + fwd = _build_path("Route towards destination:", origin_num, "route", "snrTowards", dest_num) + if fwd: + lines.append(fwd) + back = _build_path("Route back to us:", dest_num, "routeBack", "snrBack", origin_num) + if back: + lines.append(back) + + if not lines: + msg = "Traceroute completed but no route data available." + QTimer.singleShot( + 0, + lambda: (QMessageBox.information(self, "Traceroute", msg), self._show_traceroute_window(msg)), + ) + return + + text = "\n\n".join(lines) + # Show traceroute result in both the log and a popup window (like the Tk version) + QTimer.singleShot(0, lambda: (self._append(text), self._show_traceroute_window(text))) + + def _do_traceroute_via_cli(self, dest: str) -> None: + host = self.host.strip() or HOST_DEFAULT + cmd = ["meshtastic", "--host", host, "--traceroute", dest] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=40) + except Exception as e: + QTimer.singleShot( + 0, + lambda: QMessageBox.critical(self, "Traceroute", f"Failed to run meshtastic CLI: {e}"), + ) + return + out = (proc.stdout or "") + ("\n" + (proc.stderr or "") if proc.stderr else "") + if not out.strip(): + msg = "No output from meshtastic traceroute." + QTimer.singleShot( + 0, + lambda: (QMessageBox.information(self, "Traceroute", msg), self._show_traceroute_window(msg)), + ) + return + # Also log traceroute CLI output, then show it in a popup window + QTimer.singleShot(0, lambda: (self._append(out), self._show_traceroute_window(out))) + + def _show_traceroute_window(self, text: str) -> None: + dlg = QDialog(self) + dlg.setWindowTitle("Traceroute") + layout = QVBoxLayout(dlg) + view = QPlainTextEdit(dlg) + view.setReadOnly(True) + view.setPlainText(text.strip() or "No traceroute data.") + layout.addWidget(view) + btn = QPushButton("Close", dlg) + btn.clicked.connect(dlg.accept) + layout.addWidget(btn) + dlg.resize(700, 400) + dlg.exec() + + # ------------------------------------------------------------------ + # Node details & map + # ------------------------------------------------------------------ + def _cm_show_node_details(self) -> None: + self.show_raw_node(friendly=True) + + def _cm_open_map(self) -> None: + nid = self._get_selected_node_id() + iface = self.iface + if not nid or not iface or not getattr(iface, "nodes", None): + QMessageBox.information(self, "Map", "No node selected.") + return + node = getattr(iface, "nodes", {}).get(nid, {}) + lat, lon = self._extract_latlon(node) + if lat is None or lon is None: + QMessageBox.information(self, "Map", "Selected node has no GPS position.") + return + url = f"https://www.google.com/maps/search/?api=1&query={lat},{lon}" + self._open_browser_url(url) + + def show_raw_node(self, friendly: bool = False) -> None: + nid = self._get_selected_node_id() + iface = self.iface + if not nid or not iface or not getattr(iface, "nodes", None): + QMessageBox.information(self, "Node", "No node selected.") + return + node = getattr(iface, "nodes", {}).get(nid, {}) + + dlg = QDialog(self) + dlg.setWindowTitle(f"Node: {self._node_label(nid)}") + layout = QVBoxLayout(dlg) + txt = QPlainTextEdit(dlg) + txt.setReadOnly(True) + layout.addWidget(txt) + btn = QPushButton("Close", dlg) + btn.clicked.connect(dlg.accept) + layout.addWidget(btn) + + if friendly: + def fmt_val(v, indent=0): + pad = " " * indent + if isinstance(v, dict): + lines = [] + for k, vv in v.items(): + if isinstance(vv, (dict, list)): + lines.append(f"{pad}{k}:") + lines.append(fmt_val(vv, indent + 1)) + else: + lines.append(f"{pad}{k}: {vv}") + return "\n".join(lines) + elif isinstance(v, list): + lines = [] + for i, item in enumerate(v): + if isinstance(item, (dict, list)): + lines.append(f"{pad}- [{i}]") + lines.append(fmt_val(item, indent + 1)) + else: + lines.append(f"{pad}- {item}") + return "\n".join(lines) + else: + return f"{pad}{v}" + + user = (node or {}).get("user") or {} + pos = (node or {}).get("position") or {} + caps = (node or {}).get("capabilities") or {} + config = (node or {}).get("config") or {} + + node_id = user.get("id") or node.get("id") or nid + macaddr = user.get("macaddr") or node.get("macaddr") or "" + publickey = user.get("publicKey") or node.get("publicKey") or "" + hw = user.get("hwModel", "") + + lines = [ + f"Name: {user.get('shortName', '')} {user.get('longName', '')}".strip(), + f"ID: {node_id}", + f"MAC: {macaddr}", + f"HW: {hw}", + f"Public key: {publickey}", + "", + f"Last heard: {_fmt_ago(self._get_lastheard_epoch(nid, node))}", + "", + "Position:", + fmt_val(pos, 1), + ] + if caps: + lines.append("Capabilities:") + lines.append(fmt_val(caps, 1)) + if config: + lines.append("Config:") + lines.append(fmt_val(config, 1)) + lines.append("RAW fields:") + skip = {"user", "position", "capabilities", "config"} + other = {k: v for k, v in (node or {}).items() if k not in skip} + lines.append(fmt_val(other, 1)) + txt.setPlainText("\n".join(lines)) + else: + txt.setPlainText(json.dumps(node, indent=2, default=str)) + + dlg.resize(700, 500) + dlg.exec() + + # ------------------------------------------------------------------ + # Radio config (Tools menu) + # ------------------------------------------------------------------ + def show_radio_config_window(self) -> None: + iface = self.iface + if not iface or not getattr(iface, "localNode", None): + QMessageBox.information(self, "Radio config", "Connect to a device first.") + return + ln = getattr(iface, "localNode") + try: + if getattr(ln, "localConfig", None) is None and hasattr(ln, "waitForConfig"): + ln.waitForConfig("localConfig") + except Exception: + pass + cfg = getattr(ln, "localConfig", None) + mod = getattr(ln, "moduleConfig", None) + if cfg is None and mod is None: + QMessageBox.information(self, "Radio config", "No config available yet from device.") + return + + dlg = QDialog(self) + dlg.setWindowTitle("Radio + module config (read‑only)") + layout = QVBoxLayout(dlg) + txt = QPlainTextEdit(dlg) + txt.setReadOnly(True) + layout.addWidget(txt) + btn = QPushButton("Close", dlg) + btn.clicked.connect(dlg.accept) + layout.addWidget(btn) + + lines: List[str] = [] + if cfg is not None: + lines.append("localConfig:") + try: + lines.append(str(cfg)) + except Exception: + lines.append(repr(cfg)) + if mod is not None: + lines.append("") + lines.append("moduleConfig:") + try: + lines.append(str(mod)) + except Exception: + lines.append(repr(mod)) + + txt.setPlainText("\n".join(lines)) + dlg.resize(800, 600) + dlg.exec() + + # ------------------------------------------------------------------ + # Neighbors + channel editor + # ------------------------------------------------------------------ + def show_neighbors_window(self) -> None: + iface = self.iface + if not iface or not getattr(iface, "nodes", None): + QMessageBox.information(self, "Neighbors", "No interface or nodes available.") + return + + dlg = QDialog(self) + dlg.setWindowTitle("Neighbor table") + layout = QVBoxLayout(dlg) + table = QTableWidget(0, 4, dlg) + table.setHorizontalHeaderLabels(["From node", "To node", "SNR", "Last heard"]) + table.horizontalHeader().setStretchLastSection(True) + layout.addWidget(table) + + rows = 0 + try: + nodes_snapshot = dict(getattr(iface, "nodes", {}) or {}) + except Exception: + nodes_snapshot = {} + + for node_id, node in nodes_snapshot.items(): + from_label = self._node_label(str(node_id)) + neighbors = (node or {}).get("neighbors") or [] + if isinstance(neighbors, dict): + neighbors = neighbors.values() + for n in neighbors: + try: + to_id = (n or {}).get("nodeId") or (n or {}).get("id") or "" + to_label = self._node_label(str(to_id)) if to_id else str(to_id) + snr = (n or {}).get("snr") + last = (n or {}).get("lastHeard") + try: + last_epoch = float(last) if last is not None else None + except Exception: + last_epoch = None + last_str = _fmt_ago(last_epoch) + except Exception: + continue + + row = table.rowCount() + table.insertRow(row) + for col, val in enumerate([from_label, to_label, snr if snr is not None else "-", last_str]): + item = QTableWidgetItem(str(val)) + table.setItem(row, col, item) + rows += 1 + + if rows == 0: + dlg.close() + QMessageBox.information(self, "Neighbors", "No neighbor information found on nodes.") + return + + dlg.resize(700, 400) + dlg.exec() + + def show_channel_editor_window(self) -> None: + iface = self.iface + if not iface or not getattr(iface, "localNode", None): + QMessageBox.information(self, "Channels", "Connect to a device first.") + return + ln = getattr(iface, "localNode") + try: + if getattr(ln, "channels", None) in (None, {}) and hasattr(ln, "waitForConfig"): + ln.waitForConfig("channels") + except Exception: + pass + chans = getattr(ln, "channels", None) + if not chans: + QMessageBox.information(self, "Channels", "No channels available from device.") + return + + dlg = QDialog(self) + dlg.setWindowTitle("Channel editor") + layout = QVBoxLayout(dlg) + + row = QHBoxLayout() + list_widget = QListWidget(dlg) + row.addWidget(list_widget, 1) + + right = QVBoxLayout() + lbl = QLabel("Channel name:", dlg) + right.addWidget(lbl) + name_edit = QLineEdit(dlg) + right.addWidget(name_edit) + row.addLayout(right, 2) + + layout.addLayout(row) + + btn_row = QHBoxLayout() + btn_save = QPushButton("Save name to device", dlg) + btn_close = QPushButton("Close", dlg) + btn_row.addWidget(btn_save) + btn_row.addStretch(1) + btn_row.addWidget(btn_close) + layout.addLayout(btn_row) + + def _channel_name(ch) -> str: + try: + s = getattr(ch, "settings", None) + nm = (getattr(s, "name", "") or "").strip() if s is not None else "" + if nm: + return nm + except Exception: + pass + try: + nm = (getattr(ch, "name", "") or "").strip() + if nm: + return nm + except Exception: + pass + return "" + + for i, ch in enumerate(chans): + if ch is None: + label = f"{i}: (empty)" + else: + nm = _channel_name(ch) + label = f"{i}: {nm or '(no name)'}" + list_widget.addItem(label) + + def on_select() -> None: + idx = list_widget.currentRow() + if idx < 0 or idx >= len(chans): + name_edit.setText("") + return + ch = chans[idx] + if ch is None: + name_edit.setText("") + return + name_edit.setText(_channel_name(ch)) + + list_widget.currentRowChanged.connect(lambda _row: on_select()) + + def save_name() -> None: + idx = list_widget.currentRow() + if idx < 0 or idx >= len(chans): + QMessageBox.information(dlg, "Channel editor", "Select a channel first.") + return + ch = chans[idx] + if ch is None: + QMessageBox.information(dlg, "Channel editor", "Selected channel slot is empty.") + return + new_name = name_edit.text().strip() + try: + if getattr(ch, "settings", None) is not None and hasattr(ch.settings, "name"): + ch.settings.name = new_name + elif hasattr(ch, "name"): + ch.name = new_name + ln.writeChannel(idx) + self._append(f"[admin] Renamed channel {idx} to '{new_name}'") + self._update_channels_from_iface() + item = list_widget.item(idx) + if item is not None: + item.setText(f"{idx}: {new_name or '(no name)'}") + except Exception as e: + QMessageBox.critical(dlg, "Channel editor", f"Failed to write channel: {e}") + + btn_save.clicked.connect(save_name) + btn_close.clicked.connect(dlg.accept) + + dlg.resize(600, 400) + dlg.exec() + + # ------------------------------------------------------------------ + # Theme + # ------------------------------------------------------------------ + def apply_theme(self, mode: str = "light") -> None: + self.current_theme = mode + is_dark = mode == "dark" + bg = "#1e1e1e" if is_dark else "#f5f5f5" + fg = "#ffffff" if is_dark else "#000000" + acc = "#2d2d2d" if is_dark else "#ffffff" + sel = "#555555" if is_dark else "#cce0ff" + + style_sheet = f""" + QMainWindow {{ + background-color: {bg}; + color: {fg}; + }} + QWidget {{ + background-color: {bg}; + color: {fg}; + }} + QTextEdit, QPlainTextEdit {{ + background-color: {acc}; + color: {fg}; + }} + QLineEdit, QComboBox {{ + background-color: {acc}; + color: {fg}; + }} + QTableWidget {{ + background-color: {acc}; + color: {fg}; + gridline-color: #444444; + }} + QHeaderView::section {{ + background-color: {acc}; + color: {fg}; + }} + QPushButton {{ + background-color: {acc}; + color: {fg}; + }} + QMenuBar {{ + background-color: {bg}; + color: {fg}; + }} + QMenuBar::item:selected {{ + background-color: {sel}; + color: {fg}; + }} + QMenu {{ + background-color: {bg}; + color: {fg}; + }} + QMenu::item:selected {{ + background-color: {sel}; + color: {fg}; + }} + QTableWidget::item:selected {{ + background-color: {sel}; + color: {fg}; + }} + """ + self.setStyleSheet(style_sheet) + + # ------------------------------------------------------------------ + # Misc + # ------------------------------------------------------------------ + def _show_about(self) -> None: + QMessageBox.information( + self, + "About Meshtastic Client (v1)", + "Meshtastic Client (v1)\n\n" + "You can find the sourcecode on\nhttps://github.com/dk98174003/Meshtastic-Client\n\n" + "Have fun\n" + "Knud Schrøder ;O)", + ) + + def closeEvent(self, event) -> None: # type: ignore[override] + try: + self.disconnect() + finally: + super().closeEvent(event) + + +class NodeChatDialog(QDialog): + """Simple per-node chat window.""" + + def __init__(self, app: MeshtasticMainWindow, node_id: str, label: str) -> None: + super().__init__(app) + self.app = app + self.node_id = node_id + self.label = label + self.setWindowTitle(f"Chat: {label}") + + layout = QVBoxLayout(self) + self.txt = QPlainTextEdit(self) + self.txt.setReadOnly(True) + layout.addWidget(self.txt) + + row = QHBoxLayout() + self.entry = QLineEdit(self) + self.entry.setPlaceholderText("Type message and press Enter…") + row.addWidget(self.entry, 1) + self.btn_send = QPushButton("Send", self) + row.addWidget(self.btn_send) + layout.addLayout(row) + + self.btn_send.clicked.connect(self._send) + self.entry.returnPressed.connect(self._send) + + self.resize(500, 350) + + def append_line(self, line: str) -> None: + ts = datetime.datetime.now().strftime("%H:%M:%S") + self.txt.appendPlainText(f"[{ts}] {line}") + + def _send(self) -> None: + msg = self.entry.text().strip() + if not msg: + return + ok = self.app._send_text_to_node(self.node_id, msg) + if ok: + self.entry.clear() + + def closeEvent(self, event) -> None: # type: ignore[override] + key = str(self.node_id) + try: + if key in self.app._per_node_chats: + del self.app._per_node_chats[key] + except Exception: + pass + super().closeEvent(event) + + +def main() -> None: + app = QApplication(sys.argv) + win = MeshtasticMainWindow() + win.resize(1500, 820) + win.show() + sys.exit(app.exec()) + + +if __name__ == "__main__": + main() diff --git a/LinuxMintQT/rebuilddeb.sh b/LinuxMintQT/rebuilddeb.sh new file mode 100644 index 0000000..9fe74d2 --- /dev/null +++ b/LinuxMintQT/rebuilddeb.sh @@ -0,0 +1,159 @@ +#!/bin/bash +# +# Rebuild .deb for Meshtastic client +# - Installs a private venv with PySide6 + meshtastic + pubsub +# - Installs launcher, icon and .desktop file +# +# Requirements on the build machine: +# sudo apt install python3-venv ruby-full +# sudo gem install --no-document fpm +# +# Run from anywhere: +# cd /home/knud/Desktop/Meshtastic/Meshtastic_client +# ./rebuilddeb.sh +# + +set -e + +### CONFIG ############################################################ + +APP_NAME="meshtastic-client" +APP_VERSION=$(date +"1.0.%Y%m%d%H%M") +PROJECT_DIR="/home/knud/Desktop/Meshtastic/Meshtastic_client" +BUILD_DIR="$PROJECT_DIR/build" +ICON_NAME="meshtastic.png" +MAIN_PY_QT="meshtastic_client.py" + +###################################################################### + +echo "=== Building $APP_NAME version $APP_VERSION (v1) ===" +cd "$PROJECT_DIR" || exit 1 + +### CLEAN OLD BUILD + DEB ############################################ + +echo "[1/9] Cleaning old build + .deb files..." +rm -rf "$BUILD_DIR" +rm -f "$PROJECT_DIR"/*.deb + +### CREATE FOLDER STRUCTURE ########################################## + +echo "[2/9] Creating folder structure in $BUILD_DIR ..." + +# Program + venv +mkdir -p "$BUILD_DIR/usr/share/$APP_NAME" + +# Launcher +mkdir -p "$BUILD_DIR/usr/bin" + +# Desktop file +mkdir -p "$BUILD_DIR/usr/share/applications" + +# Icon (512x512; other sizes can be added later) +mkdir -p "$BUILD_DIR/usr/share/icons/hicolor/512x512/apps" + +### COPY PROGRAM FILES ################################################ + +echo "[3/9] Copying program + icon..." + +if [ ! -f "$PROJECT_DIR/$MAIN_PY_QT" ]; then + echo "ERROR: $MAIN_PY_QT not found in $PROJECT_DIR" + exit 1 +fi + +cp "$PROJECT_DIR/$MAIN_PY_QT" "$BUILD_DIR/usr/share/$APP_NAME/" + +if [ -f "$PROJECT_DIR/$ICON_NAME" ]; then + cp "$PROJECT_DIR/$ICON_NAME" \ + "$BUILD_DIR/usr/share/icons/hicolor/512x512/apps/${APP_NAME}.png" +else + echo "WARNING: $ICON_NAME not found – package will build without icon." +fi + +### CREATE VENV + INSTALL PYTHON DEPENDENCIES ######################## + +echo "[4/9] Creating venv and installing Python dependencies..." + +VENV_PATH="$BUILD_DIR/usr/share/$APP_NAME/venv" + +python3 -m venv "$VENV_PATH" + +# shellcheck disable=SC1090 +source "$VENV_PATH/bin/activate" + +pip install --upgrade pip + +# Core libs for this app +pip install meshtastic PySide6 pubsub + +deactivate + +### CREATE LAUNCHER SCRIPT ########################################### + +echo "[5/9] Creating launcher script..." + +LAUNCHER="$BUILD_DIR/usr/bin/$APP_NAME" + +cat > "$LAUNCHER" <&2 + exit 1 +fi + +exec "\$VENV/bin/python" "\$APP_DIR/$MAIN_PY_QT" "\$@" +EOF + +chmod +x "$LAUNCHER" + +### CREATE .DESKTOP FILE ############################################# + +echo "[6/9] Creating .desktop file..." + +DESKTOP_FILE="$BUILD_DIR/usr/share/applications/${APP_NAME}.desktop" + +cat > "$DESKTOP_FILE" </dev/null 2>&1 && tree "$BUILD_DIR" || true + +### BUILD .DEB PACKAGE ############################################### + +echo "[9/9] Building .deb using fpm..." + +fpm -s dir -t deb -n "$APP_NAME" -v "$APP_VERSION" \ + --description "Meshtastic Desktop Client (v1)" \ + --architecture amd64 \ + -C "$BUILD_DIR" \ + . + +echo "======================================================" +echo " Build finished!" +echo " Created package: ${APP_NAME}_${APP_VERSION}_amd64.deb" +echo "======================================================" + diff --git a/LinuxMintQT/run-client b/LinuxMintQT/run-client new file mode 100644 index 0000000..78d855d --- /dev/null +++ b/LinuxMintQT/run-client @@ -0,0 +1,7 @@ +#!/bin/bash +cd /home/knud/Desktop/Meshtastic/Meshtastic_client || exit +set -a +source .envrc +set +a +python3 meshtastic_client.py +