adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-10 19:03:22 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor LetsMesh handler initialization and add live stats support

Browse Source
This commit is contained in:
Lloyd
2025-11-19 11:28:00 +00:00
parent 1793973a84
commit 3a2466f953
3 changed files with 68 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
repeater/data_acquisition/letsmesh_handler.py
+36 -31
View File
@@ -6,6 +6,7 @@ import paho.mqtt.client as mqtt
from datetime import datetime, timedelta, UTC
from nacl.signing import SigningKey
from typing import Callable, Optional
from .. import __version__
# --------------------------------------------------------------------
@@ -19,12 +20,6 @@ def b64url(x: bytes) -> str:
# Let's Mesh MQTT Broker List (WebSocket Secure)
# --------------------------------------------------------------------
LETSMESH_BROKERS = [
# {
# "name": "test",
# "host": "localhost",
# "port": 8883,
# "audience": "mqtt.yourdomain.com"
# },
{
"name": "Europe (LetsMesh v1)",
"host": "mqtt-eu-v1.letsmesh.net",
@@ -61,15 +56,20 @@ class MeshCoreToMqttJwtPusher:
self,
private_key: str,
public_key: str,
iata_code: str,
broker_index: int = 0,
topic_prefix: str = "meshcore",
config: dict,
jwt_expiry_minutes: int = 10,
use_tls: bool = True,
status_interval: int = 60, # Heartbeat interval in seconds
node_name: str = None,
radio_config: str = None,
stats_provider: Optional[Callable[[], dict]] = None,
):
# Extract values from config
from ..config import get_node_info
node_info = get_node_info(config)
iata_code = node_info["iata_code"]
broker_index = node_info["broker_index"]
status_interval = node_info["status_interval"]
node_name = node_info["node_name"]
radio_config = node_info["radio_config"]
if broker_index >= len(LETSMESH_BROKERS):
raise ValueError(f"Invalid broker_index {broker_index}")
@@ -78,20 +78,15 @@ class MeshCoreToMqttJwtPusher:
self.private_key_hex = private_key
self.public_key = public_key.upper()
self.iata_code = iata_code
self.topic_prefix = topic_prefix
self.jwt_expiry_minutes = jwt_expiry_minutes
self.use_tls = use_tls
self.status_interval = status_interval
self.app_version = __version__
self.node_name = node_name or "PyMC-Repeater"
self.radio_config = radio_config or "0.0,0.0,0,0"
self.node_name = node_name
self.radio_config = radio_config
self.stats_provider = stats_provider
self._status_task = None
self._running = False
self._packet_stats = {
"packets_sent": 0,
"packets_received": 0,
"start_time": datetime.now(UTC)
}
# MQTT WebSocket client
self.client = mqtt.Client(
@@ -237,7 +232,7 @@ class MeshCoreToMqttJwtPusher:
}
def _topic(self, subtopic: str) -> str:
return f"{self.topic_prefix}/{self.iata_code}/{self.public_key}/{subtopic}"
return f"meshcore/{self.iata_code}/{self.public_key}/{subtopic}"
def publish_packet(self, pkt: dict, subtopic="packets", retain=False):
return self.publish(subtopic, self._process_packet(pkt), retain)
@@ -248,11 +243,16 @@ class MeshCoreToMqttJwtPusher:
"data": raw_hex,
"bytes": len(raw_hex) // 2
}
self._packet_stats["packets_sent"] += 1
return self.publish_packet(pkt, subtopic, retain)
def publish_status(self, state: str = "online", location: dict = None, extra_stats: dict = None,
origin: str = None, radio_config: str = None):
def publish_status(
self,
state: str = "online",
location: Optional[dict] = None,
extra_stats: Optional[dict] = None,
origin: Optional[str] = None,
radio_config: Optional[str] = None,
):
"""
Publish device status/heartbeat message
@@ -263,21 +263,26 @@ class MeshCoreToMqttJwtPusher:
origin: Node name/description
radio_config: Radio configuration string (freq,bw,sf,cr)
"""
uptime_secs = int((datetime.now(UTC) - self._packet_stats["start_time"]).total_seconds())
# Get live stats from provider if available
if self.stats_provider:
live_stats = self.stats_provider()
else:
live_stats = {
"uptime_secs": 0,
"packets_sent": 0,
"packets_received": 0
}
status = {
"status": state,
"timestamp": datetime.now(UTC).isoformat(),
"origin": origin or "PyMC-Repeater",
"origin": origin or self.node_name,
"origin_id": self.public_key,
# "model": self.model,
"firmware_version": self.app_version,
"radio": radio_config or "0.0,0.0,0,0",
"radio": radio_config or self.radio_config,
"client_version": f"pyMC_repeater_{self.app_version}",
"stats": {
"uptime_secs": uptime_secs,
"packets_sent": self._packet_stats["packets_sent"],
"packets_received": self._packet_stats["packets_received"],
**live_stats,
"errors": 0,
"queue_len": 0,
**(extra_stats or {})
repeater/data_acquisition/storage_collector.py
+31 -31
View File
@@ -15,8 +15,9 @@ logger = logging.getLogger("StorageCollector")
class StorageCollector:
def __init__(self, config: dict, local_identity=None):
def __init__(self, config: dict, local_identity=None, repeater_handler=None):
self.config = config
self.repeater_handler = repeater_handler
self.storage_dir = Path(config.get("storage_dir", "/var/lib/pymc_repeater"))
self.storage_dir.mkdir(parents=True, exist_ok=True)
@@ -28,41 +29,40 @@ class StorageCollector:
# Initialize LetsMesh handler if configured
self.letsmesh_handler = None
letsmesh_config = config.get("letsmesh", {})
if letsmesh_config.get("enabled", False):
if config.get("letsmesh", {}).get("enabled", False) and local_identity:
try:
if not local_identity:
logger.error("Cannot initialize LetsMesh: No local_identity provided")
else:
identity_key = config.get("mesh", {}).get("identity_key")
if not identity_key:
logger.error("Cannot initialize LetsMesh: No identity_key found in mesh config")
else:
from ..config import get_node_info
private_key_hex = identity_key.hex()
public_key_hex = local_identity.get_public_key().hex()
# Get all config from config module
node_info = get_node_info(self.config)
self.letsmesh_handler = MeshCoreToMqttJwtPusher(
private_key=private_key_hex,
public_key=public_key_hex,
iata_code=node_info["iata_code"],
broker_index=node_info["broker_index"],
status_interval=node_info["status_interval"],
node_name=node_info["node_name"],
radio_config=node_info["radio_config"]
)
self.letsmesh_handler.connect()
logger.info(f"LetsMesh handler initialized with public key: {public_key_hex[:16]}...")
# Get keys from local_identity (signing_key.encode() is the private key seed)
private_key_hex = local_identity.signing_key.encode().hex()
public_key_hex = local_identity.get_public_key().hex()
self.letsmesh_handler = MeshCoreToMqttJwtPusher(
private_key=private_key_hex,
public_key=public_key_hex,
config=config,
stats_provider=self._get_live_stats
)
self.letsmesh_handler.connect()
logger.info(f"LetsMesh handler initialized with public key: {public_key_hex[:16]}...")
except Exception as e:
logger.error(f"Failed to initialize LetsMesh handler: {e}")
self.letsmesh_handler = None
def _get_live_stats(self) -> dict:
"""Get live stats from RepeaterHandler"""
if not self.repeater_handler:
return {
"uptime_secs": 0,
"packets_sent": 0,
"packets_received": 0
}
uptime_secs = int(time.time() - self.repeater_handler.start_time)
return {
"uptime_secs": uptime_secs,
"packets_sent": self.repeater_handler.forwarded_count,
"packets_received": self.repeater_handler.rx_count
}
def record_packet(self, packet_record: dict):
logger.debug(f"Recording packet: type={packet_record.get('type')}, transmitted={packet_record.get('transmitted')}")
repeater/engine.py
+1 -1
View File
@@ -78,7 +78,7 @@ class RepeaterHandler(BaseHandler):
try:
local_identity = dispatcher.local_identity if dispatcher else None
self.storage = StorageCollector(config, local_identity)
self.storage = StorageCollector(config, local_identity, repeater_handler=self)
logger.info("StorageCollector initialized successfully")
except Exception as e:
logger.error(f"Failed to initialize StorageCollector: {e}")