an0n.eu/meshcore-gui
1
0
Fork 0
Code Pull Requests Activity

feature/archive and contactlist

Browse Source
This commit is contained in:
pe1hvh
2026-02-08 21:29:52 +01:00
parent b2399235b5
commit f9cc91ebc8
12 changed files with 579 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
CHANGELOG.md
View File

@@ -10,7 +10,39 @@ Format follows [Keep a Changelog](https://keepachangelog.com/) and [Semantic Ver
<!-- ADDED: Nieuw v5.3.0 entry bovenaan -->
## [5.3.0] - 2026-02-08 — Documentation Review & Route Table Fix
## [5.4.0] - 2026-02-08 — Contact Maintenance Feature
### Added
- ✅ **Pin/Unpin contacts** (Iteratie A) — Toggle to pin individual contacts, protecting them from bulk deletion
- Persistent pin state stored in `~/.meshcore-gui/cache/<ADDRESS>_pins.json`
- Pinned contacts visually marked with yellow background
- Pinned contacts sorted to top of contact list
- Pin state survives app restart
- New service: `services/pin_store.py` — JSON-backed persistent pin storage
- ✅ **Bulk delete unpinned contacts** (Iteratie B) — Remove all unpinned contacts from device in one action
- "🧹 Clean up" button in contacts panel with confirmation dialog
- Shows count of contacts to be removed vs. pinned contacts kept
- Progress status updates during removal
- Automatic device resync after completion
- New service: `services/contact_cleaner.py` — ContactCleanerService with purge statistics
- ✅ **Auto-add contacts toggle** (Iteratie C) — Control whether device automatically adds new contacts from mesh adverts
- "📥 Auto-add" checkbox in contacts panel (next to Clean up button)
- Syncs with device via `set_manual_add_contacts()` SDK call
- Inverted logic handled internally (UI "Auto-add ON" = `set_manual_add_contacts(false)`)
- Optimistic update with automatic rollback on BLE failure
- State synchronized from device on each GUI update cycle
### Changed
- 🔄 `contacts_panel.py`: Added pin checkbox per contact, purge button, auto-add toggle, DM dialog (all existing functionality preserved)
- 🔄 `commands.py`: Added `purge_unpinned` and `set_auto_add` command handlers
- 🔄 `shared_data.py`: Added `auto_add_enabled` field with thread-safe getter/setter
- 🔄 `protocols.py`: Added `set_auto_add_enabled` and `is_auto_add_enabled` to Writer and Reader protocols
- 🔄 `dashboard.py`: Passes `PinStore` and `set_auto_add_enabled` callback to ContactsPanel
- 🔄 **UI language**: All Dutch strings in `contacts_panel.py` and `commands.py` translated to English
---
### Fixed
- 🐛 **Route table names and IDs not displayed** — Route tables in both current messages (RoutePage) and archive messages (ArchivePage) now correctly show node names and public key IDs for sender, repeaters and receiver

22
README.md
View File

@@ -26,6 +26,7 @@ Under the hood it uses `bleak` for Bluetooth Low Energy (which talks to BlueZ on
- **Interactive Map** — Leaflet map with markers for own position and contacts
- **Channel Messages** — Send and receive messages on channels
- **Direct Messages** — Click on a contact to send a DM
- **Contact Maintenance** — Pin/unpin contacts to protect them from deletion, bulk-delete unpinned contacts from the device, and toggle automatic contact addition from mesh adverts
- **Message Filtering** — Filter messages per channel via checkboxes
- **Message Route Visualization** — Click any message to open a detailed route page showing the path (hops) through the mesh network on an interactive map, with a hop summary, route table and reply panel
- **Message Archive** — All messages and RX log entries are persisted to disk with configurable retention. Browse archived messages via the archive viewer with filters (channel, time range, text search), pagination and inline route tables
@@ -208,6 +209,9 @@ The GUI opens automatically in your browser at `http://localhost:8080`
### Contacts
- List of known nodes with type and location
- Click on a contact to send a DM
- **Pin/Unpin**: Checkbox per contact to pin it — pinned contacts are sorted to the top and visually marked with a yellow background. Pin state is persisted locally and survives app restart.
- **Bulk delete**: "🧹 Clean up" button removes all unpinned contacts from the device in one action, with a confirmation dialog showing how many will be removed vs. kept.
- **Auto-add toggle**: "📥 Auto-add" checkbox controls whether the device automatically adds new contacts when it receives adverts from other mesh nodes. Disabled by default to prevent the contact list from filling up.
### Map
- OpenStreetMap with markers for own position and contacts
@@ -336,17 +340,17 @@ The built-in bot automatically replies to messages containing recognised keyword
│ safe) │ │ (~/.meshcore- │
└──────┬──────┘ │ gui/cache/) │
│ └───────────────┘
┌──────┴──────┐
│ Message │
│ Archive │
│ (~/.meshcore│
│ -gui/ │
┌──────┴──────┐ ┌───────────────┐
│ Message │ │ PinStore │
│ Archive │ │ Contact │
│ (~/.meshcore│ │ Cleaner │
│ -gui/ │ └───────────────┘
│ archive/) │
└─────────────┘
```
- **BLEWorker**: Runs in separate thread with its own asyncio loop, with background retry for missing channel keys
- **CommandHandler**: Executes commands (send message, advert, refresh)
- **CommandHandler**: Executes commands (send message, advert, refresh, purge unpinned, set auto-add)
- **EventHandler**: Processes incoming BLE events (messages, RX log)
- **PacketDecoder**: Decodes raw LoRa packets and extracts route data
- **MeshBot**: Keyword-triggered auto-reply on configured channels
@@ -354,6 +358,8 @@ The built-in bot automatically replies to messages containing recognised keyword
- **DeviceCache**: Local JSON cache per device for instant startup and offline resilience
- **MessageArchive**: Persistent storage for messages and RX log with configurable retention and automatic cleanup
<!-- ADDED: MessageArchive component description -->
- **PinStore**: Persistent pin state storage per device (JSON-backed)
- **ContactCleanerService**: Bulk-delete logic for unpinned contacts with statistics
- **SharedData**: Thread-safe data sharing between BLE and GUI via Protocol interfaces
- **DashboardPage**: Main GUI with modular panels (device, contacts, map, messages, etc.)
- **RoutePage**: Standalone route visualization page opened per message
@@ -474,7 +480,7 @@ meshcore-gui/
│ │ └── panels/ # Modular UI panels
│ │ ├── __init__.py
│ │ ├── device_panel.py # Device info display
│ │ ├── contacts_panel.py # Contacts list with DM support
│ │ ├── contacts_panel.py # Contacts list with DM, pin/unpin, bulk delete, auto-add toggle
│ │ ├── map_panel.py # Leaflet map
│ │ ├── input_panel.py # Message input and channel select
│ │ ├── filter_panel.py # Channel filters and bot toggle
@@ -485,8 +491,10 @@ meshcore-gui/
│ ├── __init__.py
│ ├── bot.py # Keyword-triggered auto-reply bot
│ ├── cache.py # Local JSON cache per BLE device
│ ├── contact_cleaner.py # Bulk-delete logic for unpinned contacts
│ ├── dedup.py # Message deduplication
│ ├── message_archive.py # Persistent message and RX log archive
│ ├── pin_store.py # Persistent pin state storage per device
│ └── route_builder.py # Route data construction
├── docs/
│ ├── TROUBLESHOOTING.md # BLE troubleshooting guide (Linux)

BIN
docs/MeshCore_GUI_Design.docx
View File

Binary file not shown.

4
meshcore_gui.py
View File

@@ -35,6 +35,7 @@ from meshcore_gui.core.shared_data import SharedData
from meshcore_gui.gui.dashboard import DashboardPage
from meshcore_gui.gui.route_page import RoutePage
from meshcore_gui.gui.archive_page import ArchivePage
from meshcore_gui.services.pin_store import PinStore
# Global instances (needed by NiceGUI page decorators)
@@ -107,7 +108,8 @@ def main():
# Assemble components
_shared = SharedData(ble_address)
_dashboard = DashboardPage(_shared)
_pin_store = PinStore(ble_address)
_dashboard = DashboardPage(_shared, _pin_store)
_route_page = RoutePage(_shared)
_archive_page = ArchivePage(_shared)

4
meshcore_gui/__main__.py
View File

@@ -36,6 +36,7 @@ from meshcore_gui.core.shared_data import SharedData
from meshcore_gui.gui.dashboard import DashboardPage
from meshcore_gui.gui.route_page import RoutePage
from meshcore_gui.gui.archive_page import ArchivePage
from meshcore_gui.services.pin_store import PinStore
# Global instances (needed by NiceGUI page decorators)
@@ -108,7 +109,8 @@ def main():
# Assemble components
_shared = SharedData(ble_address)
_dashboard = DashboardPage(_shared)
_pin_store = PinStore(ble_address)
_dashboard = DashboardPage(_shared, _pin_store)
_route_page = RoutePage(_shared)
_archive_page = ArchivePage(_shared)

129
meshcore_gui/ble/commands.py
View File

@@ -6,10 +6,11 @@ of work. New commands can be registered without modifying existing
code (Open/Closed Principle).
"""
import asyncio
from datetime import datetime
from typing import Dict, Optional
from typing import Dict, List, Optional
from meshcore import MeshCore
from meshcore import MeshCore, EventType
from meshcore_gui.config import debug_print
from meshcore_gui.core.models import Message
@@ -34,6 +35,8 @@ class CommandHandler:
'send_dm': self._cmd_send_dm,
'send_advert': self._cmd_send_advert,
'refresh': self._cmd_refresh,
'purge_unpinned': self._cmd_purge_unpinned,
'set_auto_add': self._cmd_set_auto_add,
}
async def process_all(self) -> None:
@@ -102,6 +105,128 @@ class CommandHandler:
if self._load_data_callback:
await self._load_data_callback()
async def _cmd_purge_unpinned(self, cmd: Dict) -> None:
"""Remove unpinned contacts from the MeshCore device.
Iterates the list of public keys, calls ``remove_contact``
for each one with a short delay between calls to avoid
overwhelming the BLE link. After completion, triggers a
full refresh so the GUI reflects the new state.
Expected command dict::
{
'action': 'purge_unpinned',
'pubkeys': ['aabbcc...', ...],
}
"""
pubkeys: List[str] = cmd.get('pubkeys', [])
if not pubkeys:
self._shared.set_status("⚠️ No contacts to remove")
return
total = len(pubkeys)
removed = 0
errors = 0
self._shared.set_status(
f"🗑️ Removing {total} contacts..."
)
debug_print(f"Purge: starting removal of {total} contacts")
for i, pubkey in enumerate(pubkeys, 1):
try:
r = await self._mc.commands.remove_contact(pubkey)
if r.type == EventType.ERROR:
errors += 1
debug_print(
f"Purge: remove_contact({pubkey[:16]}) "
f"returned ERROR"
)
else:
removed += 1
debug_print(
f"Purge: removed {pubkey[:16]} "
f"({i}/{total})"
)
except Exception as exc:
errors += 1
debug_print(
f"Purge: remove_contact({pubkey[:16]}) "
f"exception: {exc}"
)
# Update status with progress
self._shared.set_status(
f"🗑️ Removing... {i}/{total}"
)
# Brief pause between BLE calls to avoid congestion
if i < total:
await asyncio.sleep(0.5)
# Summary
if errors:
status = (
f"⚠️ {removed} contacts removed, "
f"{errors} failed"
)
else:
status = f"✅ {removed} contacts removed from device"
self._shared.set_status(status)
print(f"Purge: {status}")
# Resync with device to confirm new state
if self._load_data_callback:
await self._load_data_callback()
async def _cmd_set_auto_add(self, cmd: Dict) -> None:
"""Toggle auto-add contacts on the MeshCore device.
The SDK function ``set_manual_add_contacts(true)`` means
*manual mode* (auto-add OFF). The UI toggle is inverted:
toggle ON = auto-add ON = ``set_manual_add_contacts(false)``.
On failure the SharedData flag is rolled back so the GUI
checkbox reverts on the next update cycle.
Expected command dict::
{
'action': 'set_auto_add',
'enabled': True/False,
}
"""
enabled: bool = cmd.get('enabled', False)
# Invert: UI "auto-add ON" → manual_add = False
manual_add = not enabled
try:
r = await self._mc.commands.set_manual_add_contacts(manual_add)
if r.type == EventType.ERROR:
# Rollback
self._shared.set_auto_add_enabled(not enabled)
self._shared.set_status(
"⚠️ Failed to change auto-add setting"
)
debug_print(
f"set_auto_add: ERROR response, rolled back to "
f"{'enabled' if not enabled else 'disabled'}"
)
else:
self._shared.set_auto_add_enabled(enabled)
state = "ON" if enabled else "OFF"
self._shared.set_status(f"✅ Auto-add contacts: {state}")
debug_print(f"set_auto_add: success → {state}")
except Exception as exc:
# Rollback
self._shared.set_auto_add_enabled(not enabled)
self._shared.set_status(
f"⚠️ Auto-add error: {exc}"
)
debug_print(f"set_auto_add exception: {exc}")
# ------------------------------------------------------------------
# Callback for refresh (set by BLEWorker after construction)
# ------------------------------------------------------------------

3
meshcore_gui/core/protocols.py
View File

@@ -59,6 +59,8 @@ class SharedDataWriter(Protocol):
def get_contact_by_name(self, name: str) -> Optional[tuple]: ...
def is_bot_enabled(self) -> bool: ...
def put_command(self, cmd: Dict) -> None: ...
def set_auto_add_enabled(self, enabled: bool) -> None: ...
def is_auto_add_enabled(self) -> bool: ...
# ----------------------------------------------------------------------
@@ -78,6 +80,7 @@ class SharedDataReader(Protocol):
def mark_gui_initialized(self) -> None: ...
def put_command(self, cmd: Dict) -> None: ...
def set_bot_enabled(self, enabled: bool) -> None: ...
def set_auto_add_enabled(self, enabled: bool) -> None: ...
# ----------------------------------------------------------------------

19
meshcore_gui/core/shared_data.py
View File

@@ -62,6 +62,9 @@ class SharedData:
# BOT enabled flag (toggled from GUI)
self.bot_enabled: bool = False
# Auto-add contacts flag (synced with device)
self.auto_add_enabled: bool = False
# Message archive (persistent storage)
self.archive: Optional[MessageArchive] = None
if ble_address:
@@ -121,6 +124,21 @@ class SharedData:
with self.lock:
return self.bot_enabled
# ------------------------------------------------------------------
# Auto-add contacts
# ------------------------------------------------------------------
def set_auto_add_enabled(self, enabled: bool) -> None:
"""Set auto-add contacts flag (thread-safe)."""
with self.lock:
self.auto_add_enabled = enabled
debug_print(f"Auto-add {'enabled' if enabled else 'disabled'}")
def is_auto_add_enabled(self) -> bool:
"""Get auto-add contacts flag (thread-safe)."""
with self.lock:
return self.auto_add_enabled
# ------------------------------------------------------------------
# Command queue
# ------------------------------------------------------------------
@@ -215,6 +233,7 @@ class SharedData:
'rxlog_updated': self.rxlog_updated,
'gui_initialized': self.gui_initialized,
'bot_enabled': self.bot_enabled,
'auto_add_enabled': self.auto_add_enabled,
# Archive (for archive viewer)
'archive': self.archive,
}

8
meshcore_gui/gui/dashboard.py
View File

@@ -21,6 +21,7 @@ from meshcore_gui.gui.panels import (
MessagesPanel,
RxLogPanel,
)
from meshcore_gui.services.pin_store import PinStore
# Suppress the harmless "Client has been deleted" warning that NiceGUI
@@ -39,8 +40,9 @@ class DashboardPage:
shared: SharedDataReader for data access and command dispatch.
"""
def __init__(self, shared: SharedDataReader) -> None:
def __init__(self, shared: SharedDataReader, pin_store: PinStore) -> None:
self._shared = shared
self._pin_store = pin_store
# Panels (created fresh on each render)
self._device: DevicePanel | None = None
@@ -69,7 +71,7 @@ class DashboardPage:
# Create panel instances
put_cmd = self._shared.put_command
self._device = DevicePanel()
self._contacts = ContactsPanel(put_cmd)
self._contacts = ContactsPanel(put_cmd, self._pin_store, self._shared.set_auto_add_enabled)
self._map = MapPanel()
self._input = InputPanel(put_cmd)
self._filter = FilterPanel(self._shared.set_bot_enabled)
@@ -88,7 +90,7 @@ class DashboardPage:
# Three-column layout
with ui.row().classes('w-full h-full gap-2 p-2'):
# Left column
with ui.column().classes('w-64 gap-2'):
with ui.column().classes('w-72 gap-2'):
self._device.render()
self._contacts.render()

186
meshcore_gui/gui/panels/contacts_panel.py
View File

@@ -1,10 +1,12 @@
"""Contacts panel — list of known mesh nodes with click-to-DM."""
from typing import Callable, Dict
from typing import Callable, Dict, Optional
from nicegui import ui
from meshcore_gui.gui.constants import TYPE_ICONS, TYPE_NAMES
from meshcore_gui.services.contact_cleaner import ContactCleanerService
from meshcore_gui.services.pin_store import PinStore
class ContactsPanel:
@@ -12,27 +14,65 @@ class ContactsPanel:
Args:
put_command: Callable to enqueue a command dict for the BLE worker.
pin_store: PinStore for persistent pin state.
"""
def __init__(self, put_command: Callable[[Dict], None]) -> None:
def __init__(
self,
put_command: Callable[[Dict], None],
pin_store: PinStore,
set_auto_add_enabled: Callable[[bool], None],
) -> None:
self._put_command = put_command
self._pin_store = pin_store
self._set_auto_add_enabled = set_auto_add_enabled
self._cleaner = ContactCleanerService(pin_store)
self._container = None
self._auto_add_checkbox = None
self._last_data: Optional[Dict] = None
def render(self) -> None:
with ui.card().classes('w-full'):
ui.label('👥 Contacts').classes('font-bold text-gray-600')
self._container = ui.column().classes(
'w-full gap-1 max-h-96 overflow-y-auto'
'w-full gap-0 max-h-96 overflow-y-auto'
)
with ui.row().classes('w-full gap-2 mt-2 items-center'):
ui.button(
'🧹 Clean up',
on_click=self._open_purge_dialog,
)
self._auto_add_checkbox = ui.checkbox(
'📥 Auto-add',
value=False,
on_change=self._on_auto_add_change,
)
def update(self, data: Dict) -> None:
if not self._container:
return
self._last_data = data
# Sync auto-add checkbox with device state
if self._auto_add_checkbox is not None:
device_state = data.get('auto_add_enabled', False)
if self._auto_add_checkbox.value != device_state:
self._auto_add_checkbox.set_value(device_state)
self._container.clear()
# Sort: pinned contacts first, then alphabetical within each group
contacts_items = list(data['contacts'].items())
contacts_items.sort(
key=lambda item: (
0 if self._pin_store.is_pinned(item[0]) else 1,
item[1].get('adv_name', item[0][:12]).lower(),
)
)
with self._container:
for key, contact in data['contacts'].items():
for key, contact in contacts_items:
ctype = contact.get('type', 0)
icon = TYPE_ICONS.get(ctype, '○')
name = contact.get('adv_name', key[:12])
@@ -40,6 +80,7 @@ class ContactsPanel:
lat = contact.get('adv_lat', 0)
lon = contact.get('adv_lon', 0)
has_loc = lat != 0 or lon != 0
pinned = self._pin_store.is_pinned(key)
tooltip = (
f"{name}\nType: {type_name}\n"
@@ -48,17 +89,132 @@ class ContactsPanel:
if has_loc:
tooltip += f"\nLat: {lat:.4f}\nLon: {lon:.4f}"
with ui.row().classes(
'w-full items-center gap-2 p-1 '
'hover:bg-gray-100 rounded cursor-pointer'
).on('click', lambda e, k=key, n=name: self._open_dm_dialog(k, n)):
ui.label(icon).classes('text-sm')
ui.label(name[:15]).classes(
'text-sm flex-grow truncate'
).tooltip(tooltip)
ui.label(type_name).classes('text-xs text-gray-500')
if has_loc:
ui.label('📍').classes('text-xs')
row_classes = (
'w-full items-center gap-1 py-0 px-1 '
'rounded no-wrap '
)
if pinned:
row_classes += 'bg-yellow-50'
# Outer row: checkbox + clickable contact info
with ui.row().classes(row_classes):
# Pin checkbox — click.stop prevents DM dialog opening
cb = ui.checkbox(
value=pinned,
).props('dense size=xs').on(
'click.stop', lambda e: None,
)
cb.on_value_change(
lambda e, k=key: self._toggle_pin(k)
)
# Clickable area for DM
with ui.row().classes(
'items-center gap-0.5 flex-grow '
'cursor-pointer hover:bg-gray-100 rounded py-0 px-1'
).on(
'click',
lambda e, k=key, n=name: self._open_dm_dialog(k, n),
):
ui.label(icon).classes('text-sm')
ui.label(name[:15]).classes(
'text-sm flex-grow truncate'
).tooltip(tooltip)
ui.label(type_name).classes('text-xs text-gray-500')
loc_icon = '📍' if has_loc else '✖'
loc_cls = 'text-xs w-4 text-center'
if not has_loc:
loc_cls += ' text-red-400'
ui.label(loc_icon).classes(loc_cls)
# ------------------------------------------------------------------
# Pin toggle
# ------------------------------------------------------------------
def _toggle_pin(self, pubkey: str) -> None:
"""Toggle pin state for a contact and refresh the list."""
if self._pin_store.is_pinned(pubkey):
self._pin_store.unpin(pubkey)
else:
self._pin_store.pin(pubkey)
# Re-render with last known data so sort order and visuals update
if self._last_data:
self.update(self._last_data)
# ------------------------------------------------------------------
# Auto-add toggle
# ------------------------------------------------------------------
def _on_auto_add_change(self, e) -> None:
"""Handle auto-add checkbox toggle.
Optimistically updates SharedData and sends the BLE command.
On failure, the command handler rolls back SharedData and the
next GUI update cycle will revert the checkbox.
"""
enabled = e.value
self._set_auto_add_enabled(enabled)
self._put_command({
'action': 'set_auto_add',
'enabled': enabled,
})
# ------------------------------------------------------------------
# Purge unpinned contacts
# ------------------------------------------------------------------
def _open_purge_dialog(self) -> None:
"""Open confirmation dialog for bulk-deleting unpinned contacts."""
if not self._last_data:
ui.notify('No contacts loaded', type='warning')
return
contacts = self._last_data.get('contacts', {})
if not contacts:
ui.notify('No contacts found', type='warning')
return
stats = self._cleaner.get_purge_stats(contacts)
if stats.unpinned_count == 0:
ui.notify(
'All contacts are pinned — nothing to remove',
type='info',
)
return
with ui.dialog() as dialog, ui.card().classes('w-96'):
ui.label('🧹 Clean up contacts').classes(
'font-bold text-lg'
)
ui.label(
f'{stats.unpinned_count} contacts will be removed from device.\n'
f'{stats.pinned_count} pinned contacts will be kept.'
).classes('whitespace-pre-line my-2')
with ui.row().classes('w-full justify-end gap-2 mt-4'):
ui.button('Cancel', on_click=dialog.close).props(
'flat'
)
def confirm_purge():
self._put_command({
'action': 'purge_unpinned',
'pubkeys': stats.unpinned_keys,
})
dialog.close()
ui.notify(
f'Removing {stats.unpinned_count} '
f'contacts...',
type='info',
)
ui.button(
'Remove',
on_click=confirm_purge,
).classes('bg-red-500 text-white')
dialog.open()
# ------------------------------------------------------------------
# DM dialog

75
meshcore_gui/services/contact_cleaner.py Normal file
View File

@@ -0,0 +1,75 @@
"""
Contact cleaner service for MeshCore GUI.
Provides business logic for bulk-deleting unpinned contacts from the
MeshCore device. All decision logic (which contacts to purge, counting
pinned vs unpinned) lives here — the GUI only calls this service and
displays results.
Thread safety
~~~~~~~~~~~~~
Methods read from SharedData (thread-safe) and PinStore (thread-safe).
No mutable state is stored in this service.
"""
from dataclasses import dataclass
from typing import Dict, List, Set
from meshcore_gui.services.pin_store import PinStore
@dataclass
class PurgeStats:
"""Statistics for a planned contact purge operation.
Attributes:
unpinned_keys: Public keys of contacts that will be removed.
pinned_count: Number of pinned contacts that will be kept.
total_count: Total number of contacts on the device.
"""
unpinned_keys: List[str]
pinned_count: int
total_count: int
@property
def unpinned_count(self) -> int:
"""Number of contacts that will be removed."""
return len(self.unpinned_keys)
class ContactCleanerService:
"""Business logic for bulk-deleting unpinned contacts.
Args:
pin_store: PinStore instance for checking pin status.
"""
def __init__(self, pin_store: PinStore) -> None:
self._pin_store = pin_store
def get_purge_stats(self, contacts: Dict) -> PurgeStats:
"""Calculate which contacts would be purged.
Iterates all contacts and separates them into pinned (kept)
and unpinned (to be removed).
Args:
contacts: Contacts dict from SharedData snapshot
(``{pubkey: contact_dict}``).
Returns:
PurgeStats with the list of unpinned keys and counts.
"""
pinned_keys: Set[str] = self._pin_store.get_pinned()
unpinned_keys: List[str] = []
for pubkey in contacts:
if pubkey not in pinned_keys:
unpinned_keys.append(pubkey)
return PurgeStats(
unpinned_keys=unpinned_keys,
pinned_count=len(contacts) - len(unpinned_keys),
total_count=len(contacts),
)

125
meshcore_gui/services/pin_store.py Normal file
View File

@@ -0,0 +1,125 @@
"""
Persistent pin store for MeshCore GUI.
Stores a set of pinned contact public keys per BLE device.
Pin status is purely app-side and is not stored on the device.
Storage location
~~~~~~~~~~~~~~~~
``~/.meshcore-gui/pins/<ADDRESS>.json``
Thread safety
~~~~~~~~~~~~~
All methods use an internal lock for thread-safe operation.
"""
import json
import threading
from pathlib import Path
from typing import Set
from meshcore_gui.config import debug_print
PINS_DIR = Path.home() / ".meshcore-gui" / "pins"
class PinStore:
"""Persistent storage for pinned contact public keys.
Args:
ble_address: BLE address string (used to derive filename).
"""
def __init__(self, ble_address: str) -> None:
self._lock = threading.Lock()
safe_name = (
ble_address
.replace("literal:", "")
.replace(":", "_")
.replace("/", "_")
)
self._path = PINS_DIR / f"{safe_name}_pins.json"
self._pinned: Set[str] = set()
self._load()
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def is_pinned(self, pubkey: str) -> bool:
"""Check if a contact is pinned.
Args:
pubkey: Full public key (hex string).
Returns:
True if the contact is pinned.
"""
with self._lock:
return pubkey in self._pinned
def pin(self, pubkey: str) -> None:
"""Pin a contact.
Args:
pubkey: Full public key (hex string).
"""
with self._lock:
self._pinned.add(pubkey)
self._save()
debug_print(f"PinStore: pinned {pubkey[:16]}")
def unpin(self, pubkey: str) -> None:
"""Unpin a contact.
Args:
pubkey: Full public key (hex string).
"""
with self._lock:
self._pinned.discard(pubkey)
self._save()
debug_print(f"PinStore: unpinned {pubkey[:16]}")
def get_pinned(self) -> Set[str]:
"""Return a copy of the set of pinned public keys.
Returns:
Set of pinned public key hex strings.
"""
with self._lock:
return self._pinned.copy()
# ------------------------------------------------------------------
# Persistence
# ------------------------------------------------------------------
def _load(self) -> None:
"""Load pinned contacts from disk."""
if not self._path.exists():
debug_print(f"PinStore: no file at {self._path}")
return
try:
data = json.loads(self._path.read_text(encoding="utf-8"))
self._pinned = set(data.get("pinned", []))
debug_print(
f"PinStore: loaded {len(self._pinned)} pinned contacts"
)
except (json.JSONDecodeError, OSError) as exc:
debug_print(f"PinStore: load error: {exc}")
self._pinned = set()
def _save(self) -> None:
"""Write pinned contacts to disk."""
try:
PINS_DIR.mkdir(parents=True, exist_ok=True)
data = {"pinned": sorted(self._pinned)}
self._path.write_text(
json.dumps(data, indent=2, ensure_ascii=False),
encoding="utf-8",
)
debug_print(f"PinStore: saved {len(self._pinned)} pins")
except OSError as exc:
debug_print(f"PinStore: save error: {exc}")