mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-12 11:52:55 +02:00
feat: Merge mqtt handler and letsmesh handlers
This commit is contained in:
Rigear
+8
-18
@@ -11,13 +11,13 @@ logger = logging.getLogger("Config")
|
||||
|
||||
def get_node_info(config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Extract node name, radio configuration, and LetsMesh settings from config.
|
||||
Extract node name, radio configuration, and MQTT settings from config.
|
||||
|
||||
Args:
|
||||
config: Configuration dictionary
|
||||
|
||||
Returns:
|
||||
Dictionary with node_name, radio_config, and LetsMesh configuration
|
||||
Dictionary with node_name, radio_config, and MQTT configuration
|
||||
"""
|
||||
node_name = config.get("repeater", {}).get("node_name", "PyMC-Repeater")
|
||||
radio_config = config.get("radio", {})
|
||||
@@ -30,26 +30,16 @@ def get_node_info(config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
radio_bw_khz = radio_bw / 1_000
|
||||
radio_config_str = f"{radio_freq_mhz},{radio_bw_khz},{radio_sf},{radio_cr}"
|
||||
|
||||
letsmesh_config = config.get("letsmesh", {})
|
||||
|
||||
from pymc_core.protocol.utils import PAYLOAD_TYPES
|
||||
|
||||
disallowed_types = letsmesh_config.get("disallowed_packet_types", [])
|
||||
type_name_map = {name: code for code, name in PAYLOAD_TYPES.items()}
|
||||
|
||||
disallowed_hex = [type_name_map.get(name.upper(), None) for name in disallowed_types]
|
||||
disallowed_hex = [val for val in disallowed_hex if val is not None] # Filter out invalid names
|
||||
mqtt_config = config.get("mqtt", {})
|
||||
|
||||
return {
|
||||
"node_name": node_name,
|
||||
"radio_config": radio_config_str,
|
||||
"iata_code": letsmesh_config.get("iata_code", "TEST"),
|
||||
"broker_index": letsmesh_config.get("broker_index", 0),
|
||||
"status_interval": letsmesh_config.get("status_interval", 60),
|
||||
"model": letsmesh_config.get("model", "PyMC-Repeater"),
|
||||
"disallowed_packet_types": disallowed_hex,
|
||||
"email": letsmesh_config.get("email", ""),
|
||||
"owner": letsmesh_config.get("owner", ""),
|
||||
"iata_code": mqtt_config.get("iata_code", "TEST"),
|
||||
"status_interval": mqtt_config.get("status_interval", 60),
|
||||
"model": mqtt_config.get("model", "PyMC-Repeater"),
|
||||
"email": mqtt_config.get("email", ""),
|
||||
"owner": mqtt_config.get("owner", ""),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from .mqtt_handler import MQTTHandler
|
||||
#from .mqtt_handler import MQTTHandler
|
||||
from .rrdtool_handler import RRDToolHandler
|
||||
from .sqlite_handler import SQLiteHandler
|
||||
from .storage_collector import StorageCollector
|
||||
|
||||
__all__ = ["SQLiteHandler", "RRDToolHandler", "MQTTHandler", "StorageCollector"]
|
||||
#__all__ = ["SQLiteHandler", "RRDToolHandler", "MQTTHandler", "StorageCollector"]
|
||||
__all__ = ["SQLiteHandler", "RRDToolHandler", "StorageCollector"]
|
||||
|
||||
@@ -1,694 +0,0 @@
|
||||
import base64
|
||||
import binascii
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Callable, Dict, List, Optional
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
from nacl.signing import SigningKey
|
||||
|
||||
# Try to import datetime.UTC (Python 3.11+) otherwise fallback to timezone.utc
|
||||
try:
|
||||
from datetime import UTC
|
||||
except Exception:
|
||||
from datetime import timezone
|
||||
UTC = timezone.utc
|
||||
|
||||
from repeater import __version__
|
||||
|
||||
# Try to import paho-mqtt error code mappings
|
||||
try:
|
||||
from paho.mqtt.reasoncodes import ReasonCode
|
||||
|
||||
HAS_REASON_CODES = True
|
||||
except ImportError:
|
||||
HAS_REASON_CODES = False
|
||||
|
||||
logger = logging.getLogger("LetsMeshHandler")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# Helper: Base64URL without padding
|
||||
# --------------------------------------------------------------------
|
||||
def b64url(x: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(x).rstrip(b"=").decode()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# Let's Mesh MQTT Broker List (WebSocket Secure)
|
||||
# --------------------------------------------------------------------
|
||||
LETSMESH_BROKERS = [
|
||||
{
|
||||
"name": "Europe (LetsMesh v1)",
|
||||
"host": "mqtt-eu-v1.letsmesh.net",
|
||||
"port": 443,
|
||||
"audience": "mqtt-eu-v1.letsmesh.net",
|
||||
},
|
||||
{
|
||||
"name": "US West (LetsMesh v1)",
|
||||
"host": "mqtt-us-v1.letsmesh.net",
|
||||
"port": 443,
|
||||
"audience": "mqtt-us-v1.letsmesh.net",
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Single Broker Connection Manager
|
||||
# ====================================================================
|
||||
class _BrokerConnection:
|
||||
"""
|
||||
Manages a single MQTT broker connection with independent lifecycle.
|
||||
Internal class - not exposed publicly.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
broker: dict,
|
||||
local_identity,
|
||||
public_key: str,
|
||||
iata_code: str,
|
||||
jwt_expiry_minutes: int,
|
||||
use_tls: bool,
|
||||
email: str,
|
||||
owner: str,
|
||||
broker_index: int = 0,
|
||||
on_connect_callback: Optional[Callable] = None,
|
||||
on_disconnect_callback: Optional[Callable] = None,
|
||||
):
|
||||
self.broker = broker
|
||||
self.local_identity = local_identity
|
||||
self.public_key = public_key.upper()
|
||||
self.iata_code = iata_code
|
||||
self.jwt_expiry_minutes = jwt_expiry_minutes
|
||||
self.broker_index = broker_index
|
||||
self.use_tls = use_tls
|
||||
self.email = email
|
||||
self.owner = owner
|
||||
self._on_connect_callback = on_connect_callback
|
||||
self._on_disconnect_callback = on_disconnect_callback
|
||||
self._connect_time = None
|
||||
self._tls_verified = False
|
||||
self._running = False
|
||||
self._reconnect_attempts = 0
|
||||
self._reconnect_timer = None
|
||||
self._max_reconnect_delay = 300 # 5 minutes max
|
||||
self._jwt_refresh_timer = None
|
||||
client_id = f"meshcore_{self.public_key}_{broker['host']}"
|
||||
self.client = mqtt.Client(client_id=client_id, transport="websockets")
|
||||
self.client.on_connect = self._on_connect
|
||||
self.client.on_disconnect = self._on_disconnect
|
||||
|
||||
def _generate_jwt(self) -> str:
|
||||
"""Generate MeshCore-style Ed25519 JWT token"""
|
||||
now = datetime.now(UTC)
|
||||
|
||||
header = {"alg": "Ed25519", "typ": "JWT"}
|
||||
|
||||
payload = {
|
||||
"publicKey": self.public_key.upper(),
|
||||
"aud": self.broker["audience"],
|
||||
"iat": int(now.timestamp()),
|
||||
"exp": int((now + timedelta(minutes=self.jwt_expiry_minutes)).timestamp()),
|
||||
}
|
||||
|
||||
# Only include email/owner for verified TLS connections
|
||||
if self.use_tls and self._tls_verified and (self.email or self.owner):
|
||||
payload["email"] = self.email
|
||||
payload["owner"] = self.owner
|
||||
else:
|
||||
payload["email"] = ""
|
||||
payload["owner"] = ""
|
||||
|
||||
# Encode header and payload (compact JSON - no spaces)
|
||||
header_b64 = b64url(json.dumps(header, separators=(",", ":")).encode())
|
||||
payload_b64 = b64url(json.dumps(payload, separators=(",", ":")).encode())
|
||||
|
||||
signing_input = f"{header_b64}.{payload_b64}".encode()
|
||||
|
||||
# Sign using LocalIdentity (supports both standard and firmware keys)
|
||||
try:
|
||||
signature = self.local_identity.sign(signing_input)
|
||||
except Exception as e:
|
||||
logger.error(f"JWT signing failed for {self.broker['name']}: {e}")
|
||||
logger.error(f" - public_key: {self.public_key}")
|
||||
logger.error(f" - signing_input length: {len(signing_input)}")
|
||||
raise
|
||||
|
||||
signature_hex = binascii.hexlify(signature).decode()
|
||||
token = f"{header_b64}.{payload_b64}.{signature_hex}"
|
||||
|
||||
logger.debug(f"JWT token generated for {self.broker['name']}: {token[:50]}...")
|
||||
|
||||
return token
|
||||
|
||||
def _on_connect(self, client, userdata, flags, rc):
|
||||
"""MQTT connection callback"""
|
||||
if rc == 0:
|
||||
logger.info(f"Connected to {self.broker['name']}")
|
||||
self._running = True
|
||||
self._reconnect_attempts = 0 # Reset counter on success
|
||||
self._schedule_jwt_refresh() # Schedule proactive JWT refresh
|
||||
if self._on_connect_callback:
|
||||
self._on_connect_callback(self.broker["name"])
|
||||
else:
|
||||
error_msg = get_mqtt_error_message(rc, is_disconnect=False)
|
||||
logger.error(f"Failed to connect to {self.broker['name']}: {error_msg}")
|
||||
self._schedule_reconnect()
|
||||
|
||||
def _on_disconnect(self, client, userdata, rc):
|
||||
"""MQTT disconnection callback"""
|
||||
was_running = self._running
|
||||
self._running = False
|
||||
|
||||
if rc != 0: # Unexpected disconnect
|
||||
error_msg = get_mqtt_error_message(rc, is_disconnect=True)
|
||||
logger.warning(f"Disconnected from {self.broker['name']} (rc={rc}): {error_msg}")
|
||||
if was_running: # Only reconnect if we were intentionally connected
|
||||
self._schedule_reconnect(reason=error_msg)
|
||||
else:
|
||||
logger.info(f"Clean disconnect from {self.broker['name']}")
|
||||
|
||||
if self._on_disconnect_callback:
|
||||
self._on_disconnect_callback(self.broker["name"])
|
||||
|
||||
def _schedule_reconnect(self, reason: str = "connection lost"):
|
||||
"""Schedule reconnection with exponential backoff"""
|
||||
if self._reconnect_timer:
|
||||
self._reconnect_timer.cancel()
|
||||
|
||||
# Exponential backoff: 5s, 10s, 20s, 40s, 80s, up to max
|
||||
delay = min(5 * (2**self._reconnect_attempts), self._max_reconnect_delay)
|
||||
self._reconnect_attempts += 1
|
||||
|
||||
logger.info(
|
||||
f"Scheduling reconnect to {self.broker['name']} in {delay}s (attempt {self._reconnect_attempts}, reason: {reason})"
|
||||
)
|
||||
self._reconnect_timer = threading.Timer(delay, lambda: self._attempt_reconnect(reason))
|
||||
self._reconnect_timer.daemon = True
|
||||
self._reconnect_timer.start()
|
||||
|
||||
def _attempt_reconnect(self, reason: str = "connection lost"):
|
||||
"""Attempt to reconnect to broker with fresh JWT"""
|
||||
try:
|
||||
logger.info(f"Attempting reconnection to {self.broker['name']} (reason: {reason})...")
|
||||
|
||||
# Stop the loop if it's still running (websocket mode requires clean restart)
|
||||
try:
|
||||
self.client.loop_stop()
|
||||
except:
|
||||
pass
|
||||
|
||||
self._set_jwt_credentials()
|
||||
|
||||
# Reconnect and restart loop
|
||||
self.client.connect(self.broker["host"], self.broker["port"], keepalive=60)
|
||||
self.client.loop_start()
|
||||
self._loop_running = True
|
||||
except Exception as e:
|
||||
logger.error(f"Reconnection failed for {self.broker['name']}: {e}")
|
||||
self._schedule_reconnect() # Try again later
|
||||
|
||||
def _set_jwt_credentials(self):
|
||||
"""Set JWT token credentials before connecting (CONNECT handshake only)"""
|
||||
try:
|
||||
token = self._generate_jwt()
|
||||
username = f"v1_{self.public_key}"
|
||||
self.client.username_pw_set(username=username, password=token)
|
||||
self._connect_time = datetime.now(UTC)
|
||||
logger.debug(f"JWT credentials set for {self.broker['name']}")
|
||||
logger.debug(f"Using username: {username}")
|
||||
logger.debug(f"Public key: {self.public_key[:16]}...{self.public_key[-16:]}")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to set JWT credentials for {self.broker['name']}: {e}")
|
||||
raise
|
||||
|
||||
def connect(self):
|
||||
"""Establish connection to broker"""
|
||||
# Conditional TLS setup
|
||||
if self.use_tls:
|
||||
import ssl
|
||||
|
||||
self.client.tls_set(cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS_CLIENT)
|
||||
self.client.tls_insecure_set(False)
|
||||
self._tls_verified = True
|
||||
protocol = "wss"
|
||||
else:
|
||||
protocol = "ws"
|
||||
|
||||
# Set JWT credentials before CONNECT handshake
|
||||
self._set_jwt_credentials()
|
||||
|
||||
logger.info(
|
||||
f"Connecting to {self.broker['name']} "
|
||||
f"({protocol}://{self.broker['host']}:{self.broker['port']}) ..."
|
||||
)
|
||||
|
||||
self.client.connect(self.broker["host"], self.broker["port"], keepalive=60)
|
||||
self.client.loop_start()
|
||||
self._loop_running = True
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from broker"""
|
||||
self._running = False
|
||||
self._loop_running = False
|
||||
|
||||
# Cancel any pending timers
|
||||
if self._reconnect_timer:
|
||||
self._reconnect_timer.cancel()
|
||||
self._reconnect_timer = None
|
||||
if self._jwt_refresh_timer:
|
||||
self._jwt_refresh_timer.cancel()
|
||||
self._jwt_refresh_timer = None
|
||||
|
||||
self.client.loop_stop()
|
||||
self.client.disconnect()
|
||||
logger.info(f"Disconnected from {self.broker['name']}")
|
||||
|
||||
def publish(self, topic: str, payload: str, retain: bool = False):
|
||||
"""Publish message to broker"""
|
||||
if self._running:
|
||||
result = self.client.publish(topic, payload, retain=retain)
|
||||
return result
|
||||
return None
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if connection is active"""
|
||||
return self._running
|
||||
|
||||
def has_pending_reconnect(self) -> bool:
|
||||
"""Check if a reconnection is scheduled"""
|
||||
return self._reconnect_timer is not None and self._reconnect_timer.is_alive()
|
||||
|
||||
def should_reconnect_for_token_expiry(self) -> bool:
|
||||
"""Check if connection should be reconnected due to JWT expiry (at 80% of lifetime)"""
|
||||
if not self._connect_time:
|
||||
return False
|
||||
elapsed = (datetime.now(UTC) - self._connect_time).total_seconds()
|
||||
expiry_seconds = self.jwt_expiry_minutes * 60
|
||||
# Stagger refresh by 5% per broker to prevent simultaneous disconnects
|
||||
# Broker 0: 80%, Broker 1: 85%, Broker 2: 90%, etc.
|
||||
stagger_offset = self.broker_index * 0.05
|
||||
refresh_threshold = 0.80 + stagger_offset
|
||||
return elapsed >= expiry_seconds * refresh_threshold
|
||||
|
||||
def _schedule_jwt_refresh(self):
|
||||
"""Schedule proactive JWT refresh before token expires"""
|
||||
if self._jwt_refresh_timer:
|
||||
self._jwt_refresh_timer.cancel()
|
||||
|
||||
expiry_seconds = self.jwt_expiry_minutes * 60
|
||||
# Stagger refresh by 5% per broker to prevent simultaneous disconnects
|
||||
# Broker 0: 80%, Broker 1: 85%, Broker 2: 90%, etc.
|
||||
stagger_offset = self.broker_index * 0.05
|
||||
refresh_threshold = 0.80 + stagger_offset
|
||||
refresh_delay = expiry_seconds * refresh_threshold
|
||||
|
||||
logger.info(
|
||||
f"JWT refresh scheduled for {self.broker['name']} in {refresh_delay:.0f}s "
|
||||
f"({refresh_threshold*100:.0f}% of {self.jwt_expiry_minutes}min token lifetime)"
|
||||
)
|
||||
self._jwt_refresh_timer = threading.Timer(refresh_delay, self.reconnect_for_token_expiry)
|
||||
self._jwt_refresh_timer.daemon = True
|
||||
self._jwt_refresh_timer.start()
|
||||
|
||||
def reconnect_for_token_expiry(self):
|
||||
"""Proactively reconnect with new JWT before current one expires"""
|
||||
if not self._running:
|
||||
return
|
||||
|
||||
logger.info(f"JWT token expiring soon for {self.broker['name']}, refreshing...")
|
||||
self._running = False
|
||||
self._jwt_refresh_timer = None
|
||||
|
||||
self._schedule_reconnect(reason="JWT token expiry")
|
||||
self.client.disconnect()
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# MeshCore → MQTT Publisher with Ed25519 auth token
|
||||
# ====================================================================
|
||||
class MeshCoreToMqttJwtPusher:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
local_identity,
|
||||
config: dict,
|
||||
jwt_expiry_minutes: int = 10,
|
||||
use_tls: bool = True,
|
||||
stats_provider: Optional[Callable[[], dict]] = None,
|
||||
):
|
||||
# Store local identity and get public key
|
||||
self.local_identity = local_identity
|
||||
public_key = local_identity.get_public_key().hex().upper()
|
||||
|
||||
# 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.get("broker_index")
|
||||
self.email = node_info.get("email", "")
|
||||
self.owner = node_info.get("owner", "")
|
||||
status_interval = node_info["status_interval"]
|
||||
node_name = node_info["node_name"]
|
||||
radio_config = node_info["radio_config"]
|
||||
|
||||
# Get additional brokers from config (optional)
|
||||
letsmesh_config = config.get("letsmesh", {})
|
||||
additional_brokers = letsmesh_config.get("additional_brokers", [])
|
||||
|
||||
# Determine which brokers to connect to
|
||||
if broker_index == -2:
|
||||
# Custom brokers only - no built-in brokers
|
||||
self.brokers = []
|
||||
logger.info("Custom broker mode: using only user-defined brokers")
|
||||
elif broker_index is None or broker_index == -1:
|
||||
# Connect to all built-in brokers + additional ones
|
||||
self.brokers = LETSMESH_BROKERS.copy()
|
||||
logger.info(
|
||||
f"Multi-broker mode: connecting to all {len(LETSMESH_BROKERS)} built-in brokers"
|
||||
)
|
||||
else:
|
||||
|
||||
if broker_index >= len(LETSMESH_BROKERS):
|
||||
raise ValueError(f"Invalid broker_index {broker_index}")
|
||||
self.brokers = [LETSMESH_BROKERS[broker_index]]
|
||||
logger.info(f"Single broker mode: connecting to {self.brokers[0]['name']}")
|
||||
|
||||
# Add additional brokers from config
|
||||
if additional_brokers:
|
||||
for broker_config in additional_brokers:
|
||||
if all(k in broker_config for k in ["name", "host", "port", "audience"]):
|
||||
self.brokers.append(broker_config)
|
||||
logger.info(f"Added custom broker: {broker_config['name']}")
|
||||
else:
|
||||
logger.warning(f"Skipping invalid broker config: {broker_config}")
|
||||
|
||||
# Validate that we have at least one broker
|
||||
if not self.brokers:
|
||||
raise ValueError(
|
||||
"No brokers configured. Either set broker_index to a valid value "
|
||||
"or provide additional_brokers in config."
|
||||
)
|
||||
|
||||
self.local_identity = local_identity
|
||||
self.public_key = public_key
|
||||
self.iata_code = iata_code
|
||||
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
|
||||
self.radio_config = radio_config
|
||||
self.stats_provider = stats_provider
|
||||
self._status_task = None
|
||||
self._running = False
|
||||
self._lock = threading.Lock()
|
||||
|
||||
# Create broker connections
|
||||
self.connections: List[_BrokerConnection] = []
|
||||
for idx, broker in enumerate(self.brokers):
|
||||
conn = _BrokerConnection(
|
||||
broker=broker,
|
||||
local_identity=self.local_identity,
|
||||
public_key=self.public_key,
|
||||
iata_code=self.iata_code,
|
||||
jwt_expiry_minutes=self.jwt_expiry_minutes,
|
||||
use_tls=self.use_tls,
|
||||
email=self.email,
|
||||
owner=self.owner,
|
||||
broker_index=idx,
|
||||
on_connect_callback=self._on_broker_connected,
|
||||
on_disconnect_callback=self._on_broker_disconnected,
|
||||
)
|
||||
self.connections.append(conn)
|
||||
|
||||
logger.info(f"Initialized with {len(self.connections)} broker connection(s)")
|
||||
|
||||
def _on_broker_connected(self, broker_name: str):
|
||||
"""Callback when a broker connects"""
|
||||
# Publish initial status on first connection
|
||||
if not self._status_task and self.status_interval > 0:
|
||||
self._running = True
|
||||
self.publish_status(
|
||||
state="online", origin=self.node_name, radio_config=self.radio_config
|
||||
)
|
||||
# Start heartbeat thread
|
||||
self._status_task = threading.Thread(target=self._status_heartbeat_loop, daemon=True)
|
||||
self._status_task.start()
|
||||
logger.info(f"Started status heartbeat (interval: {self.status_interval}s)")
|
||||
|
||||
def _on_broker_disconnected(self, broker_name: str):
|
||||
"""Callback when a broker disconnects"""
|
||||
# Check if all connections are down AND none have pending reconnects
|
||||
all_down = all(not conn.is_connected() for conn in self.connections)
|
||||
any_reconnecting = any(conn.has_pending_reconnect() for conn in self.connections)
|
||||
|
||||
if all_down and not any_reconnecting:
|
||||
logger.warning("All broker connections lost with no pending reconnects")
|
||||
elif all_down:
|
||||
logger.info("All brokers temporarily disconnected, reconnects pending")
|
||||
|
||||
def connect(self):
|
||||
"""Establish connections to all configured brokers"""
|
||||
for idx, conn in enumerate(self.connections):
|
||||
try:
|
||||
if idx == 0:
|
||||
# Connect first broker immediately
|
||||
conn.connect()
|
||||
else:
|
||||
# Stagger additional brokers using background timers
|
||||
delay = idx * 30
|
||||
logger.info(f"Staggering connection to {conn.broker['name']} by {delay}s")
|
||||
timer = threading.Timer(delay, lambda c=conn: self._delayed_connect(c))
|
||||
timer.daemon = True
|
||||
timer.start()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to {conn.broker['name']}: {e}")
|
||||
|
||||
def _delayed_connect(self, conn):
|
||||
"""Connect a broker after a delay (called by timer)"""
|
||||
try:
|
||||
conn.connect()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to {conn.broker['name']}: {e}")
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from all brokers"""
|
||||
# Stop the heartbeat loop
|
||||
self._running = False
|
||||
|
||||
# Publish offline status before disconnecting
|
||||
self.publish_status(state="offline", origin=self.node_name, radio_config=self.radio_config)
|
||||
|
||||
import time
|
||||
|
||||
time.sleep(0.5) # Give time for messages to be sent
|
||||
|
||||
# Disconnect all brokers
|
||||
for conn in self.connections:
|
||||
try:
|
||||
conn.disconnect()
|
||||
except Exception as e:
|
||||
logger.error(f"Error disconnecting from {conn.broker['name']}: {e}")
|
||||
|
||||
logger.info("Disconnected from all brokers")
|
||||
|
||||
def _status_heartbeat_loop(self):
|
||||
"""Background thread that publishes periodic status updates"""
|
||||
import time
|
||||
|
||||
while self._running:
|
||||
try:
|
||||
# Publish status (JWT refresh now handled by individual broker timers)
|
||||
self.publish_status(
|
||||
state="online", origin=self.node_name, radio_config=self.radio_config
|
||||
)
|
||||
logger.debug(f"Status heartbeat sent (next in {self.status_interval}s)")
|
||||
|
||||
time.sleep(self.status_interval)
|
||||
except Exception as e:
|
||||
logger.error(f"Status heartbeat error: {e}")
|
||||
time.sleep(self.status_interval)
|
||||
|
||||
# ----------------------------------------------------------------
|
||||
# Packet helpers
|
||||
# ----------------------------------------------------------------
|
||||
def _process_packet(self, pkt: dict) -> dict:
|
||||
return {"timestamp": datetime.now(UTC).isoformat(), "origin_id": self.public_key, **pkt}
|
||||
|
||||
def _topic(self, subtopic: str) -> str:
|
||||
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)
|
||||
|
||||
def publish_raw_data(self, raw_hex: str, subtopic="raw", retain=False):
|
||||
pkt = {"type": "raw", "data": raw_hex, "bytes": len(raw_hex) // 2}
|
||||
return self.publish_packet(pkt, subtopic, retain)
|
||||
|
||||
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
|
||||
|
||||
Args:
|
||||
state: Device state (online/offline)
|
||||
location: Optional dict with latitude/longitude
|
||||
extra_stats: Optional additional statistics to include
|
||||
origin: Node name/description
|
||||
radio_config: Radio configuration string (freq,bw,sf,cr)
|
||||
"""
|
||||
# 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 self.node_name,
|
||||
"origin_id": self.public_key,
|
||||
"model": "PyMC-Repeater",
|
||||
"firmware_version": self.app_version,
|
||||
"radio": radio_config or self.radio_config,
|
||||
"client_version": f"pyMC_repeater/{self.app_version}",
|
||||
"stats": {**live_stats, "errors": 0, "queue_len": 0, **(extra_stats or {})},
|
||||
}
|
||||
|
||||
if location:
|
||||
status["location"] = location
|
||||
|
||||
return self.publish("status", status, retain=False)
|
||||
|
||||
def publish(self, subtopic: str, payload: dict, retain: bool = False):
|
||||
"""Publish message to all connected brokers"""
|
||||
topic = self._topic(subtopic)
|
||||
message = json.dumps(payload)
|
||||
|
||||
results = []
|
||||
with self._lock:
|
||||
for conn in self.connections:
|
||||
if conn.is_connected():
|
||||
result = conn.publish(topic, message, retain=retain)
|
||||
results.append((conn.broker["name"], result))
|
||||
logger.debug(f"Published to {conn.broker['name']}/{topic}")
|
||||
|
||||
if not results:
|
||||
logger.warning(f"No active broker connections for publishing to {topic}")
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Helper Functions
|
||||
# ====================================================================
|
||||
|
||||
|
||||
def get_mqtt_error_message(rc: int, is_disconnect: bool = False) -> str:
|
||||
"""
|
||||
Get human-readable MQTT error message.
|
||||
|
||||
Args:
|
||||
rc: Return code from paho-mqtt
|
||||
is_disconnect: True if from on_disconnect, False if from on_connect
|
||||
|
||||
Returns:
|
||||
Human-readable error message
|
||||
"""
|
||||
if HAS_REASON_CODES:
|
||||
try:
|
||||
# ReasonCode object has getName() method and value property
|
||||
reason = ReasonCode(mqtt.CONNACK if not is_disconnect else mqtt.DISCONNECT, identifier=rc)
|
||||
name = reason.getName() if hasattr(reason, 'getName') else str(reason)
|
||||
return f"{name} (code {rc})"
|
||||
except Exception as e:
|
||||
# Log the exception for debugging
|
||||
logger.debug(f"Could not decode reason code {rc}: {e}")
|
||||
|
||||
# Fallback to manual mappings - Extended with MQTT v5 codes
|
||||
connect_errors = {
|
||||
0: "Connection accepted",
|
||||
1: "Incorrect protocol version",
|
||||
2: "Invalid client identifier",
|
||||
3: "Server unavailable",
|
||||
4: "Bad username or password (JWT invalid)",
|
||||
5: "Not authorized (JWT signature/format invalid)",
|
||||
# MQTT v5 codes
|
||||
128: "Unspecified error",
|
||||
129: "Malformed packet",
|
||||
130: "Protocol error",
|
||||
131: "Implementation specific error",
|
||||
132: "Unsupported protocol version",
|
||||
133: "Client identifier not valid",
|
||||
134: "Bad username or password",
|
||||
135: "Not authorized",
|
||||
136: "Server unavailable",
|
||||
137: "Server busy",
|
||||
138: "Banned",
|
||||
140: "Bad authentication method",
|
||||
144: "Topic name invalid",
|
||||
149: "Packet too large",
|
||||
151: "Quota exceeded",
|
||||
153: "Payload format invalid",
|
||||
154: "Retain not supported",
|
||||
155: "QoS not supported",
|
||||
156: "Use another server",
|
||||
157: "Server moved",
|
||||
159: "Connection rate exceeded",
|
||||
}
|
||||
|
||||
disconnect_errors = {
|
||||
0: "Normal disconnect",
|
||||
1: "Unacceptable protocol version",
|
||||
2: "Identifier rejected",
|
||||
3: "Server unavailable",
|
||||
4: "Bad username or password",
|
||||
5: "Not authorized",
|
||||
7: "Connection lost / network error",
|
||||
16: "Connection lost / protocol error",
|
||||
17: "Client timeout",
|
||||
# MQTT v5 codes
|
||||
4: "Disconnect with Will message",
|
||||
128: "Unspecified error",
|
||||
129: "Malformed packet",
|
||||
130: "Protocol error",
|
||||
131: "Implementation specific error",
|
||||
135: "Not authorized",
|
||||
137: "Server busy",
|
||||
139: "Server shutting down",
|
||||
141: "Keep alive timeout",
|
||||
142: "Session taken over",
|
||||
143: "Topic filter invalid",
|
||||
144: "Topic name invalid",
|
||||
147: "Receive maximum exceeded",
|
||||
148: "Topic alias invalid",
|
||||
149: "Packet too large",
|
||||
150: "Message rate too high",
|
||||
151: "Quota exceeded",
|
||||
152: "Administrative action",
|
||||
153: "Payload format invalid",
|
||||
154: "Retain not supported",
|
||||
155: "QoS not supported",
|
||||
156: "Use another server",
|
||||
157: "Server moved",
|
||||
158: "Shared subscriptions not supported",
|
||||
159: "Connection rate exceeded",
|
||||
160: "Maximum connect time",
|
||||
161: "Subscription identifiers not supported",
|
||||
162: "Wildcard subscriptions not supported",
|
||||
}
|
||||
|
||||
error_dict = disconnect_errors if is_disconnect else connect_errors
|
||||
return error_dict.get(rc, f"Unknown error code {rc}")
|
||||
@@ -1,133 +1,720 @@
|
||||
import base64
|
||||
import binascii
|
||||
import json
|
||||
import logging
|
||||
import ssl
|
||||
from typing import Any, Dict, Optional
|
||||
import string
|
||||
import threading
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Callable, Dict, List, Optional
|
||||
|
||||
import paho.mqtt.client as mqtt
|
||||
from nacl.signing import SigningKey
|
||||
|
||||
# Try to import datetime.UTC (Python 3.11+) otherwise fallback to timezone.utc
|
||||
try:
|
||||
import paho.mqtt.client as mqtt
|
||||
from datetime import UTC
|
||||
except Exception:
|
||||
from datetime import timezone
|
||||
UTC = timezone.utc
|
||||
|
||||
MQTT_AVAILABLE = True
|
||||
from repeater import __version__
|
||||
|
||||
# Try to import paho-mqtt error code mappings
|
||||
try:
|
||||
from paho.mqtt.reasoncodes import ReasonCode
|
||||
|
||||
HAS_REASON_CODES = True
|
||||
except ImportError:
|
||||
MQTT_AVAILABLE = False
|
||||
|
||||
from .storage_utils import PacketRecord
|
||||
HAS_REASON_CODES = False
|
||||
|
||||
logger = logging.getLogger("MQTTHandler")
|
||||
|
||||
|
||||
class MQTTHandler:
|
||||
def __init__(self, mqtt_config: dict, node_name: str = "unknown", node_id: str = "unknown"):
|
||||
self.mqtt_config = mqtt_config
|
||||
self.node_name = node_name
|
||||
self.node_id = node_id
|
||||
self.client = None
|
||||
self.available = MQTT_AVAILABLE
|
||||
self._init_client()
|
||||
# --------------------------------------------------------------------
|
||||
# Helper: Base64URL without padding
|
||||
# --------------------------------------------------------------------
|
||||
def b64url(x: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(x).rstrip(b"=").decode()
|
||||
|
||||
def _init_client(self):
|
||||
if not self.available or not self.mqtt_config.get("enabled", False):
|
||||
logger.info("MQTT disabled or not available")
|
||||
return
|
||||
|
||||
try:
|
||||
# Use WebSocket transport if configured, otherwise use standard TCP
|
||||
transport = "websockets" if self.mqtt_config.get("use_websockets", False) else "tcp"
|
||||
self.client = mqtt.Client(transport=transport)
|
||||
|
||||
if transport == "websockets":
|
||||
logger.info("Using WebSocket transport for MQTT")
|
||||
|
||||
# Configure TLS/SSL if enabled
|
||||
tls_config = self.mqtt_config.get("tls", {})
|
||||
if tls_config.get("enabled", False):
|
||||
tls_params = {
|
||||
"cert_reqs": ssl.CERT_REQUIRED,
|
||||
"tls_version": ssl.PROTOCOL_TLS,
|
||||
}
|
||||
|
||||
# CA certificate for server verification (optional - uses system certs if not specified)
|
||||
ca_cert = tls_config.get("ca_cert")
|
||||
if ca_cert:
|
||||
tls_params["ca_certs"] = ca_cert
|
||||
logger.info("Using custom CA certificate for MQTT TLS")
|
||||
else:
|
||||
logger.info("Using system default CA certificates for MQTT TLS")
|
||||
|
||||
# Client certificate and key (for mutual TLS)
|
||||
client_cert = tls_config.get("client_cert")
|
||||
client_key = tls_config.get("client_key")
|
||||
if client_cert:
|
||||
tls_params["certfile"] = client_cert
|
||||
if client_key:
|
||||
tls_params["keyfile"] = client_key
|
||||
|
||||
# Allow insecure connections (skip cert verification)
|
||||
if tls_config.get("insecure", False):
|
||||
tls_params["cert_reqs"] = ssl.CERT_NONE
|
||||
logger.warning("MQTT TLS certificate verification disabled (insecure mode)")
|
||||
|
||||
self.client.tls_set(**tls_params)
|
||||
logger.info("MQTT TLS/SSL configured")
|
||||
|
||||
username = self.mqtt_config.get("username")
|
||||
password = self.mqtt_config.get("password")
|
||||
if username:
|
||||
self.client.username_pw_set(username, password)
|
||||
|
||||
broker = self.mqtt_config.get("broker", "localhost")
|
||||
port = self.mqtt_config.get("port", 1883)
|
||||
|
||||
secure = "(TLS)" if tls_config.get("enabled", False) else ""
|
||||
logger.info(f"Connecting to MQTT broker {broker}:{port} {secure}...")
|
||||
|
||||
self.client.connect(broker, port, 60)
|
||||
self.client.loop_start()
|
||||
logger.info(f"MQTT client successfully connected")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize MQTT: {e}")
|
||||
self.client = None
|
||||
|
||||
def publish(self, record: dict, record_type: str):
|
||||
"""
|
||||
Publish record to MQTT.
|
||||
Packets MUST use PacketRecord format. Non-packet records use original format.
|
||||
# # --------------------------------------------------------------------
|
||||
# # Let's Mesh MQTT Broker List (WebSocket Secure)
|
||||
# # --------------------------------------------------------------------
|
||||
# LETSMESH_BROKERS = [
|
||||
# {
|
||||
# "name": "Europe (LetsMesh v1)",
|
||||
# "host": "mqtt-eu-v1.letsmesh.net",
|
||||
# "port": 443,
|
||||
# "audience": "mqtt-eu-v1.letsmesh.net",
|
||||
# "use_jwt_auth": True,
|
||||
# "transport": "websockets",
|
||||
# "enabled": True,
|
||||
# },
|
||||
# {
|
||||
# "name": "US West (LetsMesh v1)",
|
||||
# "host": "mqtt-us-v1.letsmesh.net",
|
||||
# "port": 443,
|
||||
# "audience": "mqtt-us-v1.letsmesh.net",
|
||||
# "use_jwt_auth": True,
|
||||
# "transport": "websockets",
|
||||
# "enabled": True,
|
||||
# },
|
||||
# ]
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Single Broker Connection Manager
|
||||
# ====================================================================
|
||||
class _BrokerConnection:
|
||||
"""
|
||||
Manages a single MQTT broker connection with independent lifecycle.
|
||||
Internal class - not exposed publicly.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
broker: dict,
|
||||
local_identity,
|
||||
public_key: str,
|
||||
iata_code: str,
|
||||
jwt_expiry_minutes: int,
|
||||
use_tls: bool,
|
||||
email: str,
|
||||
owner: str,
|
||||
on_connect_callback: Optional[Callable] = None,
|
||||
on_disconnect_callback: Optional[Callable] = None
|
||||
):
|
||||
self.broker = broker
|
||||
self.local_identity = local_identity
|
||||
self.public_key = public_key.upper()
|
||||
self.iata_code = iata_code
|
||||
self.jwt_expiry_minutes = jwt_expiry_minutes
|
||||
self.use_tls = use_tls
|
||||
self.email = email
|
||||
self.owner = owner
|
||||
self._on_connect_callback = on_connect_callback
|
||||
self._on_disconnect_callback = on_disconnect_callback
|
||||
self._connect_time = None
|
||||
self._tls_verified = False
|
||||
self._running = False
|
||||
self._reconnect_attempts = 0
|
||||
self._reconnect_timer = None
|
||||
self._max_reconnect_delay = 300 # 5 minutes max
|
||||
self._jwt_refresh_timer = None
|
||||
self.transport= broker.get('transport', 'websockets')
|
||||
client_id = f"meshcore_{self.public_key}_{broker['host']}"
|
||||
self.client = mqtt.Client(client_id=client_id, transport=self.transport)
|
||||
self.client.on_connect = self._on_connect
|
||||
self.client.on_disconnect = self._on_disconnect
|
||||
self.use_jwt_auth = broker.get('use_jwt_auth', False)
|
||||
self.username = broker.get('username', None)
|
||||
self.password = broker.get('password', None)
|
||||
|
||||
Args:
|
||||
record: The record dictionary to publish
|
||||
record_type: Type of record (packet, advert, noise_floor, etc.)
|
||||
"""
|
||||
if not self.client:
|
||||
return
|
||||
|
||||
from pymc_core.protocol.utils import PAYLOAD_TYPES
|
||||
|
||||
disallowed_types = broker.get("disallowed_packet_types", [])
|
||||
type_name_map = {name: code for code, name in PAYLOAD_TYPES.items()}
|
||||
|
||||
self.disallowed_hex = [type_name_map.get(name.upper(), None) for name in disallowed_types]
|
||||
self.disallowed_hex = [val for val in self.disallowed_hex if val is not None] # Filter out invalid names
|
||||
|
||||
|
||||
def _generate_jwt(self) -> str:
|
||||
"""Generate MeshCore-style Ed25519 JWT token"""
|
||||
now = datetime.now(UTC)
|
||||
|
||||
header = {"alg": "Ed25519", "typ": "JWT"}
|
||||
|
||||
payload = {
|
||||
"publicKey": self.public_key.upper(),
|
||||
"aud": self.broker["audience"],
|
||||
"iat": int(now.timestamp()),
|
||||
"exp": int((now + timedelta(minutes=self.jwt_expiry_minutes)).timestamp()),
|
||||
}
|
||||
|
||||
if "audience" in self.broker:
|
||||
payload["aud"] = self.broker["audience"]
|
||||
|
||||
# Only include email/owner for verified TLS connections
|
||||
if self.use_tls and self._tls_verified and (self.email or self.owner):
|
||||
payload["email"] = self.email
|
||||
payload["owner"] = self.owner
|
||||
else:
|
||||
payload["email"] = ""
|
||||
payload["owner"] = ""
|
||||
|
||||
# Encode header and payload (compact JSON - no spaces)
|
||||
header_b64 = b64url(json.dumps(header, separators=(",", ":")).encode())
|
||||
payload_b64 = b64url(json.dumps(payload, separators=(",", ":")).encode())
|
||||
|
||||
signing_input = f"{header_b64}.{payload_b64}".encode()
|
||||
|
||||
# Sign using LocalIdentity (supports both standard and firmware keys)
|
||||
try:
|
||||
base_topic = self.mqtt_config.get("base_topic", "meshcore/repeater")
|
||||
topic = f"{base_topic}/{self.node_name}/{record_type}"
|
||||
signature = self.local_identity.sign(signing_input)
|
||||
except Exception as e:
|
||||
logger.error(f"JWT signing failed for {self.broker['name']}: {e}")
|
||||
logger.error(f" - public_key: {self.public_key}")
|
||||
logger.error(f" - signing_input length: {len(signing_input)}")
|
||||
raise
|
||||
|
||||
if record_type == "packet":
|
||||
packet_record = PacketRecord.from_packet_record(
|
||||
record, origin=self.node_name, origin_id=self.node_id
|
||||
)
|
||||
if not packet_record:
|
||||
logger.debug(
|
||||
"Skipping MQTT publish: packet missing required data for PacketRecord"
|
||||
)
|
||||
return
|
||||
signature_hex = binascii.hexlify(signature).decode()
|
||||
token = f"{header_b64}.{payload_b64}.{signature_hex}"
|
||||
|
||||
payload = packet_record.to_dict()
|
||||
logger.debug("Publishing packet using PacketRecord format")
|
||||
logger.debug(f"JWT token generated for {self.broker['name']}: {token[:50]}...")
|
||||
|
||||
return token
|
||||
|
||||
def _on_connect(self, client, userdata, flags, rc):
|
||||
"""MQTT connection callback"""
|
||||
if rc == 0:
|
||||
logger.info(f"Connected to {self.broker['name']}")
|
||||
self._running = True
|
||||
self._reconnect_attempts = 0 # Reset counter on success
|
||||
if self.use_jwt_auth:
|
||||
self._schedule_jwt_refresh() # Schedule proactive JWT refresh
|
||||
if self._on_connect_callback:
|
||||
self._on_connect_callback(self.broker["name"])
|
||||
else:
|
||||
error_msg = get_mqtt_error_message(rc, is_disconnect=False)
|
||||
logger.error(f"Failed to connect to {self.broker['name']}: {error_msg}")
|
||||
self._schedule_reconnect()
|
||||
|
||||
def _on_disconnect(self, client, userdata, rc):
|
||||
"""MQTT disconnection callback"""
|
||||
was_running = self._running
|
||||
self._running = False
|
||||
|
||||
if rc != 0: # Unexpected disconnect
|
||||
error_msg = get_mqtt_error_message(rc, is_disconnect=True)
|
||||
logger.warning(f"Disconnected from {self.broker['name']} (rc={rc}): {error_msg}")
|
||||
if was_running: # Only reconnect if we were intentionally connected
|
||||
self._schedule_reconnect(reason=error_msg)
|
||||
else:
|
||||
logger.info(f"Clean disconnect from {self.broker['name']}")
|
||||
|
||||
if self._on_disconnect_callback:
|
||||
self._on_disconnect_callback(self.broker["name"])
|
||||
|
||||
def _schedule_reconnect(self, reason: str = "connection lost"):
|
||||
"""Schedule reconnection with exponential backoff"""
|
||||
if self._reconnect_timer:
|
||||
self._reconnect_timer.cancel()
|
||||
|
||||
# Exponential backoff: 5s, 10s, 20s, 40s, 80s, up to max
|
||||
delay = min(5 * (2**self._reconnect_attempts), self._max_reconnect_delay)
|
||||
self._reconnect_attempts += 1
|
||||
|
||||
logger.info(
|
||||
f"Scheduling reconnect to {self.broker['name']} in {delay}s (attempt {self._reconnect_attempts}, reason: {reason})"
|
||||
)
|
||||
self._reconnect_timer = threading.Timer(delay, lambda: self._attempt_reconnect(reason))
|
||||
self._reconnect_timer.daemon = True
|
||||
self._reconnect_timer.start()
|
||||
|
||||
def _attempt_reconnect(self, reason: str = "connection lost"):
|
||||
"""Attempt to reconnect to broker with fresh JWT"""
|
||||
try:
|
||||
logger.info(f"Attempting reconnection to {self.broker['name']} (reason: {reason})...")
|
||||
|
||||
# Stop the loop if it's still running (websocket mode requires clean restart)
|
||||
try:
|
||||
self.client.loop_stop()
|
||||
except:
|
||||
pass
|
||||
|
||||
self._set_credentials()
|
||||
|
||||
# Reconnect and restart loop
|
||||
self.client.connect(self.broker["host"], self.broker["port"], keepalive=60)
|
||||
self.client.loop_start()
|
||||
self._loop_running = True
|
||||
except Exception as e:
|
||||
logger.error(f"Reconnection failed for {self.broker['name']}: {e}")
|
||||
self._schedule_reconnect() # Try again later
|
||||
|
||||
def _set_credentials(self):
|
||||
"""Set credentials before connecting (CONNECT handshake only)"""
|
||||
try:
|
||||
if self.use_jwt_auth:
|
||||
logger.debug(f"Generating JWT credentials for {self.broker['name']}...")
|
||||
token = self._generate_jwt()
|
||||
username = f"v1_{self.public_key}"
|
||||
self.client.username_pw_set(username=username, password=token)
|
||||
logger.debug(f"Credentials set for {self.broker['name']}")
|
||||
logger.debug(f"Using username: {username}")
|
||||
logger.debug(f"Public key: {self.public_key[:16]}...{self.public_key[-16:]}")
|
||||
elif self.username and self.password:
|
||||
logger.info(f"Using provided credentials for {self.broker['name']} (username: {self.username})")
|
||||
self.client.username_pw_set(username=self.username, password=self.password)
|
||||
else:
|
||||
payload = {k: v for k, v in record.items() if v is not None}
|
||||
logger.info(f"No credentials set for {self.broker['name']} (JWT auth disabled and no username/password provided)")
|
||||
|
||||
message = json.dumps(payload, default=str)
|
||||
self.client.publish(topic, message, qos=0, retain=False)
|
||||
logger.debug(f"Published to {topic}")
|
||||
self._connect_time = datetime.now(UTC)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to publish to MQTT: {e}")
|
||||
logger.error(f"Failed to set JWT credentials for {self.broker['name']}: {e}")
|
||||
raise
|
||||
|
||||
def close(self):
|
||||
if self.client:
|
||||
self.client.loop_stop()
|
||||
self.client.disconnect()
|
||||
logger.info("MQTT client disconnected")
|
||||
def connect(self):
|
||||
"""Establish connection to broker"""
|
||||
# Conditional TLS setup
|
||||
if self.transport == "websockets":
|
||||
if self.use_tls:
|
||||
import ssl
|
||||
|
||||
self.client.tls_set(cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS_CLIENT)
|
||||
self.client.tls_insecure_set(False)
|
||||
self._tls_verified = True
|
||||
protocol = "wss"
|
||||
else:
|
||||
protocol = "ws"
|
||||
elif self.transport == "tcp":
|
||||
protocol = "mqtt"
|
||||
else:
|
||||
raise ValueError(f"Invalid transport '{self.transport}' for {self.broker['name']}")
|
||||
|
||||
# Set JWT credentials before CONNECT handshake
|
||||
self._set_credentials()
|
||||
|
||||
logger.info(
|
||||
f"Connecting to {self.broker['name']} "
|
||||
f"({protocol}://{self.broker['host']}:{self.broker['port']}) ..."
|
||||
)
|
||||
|
||||
self.client.connect(self.broker["host"], self.broker["port"], keepalive=60)
|
||||
self.client.loop_start()
|
||||
self._loop_running = True
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from broker"""
|
||||
self._running = False
|
||||
self._loop_running = False
|
||||
|
||||
# Cancel any pending timers
|
||||
if self._reconnect_timer:
|
||||
self._reconnect_timer.cancel()
|
||||
self._reconnect_timer = None
|
||||
if self._jwt_refresh_timer:
|
||||
self._jwt_refresh_timer.cancel()
|
||||
self._jwt_refresh_timer = None
|
||||
|
||||
self.client.loop_stop()
|
||||
self.client.disconnect()
|
||||
logger.info(f"Disconnected from {self.broker['name']}")
|
||||
|
||||
def publish(self, topic: str, payload: str, retain: bool = False):
|
||||
"""Publish message to broker"""
|
||||
if self._running:
|
||||
result = self.client.publish(topic, payload, retain=retain)
|
||||
return result
|
||||
return None
|
||||
|
||||
def is_connected(self) -> bool:
|
||||
"""Check if connection is active"""
|
||||
return self._running
|
||||
|
||||
def has_pending_reconnect(self) -> bool:
|
||||
"""Check if a reconnection is scheduled"""
|
||||
return self._reconnect_timer is not None and self._reconnect_timer.is_alive()
|
||||
|
||||
def should_reconnect_for_token_expiry(self) -> bool:
|
||||
"""Check if connection should be reconnected due to JWT expiry (at 80% of lifetime)"""
|
||||
if not self._connect_time:
|
||||
return False
|
||||
elapsed = (datetime.now(UTC) - self._connect_time).total_seconds()
|
||||
expiry_seconds = self.jwt_expiry_minutes * 60
|
||||
# Stagger refresh by 5% per broker to prevent simultaneous disconnects
|
||||
# Broker 0: 80%, Broker 1: 85%, Broker 2: 90%, etc.
|
||||
stagger_offset = self.broker_index * 0.05
|
||||
refresh_threshold = 0.80 + stagger_offset
|
||||
return elapsed >= expiry_seconds * refresh_threshold
|
||||
|
||||
def _schedule_jwt_refresh(self):
|
||||
"""Schedule proactive JWT refresh before token expires"""
|
||||
if self._jwt_refresh_timer:
|
||||
self._jwt_refresh_timer.cancel()
|
||||
|
||||
expiry_seconds = self.jwt_expiry_minutes * 60
|
||||
# Stagger refresh by 5% per broker to prevent simultaneous disconnects
|
||||
# Broker 0: 80%, Broker 1: 85%, Broker 2: 90%, etc.
|
||||
stagger_offset = self.broker_index * 0.05
|
||||
refresh_threshold = 0.80 + stagger_offset
|
||||
refresh_delay = expiry_seconds * refresh_threshold
|
||||
|
||||
logger.info(
|
||||
f"JWT refresh scheduled for {self.broker['name']} in {refresh_delay:.0f}s "
|
||||
f"({refresh_threshold*100:.0f}% of {self.jwt_expiry_minutes}min token lifetime)"
|
||||
)
|
||||
self._jwt_refresh_timer = threading.Timer(refresh_delay, self.reconnect_for_token_expiry)
|
||||
self._jwt_refresh_timer.daemon = True
|
||||
self._jwt_refresh_timer.start()
|
||||
|
||||
def reconnect_for_token_expiry(self):
|
||||
"""Proactively reconnect with new JWT before current one expires"""
|
||||
if not self._running:
|
||||
return
|
||||
|
||||
logger.info(f"JWT token expiring soon for {self.broker['name']}, refreshing...")
|
||||
self._running = False
|
||||
self._jwt_refresh_timer = None
|
||||
|
||||
self._schedule_reconnect(reason="JWT token expiry")
|
||||
self.client.disconnect()
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# MeshCore → MQTT Publisher with Ed25519 auth token
|
||||
# ====================================================================
|
||||
class MeshCoreToMqttPusher:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
local_identity,
|
||||
config: dict,
|
||||
jwt_expiry_minutes: int = 10,
|
||||
use_tls: bool = True,
|
||||
stats_provider: Optional[Callable[[], dict]] = None,
|
||||
):
|
||||
# Store local identity and get public key
|
||||
self.local_identity = local_identity
|
||||
public_key = local_identity.get_public_key().hex().upper()
|
||||
|
||||
# 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.get("broker_index")
|
||||
self.email = node_info.get("email", "")
|
||||
self.owner = node_info.get("owner", "")
|
||||
status_interval = node_info["status_interval"]
|
||||
node_name = node_info["node_name"]
|
||||
radio_config = node_info["radio_config"]
|
||||
|
||||
# Get additional brokers from config (optional)
|
||||
mqtt_config = config.get("mqtt", {})
|
||||
brokers = mqtt_config.get("brokers", [])
|
||||
|
||||
# Add additional brokers from config
|
||||
if brokers:
|
||||
for broker_config in brokers:
|
||||
if all(k in broker_config for k in ["name", "host", "port", "enabled"]):
|
||||
if broker_config["enabled"]:
|
||||
self.brokers.append(broker_config)
|
||||
logger.info(f"Added broker: {broker_config['name']}")
|
||||
else:
|
||||
logger.info(f"Broker disabled in config, skipping: {broker_config['name']}")
|
||||
else:
|
||||
logger.warning(f"Skipping invalid broker config: {broker_config}")
|
||||
|
||||
# Validate that we have at least one broker
|
||||
# if not self.brokers:
|
||||
# raise ValueError(
|
||||
# "No brokers configured. Either set broker_index to a valid value "
|
||||
# "or provide additional_brokers in config."
|
||||
# )
|
||||
|
||||
self.local_identity = local_identity
|
||||
self.public_key = public_key
|
||||
self.iata_code = iata_code
|
||||
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
|
||||
self.radio_config = radio_config
|
||||
self.stats_provider = stats_provider
|
||||
self._status_task = None
|
||||
self._running = False
|
||||
self._lock = threading.Lock()
|
||||
|
||||
# Create broker connections
|
||||
self.connections: List[_BrokerConnection] = []
|
||||
for idx, broker in enumerate(self.brokers):
|
||||
conn = _BrokerConnection(
|
||||
broker=broker,
|
||||
local_identity=self.local_identity,
|
||||
public_key=self.public_key,
|
||||
iata_code=self.iata_code,
|
||||
jwt_expiry_minutes=self.jwt_expiry_minutes,
|
||||
use_tls=self.use_tls,
|
||||
email=self.email,
|
||||
owner=self.owner,
|
||||
broker_index=idx,
|
||||
on_connect_callback=self._on_broker_connected,
|
||||
on_disconnect_callback=self._on_broker_disconnected,
|
||||
)
|
||||
self.connections.append(conn)
|
||||
|
||||
logger.info(f"Initialized with {len(self.connections)} broker connection(s)")
|
||||
|
||||
def _on_broker_connected(self, broker_name: str):
|
||||
"""Callback when a broker connects"""
|
||||
# Publish initial status on first connection
|
||||
if not self._status_task and self.status_interval > 0:
|
||||
self._running = True
|
||||
logger.info(f"Publishing initial status for {broker_name}...")
|
||||
self.publish_status(
|
||||
state="online", origin=self.node_name, radio_config=self.radio_config
|
||||
)
|
||||
# Start heartbeat thread
|
||||
self._status_task = threading.Thread(target=self._status_heartbeat_loop, daemon=True)
|
||||
self._status_task.start()
|
||||
logger.info(f"Started status heartbeat (interval: {self.status_interval}s)")
|
||||
|
||||
def _on_broker_disconnected(self, broker_name: str):
|
||||
"""Callback when a broker disconnects"""
|
||||
# Check if all connections are down AND none have pending reconnects
|
||||
all_down = all(not conn.is_connected() for conn in self.connections)
|
||||
any_reconnecting = any(conn.has_pending_reconnect() for conn in self.connections)
|
||||
|
||||
if all_down and not any_reconnecting:
|
||||
logger.warning("All broker connections lost with no pending reconnects")
|
||||
elif all_down:
|
||||
logger.info("All brokers temporarily disconnected, reconnects pending")
|
||||
|
||||
def connect(self):
|
||||
"""Establish connections to all configured brokers"""
|
||||
for idx, conn in enumerate(self.connections):
|
||||
try:
|
||||
if idx == 0:
|
||||
# Connect first broker immediately
|
||||
conn.connect()
|
||||
else:
|
||||
# Stagger additional brokers using background timers
|
||||
delay = idx * 30
|
||||
logger.info(f"Staggering connection to {conn.broker['name']} by {delay}s")
|
||||
timer = threading.Timer(delay, lambda c=conn: self._delayed_connect(c))
|
||||
timer.daemon = True
|
||||
timer.start()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to {conn.broker['name']}: {e}")
|
||||
|
||||
def _delayed_connect(self, conn):
|
||||
"""Connect a broker after a delay (called by timer)"""
|
||||
try:
|
||||
conn.connect()
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to connect to {conn.broker['name']}: {e}")
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from all brokers"""
|
||||
# Stop the heartbeat loop
|
||||
self._running = False
|
||||
|
||||
# Publish offline status before disconnecting
|
||||
self.publish_status(state="offline", origin=self.node_name, radio_config=self.radio_config)
|
||||
|
||||
import time
|
||||
|
||||
time.sleep(0.5) # Give time for messages to be sent
|
||||
|
||||
# Disconnect all brokers
|
||||
for conn in self.connections:
|
||||
try:
|
||||
conn.disconnect()
|
||||
except Exception as e:
|
||||
logger.error(f"Error disconnecting from {conn.broker['name']}: {e}")
|
||||
|
||||
logger.info("Disconnected from all brokers")
|
||||
|
||||
def _status_heartbeat_loop(self):
|
||||
"""Background thread that publishes periodic status updates"""
|
||||
import time
|
||||
|
||||
while self._running:
|
||||
try:
|
||||
# Publish status (JWT refresh now handled by individual broker timers)
|
||||
self.publish_status(
|
||||
state="online", origin=self.node_name, radio_config=self.radio_config
|
||||
)
|
||||
logger.debug(f"Status heartbeat sent (next in {self.status_interval}s)")
|
||||
|
||||
time.sleep(self.status_interval)
|
||||
except Exception as e:
|
||||
logger.error(f"Status heartbeat error: {e}")
|
||||
time.sleep(self.status_interval)
|
||||
|
||||
# ----------------------------------------------------------------
|
||||
# Packet helpers
|
||||
# ----------------------------------------------------------------
|
||||
def _process_packet(self, pkt: dict) -> dict:
|
||||
return {"timestamp": datetime.now(UTC).isoformat(), "origin_id": self.public_key, **pkt}
|
||||
|
||||
def _topic(self, subtopic: str) -> str:
|
||||
return f"meshcore/{self.iata_code}/{self.public_key}/{subtopic}"
|
||||
|
||||
def publish_packet(self, pkt: dict, packet_type: string, subtopic="packets", retain=False):
|
||||
if packet_type in self.disallowed_packet_types:
|
||||
logger.debug(f"Skipped publishing packet type 0x{packet_type:02X} (disallowed)")
|
||||
return
|
||||
|
||||
return self.publish(subtopic, self._process_packet(pkt), retain)
|
||||
|
||||
def publish_raw_data(self, raw_hex: str, subtopic="raw", retain=False):
|
||||
pkt = {"type": "raw", "data": raw_hex, "bytes": len(raw_hex) // 2}
|
||||
return self.publish_packet(pkt, "raw", subtopic, retain)
|
||||
|
||||
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
|
||||
|
||||
Args:
|
||||
state: Device state (online/offline)
|
||||
location: Optional dict with latitude/longitude
|
||||
extra_stats: Optional additional statistics to include
|
||||
origin: Node name/description
|
||||
radio_config: Radio configuration string (freq,bw,sf,cr)
|
||||
"""
|
||||
# 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 self.node_name,
|
||||
"origin_id": self.public_key,
|
||||
"model": "PyMC-Repeater",
|
||||
"firmware_version": self.app_version,
|
||||
"radio": radio_config or self.radio_config,
|
||||
"client_version": f"pyMC_repeater/{self.app_version}",
|
||||
"stats": {**live_stats, "errors": 0, "queue_len": 0, **(extra_stats or {})},
|
||||
}
|
||||
|
||||
if location:
|
||||
status["location"] = location
|
||||
|
||||
return self.publish("status", status, retain=True, qos=1)
|
||||
|
||||
def publish(self, subtopic: str, payload: dict, retain: bool = False):
|
||||
"""Publish message to all connected brokers"""
|
||||
topic = self._topic(subtopic)
|
||||
message = json.dumps(payload)
|
||||
|
||||
results = []
|
||||
with self._lock:
|
||||
for conn in self.connections:
|
||||
if conn.is_connected():
|
||||
result = conn.publish(topic, message, retain=retain)
|
||||
results.append((conn.broker["name"], result))
|
||||
logger.debug(f"Published to {conn.broker['name']}/{topic}")
|
||||
|
||||
if not results:
|
||||
logger.warning(f"No active broker connections for publishing to {topic}")
|
||||
|
||||
return results
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Helper Functions
|
||||
# ====================================================================
|
||||
|
||||
|
||||
def get_mqtt_error_message(rc: int, is_disconnect: bool = False) -> str:
|
||||
"""
|
||||
Get human-readable MQTT error message.
|
||||
|
||||
Args:
|
||||
rc: Return code from paho-mqtt
|
||||
is_disconnect: True if from on_disconnect, False if from on_connect
|
||||
|
||||
Returns:
|
||||
Human-readable error message
|
||||
"""
|
||||
if HAS_REASON_CODES:
|
||||
try:
|
||||
# ReasonCode object has getName() method and value property
|
||||
reason = ReasonCode(mqtt.CONNACK if not is_disconnect else mqtt.DISCONNECT, identifier=rc)
|
||||
name = reason.getName() if hasattr(reason, 'getName') else str(reason)
|
||||
return f"{name} (code {rc})"
|
||||
except Exception as e:
|
||||
# Log the exception for debugging
|
||||
logger.debug(f"Could not decode reason code {rc}: {e}")
|
||||
|
||||
# Fallback to manual mappings - Extended with MQTT v5 codes
|
||||
connect_errors = {
|
||||
0: "Connection accepted",
|
||||
1: "Incorrect protocol version",
|
||||
2: "Invalid client identifier",
|
||||
3: "Server unavailable",
|
||||
4: "Bad username or password (JWT invalid)",
|
||||
5: "Not authorized (JWT signature/format invalid)",
|
||||
# MQTT v5 codes
|
||||
128: "Unspecified error",
|
||||
129: "Malformed packet",
|
||||
130: "Protocol error",
|
||||
131: "Implementation specific error",
|
||||
132: "Unsupported protocol version",
|
||||
133: "Client identifier not valid",
|
||||
134: "Bad username or password",
|
||||
135: "Not authorized",
|
||||
136: "Server unavailable",
|
||||
137: "Server busy",
|
||||
138: "Banned",
|
||||
140: "Bad authentication method",
|
||||
144: "Topic name invalid",
|
||||
149: "Packet too large",
|
||||
151: "Quota exceeded",
|
||||
153: "Payload format invalid",
|
||||
154: "Retain not supported",
|
||||
155: "QoS not supported",
|
||||
156: "Use another server",
|
||||
157: "Server moved",
|
||||
159: "Connection rate exceeded",
|
||||
}
|
||||
|
||||
disconnect_errors = {
|
||||
0: "Normal disconnect",
|
||||
1: "Unacceptable protocol version",
|
||||
2: "Identifier rejected",
|
||||
3: "Server unavailable",
|
||||
4: "Bad username or password",
|
||||
5: "Not authorized",
|
||||
7: "Connection lost / network error",
|
||||
16: "Connection lost / protocol error",
|
||||
17: "Client timeout",
|
||||
# MQTT v5 codes
|
||||
4: "Disconnect with Will message",
|
||||
128: "Unspecified error",
|
||||
129: "Malformed packet",
|
||||
130: "Protocol error",
|
||||
131: "Implementation specific error",
|
||||
135: "Not authorized",
|
||||
137: "Server busy",
|
||||
139: "Server shutting down",
|
||||
141: "Keep alive timeout",
|
||||
142: "Session taken over",
|
||||
143: "Topic filter invalid",
|
||||
144: "Topic name invalid",
|
||||
147: "Receive maximum exceeded",
|
||||
148: "Topic alias invalid",
|
||||
149: "Packet too large",
|
||||
150: "Message rate too high",
|
||||
151: "Quota exceeded",
|
||||
152: "Administrative action",
|
||||
153: "Payload format invalid",
|
||||
154: "Retain not supported",
|
||||
155: "QoS not supported",
|
||||
156: "Use another server",
|
||||
157: "Server moved",
|
||||
158: "Shared subscriptions not supported",
|
||||
159: "Connection rate exceeded",
|
||||
160: "Maximum connect time",
|
||||
161: "Subscription identifiers not supported",
|
||||
162: "Wildcard subscriptions not supported",
|
||||
}
|
||||
|
||||
error_dict = disconnect_errors if is_disconnect else connect_errors
|
||||
return error_dict.get(rc, f"Unknown error code {rc}")
|
||||
|
||||
@@ -5,8 +5,8 @@ from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from .letsmesh_handler import MeshCoreToMqttJwtPusher
|
||||
from .mqtt_handler import MQTTHandler
|
||||
from .mqtt_handler import MeshCoreToMqttPusher
|
||||
#from .old_mqtt_handler import MQTTHandler
|
||||
from .rrdtool_handler import RRDToolHandler
|
||||
from .sqlite_handler import SQLiteHandler
|
||||
from .storage_utils import PacketRecord
|
||||
@@ -32,40 +32,40 @@ class StorageCollector:
|
||||
|
||||
self.sqlite_handler = SQLiteHandler(self.storage_dir)
|
||||
self.rrd_handler = RRDToolHandler(self.storage_dir)
|
||||
self.mqtt_handler = MQTTHandler(config.get("mqtt", {}), node_name, node_id)
|
||||
# self.old_mqtt_handler = MQTTHandler(config.get("mqtt", {}), node_name, node_id)
|
||||
|
||||
# Initialize LetsMesh handler if configured
|
||||
self.letsmesh_handler = None
|
||||
if config.get("letsmesh", {}).get("enabled", False) and local_identity:
|
||||
# Initialize MQTT handler if configured
|
||||
self.mqtt_handler = None
|
||||
if config.get("mqtt", {}) and local_identity:
|
||||
try:
|
||||
# Pass local_identity directly (supports both standard and firmware keys)
|
||||
self.letsmesh_handler = MeshCoreToMqttJwtPusher(
|
||||
self.mqtt_handler = MeshCoreToMqttPusher(
|
||||
local_identity=local_identity,
|
||||
config=config,
|
||||
stats_provider=self._get_live_stats,
|
||||
)
|
||||
self.letsmesh_handler.connect()
|
||||
self.mqtt_handler.connect()
|
||||
|
||||
# Get disallowed packet types from config
|
||||
from ..config import get_node_info
|
||||
|
||||
node_info = get_node_info(config)
|
||||
self.disallowed_packet_types = set(node_info["disallowed_packet_types"])
|
||||
#node_info = get_node_info(config)
|
||||
#self.disallowed_packet_types = set(node_info["disallowed_packet_types"])
|
||||
|
||||
public_key_hex = local_identity.get_public_key().hex()
|
||||
logger.info(
|
||||
f"LetsMesh handler initialized with public key: {public_key_hex[:16]}..."
|
||||
f"MQTT handler initialized with public key: {public_key_hex[:16]}..."
|
||||
)
|
||||
if self.disallowed_packet_types:
|
||||
logger.info(f"Disallowed packet types: {sorted(self.disallowed_packet_types)}")
|
||||
else:
|
||||
logger.info("All packet types allowed")
|
||||
#if self.disallowed_packet_types:
|
||||
# logger.info(f"Disallowed packet types: {sorted(self.disallowed_packet_types)}")
|
||||
#else:
|
||||
# logger.info("All packet types allowed")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize LetsMesh handler: {e}")
|
||||
self.letsmesh_handler = None
|
||||
self.disallowed_packet_types = set()
|
||||
else:
|
||||
self.disallowed_packet_types = set()
|
||||
logger.error(f"Failed to initialize MQTT handler: {e}")
|
||||
self.mqtt_handler = None
|
||||
#self.disallowed_packet_types = set()
|
||||
#else:
|
||||
# self.disallowed_packet_types = set()
|
||||
|
||||
# Initialize hardware stats collector
|
||||
from .hardware_stats import HardwareStatsCollector
|
||||
@@ -146,7 +146,7 @@ class StorageCollector:
|
||||
self.sqlite_handler.store_packet(packet_record)
|
||||
cumulative_counts = self.sqlite_handler.get_cumulative_counts()
|
||||
self.rrd_handler.update_packet_metrics(packet_record, cumulative_counts)
|
||||
self.mqtt_handler.publish(packet_record, "packet")
|
||||
#self.old_mqtt_handler.publish(packet_record, "packet")
|
||||
|
||||
# Broadcast to WebSocket clients for real-time updates
|
||||
if self.websocket_available:
|
||||
@@ -170,17 +170,17 @@ class StorageCollector:
|
||||
except Exception as e:
|
||||
logger.debug(f"WebSocket broadcast failed: {e}")
|
||||
|
||||
# Publish to LetsMesh if enabled (skip invalid packets if requested)
|
||||
if skip_letsmesh_if_invalid and packet_record.get("drop_reason"):
|
||||
logger.debug(
|
||||
f"Skipping LetsMesh publish for packet with drop_reason: {packet_record.get('drop_reason')}"
|
||||
)
|
||||
else:
|
||||
self._publish_to_letsmesh(packet_record)
|
||||
# # Publish to LetsMesh if enabled (skip invalid packets if requested)
|
||||
# if skip_letsmesh_if_invalid and packet_record.get("drop_reason"):
|
||||
# logger.debug(
|
||||
# f"Skipping LetsMesh publish for packet with drop_reason: {packet_record.get('drop_reason')}"
|
||||
# )
|
||||
# else:
|
||||
self._publish_to_letsmesh(packet_record)
|
||||
|
||||
def _publish_to_letsmesh(self, packet_record: dict):
|
||||
"""Publish packet to LetsMesh broker if enabled and allowed"""
|
||||
if not self.letsmesh_handler:
|
||||
if not self.mqtt_handler:
|
||||
return
|
||||
|
||||
try:
|
||||
@@ -189,17 +189,17 @@ class StorageCollector:
|
||||
logger.error("Cannot publish to LetsMesh: packet_record missing 'type' field")
|
||||
return
|
||||
|
||||
if packet_type in self.disallowed_packet_types:
|
||||
logger.debug(f"Skipped publishing packet type 0x{packet_type:02X} (disallowed)")
|
||||
return
|
||||
# if packet_type in self.disallowed_packet_types:
|
||||
# logger.debug(f"Skipped publishing packet type 0x{packet_type:02X} (disallowed)")
|
||||
# return
|
||||
|
||||
node_name = self.config.get("repeater", {}).get("node_name", "Unknown")
|
||||
packet = PacketRecord.from_packet_record(
|
||||
packet_record, origin=node_name, origin_id=self.letsmesh_handler.public_key
|
||||
packet_record, origin=node_name, origin_id=self.mqtt_handler.public_key
|
||||
)
|
||||
|
||||
if packet:
|
||||
self.letsmesh_handler.publish_packet(packet.to_dict())
|
||||
self.mqtt_handler.publish_packet(packet.to_dict(), packet_type)
|
||||
logger.debug(f"Published packet type 0x{packet_type:02X} to LetsMesh")
|
||||
else:
|
||||
logger.debug("Skipped LetsMesh publish: packet missing raw_packet data")
|
||||
@@ -209,18 +209,18 @@ class StorageCollector:
|
||||
|
||||
def record_advert(self, advert_record: dict):
|
||||
self.sqlite_handler.store_advert(advert_record)
|
||||
self.mqtt_handler.publish(advert_record, "advert")
|
||||
#self.old_mqtt_handler.publish(advert_record, "advert")
|
||||
|
||||
def record_noise_floor(self, noise_floor_dbm: float):
|
||||
noise_record = {"timestamp": time.time(), "noise_floor_dbm": noise_floor_dbm}
|
||||
self.sqlite_handler.store_noise_floor(noise_record)
|
||||
self.mqtt_handler.publish(noise_record, "noise_floor")
|
||||
#self.old_mqtt_handler.publish(noise_record, "noise_floor")
|
||||
|
||||
def record_crc_errors(self, count: int):
|
||||
"""Record a batch of CRC errors detected since last poll."""
|
||||
crc_record = {"timestamp": time.time(), "count": count}
|
||||
self.sqlite_handler.store_crc_errors(crc_record)
|
||||
self.mqtt_handler.publish(crc_record, "crc_errors")
|
||||
#self.old_mqtt_handler.publish(crc_record, "crc_errors")
|
||||
|
||||
def get_crc_error_count(self, hours: int = 24) -> int:
|
||||
return self.sqlite_handler.get_crc_error_count(hours)
|
||||
@@ -305,13 +305,13 @@ class StorageCollector:
|
||||
return self.sqlite_handler.get_noise_floor_stats(hours)
|
||||
|
||||
def close(self):
|
||||
self.mqtt_handler.close()
|
||||
if self.letsmesh_handler:
|
||||
#self.old_mqtt_handler.close()
|
||||
if self.mqtt_handler:
|
||||
try:
|
||||
self.letsmesh_handler.disconnect()
|
||||
logger.info("LetsMesh handler disconnected")
|
||||
self.mqtt_handler.disconnect()
|
||||
logger.info("MQTT handler disconnected")
|
||||
except Exception as e:
|
||||
logger.error(f"Error disconnecting LetsMesh handler: {e}")
|
||||
logger.error(f"Error disconnecting MQTT handler: {e}")
|
||||
|
||||
def create_transport_key(
|
||||
self,
|
||||
|
||||
+1
-1
@@ -1138,7 +1138,7 @@ class RepeaterHandler(BaseHandler):
|
||||
"unscoped_flood_allow": self.config.get("mesh", {}).get("unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True)),
|
||||
"path_hash_mode": self.config.get("mesh", {}).get("path_hash_mode", 0),
|
||||
},
|
||||
"letsmesh": self.config.get("letsmesh", {}),
|
||||
#"mqtt": self.config.get("mqtt", {}),
|
||||
},
|
||||
"public_key": None,
|
||||
}
|
||||
|
||||
@@ -54,8 +54,8 @@ logger = logging.getLogger("HTTPServer")
|
||||
# POST /api/update_duty_cycle_config {"enabled": true, "on_time": 300, "off_time": 60} - Update duty cycle config
|
||||
# POST /api/update_radio_config - Update radio configuration
|
||||
# POST /api/update_advert_rate_limit_config - Update advert rate limiting settings
|
||||
# GET /api/letsmesh_status - Get LetsMesh Observer connection status
|
||||
# POST /api/update_letsmesh_config - Update LetsMesh Observer configuration
|
||||
# GET /api/mqtt_status - Get MQTT Observer connection status
|
||||
# POST /api/update_mqtt_config - Update MQTT Observer configuration
|
||||
|
||||
# Packets
|
||||
# GET /api/packet_stats?hours=24 - Get packet statistics
|
||||
@@ -999,18 +999,18 @@ class APIEndpoints:
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
def letsmesh_status(self):
|
||||
"""Get LetsMesh connection status and configuration."""
|
||||
def mqtt_status(self):
|
||||
"""Get MQTT connection status and configuration."""
|
||||
self._set_cors_headers()
|
||||
try:
|
||||
letsmesh_cfg = self.config.get("letsmesh", {})
|
||||
enabled = letsmesh_cfg.get("enabled", False)
|
||||
mqtt_cfg = self.config.get("mqtt", {})
|
||||
enabled = mqtt_cfg.get("enabled", False)
|
||||
|
||||
# Walk the chain to the letsmesh_handler
|
||||
# Walk the chain to the mqtt_handler
|
||||
handler = None
|
||||
try:
|
||||
storage = self._get_storage()
|
||||
handler = getattr(storage, "letsmesh_handler", None)
|
||||
handler = getattr(storage, "mqtt_handler", None)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
@@ -1030,100 +1030,98 @@ class APIEndpoints:
|
||||
"brokers": connected_brokers,
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting LetsMesh status: {e}")
|
||||
logger.error(f"Error getting MQTT status: {e}")
|
||||
return self._error(str(e))
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
@cherrypy.tools.json_in()
|
||||
def update_letsmesh_config(self):
|
||||
"""Update LetsMesh Observer configuration.
|
||||
# @cherrypy.expose
|
||||
# @cherrypy.tools.json_out()
|
||||
# @cherrypy.tools.json_in()
|
||||
# def update_mqtt_config(self):
|
||||
# """Update MQTT Observer configuration.
|
||||
|
||||
POST /api/update_letsmesh_config
|
||||
Body: {
|
||||
"enabled": true,
|
||||
"iata_code": "SFO",
|
||||
"broker_index": 0,
|
||||
"status_interval": 300,
|
||||
"owner": "Callsign",
|
||||
"email": "user@example.com",
|
||||
"disallowed_packet_types": ["ACK"]
|
||||
}
|
||||
"""
|
||||
self._set_cors_headers()
|
||||
# POST /api/update_mqtt_config
|
||||
# Body: {
|
||||
# "iata_code": "SFO",
|
||||
# "status_interval": 300,
|
||||
# "owner": "Callsign",
|
||||
# "email": "user@example.com",
|
||||
# "disallowed_packet_types": ["ACK"]
|
||||
# }
|
||||
# """
|
||||
# self._set_cors_headers()
|
||||
|
||||
if cherrypy.request.method == "OPTIONS":
|
||||
return ""
|
||||
# if cherrypy.request.method == "OPTIONS":
|
||||
# return ""
|
||||
|
||||
try:
|
||||
self._require_post()
|
||||
data = cherrypy.request.json or {}
|
||||
# try:
|
||||
# self._require_post()
|
||||
# data = cherrypy.request.json or {}
|
||||
|
||||
if not data:
|
||||
return self._error("No configuration updates provided")
|
||||
# if not data:
|
||||
# return self._error("No configuration updates provided")
|
||||
|
||||
letsmesh_updates = {}
|
||||
# letsmesh_updates = {}
|
||||
|
||||
if "enabled" in data:
|
||||
letsmesh_updates["enabled"] = bool(data["enabled"])
|
||||
if "iata_code" in data:
|
||||
letsmesh_updates["iata_code"] = str(data["iata_code"]).strip()
|
||||
if "broker_index" in data:
|
||||
letsmesh_updates["broker_index"] = int(data["broker_index"])
|
||||
if "status_interval" in data:
|
||||
letsmesh_updates["status_interval"] = max(60, int(data["status_interval"]))
|
||||
if "owner" in data:
|
||||
letsmesh_updates["owner"] = str(data["owner"]).strip()
|
||||
if "email" in data:
|
||||
letsmesh_updates["email"] = str(data["email"]).strip()
|
||||
if "disallowed_packet_types" in data:
|
||||
letsmesh_updates["disallowed_packet_types"] = list(data["disallowed_packet_types"])
|
||||
if "additional_brokers" in data:
|
||||
brokers = data["additional_brokers"]
|
||||
if not isinstance(brokers, list):
|
||||
return self._error("additional_brokers must be a list")
|
||||
validated = []
|
||||
for i, b in enumerate(brokers):
|
||||
if not isinstance(b, dict):
|
||||
return self._error(f"Broker at index {i} must be an object")
|
||||
for field in ("name", "host", "audience"):
|
||||
if not b.get(field, "").strip():
|
||||
return self._error(f"Broker at index {i} missing required field: {field}")
|
||||
try:
|
||||
port = int(b.get("port", 443))
|
||||
except (ValueError, TypeError):
|
||||
return self._error(f"Broker at index {i} has invalid port")
|
||||
validated.append({
|
||||
"name": str(b["name"]).strip(),
|
||||
"host": str(b["host"]).strip(),
|
||||
"port": port,
|
||||
"audience": str(b["audience"]).strip(),
|
||||
})
|
||||
letsmesh_updates["additional_brokers"] = validated
|
||||
# if "enabled" in data:
|
||||
# letsmesh_updates["enabled"] = bool(data["enabled"])
|
||||
# if "iata_code" in data:
|
||||
# letsmesh_updates["iata_code"] = str(data["iata_code"]).strip()
|
||||
# if "broker_index" in data:
|
||||
# letsmesh_updates["broker_index"] = int(data["broker_index"])
|
||||
# if "status_interval" in data:
|
||||
# letsmesh_updates["status_interval"] = max(60, int(data["status_interval"]))
|
||||
# if "owner" in data:
|
||||
# letsmesh_updates["owner"] = str(data["owner"]).strip()
|
||||
# if "email" in data:
|
||||
# letsmesh_updates["email"] = str(data["email"]).strip()
|
||||
# if "disallowed_packet_types" in data:
|
||||
# letsmesh_updates["disallowed_packet_types"] = list(data["disallowed_packet_types"])
|
||||
# if "additional_brokers" in data:
|
||||
# brokers = data["additional_brokers"]
|
||||
# if not isinstance(brokers, list):
|
||||
# return self._error("additional_brokers must be a list")
|
||||
# validated = []
|
||||
# for i, b in enumerate(brokers):
|
||||
# if not isinstance(b, dict):
|
||||
# return self._error(f"Broker at index {i} must be an object")
|
||||
# for field in ("name", "host", "audience"):
|
||||
# if not b.get(field, "").strip():
|
||||
# return self._error(f"Broker at index {i} missing required field: {field}")
|
||||
# try:
|
||||
# port = int(b.get("port", 443))
|
||||
# except (ValueError, TypeError):
|
||||
# return self._error(f"Broker at index {i} has invalid port")
|
||||
# validated.append({
|
||||
# "name": str(b["name"]).strip(),
|
||||
# "host": str(b["host"]).strip(),
|
||||
# "port": port,
|
||||
# "audience": str(b["audience"]).strip(),
|
||||
# })
|
||||
# letsmesh_updates["additional_brokers"] = validated
|
||||
|
||||
if not letsmesh_updates:
|
||||
return self._error("No valid settings provided")
|
||||
# if not letsmesh_updates:
|
||||
# return self._error("No valid settings provided")
|
||||
|
||||
result = self.config_manager.update_and_save(
|
||||
updates={"letsmesh": letsmesh_updates},
|
||||
live_update=False, # Restart required for LetsMesh handler changes
|
||||
)
|
||||
# result = self.config_manager.update_and_save(
|
||||
# updates={"letsmesh": letsmesh_updates},
|
||||
# live_update=False, # Restart required for LetsMesh handler changes
|
||||
# )
|
||||
|
||||
if result.get("success"):
|
||||
logger.info(f"LetsMesh config updated: {list(letsmesh_updates.keys())}")
|
||||
return self._success({
|
||||
"persisted": result.get("saved", False),
|
||||
"restart_required": True,
|
||||
"message": "Observer settings saved. Restart the service for changes to take effect.",
|
||||
})
|
||||
else:
|
||||
return self._error(result.get("error", "Failed to update LetsMesh configuration"))
|
||||
# if result.get("success"):
|
||||
# logger.info(f"LetsMesh config updated: {list(letsmesh_updates.keys())}")
|
||||
# return self._success({
|
||||
# "persisted": result.get("saved", False),
|
||||
# "restart_required": True,
|
||||
# "message": "Observer settings saved. Restart the service for changes to take effect.",
|
||||
# })
|
||||
# else:
|
||||
# return self._error(result.get("error", "Failed to update LetsMesh configuration"))
|
||||
|
||||
except cherrypy.HTTPError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating LetsMesh config: {e}")
|
||||
return self._error(str(e))
|
||||
# except cherrypy.HTTPError:
|
||||
# raise
|
||||
# except Exception as e:
|
||||
# logger.error(f"Error updating LetsMesh config: {e}")
|
||||
# return self._error(str(e))
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
|
||||
Reference in New Issue
Block a user