adam/meshcore-gui
1
0
Fork 0
mirror of https://github.com/pe1hvh/meshcore-gui.git synced 2026-08-08 09:52:58 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(channels): add hashtag and private channel management dialog

Browse Source 
WHAT: New ＋ Add Channel button in the Messages submenu opens a dialog
supporting three modes — Hashtag (key derived from name), Private New
(random key + QR export), Private Existing (paste hex key).

WHY: Channels could only be added via firmware/external tools. The new
dialog covers all user scenarios without changing the BLE worker or
existing panels.

NOTES: Requires `qrcode[pil]` for QR rendering (graceful degradation if
absent). Channel re-discovery is triggered automatically on success.
This commit is contained in:
pe1hvh
2026-04-04 11:40:14 +02:00
parent 324572f4b3
commit a62b8d1733
12 changed files with 599 additions and 1348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+36
View File
@@ -10,6 +10,42 @@ Format follows [Keep a Changelog](https://keepachangelog.com/) and [Semantic Ver
---
## [1.16.0] - 2026-04-04
### ADDED
- **Add Channel dialog** (`gui/panels/channel_panel.py`): new `ChannelPanel` class
that renders a `ui.dialog` with three modes — Hashtag, Private New, and
Private Existing — accessible via a `＋ Add Channel` button at the bottom of
the Messages submenu.
- **ChannelService** (`services/channel_service.py`): pure-Python business logic
for channel key management. Provides `generate_secret()`, `derive_hashtag_key()`,
`build_qr_url()` and `generate_qr_base64()`. No GUI or BLE dependencies.
- **`_cmd_add_channel`** (`ble/commands.py`): new BLE command handler that calls
`mc.commands.set_channel(idx, name, secret)` and triggers a full channel
re-discovery on success so the GUI immediately reflects the new channel.
- **`＋ Add Channel` submenu button** (`gui/dashboard.py`): added to the Messages
submenu — present on initial render and preserved through all dynamic rebuilds.
- **QR code sharing**: after adding a new private channel the dialog shows a
scannable QR code (`meshcore://channel/add?name=…&secret=…`) and a
copy-to-clipboard button for the hex key.
### CHANGED
- `gui/panels/__init__.py`: `ChannelPanel` added to re-exports.
- `gui/dashboard.py`: `ChannelPanel` imported, instantiated in `render()`,
updated in `_update_ui()` and opened from the `＋ Add Channel` submenu button.
- `ble/commands.py`: `add_channel` registered in the handler dict.
### RATIONALE
- Channels could previously only be added by flashing or using a separate tool.
The dialog covers all three user scenarios (hashtag join, private create, private
join) without requiring any changes to the BLE worker's discovery logic.
### IMPACT
- No existing command handlers modified. No existing panel logic altered.
Pure addition — all existing functionality unaffected.
---
## [1.15.0] - 2026-03-16
### ADDED
README.md
+14 -1
View File
@@ -119,6 +119,7 @@ Under the hood it uses `meshcore` as the protocol layer, `meshcoredecoder` for r
<!-- ADDED: Room Server feature (v5.7.0) -->
- **Dynamic Channel Discovery** — Channels are automatically discovered from the device at startup via probing, eliminating the need to manually configure `CHANNELS_CONFIG`
<!-- ADDED: Dynamic channel discovery (v5.7.0) -->
- **Add Channel** — Add hashtag or private channels directly from the GUI via the `＋ Add Channel` button in the Messages submenu. New private channels generate a shareable QR code and hex key for distribution to other users
- **Keyword Bot** — Built-in auto-reply bot that responds to configurable keywords on selected channels, with cooldown and loop prevention
- **Packet Decoding** — Raw LoRa packets from RX log are decoded and decrypted using channel keys, providing message hashes, path hashes and hop data
- **Message Deduplication** — Dual-strategy dedup (hash-based and content-based) prevents duplicate messages from appearing
@@ -304,7 +305,19 @@ Channels are automatically discovered from the device at startup via the serial
If you want to cache the discovered channel list to disk (for faster startup), set `CHANNEL_CACHE_ENABLED = True` in `meshcore_gui/config.py`. By default, channels are always fetched fresh from the device.
> **Note:** The maximum number of channel slots probed can be adjusted via `MAX_CHANNELS` in `config.py` (default: 8, which matches the MeshCore protocol limit).
> **Note:** The maximum number of channel slots probed can be adjusted via `MAX_CHANNELS` in `config.py` (default: 100).
#### Adding channels from the GUI
Open the **Messages** section in the left drawer and click **＋ Add Channel** at the bottom of the channel list. A dialog appears with three modes:
| Mode | When to use | Key input |
|---|---|---|
| **# Hashtag** | Join a public topic channel (e.g. `#localmesh`) | None — key is derived from the name |
| **🔒 Private – New** | Create a new private group channel | Click *Generate key*, then share the QR or hex key |
| **🔒 Private – Existing** | Join a private channel someone shared with you | Paste the 32-char hex key |
After clicking **Add Channel**, the device is updated and channel discovery re-runs automatically. A new private channel shows a scannable QR code (`meshcore://channel/add?name=…&secret=…`) that the official MeshCore app can read directly.
### 6.4. Start the GUI
meshcore_gui/ble/commands.py
+55
View File
@@ -51,6 +51,7 @@ class CommandHandler:
'logout_room': self._cmd_logout_room,
'send_room_msg': self._cmd_send_room_msg,
'load_room_history': self._cmd_load_room_history,
'add_channel': self._cmd_add_channel,
}
async def process_all(self) -> None:
@@ -547,6 +548,60 @@ class CommandHandler:
)
debug_print(f"send_room_msg exception: {exc}")
async def _cmd_add_channel(self, cmd: Dict) -> None:
"""Add or update a channel slot on the MeshCore device.
Calls ``mc.commands.set_channel()`` and — on success — triggers a
full channel re-discovery so the GUI immediately reflects the new
channel in the submenu and filter checkboxes.
The library's ``set_channel`` handles two cases automatically:
- ``channel_name.startswith('#')`` or ``secret=None`` → key derived
from ``SHA-256(name)[:16]`` (hashtag channel).
- Explicit 16-byte ``secret`` → used verbatim (private channel).
Expected command dict::
{
'action': 'add_channel',
'idx': int, # target channel slot (1–99)
'name': str, # channel name
'secret_hex': str, # 32-char hex for private; '' for hashtag
}
"""
idx: int = int(cmd.get('idx', 1))
name: str = (cmd.get('name') or '').strip()
secret_hex: str = (cmd.get('secret_hex') or '').strip()
if not name:
debug_print('add_channel: no name provided, skipping')
return
# Resolve secret: empty string → None (library derives from name)
secret_bytes: Optional[bytes] = None
if secret_hex:
try:
secret_bytes = bytes.fromhex(secret_hex)
except ValueError:
self._shared.set_status('⚠️ Invalid channel secret (not valid hex)')
debug_print(f'add_channel: bad hex secret for [{idx}] {name}')
return
try:
r = await self._mc.commands.set_channel(idx, name, secret_bytes)
if r.type == EventType.ERROR:
self._shared.set_status(f"⚠️ Failed to add channel '{name}'")
debug_print(f'add_channel: device returned ERROR for [{idx}] {name}')
else:
self._shared.set_status(f"✅ Channel [{idx}] '{name}' added")
debug_print(f'add_channel: success [{idx}] {name}')
# Re-discover channels so the GUI updates immediately
if self._load_data_callback:
await self._load_data_callback()
except Exception as exc:
self._shared.set_status(f'⚠️ Add channel error: {exc}')
debug_print(f'add_channel exception: {exc}')
# ------------------------------------------------------------------
# Callback for refresh (set by SerialWorker after construction)
# ------------------------------------------------------------------
meshcore_gui/ble/events.py.bak
-379
View File
@@ -1,379 +0,0 @@
"""
Device event callbacks for MeshCore GUI.
Handles ``CHANNEL_MSG_RECV``, ``CONTACT_MSG_RECV`` and ``RX_LOG_DATA``
events from the MeshCore library. Extracted from ``SerialWorker`` so the
worker only deals with connection lifecycle.
"""
from typing import Dict, Optional
from meshcore_gui.config import debug_print
from meshcore_gui.core.models import Message, RxLogEntry
from meshcore_gui.core.protocols import SharedDataWriter
from meshcore_gui.ble.packet_decoder import PacketDecoder, PayloadType
from meshcore_gui.services.bot import MeshBot
from meshcore_gui.services.dedup import DualDeduplicator
class EventHandler:
"""Processes device events and writes results to shared data.
Args:
shared: SharedDataWriter for storing messages and RX log.
decoder: PacketDecoder for raw LoRa packet decryption.
dedup: DualDeduplicator for message deduplication.
bot: MeshBot for auto-reply logic.
"""
# Maximum entries in the path cache before oldest are evicted.
_PATH_CACHE_MAX = 200
def __init__(
self,
shared: SharedDataWriter,
decoder: PacketDecoder,
dedup: DualDeduplicator,
bot: MeshBot,
) -> None:
self._shared = shared
self._decoder = decoder
self._dedup = dedup
self._bot = bot
# Cache: message_hash → path_hashes (from RX_LOG decode).
# Used by on_channel_msg fallback to recover hashes that the
# CHANNEL_MSG_RECV event does not provide.
self._path_cache: Dict[str, list] = {}
# ------------------------------------------------------------------
# Helpers — resolve names at receive time
# ------------------------------------------------------------------
def _resolve_path_names(self, path_hashes: list) -> list:
"""Resolve 2-char path hashes to display names.
Performs a contact lookup for each hash *now* so the names are
captured at receive time and stored in the archive.
Args:
path_hashes: List of 2-char hex strings.
Returns:
List of display names (same length as *path_hashes*).
Unknown hashes become their uppercase hex value.
"""
names = []
for h in path_hashes:
if not h or len(h) < 2:
names.append('-')
continue
name = self._shared.get_contact_name_by_prefix(h)
# get_contact_name_by_prefix returns h[:8] as fallback,
# normalise to uppercase hex for 2-char hashes.
if name and name != h[:8]:
names.append(name)
else:
names.append(h.upper())
return names
# ------------------------------------------------------------------
# RX_LOG_DATA — the single source of truth for path info
# ------------------------------------------------------------------
def on_rx_log(self, event) -> None:
"""Handle RX log data events."""
payload = event.payload
# Extract basic RX log info
time_str = Message.now_timestamp()
snr = payload.get('snr', 0)
rssi = payload.get('rssi', 0)
payload_type = '?'
hops = payload.get('path_len', 0)
# Try to decode payload to get message_hash
message_hash = ""
rx_path_hashes: list = []
rx_path_names: list = []
rx_sender: str = ""
rx_receiver: str = self._shared.get_device_name() or ""
payload_hex = payload.get('payload', '')
decoded = None
if payload_hex:
decoded = self._decoder.decode(payload_hex)
if decoded is not None:
message_hash = decoded.message_hash
payload_type = self._decoder.get_payload_type_text(decoded.payload_type)
# Capture path info for all packet types
if decoded.path_hashes:
rx_path_hashes = decoded.path_hashes
rx_path_names = self._resolve_path_names(decoded.path_hashes)
# Use decoded path_length (from packet body) — more
# reliable than the frame-header path_len which can be 0.
if decoded.path_length:
hops = decoded.path_length
# Capture sender name when available (GroupText only)
if decoded.sender:
rx_sender = decoded.sender
# Cache path_hashes for correlation with on_channel_msg
if decoded.path_hashes and message_hash:
self._path_cache[message_hash] = decoded.path_hashes
# Evict oldest entries if cache is too large
if len(self._path_cache) > self._PATH_CACHE_MAX:
oldest = next(iter(self._path_cache))
del self._path_cache[oldest]
# Process decoded message if it's a group text
if decoded.payload_type == PayloadType.GroupText and decoded.is_decrypted:
if decoded.channel_idx is None:
# The channel hash could not be resolved to a channel index
# (PacketDecoder._hash_to_idx lookup returned None).
# Marking dedup here would suppress on_channel_msg, which
# carries a valid channel_idx from the device event — the only
# path through which the bot can pass Guard 2 and respond.
# Skip the entire block; on_channel_msg handles message + bot.
# Path info is already in _path_cache for on_channel_msg to use.
debug_print(
f"RX_LOG → GroupText decrypted but channel_idx unresolved "
f"(hash={decoded.message_hash}); deferring to on_channel_msg"
)
else:
self._dedup.mark_hash(decoded.message_hash)
self._dedup.mark_content(
decoded.sender, decoded.channel_idx, decoded.text,
)
sender_pubkey = ''
if decoded.sender:
match = self._shared.get_contact_by_name(decoded.sender)
if match:
sender_pubkey, _contact = match
snr_msg = self._extract_snr(payload)
self._shared.add_message(Message.incoming(
decoded.sender,
decoded.text,
decoded.channel_idx,
time=time_str,
snr=snr_msg,
path_len=decoded.path_length,
sender_pubkey=sender_pubkey,
path_hashes=decoded.path_hashes,
path_names=rx_path_names,
message_hash=decoded.message_hash,
))
debug_print(
f"RX_LOG → message: hash={decoded.message_hash}, "
f"sender={decoded.sender!r}, ch={decoded.channel_idx}, "
f"path={decoded.path_hashes}, "
f"path_names={rx_path_names}"
)
self._bot.check_and_reply(
sender=decoded.sender,
text=decoded.text,
channel_idx=decoded.channel_idx,
snr=snr_msg,
path_len=decoded.path_length,
path_hashes=decoded.path_hashes,
)
# Add RX log entry with message_hash and path info (if available)
# ── Fase 1 Observer: raw packet metadata ──
raw_packet_len = len(payload_hex) // 2 if payload_hex else 0
raw_payload_len = max(0, raw_packet_len - 1 - hops) if payload_hex else 0
raw_route_type = "D" if hops > 0 else ("F" if payload_hex else "")
raw_packet_type_num = -1
if payload_hex and decoded is not None:
try:
raw_packet_type_num = decoded.payload_type.value
except (AttributeError, ValueError):
pass
self._shared.add_rx_log(RxLogEntry(
time=time_str,
snr=snr,
rssi=rssi,
payload_type=payload_type,
hops=hops,
message_hash=message_hash,
path_hashes=rx_path_hashes,
path_names=rx_path_names,
sender=rx_sender,
receiver=rx_receiver,
raw_payload=payload_hex,
packet_len=raw_packet_len,
payload_len=raw_payload_len,
route_type=raw_route_type,
packet_type_num=raw_packet_type_num,
))
# ------------------------------------------------------------------
# CHANNEL_MSG_RECV — fallback when RX_LOG decode missed it
# ------------------------------------------------------------------
def on_channel_msg(self, event) -> None:
"""Handle channel message events."""
payload = event.payload
debug_print(f"Channel msg payload keys: {list(payload.keys())}")
# Dedup via hash
msg_hash = payload.get('message_hash', '')
if msg_hash and self._dedup.is_hash_seen(msg_hash):
debug_print(f"Channel msg suppressed (hash): {msg_hash}")
return
# Parse sender from "SenderName: message body" format
raw_text = payload.get('text', '')
sender, msg_text = '', raw_text
if ': ' in raw_text:
name_part, body_part = raw_text.split(': ', 1)
sender = name_part.strip()
msg_text = body_part
elif raw_text:
msg_text = raw_text
# Dedup via content
ch_idx = payload.get('channel_idx')
if self._dedup.is_content_seen(sender, ch_idx, msg_text):
debug_print(f"Channel msg suppressed (content): {sender!r}")
return
debug_print(
f"Channel msg (fallback): sender={sender!r}, "
f"text={msg_text[:40]!r}"
)
sender_pubkey = ''
if sender:
match = self._shared.get_contact_by_name(sender)
if match:
sender_pubkey, _contact = match
snr = self._extract_snr(payload)
# Recover path_hashes from RX_LOG cache (CHANNEL_MSG_RECV
# does not carry them, but the preceding RX_LOG decode does).
path_hashes = self._path_cache.pop(msg_hash, []) if msg_hash else []
path_names = self._resolve_path_names(path_hashes)
self._shared.add_message(Message.incoming(
sender,
msg_text,
ch_idx,
snr=snr,
path_len=payload.get('path_len', 0),
sender_pubkey=sender_pubkey,
path_hashes=path_hashes,
path_names=path_names,
message_hash=msg_hash,
))
self._bot.check_and_reply(
sender=sender,
text=msg_text,
channel_idx=ch_idx,
snr=snr,
path_len=payload.get('path_len', 0),
)
# ------------------------------------------------------------------
# CONTACT_MSG_RECV — DMs
# ------------------------------------------------------------------
def on_contact_msg(self, event) -> None:
"""Handle direct message and room message events.
Room Server messages arrive as ``CONTACT_MSG_RECV`` with
``txt_type == 2``. The ``pubkey_prefix`` is the Room Server's
key and the ``signature`` field contains the original author's
pubkey prefix. We resolve the author name from ``signature``
so the UI shows who actually wrote the message.
"""
payload = event.payload
pubkey = payload.get('pubkey_prefix', '')
txt_type = payload.get('txt_type', 0)
signature = payload.get('signature', '')
debug_print(f"DM payload keys: {list(payload.keys())}")
# Common fields for both Room and DM messages
msg_hash = payload.get('message_hash', '')
path_hashes = self._path_cache.pop(msg_hash, []) if msg_hash else []
path_names = self._resolve_path_names(path_hashes)
# DM payloads may report path_len=255 (0xFF) meaning "unknown";
# treat as 0 when no actual path data is available.
raw_path_len = payload.get('path_len', 0)
path_len = raw_path_len if raw_path_len < 255 else 0
if path_hashes:
# Trust actual decoded hashes over the raw header value
path_len = len(path_hashes)
# --- Room Server message (txt_type 2) ---
if txt_type == 2 and signature:
# Resolve actual author from signature (author pubkey prefix)
author = self._shared.get_contact_name_by_prefix(signature)
if not author:
author = signature[:8] if signature else '?'
self._shared.add_message(Message.incoming(
author,
payload.get('text', ''),
None,
snr=self._extract_snr(payload),
path_len=path_len,
sender_pubkey=pubkey,
path_hashes=path_hashes,
path_names=path_names,
message_hash=msg_hash,
))
debug_print(
f"Room msg from {author} (sig={signature}) "
f"via room {pubkey[:12]}: "
f"{payload.get('text', '')[:30]}"
)
return
# --- Regular DM ---
sender = ''
if pubkey:
sender = self._shared.get_contact_name_by_prefix(pubkey)
if not sender:
sender = pubkey[:8] if pubkey else ''
self._shared.add_message(Message.incoming(
sender,
payload.get('text', ''),
None,
snr=self._extract_snr(payload),
path_len=path_len,
sender_pubkey=pubkey,
path_hashes=path_hashes,
path_names=path_names,
message_hash=msg_hash,
))
debug_print(f"DM received from {sender}: {payload.get('text', '')[:30]}")
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _extract_snr(payload: Dict) -> Optional[float]:
"""Extract SNR from a payload dict (handles 'SNR' and 'snr' keys)."""
raw = payload.get('SNR') or payload.get('snr')
if raw is not None:
try:
return float(raw)
except (ValueError, TypeError):
pass
return None
meshcore_gui/ble/worker.py.bak
-964
View File
@@ -1,964 +0,0 @@
"""
Communication worker for MeshCore GUI (Serial + BLE).
Runs in a separate thread with its own asyncio event loop. Connects
to the MeshCore device, wires up collaborators, and runs the command
processing loop.
Transport selection
~~~~~~~~~~~~~~~~~~~~
The :func:`create_worker` factory returns the appropriate worker class
based on the device identifier:
- ``/dev/ttyACM0`` → :class:`SerialWorker` (USB serial)
- ``literal:AA:BB:CC:DD:EE:FF`` → :class:`BLEWorker` (Bluetooth LE)
Both workers share the same base class (:class:`_BaseWorker`) which
implements the main loop, event wiring, data loading and caching.
Command execution → :mod:`meshcore_gui.ble.commands`
Event handling → :mod:`meshcore_gui.ble.events`
Packet decoding → :mod:`meshcore_gui.ble.packet_decoder`
PIN agent (BLE) → :mod:`meshcore_gui.ble.ble_agent`
Reconnect (BLE) → :mod:`meshcore_gui.ble.ble_reconnect`
Bot logic → :mod:`meshcore_gui.services.bot`
Deduplication → :mod:`meshcore_gui.services.dedup`
Cache → :mod:`meshcore_gui.services.cache`
Author: PE1HVH
SPDX-License-Identifier: MIT
"""
import abc
import asyncio
import threading
import time
from typing import Dict, List, Optional, Set
from meshcore import MeshCore, EventType
import meshcore_gui.config as _config
from meshcore_gui.config import (
DEFAULT_TIMEOUT,
CHANNEL_CACHE_ENABLED,
CONTACT_REFRESH_SECONDS,
MAX_CHANNELS,
RECONNECT_BASE_DELAY,
RECONNECT_MAX_RETRIES,
debug_data,
debug_print,
pp,
)
from meshcore_gui.core.protocols import SharedDataWriter
from meshcore_gui.ble.commands import CommandHandler
from meshcore_gui.ble.events import EventHandler
from meshcore_gui.ble.packet_decoder import PacketDecoder
from meshcore_gui.services.bot import BotConfig, MeshBot
from meshcore_gui.services.cache import DeviceCache
from meshcore_gui.services.dedup import DualDeduplicator
from meshcore_gui.services.device_identity import write_device_identity
# Seconds between background retry attempts for missing channel keys.
KEY_RETRY_INTERVAL: float = 30.0
# Seconds between periodic cleanup of old archived data (24 hours).
CLEANUP_INTERVAL: float = 86400.0
# ======================================================================
# Factory
# ======================================================================
def create_worker(device_id: str, shared: SharedDataWriter, **kwargs):
"""Return the appropriate worker for *device_id*.
Keyword arguments are forwarded to the worker constructor
(e.g. ``baudrate``, ``cx_dly`` for serial).
"""
from meshcore_gui.config import is_ble_address
if is_ble_address(device_id):
return BLEWorker(device_id, shared)
return SerialWorker(
device_id,
shared,
baudrate=kwargs.get("baudrate", _config.SERIAL_BAUDRATE),
cx_dly=kwargs.get("cx_dly", _config.SERIAL_CX_DELAY),
)
# ======================================================================
# Base worker (shared by BLE and Serial)
# ======================================================================
class _BaseWorker(abc.ABC):
"""Abstract base for transport-specific workers.
Subclasses must implement:
- :pyattr:`_log_prefix` — ``"BLE"`` or ``"SERIAL"``
- :meth:`_async_main` — transport-specific startup + main loop
- :meth:`_connect` — create the :class:`MeshCore` connection
- :meth:`_reconnect` — re-establish after a disconnect
- :pyattr:`_disconnect_keywords` — error substrings that signal
a broken connection
"""
def __init__(self, device_id: str, shared: SharedDataWriter) -> None:
self.device_id = device_id
self.shared = shared
self.mc: Optional[MeshCore] = None
self.running = True
self._disconnected = False
# Local cache (one file per device)
self._cache = DeviceCache(device_id)
# Collaborators (created eagerly, wired after connection)
self._decoder = PacketDecoder()
self._dedup = DualDeduplicator(max_size=200)
self._bot = MeshBot(
config=BotConfig(),
command_sink=shared.put_command,
enabled_check=shared.is_bot_enabled,
)
# Channel indices that still need keys from device
self._pending_keys: Set[int] = set()
# Dynamically discovered channels from device
self._channels: List[Dict] = []
# ── abstract properties / methods ─────────────────────────────
@property
@abc.abstractmethod
def _log_prefix(self) -> str:
"""Short label for log messages, e.g. ``"BLE"`` or ``"SERIAL"``."""
@property
@abc.abstractmethod
def _disconnect_keywords(self) -> tuple:
"""Lowercase substrings that indicate a transport disconnect."""
@abc.abstractmethod
async def _async_main(self) -> None:
"""Transport-specific startup + main loop."""
@abc.abstractmethod
async def _connect(self) -> None:
"""Create a fresh connection and wire collaborators."""
@abc.abstractmethod
async def _reconnect(self) -> Optional[MeshCore]:
"""Attempt to re-establish the connection after a disconnect."""
# ── thread lifecycle ──────────────────────────────────────────
def start(self) -> None:
"""Start the worker in a new daemon thread."""
thread = threading.Thread(target=self._run, daemon=True)
thread.start()
debug_print(f"{self._log_prefix} worker thread started")
def _run(self) -> None:
asyncio.run(self._async_main())
# ── shared main loop (called from subclass _async_main) ───────
async def _main_loop(self) -> None:
"""Command processing + periodic tasks.
Runs until ``self.running`` is cleared or a disconnect is
detected. Subclasses call this from their ``_async_main``.
"""
last_contact_refresh = time.time()
last_key_retry = time.time()
last_cleanup = time.time()
while self.running and not self._disconnected:
try:
await self._cmd_handler.process_all()
except Exception as e:
error_str = str(e).lower()
if any(kw in error_str for kw in self._disconnect_keywords):
print(f"{self._log_prefix}: ⚠️ Connection error detected: {e}")
self._disconnected = True
break
debug_print(f"Command processing error: {e}")
now = time.time()
if now - last_contact_refresh > CONTACT_REFRESH_SECONDS:
await self._refresh_contacts()
last_contact_refresh = now
if self._pending_keys and now - last_key_retry > KEY_RETRY_INTERVAL:
await self._retry_missing_keys()
last_key_retry = now
if now - last_cleanup > CLEANUP_INTERVAL:
await self._cleanup_old_data()
last_cleanup = now
await asyncio.sleep(0.1)
async def _handle_reconnect(self) -> bool:
"""Shared reconnect logic after a disconnect.
Returns True if reconnection succeeded, False otherwise.
"""
self.shared.set_connected(False)
self.shared.set_status("🔄 Verbinding verloren — herverbinden...")
print(f"{self._log_prefix}: Verbinding verloren, start reconnect...")
self.mc = None
new_mc = await self._reconnect()
if new_mc:
self.mc = new_mc
await asyncio.sleep(1)
self._wire_collaborators()
await self._load_data()
await self.mc.start_auto_message_fetching()
self._seed_dedup_from_messages()
self.shared.set_connected(True)
self.shared.set_status("✅ Herverbonden")
print(f"{self._log_prefix}: ✅ Herverbonden en operationeel")
return True
self.shared.set_status("❌ Herverbinding mislukt — herstart nodig")
print(
f"{self._log_prefix}: ❌ Kan niet herverbinden — "
"wacht 60s en probeer opnieuw..."
)
return False
# ── collaborator wiring ───────────────────────────────────────
def _wire_collaborators(self) -> None:
"""(Re-)create handlers and subscribe to MeshCore events."""
self._evt_handler = EventHandler(
shared=self.shared,
decoder=self._decoder,
dedup=self._dedup,
bot=self._bot,
)
self._cmd_handler = CommandHandler(
mc=self.mc, shared=self.shared, cache=self._cache,
)
self._cmd_handler.set_load_data_callback(self._load_data)
self.mc.subscribe(EventType.CHANNEL_MSG_RECV, self._evt_handler.on_channel_msg)
self.mc.subscribe(EventType.CONTACT_MSG_RECV, self._evt_handler.on_contact_msg)
self.mc.subscribe(EventType.RX_LOG_DATA, self._evt_handler.on_rx_log)
self.mc.subscribe(EventType.LOGIN_SUCCESS, self._on_login_success)
# ── LOGIN_SUCCESS handler (Room Server) ───────────────────────
def _on_login_success(self, event) -> None:
payload = event.payload or {}
pubkey = payload.get("pubkey_prefix", "")
is_admin = payload.get("is_admin", False)
debug_print(f"LOGIN_SUCCESS received: pubkey={pubkey}, admin={is_admin}")
self.shared.set_status("✅ Room login OK — messages arriving over RF…")
# ── apply cache ───────────────────────────────────────────────
def _apply_cache(self) -> None:
"""Push cached data to SharedData so GUI renders immediately."""
device = self._cache.get_device()
if device:
self.shared.update_from_appstart(device)
fw = device.get("firmware_version") or device.get("ver")
if fw:
self.shared.update_from_device_query({"ver": fw})
self.shared.set_status("📦 Loaded from cache")
debug_print(f"Cache → device info: {device.get('name', '?')}")
if CHANNEL_CACHE_ENABLED:
channels = self._cache.get_channels()
if channels:
self._channels = channels
self.shared.set_channels(channels)
debug_print(f"Cache → channels: {[c['name'] for c in channels]}")
else:
debug_print("Channel cache disabled — skipping cached channels")
contacts = self._cache.get_contacts()
if contacts:
self.shared.set_contacts(contacts)
debug_print(f"Cache → contacts: {len(contacts)}")
cached_keys = self._cache.get_channel_keys()
for idx_str, secret_hex in cached_keys.items():
try:
idx = int(idx_str)
secret_bytes = bytes.fromhex(secret_hex)
if len(secret_bytes) >= 16:
self._decoder.add_channel_key(idx, secret_bytes[:16], source="cache")
debug_print(f"Cache → channel key [{idx}]")
except (ValueError, TypeError) as exc:
debug_print(f"Cache → bad channel key [{idx_str}]: {exc}")
cached_orig_name = self._cache.get_original_device_name()
if cached_orig_name:
self.shared.set_original_device_name(cached_orig_name)
debug_print(f"Cache → original device name: {cached_orig_name}")
count = self.shared.load_recent_from_archive(limit=100)
if count:
debug_print(f"Cache → {count} recent messages from archive")
self._seed_dedup_from_messages()
# ── initial data loading ──────────────────────────────────────
async def _export_device_identity(self) -> None:
"""Export device keys and write identity file for Observer.
Calls ``export_private_key()`` on the device and writes the
result to ``~/.meshcore-gui/device_identity.json`` so the
MeshCore Observer can authenticate to the MQTT broker without
manual key configuration.
"""
pfx = self._log_prefix
try:
r = await self.mc.commands.export_private_key()
if r is None:
debug_print(f"{pfx}: export_private_key returned None")
return
if r.type == EventType.PRIVATE_KEY:
prv_bytes = r.payload.get("private_key", b"")
if len(prv_bytes) == 64:
# Gather device info for the identity file
pub_key = ""
dev_name = ""
fw_ver = ""
with self.shared.lock:
pub_key = self.shared.device.public_key
dev_name = self.shared.device.name
fw_ver = self.shared.device.firmware_version
write_device_identity(
public_key=pub_key,
private_key_bytes=prv_bytes,
device_name=dev_name,
firmware_version=fw_ver,
source_device=self.device_id,
)
else:
debug_print(
f"{pfx}: export_private_key: unexpected "
f"length {len(prv_bytes)} bytes"
)
elif r.type == EventType.DISABLED:
print(
f"{pfx}: ℹ️ Private key export is disabled on device "
f"— manual key setup required for Observer MQTT"
)
else:
debug_print(
f"{pfx}: export_private_key: unexpected "
f"response type {r.type}"
)
except Exception as exc:
debug_print(f"{pfx}: export_private_key failed: {exc}")
async def _load_data(self) -> None:
"""Load device info, channels and contacts from device."""
pfx = self._log_prefix
# send_appstart — reuse result from MeshCore.connect()
self.shared.set_status("🔄 Device info...")
cached_info = self.mc.self_info
if cached_info and cached_info.get("name"):
print(f"{pfx}: send_appstart OK (from connect): {cached_info.get('name')}")
self.shared.update_from_appstart(cached_info)
self._cache.set_device(cached_info)
else:
debug_print("self_info empty after connect(), falling back to manual send_appstart")
appstart_ok = False
for i in range(3):
debug_print(f"send_appstart fallback attempt {i + 1}/3")
try:
r = await self.mc.commands.send_appstart()
if r is None:
debug_print(f"send_appstart fallback {i + 1}: received None, retrying")
await asyncio.sleep(2.0)
continue
if r.type != EventType.ERROR:
print(f"{pfx}: send_appstart OK: {r.payload.get('name')} (fallback attempt {i + 1})")
self.shared.update_from_appstart(r.payload)
self._cache.set_device(r.payload)
appstart_ok = True
break
else:
debug_print(f"send_appstart fallback {i + 1}: ERROR — payload={pp(r.payload)}")
except Exception as exc:
debug_print(f"send_appstart fallback {i + 1} exception: {exc}")
await asyncio.sleep(2.0)
if not appstart_ok:
print(f"{pfx}: ⚠️ send_appstart failed after 3 fallback attempts")
# send_device_query
for i in range(5):
debug_print(f"send_device_query attempt {i + 1}/5")
try:
r = await self.mc.commands.send_device_query()
if r is None:
debug_print(f"send_device_query attempt {i + 1}: received None response, retrying")
await asyncio.sleep(2.0)
continue
if r.type != EventType.ERROR:
fw = r.payload.get("ver", "")
print(f"{pfx}: send_device_query OK: {fw} (attempt {i + 1})")
self.shared.update_from_device_query(r.payload)
if fw:
self._cache.set_firmware_version(fw)
break
else:
debug_print(f"send_device_query attempt {i + 1}: ERROR response — payload={pp(r.payload)}")
except Exception as exc:
debug_print(f"send_device_query attempt {i + 1} exception: {exc}")
await asyncio.sleep(2.0)
# Export device identity for MeshCore Observer
await self._export_device_identity()
# Channels
await self._discover_channels()
# Contacts
self.shared.set_status("🔄 Contacts...")
debug_print("get_contacts starting")
try:
r = await self._get_contacts_with_timeout()
debug_print(f"get_contacts result: type={r.type if r else None}")
if r and r.payload:
try:
payload_len = len(r.payload)
except Exception:
payload_len = None
if payload_len is not None and payload_len > 10:
debug_print(f"get_contacts payload size={payload_len} (omitted)")
else:
debug_data("get_contacts payload", r.payload)
if r is None:
debug_print(f"{pfx}: get_contacts returned None, keeping cached contacts")
elif r.type != EventType.ERROR:
merged = self._cache.merge_contacts(r.payload)
self.shared.set_contacts(merged)
print(f"{pfx}: Contacts — {len(r.payload)} from device, {len(merged)} total (with cache)")
else:
debug_print(f"{pfx}: get_contacts failed — payload={pp(r.payload)}, keeping cached contacts")
except Exception as exc:
debug_print(f"{pfx}: get_contacts exception: {exc}")
async def _get_contacts_with_timeout(self):
"""Fetch contacts with a bounded timeout to avoid hanging refresh."""
timeout = max(DEFAULT_TIMEOUT * 2, 10.0)
try:
return await asyncio.wait_for(
self.mc.commands.get_contacts(), timeout=timeout,
)
except asyncio.TimeoutError:
self.shared.set_status("⚠️ Contacts timeout — using cached contacts")
debug_print(f"get_contacts timeout after {timeout:.0f}s")
return None
# ── channel discovery ─────────────────────────────────────────
async def _discover_channels(self) -> None:
"""Discover channels and load their keys from the device."""
pfx = self._log_prefix
self.shared.set_status("🔄 Discovering channels...")
discovered: List[Dict] = []
cached_keys = self._cache.get_channel_keys()
confirmed: list[str] = []
from_cache: list[str] = []
derived: list[str] = []
consecutive_errors = 0
for idx in range(MAX_CHANNELS):
payload = await self._try_get_channel_info(idx, max_attempts=2, delay=1.0)
if payload is None:
consecutive_errors += 1
if consecutive_errors >= 3:
debug_print(
f"Channel discovery: {consecutive_errors} consecutive "
f"empty slots at idx {idx}, stopping"
)
break
continue
consecutive_errors = 0
name = payload.get("name") or payload.get("channel_name") or ""
if not name.strip():
debug_print(f"Channel [{idx}]: response OK but no name — skipping (undefined slot)")
continue
discovered.append({"idx": idx, "name": name})
secret = payload.get("channel_secret")
secret_bytes = self._extract_secret(secret)
if secret_bytes:
self._decoder.add_channel_key(idx, secret_bytes, source="device")
self._cache.set_channel_key(idx, secret_bytes.hex())
self._pending_keys.discard(idx)
confirmed.append(f"[{idx}] {name}")
elif str(idx) in cached_keys:
from_cache.append(f"[{idx}] {name}")
print(f"{pfx}: 📦 Channel [{idx}] '{name}' — using cached key")
else:
self._decoder.add_channel_key_from_name(idx, name)
self._pending_keys.add(idx)
derived.append(f"[{idx}] {name}")
print(f"{pfx}: ⚠️ Channel [{idx}] '{name}' — name-derived key (will retry)")
await asyncio.sleep(0.3)
if not discovered:
discovered = [{"idx": 0, "name": "Public"}]
print(f"{pfx}: ⚠️ No channels discovered, using default Public channel")
self._channels = discovered
self.shared.set_channels(discovered)
if CHANNEL_CACHE_ENABLED:
self._cache.set_channels(discovered)
debug_print("Channel list cached to disk")
print(f"{pfx}: Channels discovered: {[c['name'] for c in discovered]}")
print(f"{pfx}: PacketDecoder ready — has_keys={self._decoder.has_keys}")
if confirmed:
print(f"{pfx}: ✅ Keys from device: {', '.join(confirmed)}")
if from_cache:
print(f"{pfx}: 📦 Keys from cache: {', '.join(from_cache)}")
if derived:
print(f"{pfx}: ⚠️ Name-derived keys: {', '.join(derived)}")
async def _try_get_channel_info(
self, idx: int, max_attempts: int, delay: float,
) -> Optional[Dict]:
for attempt in range(max_attempts):
try:
r = await self.mc.commands.get_channel(idx)
if r is None:
debug_print(f"get_channel({idx}) attempt {attempt + 1}/{max_attempts}: received None response, retrying")
await asyncio.sleep(delay)
continue
if r.type == EventType.ERROR:
debug_print(f"get_channel({idx}) attempt {attempt + 1}/{max_attempts}: ERROR response — payload={pp(r.payload)}")
await asyncio.sleep(delay)
continue
debug_print(f"get_channel({idx}) attempt {attempt + 1}/{max_attempts}: OK — keys={list(r.payload.keys())}")
return r.payload
except Exception as exc:
debug_print(f"get_channel({idx}) attempt {attempt + 1}/{max_attempts} error: {exc}")
await asyncio.sleep(delay)
return None
async def _try_load_channel_key(
self, idx: int, name: str, max_attempts: int, delay: float,
) -> bool:
payload = await self._try_get_channel_info(idx, max_attempts, delay)
if payload is None:
return False
secret = payload.get("channel_secret")
secret_bytes = self._extract_secret(secret)
if secret_bytes:
self._decoder.add_channel_key(idx, secret_bytes, source="device")
self._cache.set_channel_key(idx, secret_bytes.hex())
print(f"{self._log_prefix}: ✅ Channel [{idx}] '{name}' — key from device (background retry)")
self._pending_keys.discard(idx)
return True
debug_print(f"get_channel({idx}): response OK but secret unusable")
return False
async def _retry_missing_keys(self) -> None:
if not self._pending_keys:
return
pending_copy = set(self._pending_keys)
ch_map = {ch["idx"]: ch["name"] for ch in self._channels}
debug_print(f"Background key retry: trying {len(pending_copy)} channels")
for idx in pending_copy:
name = ch_map.get(idx, f"ch{idx}")
loaded = await self._try_load_channel_key(idx, name, max_attempts=1, delay=0.5)
if loaded:
self._pending_keys.discard(idx)
await asyncio.sleep(1.0)
if not self._pending_keys:
print(f"{self._log_prefix}: ✅ All channel keys now loaded!")
else:
remaining = [f"[{idx}] {ch_map.get(idx, '?')}" for idx in sorted(self._pending_keys)]
debug_print(f"Background retry: still pending: {', '.join(remaining)}")
# ── helpers ────────────────────────────────────────────────────
def _seed_dedup_from_messages(self) -> None:
"""Seed the deduplicator with messages already in SharedData."""
snapshot = self.shared.get_snapshot()
messages = snapshot.get("messages", [])
seeded = 0
for msg in messages:
if msg.message_hash:
self._dedup.mark_hash(msg.message_hash)
seeded += 1
if msg.sender and msg.text:
self._dedup.mark_content(msg.sender, msg.channel, msg.text)
seeded += 1
debug_print(f"Dedup seeded with {seeded} entries from {len(messages)} messages")
@staticmethod
def _extract_secret(secret) -> Optional[bytes]:
if secret and isinstance(secret, bytes) and len(secret) >= 16:
return secret[:16]
if secret and isinstance(secret, str) and len(secret) >= 32:
try:
raw = bytes.fromhex(secret)
if len(raw) >= 16:
return raw[:16]
except ValueError:
pass
return None
# ── periodic tasks ────────────────────────────────────────────
async def _refresh_contacts(self) -> None:
try:
r = await self._get_contacts_with_timeout()
if r is None:
debug_print("Periodic refresh: get_contacts returned None, skipping")
return
if r.type != EventType.ERROR:
merged = self._cache.merge_contacts(r.payload)
self.shared.set_contacts(merged)
debug_print(
f"Periodic refresh: {len(r.payload)} from device, "
f"{len(merged)} total"
)
except Exception as exc:
debug_print(f"Periodic contact refresh failed: {exc}")
async def _cleanup_old_data(self) -> None:
try:
if self.shared.archive:
self.shared.archive.cleanup_old_data()
stats = self.shared.archive.get_stats()
debug_print(
f"Cleanup: archive now has {stats['total_messages']} messages, "
f"{stats['total_rxlog']} rxlog entries"
)
removed = self._cache.prune_old_contacts()
if removed > 0:
contacts = self._cache.get_contacts()
self.shared.set_contacts(contacts)
debug_print(f"Cleanup: pruned {removed} old contacts")
except Exception as exc:
debug_print(f"Periodic cleanup failed: {exc}")
# ======================================================================
# Serial worker
# ======================================================================
class SerialWorker(_BaseWorker):
"""Serial communication worker (USB/UART).
Args:
port: Serial device path (e.g. ``"/dev/ttyUSB0"``).
shared: SharedDataWriter for thread-safe communication.
baudrate: Serial baudrate (default from config).
cx_dly: Connection delay for meshcore serial transport.
"""
def __init__(
self,
port: str,
shared: SharedDataWriter,
baudrate: int = _config.SERIAL_BAUDRATE,
cx_dly: float = _config.SERIAL_CX_DELAY,
) -> None:
super().__init__(port, shared)
self.port = port
self.baudrate = baudrate
self.cx_dly = cx_dly
@property
def _log_prefix(self) -> str:
return "SERIAL"
@property
def _disconnect_keywords(self) -> tuple:
return (
"not connected", "disconnected", "connection reset",
"broken pipe", "i/o error", "read failed", "write failed",
"port is closed", "port closed",
)
async def _async_main(self) -> None:
try:
while self.running:
# ── Outer loop: (re)establish a fresh serial connection ──
self._disconnected = False
await self._connect()
if not self.mc:
print("SERIAL: Initial connection failed, retrying in 30s...")
self.shared.set_status("⚠️ Connection failed — retrying...")
await asyncio.sleep(30)
continue
# ── Inner loop: run + reconnect without calling _connect() again ──
# _handle_reconnect() already creates a fresh MeshCore and loads
# data — calling _connect() on top of that would attempt to open
# the serial port a second time, causing an immediate disconnect.
while self.running:
await self._main_loop()
if not self._disconnected or not self.running:
break
ok = await self._handle_reconnect()
if ok:
# Reconnected — reset flag and go back to _main_loop,
# NOT to the outer while (which would call _connect() again).
self._disconnected = False
else:
# All reconnect attempts exhausted — wait, then let the
# outer loop call _connect() for a clean fresh start.
await asyncio.sleep(60)
break
finally:
return
async def _connect(self) -> None:
if self._cache.load():
self._apply_cache()
print("SERIAL: Cache loaded — GUI populated from disk")
else:
print("SERIAL: No cache found — waiting for device data")
self.shared.set_status(f"🔄 Connecting to {self.port}...")
try:
print(f"SERIAL: Connecting to {self.port}...")
self.mc = await MeshCore.create_serial(
self.port,
baudrate=self.baudrate,
auto_reconnect=False,
default_timeout=DEFAULT_TIMEOUT,
debug=_config.MESHCORE_LIB_DEBUG,
cx_dly=self.cx_dly,
)
if self.mc is None:
raise RuntimeError("No response from device over serial")
print("SERIAL: Connected!")
await asyncio.sleep(1)
debug_print("Post-connection sleep done, wiring collaborators")
self._wire_collaborators()
await self._load_data()
await self.mc.start_auto_message_fetching()
self.shared.set_connected(True)
self.shared.set_status("✅ Connected")
print("SERIAL: Ready!")
if self._pending_keys:
pending_names = [
f"[{ch['idx']}] {ch['name']}"
for ch in self._channels
if ch["idx"] in self._pending_keys
]
print(
f"SERIAL: ⏳ Background retry active for: "
f"{', '.join(pending_names)} (every {KEY_RETRY_INTERVAL:.0f}s)"
)
except Exception as e:
print(f"SERIAL: Connection error: {e}")
self.mc = None # ensure _async_main sees connection as failed
if self._cache.has_cache:
self.shared.set_status(f"⚠️ Offline — using cached data ({e})")
else:
self.shared.set_status(f"❌ {e}")
async def _reconnect(self) -> Optional[MeshCore]:
for attempt in range(1, RECONNECT_MAX_RETRIES + 1):
delay = RECONNECT_BASE_DELAY * attempt
print(
f"SERIAL: 🔄 Reconnect attempt {attempt}/{RECONNECT_MAX_RETRIES} "
f"in {delay:.0f}s..."
)
await asyncio.sleep(delay)
try:
mc = await MeshCore.create_serial(
self.port,
baudrate=self.baudrate,
auto_reconnect=False,
default_timeout=DEFAULT_TIMEOUT,
debug=_config.MESHCORE_LIB_DEBUG,
cx_dly=self.cx_dly,
)
if mc is None:
raise RuntimeError("No response from device over serial")
return mc
except Exception as exc:
print(f"SERIAL: ❌ Reconnect attempt {attempt} failed: {exc}")
print(f"SERIAL: ❌ Reconnect failed after {RECONNECT_MAX_RETRIES} attempts")
return None
# ======================================================================
# BLE worker
# ======================================================================
class BLEWorker(_BaseWorker):
"""BLE communication worker (Bluetooth Low Energy).
Args:
address: BLE MAC address (e.g. ``"literal:AA:BB:CC:DD:EE:FF"``).
shared: SharedDataWriter for thread-safe communication.
"""
def __init__(self, address: str, shared: SharedDataWriter) -> None:
super().__init__(address, shared)
self.address = address
# BLE PIN agent — imported lazily so serial-only installs
# don't need dbus_fast / bleak.
from meshcore_gui.ble.ble_agent import BleAgentManager
self._agent = BleAgentManager(pin=_config.BLE_PIN)
@property
def _log_prefix(self) -> str:
return "BLE"
@property
def _disconnect_keywords(self) -> tuple:
return (
"not connected", "disconnected", "dbus",
"pin or key missing", "connection reset", "broken pipe",
"failed to discover", "service discovery",
)
async def _async_main(self) -> None:
from meshcore_gui.ble.ble_reconnect import remove_bond
# Step 1: Start PIN agent BEFORE any BLE connection
await self._agent.start()
# Step 2: Remove stale bond (clean slate)
await remove_bond(self.address)
await asyncio.sleep(1)
# Step 3: Connect + main loop
try:
while self.running:
# ── Outer loop: (re)establish a fresh BLE connection ──
self._disconnected = False
await self._connect()
if not self.mc:
print("BLE: Initial connection failed, retrying in 30s...")
self.shared.set_status("⚠️ Connection failed — retrying...")
await asyncio.sleep(30)
await remove_bond(self.address)
await asyncio.sleep(1)
continue
# ── Inner loop: run + reconnect without calling _connect() again ──
# _handle_reconnect() already creates a fresh MeshCore and loads
# data — calling _connect() on top would open a second BLE session,
# causing an immediate disconnect.
while self.running:
await self._main_loop()
if not self._disconnected or not self.running:
break
ok = await self._handle_reconnect()
if ok:
# Reconnected — reset flag and go back to _main_loop,
# NOT to the outer while (which would call _connect() again).
self._disconnected = False
else:
await asyncio.sleep(60)
await remove_bond(self.address)
await asyncio.sleep(1)
break
finally:
await self._agent.stop()
async def _connect(self) -> None:
if self._cache.load():
self._apply_cache()
print("BLE: Cache loaded — GUI populated from disk")
else:
print("BLE: No cache found — waiting for BLE data")
self.shared.set_status(f"🔄 Connecting to {self.address}...")
try:
print(f"BLE: Connecting to {self.address}...")
self.mc = await MeshCore.create_ble(
self.address,
auto_reconnect=False,
default_timeout=DEFAULT_TIMEOUT,
debug=_config.MESHCORE_LIB_DEBUG,
)
print("BLE: Connected!")
await asyncio.sleep(1)
debug_print("Post-connection sleep done, wiring collaborators")
self._wire_collaborators()
await self._load_data()
await self.mc.start_auto_message_fetching()
self.shared.set_connected(True)
self.shared.set_status("✅ Connected")
print("BLE: Ready!")
if self._pending_keys:
pending_names = [
f"[{ch['idx']}] {ch['name']}"
for ch in self._channels
if ch["idx"] in self._pending_keys
]
print(
f"BLE: ⏳ Background retry active for: "
f"{', '.join(pending_names)} (every {KEY_RETRY_INTERVAL:.0f}s)"
)
except Exception as e:
print(f"BLE: Connection error: {e}")
self.mc = None # ensure _async_main sees connection as failed
if self._cache.has_cache:
self.shared.set_status(f"⚠️ Offline — using cached data ({e})")
else:
self.shared.set_status(f"❌ {e}")
async def _reconnect(self) -> Optional[MeshCore]:
from meshcore_gui.ble.ble_reconnect import reconnect_loop
async def _create_fresh_connection() -> MeshCore:
return await MeshCore.create_ble(
self.address,
auto_reconnect=False,
default_timeout=DEFAULT_TIMEOUT,
debug=_config.MESHCORE_LIB_DEBUG,
)
return await reconnect_loop(
_create_fresh_connection,
self.address,
max_retries=RECONNECT_MAX_RETRIES,
base_delay=RECONNECT_BASE_DELAY,
)
meshcore_gui/config.py
+2 -2
View File
@@ -25,7 +25,7 @@ from typing import Any, Dict, List
# ==============================================================================
VERSION: str = "1.15.0"
VERSION: str = "1.16.0"
# ==============================================================================
@@ -279,7 +279,7 @@ def debug_data(label: str, obj: Any) -> None:
# Maximum number of channel slots to probe on the device.
# MeshCore supports up to 8 channels (indices 0-7).
MAX_CHANNELS: int = 8
MAX_CHANNELS: int = 100
# Enable or disable caching of the channel list to disk.
# When False (default), channels are always fetched fresh from the
meshcore_gui/gui/dashboard.py
+16
View File
@@ -18,6 +18,7 @@ from meshcore_gui.gui.panels import (
ActionsPanel,
BbsPanel,
BotPanel,
ChannelPanel,
ContactsPanel,
DevicePanel,
MapPanel,
@@ -325,6 +326,9 @@ class DashboardPage:
self._bbs: BbsPanel | None = None
self._bot: BotPanel | None = None
# Channel add dialog panel
self._channel_panel: ChannelPanel | None = None
# Header status label
self._status_label = None
@@ -377,6 +381,8 @@ class DashboardPage:
self._bot_config_store,
self._pin_store,
)
self._channel_panel = ChannelPanel(put_cmd)
self._channel_panel.render()
# Inject DOMCA theme (fonts + CSS variables)
ui.add_head_html(_DOMCA_HEAD)
@@ -418,6 +424,10 @@ class DashboardPage:
'DM', lambda: self._navigate_panel('messages', channel='DM')
)
# Dynamic channel items populated by _update_submenus
self._make_sub_btn(
'＋ Add Channel',
lambda: self._channel_panel.open() if self._channel_panel else None,
)
# ── 🏠 ROOMS (expandable with room submenu) ───────────
with ui.expansion(
@@ -610,6 +620,10 @@ class DashboardPage:
f"[{idx}] {name}",
lambda i=idx: self._navigate_panel('messages', channel=i),
)
self._make_sub_btn(
'＋ Add Channel',
lambda: self._channel_panel.open() if self._channel_panel else None,
)
# Rebuild Archive submenu
if self._archive_sub_container:
@@ -816,6 +830,8 @@ class DashboardPage:
self._messages.update_filters(data)
self._messages.update_channel_options(data['channels'])
self._update_submenus(data)
if self._channel_panel:
self._channel_panel.update(data)
if self._active_panel == 'device':
if data['device_updated'] or is_first:
meshcore_gui/gui/panels/__init__.py
+1
View File
@@ -17,3 +17,4 @@ from meshcore_gui.gui.panels.rxlog_panel import RxLogPanel # noqa: F401
from meshcore_gui.gui.panels.room_server_panel import RoomServerPanel # noqa: F401
from meshcore_gui.gui.panels.bbs_panel import BbsPanel # noqa: F401
from meshcore_gui.gui.panels.bot_panel import BotPanel # noqa: F401
from meshcore_gui.gui.panels.channel_panel import ChannelPanel # noqa: F401
meshcore_gui/gui/panels/channel_panel.py
+336
View File
@@ -0,0 +1,336 @@
"""
Channel panel — dialog for adding hashtag and private channels.
Triggered by the ``＋ Add Channel`` button in the Messages submenu.
Three modes are supported:
Hashtag
Name must start with ``#``. The channel key is derived automatically
from the name by the MeshCore library; no manual key input is required
and no key export is offered (the name itself is the shared secret).
Private — New
Name is freely chosen. A random 16-byte key is generated on demand.
After submission the dialog shows a QR code and a copy-to-clipboard
button so the key can be shared with other users.
Private — Existing (join)
Used when another user has shared a private channel key. The user
pastes the 32-character hex key and the dialog writes it to the
device verbatim. No key export is offered.
"""
from typing import Callable, Dict, List, Optional
from nicegui import ui
from meshcore_gui.services.channel_service import (
generate_qr_base64,
generate_secret,
secret_to_hex,
)
class ChannelPanel:
"""NiceGUI dialog for adding a channel to the connected MeshCore device.
Args:
put_command: Callable to enqueue a command dict for the BLE worker.
"""
def __init__(self, put_command: Callable[[Dict], None]) -> None:
self._put_command = put_command
self._channels: List[Dict] = []
# Dialog + form widget references (populated in render())
self._dialog: Optional[ui.dialog] = None
self._mode_radio: Optional[ui.radio] = None
self._idx_input: Optional[ui.number] = None
self._name_input: Optional[ui.input] = None
# Private-mode widgets
self._hashtag_info: Optional[ui.label] = None
self._secret_section: Optional[ui.column] = None
self._secret_input: Optional[ui.input] = None
self._generate_row: Optional[ui.row] = None
self._copy_btn: Optional[ui.button] = None
# QR section (private-new only, revealed after submit)
self._qr_section: Optional[ui.column] = None
self._qr_label: Optional[ui.label] = None
self._qr_image: Optional[ui.image] = None
# Transient state
self._generated_secret: Optional[bytes] = None
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def render(self) -> None:
"""Create the dialog widget tree.
Must be called once during dashboard page rendering (inside the
NiceGUI ``@ui.page`` context) so that all widgets are bound to the
correct client session.
"""
self._dialog = ui.dialog()
with self._dialog:
with ui.card().classes('w-full').style(
'min-width: 340px; max-width: 440px; gap: 0.6rem'
):
ui.label('📡 Add Channel').classes('font-bold text-gray-600 text-base')
# ── Mode selection ──────────────────────────────────
self._mode_radio = ui.radio(
options={
'hashtag': '# Hashtag channel',
'private_new': '🔒 Private – New',
'private_existing': '🔒 Private – Existing (join)',
},
value='hashtag',
on_change=self._on_mode_change,
).classes('w-full')
# ── Channel index ────────────────────────────────────
self._idx_input = ui.number(
label='Channel index (1 – 99)',
value=1,
min=1,
max=99,
step=1,
format='%d',
).classes('w-full')
# ── Channel name ─────────────────────────────────────
self._name_input = ui.input(
label='Channel name',
placeholder='e.g. #localmesh',
).classes('w-full')
# ── Hashtag info label (hashtag mode only) ───────────
self._hashtag_info = ui.label(
'🔑 Key is derived automatically from the name. '
'Anyone who knows the name can join.'
).classes('text-xs text-gray-500')
# ── Private secret section ───────────────────────────
self._secret_section = ui.column().classes('w-full gap-1')
with self._secret_section:
self._secret_input = ui.input(
label='Secret key (32 hex chars)',
placeholder='e.g. 8b3387e9c5cdea6ac9e5edbaa115cd72',
).classes('w-full')
# Generate + copy row (private-new only)
self._generate_row = ui.row().classes('gap-2 items-center')
with self._generate_row:
ui.button(
'🎲 Generate key',
on_click=self._generate_secret,
).props('flat dense no-caps')
self._copy_btn = ui.button(
'📋 Copy key',
on_click=self._copy_key,
).props('flat dense no-caps')
# ── Action buttons ───────────────────────────────────
with ui.row().classes('gap-2 justify-end w-full'):
ui.button(
'Cancel',
on_click=self._close,
).props('flat no-caps')
ui.button(
'Add Channel',
on_click=self._submit,
).props('unelevated color=primary no-caps')
# ── QR code section (shown after private-new submit) ─
self._qr_section = ui.column().classes('w-full items-center gap-1')
with self._qr_section:
ui.separator()
self._qr_label = ui.label('').classes(
'text-xs text-gray-500 text-center'
)
self._qr_image = ui.image('').style('width: 192px; height: 192px')
ui.label(
'Scan with the MeshCore app to share this channel.'
).classes('text-xs text-gray-400 text-center')
# Apply initial visibility based on default mode
self._apply_visibility('hashtag')
self._qr_section.set_visibility(False)
def update(self, data: Dict) -> None:
"""Update the next-available channel index from the live channel list.
Called every 500 ms from the dashboard update cycle. Stores the
current channel list so ``open()`` can pre-fill a sensible index.
Args:
data: SharedData snapshot dict containing the ``channels`` list.
"""
self._channels = data.get('channels', [])
def open(self) -> None:
"""Open the dialog and reset the form to a clean state."""
if self._dialog is None:
return
self._reset_form()
self._dialog.open()
# ------------------------------------------------------------------
# Private — form logic
# ------------------------------------------------------------------
def _close(self) -> None:
"""Close the dialog."""
if self._dialog:
self._dialog.close()
def _reset_form(self) -> None:
"""Reset all fields to their defaults and hide the QR section."""
self._generated_secret = None
if self._mode_radio:
self._mode_radio.value = 'hashtag'
if self._name_input:
self._name_input.value = ''
if self._secret_input:
self._secret_input.value = ''
if self._qr_section:
self._qr_section.set_visibility(False)
if self._qr_image:
self._qr_image.source = ''
if self._qr_label:
self._qr_label.text = ''
# Pre-fill next available index
if self._channels:
next_idx = min(max(ch['idx'] for ch in self._channels) + 1, 99)
else:
next_idx = 1
if self._idx_input:
self._idx_input.value = next_idx
self._apply_visibility('hashtag')
def _on_mode_change(self, event=None) -> None:
"""React to mode-radio change — update field visibility."""
mode = self._mode_radio.value if self._mode_radio else 'hashtag'
self._generated_secret = None
if self._secret_input:
self._secret_input.value = ''
if self._qr_section:
self._qr_section.set_visibility(False)
self._apply_visibility(mode)
def _apply_visibility(self, mode: str) -> None:
"""Show/hide sections according to *mode*."""
is_hashtag = mode == 'hashtag'
is_private = mode in ('private_new', 'private_existing')
is_private_new = mode == 'private_new'
if self._hashtag_info:
self._hashtag_info.set_visibility(is_hashtag)
if self._secret_section:
self._secret_section.set_visibility(is_private)
if self._generate_row:
self._generate_row.set_visibility(is_private_new)
if self._copy_btn:
self._copy_btn.set_visibility(is_private_new)
# Adjust name placeholder to hint correct input format
if self._name_input:
placeholder = 'e.g. #localmesh' if is_hashtag else 'e.g. TeamName'
self._name_input.props(f'placeholder="{placeholder}"')
# ------------------------------------------------------------------
# Private — actions
# ------------------------------------------------------------------
def _generate_secret(self) -> None:
"""Generate a new random secret and display it in the secret field."""
self._generated_secret = generate_secret()
if self._secret_input:
self._secret_input.value = secret_to_hex(self._generated_secret)
def _copy_key(self) -> None:
"""Copy the displayed secret to the system clipboard."""
key = (self._secret_input.value or '').strip() if self._secret_input else ''
if key:
ui.run_javascript(
f'navigator.clipboard.writeText("{key}").catch(()=>{{}})'
)
ui.notify('Key copied to clipboard', type='positive', timeout=2000)
else:
ui.notify('Generate a key first', type='warning', timeout=2000)
def _submit(self) -> None:
"""Validate form inputs and queue the ``add_channel`` command."""
mode = self._mode_radio.value if self._mode_radio else 'hashtag'
name = (self._name_input.value or '').strip() if self._name_input else ''
idx = int(self._idx_input.value or 1) if self._idx_input else 1
# ── Validation ──────────────────────────────────────────────
if not name:
ui.notify('Channel name is required', type='warning', timeout=3000)
return
if mode == 'hashtag' and not name.startswith('#'):
ui.notify(
'Hashtag channel name must start with #',
type='warning',
timeout=3000,
)
return
if mode == 'private_new':
if not self._generated_secret:
ui.notify(
'Click "Generate key" to create a secret first',
type='warning',
timeout=3000,
)
return
secret_hex = secret_to_hex(self._generated_secret)
elif mode == 'private_existing':
raw = (self._secret_input.value or '').strip().lower() if self._secret_input else ''
valid_chars = set('0123456789abcdef')
if len(raw) != 32 or not all(c in valid_chars for c in raw):
ui.notify(
'Secret must be exactly 32 hexadecimal characters',
type='warning',
timeout=3000,
)
return
secret_hex = raw
else:
# Hashtag: library derives the key; pass empty string so the
# command handler passes secret=None to set_channel().
secret_hex = ''
# ── Queue command ────────────────────────────────────────────
self._put_command({
'action': 'add_channel',
'idx': idx,
'name': name,
'secret_hex': secret_hex,
})
ui.notify(f"Adding [{idx}] {name}…", type='info', timeout=2500)
# ── QR code for new private channels ─────────────────────────
if mode == 'private_new' and self._generated_secret:
qr_data = generate_qr_base64(name, self._generated_secret)
if qr_data and self._qr_image and self._qr_label and self._qr_section:
self._qr_image.source = qr_data
self._qr_label.text = f'Share key for "{name}"'
self._qr_section.set_visibility(True)
# Keep dialog open so the user can scan / copy the key
return
self._close()
meshcore_gui/services/bot.py
+2 -2
View File
@@ -57,9 +57,9 @@ BOT_COOLDOWN_SECONDS: float = 5.0
# The bot checks whether the incoming message text *contains* the keyword
# (case-insensitive). First match wins.
BOT_KEYWORDS: Dict[str, str] = {
'test': '@[{sender}], rcvd | SNR {snr} | {path}',
'#test': '@[{sender}], rcvd | SNR {snr} | {path}',
'ping': 'Pong!',
'help': 'test, ping, help',
# 'help': 'test, ping, help',
}
meshcore_gui/services/channel_service.py
+137
View File
@@ -0,0 +1,137 @@
"""
Channel management service for MeshCore GUI.
Provides pure-Python helpers for:
- Generating random private channel secrets.
- Deriving deterministic hashtag-channel keys (SHA-256 of name).
- Building MeshCore-compatible QR code URLs.
- Rendering QR codes as base64-encoded PNG data URIs for inline display.
No GUI or BLE dependencies — safe to import from any layer.
"""
import base64
import io
import os
from hashlib import sha256
from urllib.parse import urlencode
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
# The globally known public-channel secret (index 0).
# Every MeshCore device ships with this key for the default "Public" channel.
# Shared here for reference only; not used by the Add Channel dialog since
# index-0 / Public is out of scope for this feature.
PUBLIC_CHANNEL_SECRET: bytes = bytes.fromhex("8b3387e9c5cdea6ac9e5edbaa115cd72")
# ---------------------------------------------------------------------------
# Key helpers
# ---------------------------------------------------------------------------
def generate_secret() -> bytes:
"""Generate a cryptographically secure 16-byte random channel secret.
Returns:
16 random bytes suitable for a new private channel.
"""
return os.urandom(16)
def derive_hashtag_key(name: str) -> bytes:
"""Derive the deterministic key for a hashtag channel.
MeshCore computes the channel key as the first 16 bytes of
``SHA-256(name.encode('utf-8'))``. Because the ``#`` sign is part
of the name (e.g. ``"#localmesh"``), callers must pass the full
name including the hash symbol.
The library's ``set_channel()`` performs this derivation itself when
``channel_name.startswith('#')``, so this helper is provided for
informational display only.
Args:
name: Channel name including the leading ``#`` (e.g. ``"#test"``).
Returns:
16-byte derived key.
"""
return sha256(name.encode("utf-8")).digest()[:16]
def secret_to_hex(secret: bytes) -> str:
"""Convert a 16-byte channel secret to a lowercase hex string (32 chars).
Args:
secret: 16-byte channel secret.
Returns:
32-character lowercase hex string.
"""
return secret.hex()
# ---------------------------------------------------------------------------
# QR code helpers
# ---------------------------------------------------------------------------
def build_qr_url(name: str, secret: bytes) -> str:
"""Build the official MeshCore QR code URL for sharing a private channel.
Format (per MeshCore docs)::
meshcore://channel/add?name=<name>&secret=<32-hex>
This URL is recognised by the official MeshCore mobile app, allowing
recipients to join the channel by scanning the QR code.
Args:
name: Channel name (without leading ``#`` for private channels).
secret: 16-byte channel secret.
Returns:
URL string ready to be encoded into a QR code.
"""
params = {"name": name, "secret": secret.hex()}
return "meshcore://channel/add?" + urlencode(params)
def generate_qr_base64(name: str, secret: bytes) -> str:
"""Generate a QR code PNG as a base64 data URI for inline display.
Uses the ``qrcode`` library with Pillow for image rendering. Returns
an empty string if either library is unavailable, allowing callers to
degrade gracefully (hide the QR widget rather than crashing).
Args:
name: Private channel name.
secret: 16-byte channel secret.
Returns:
``data:image/png;base64,...`` string, or ``""`` on import error.
"""
try:
import qrcode # type: ignore[import]
from PIL import Image # noqa: F401 — imported for side-effect (PIL check)
url = build_qr_url(name, secret)
qr = qrcode.QRCode(
version=None,
error_correction=qrcode.constants.ERROR_CORRECT_M,
box_size=6,
border=2,
)
qr.add_data(url)
qr.make(fit=True)
img: Image.Image = qr.make_image(fill_color="black", back_color="white")
buf = io.BytesIO()
img.save(buf, format="PNG")
b64 = base64.b64encode(buf.getvalue()).decode("ascii")
return f"data:image/png;base64,{b64}"
except ImportError:
return ""
meshcore_guiAddChannel.zip
BIN
View File
Binary file not shown.