feat: Enhance Meshtastic mode with QR code support

Add QR code generation for sharing Meshtastic channel configurations.
Add qrcode[pil] dependency for QR code generation.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-01-29 14:50:53 +00:00
parent 9a8f87d501
commit 79ffc7ca85
5 changed files with 1969 additions and 6 deletions
+3
View File
@@ -23,6 +23,9 @@ pyserial>=3.5
# Meshtastic mesh network support (optional - only needed for Meshtastic features) # Meshtastic mesh network support (optional - only needed for Meshtastic features)
meshtastic>=2.0.0 meshtastic>=2.0.0
# QR code generation for Meshtastic channels (optional)
qrcode[pil]>=7.4
# Development dependencies (install with: pip install -r requirements-dev.txt) # Development dependencies (install with: pip install -r requirements-dev.txt)
# pytest>=7.0.0 # pytest>=7.0.0
# pytest-cov>=4.0.0 # pytest-cov>=4.0.0
+404
View File
@@ -607,3 +607,407 @@ def get_traceroute_results():
'results': [r.to_dict() for r in results], 'results': [r.to_dict() for r in results],
'count': len(results) 'count': len(results)
}) })
@meshtastic_bp.route('/position/request', methods=['POST'])
def request_position():
"""
Request position from a specific node.
JSON body:
{
"node_id": "!a1b2c3d4" // Required: target node ID
}
Returns:
JSON with request status.
"""
if not is_meshtastic_available():
return jsonify({
'status': 'error',
'message': 'Meshtastic SDK not installed'
}), 400
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device'
}), 400
data = request.get_json(silent=True) or {}
node_id = data.get('node_id')
if not node_id:
return jsonify({
'status': 'error',
'message': 'Node ID is required'
}), 400
success, error = client.request_position(node_id)
if success:
return jsonify({
'status': 'sent',
'node_id': node_id
})
else:
return jsonify({
'status': 'error',
'message': error or 'Failed to request position'
}), 500
@meshtastic_bp.route('/firmware/check')
def check_firmware():
"""
Check current firmware version and compare to latest release.
Returns:
JSON with current_version, latest_version, update_available, release_url.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device'
}), 400
result = client.check_firmware()
result['status'] = 'ok'
return jsonify(result)
@meshtastic_bp.route('/channels/<int:index>/qr')
def get_channel_qr(index: int):
"""
Generate QR code for a channel configuration.
Args:
index: Channel index (0-7)
Returns:
PNG image of QR code.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device'
}), 400
if not 0 <= index <= 7:
return jsonify({
'status': 'error',
'message': 'Channel index must be 0-7'
}), 400
png_data = client.generate_channel_qr(index)
if png_data:
return Response(png_data, mimetype='image/png')
else:
return jsonify({
'status': 'error',
'message': 'Failed to generate QR code. Make sure qrcode library is installed.'
}), 500
@meshtastic_bp.route('/telemetry/history')
def get_telemetry_history():
"""
Get telemetry history for a node.
Query parameters:
node_id: Node ID or number (required)
hours: Number of hours of history (default: 24)
Returns:
JSON with telemetry data points.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device',
'data': []
}), 400
node_id = request.args.get('node_id')
hours = request.args.get('hours', 24, type=int)
if not node_id:
return jsonify({
'status': 'error',
'message': 'node_id is required',
'data': []
}), 400
# Parse node ID to number
try:
if node_id.startswith('!'):
node_num = int(node_id[1:], 16)
else:
node_num = int(node_id)
except ValueError:
return jsonify({
'status': 'error',
'message': f'Invalid node_id: {node_id}',
'data': []
}), 400
history = client.get_telemetry_history(node_num, hours=hours)
return jsonify({
'status': 'ok',
'node_id': node_id,
'hours': hours,
'data': [p.to_dict() for p in history],
'count': len(history)
})
@meshtastic_bp.route('/neighbors')
def get_neighbors():
"""
Get neighbor information for mesh topology visualization.
Query parameters:
node_id: Specific node ID (optional, returns all if not provided)
Returns:
JSON with neighbor relationships.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device',
'neighbors': {}
}), 400
node_id = request.args.get('node_id')
node_num = None
if node_id:
try:
if node_id.startswith('!'):
node_num = int(node_id[1:], 16)
else:
node_num = int(node_id)
except ValueError:
return jsonify({
'status': 'error',
'message': f'Invalid node_id: {node_id}',
'neighbors': {}
}), 400
neighbors = client.get_neighbors(node_num)
# Convert to JSON-serializable format
result = {}
for num, neighbor_list in neighbors.items():
node_key = f"!{num:08x}"
result[node_key] = [n.to_dict() for n in neighbor_list]
return jsonify({
'status': 'ok',
'neighbors': result,
'node_count': len(result)
})
@meshtastic_bp.route('/pending')
def get_pending_messages():
"""
Get messages waiting for ACK.
Returns:
JSON with pending messages and their status.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device',
'messages': []
}), 400
pending = client.get_pending_messages()
return jsonify({
'status': 'ok',
'messages': [m.to_dict() for m in pending.values()],
'count': len(pending)
})
@meshtastic_bp.route('/range-test/start', methods=['POST'])
def start_range_test():
"""
Start a range test.
JSON body:
{
"count": 10, // Number of packets to send (default 10)
"interval": 5 // Seconds between packets (default 5)
}
Returns:
JSON with start status.
"""
if not is_meshtastic_available():
return jsonify({
'status': 'error',
'message': 'Meshtastic SDK not installed'
}), 400
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device'
}), 400
data = request.get_json(silent=True) or {}
count = data.get('count', 10)
interval = data.get('interval', 5)
# Validate
if not isinstance(count, int) or count < 1 or count > 100:
count = 10
if not isinstance(interval, int) or interval < 1 or interval > 60:
interval = 5
success, error = client.start_range_test(count=count, interval=interval)
if success:
return jsonify({
'status': 'started',
'count': count,
'interval': interval
})
else:
return jsonify({
'status': 'error',
'message': error or 'Failed to start range test'
}), 500
@meshtastic_bp.route('/range-test/stop', methods=['POST'])
def stop_range_test():
"""
Stop an ongoing range test.
Returns:
JSON confirmation.
"""
client = get_meshtastic_client()
if client:
client.stop_range_test()
return jsonify({'status': 'stopped'})
@meshtastic_bp.route('/range-test/status')
def get_range_test_status():
"""
Get range test status and results.
Returns:
JSON with running status and results.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device',
'running': False,
'results': []
}), 400
status = client.get_range_test_status()
return jsonify({
'status': 'ok',
**status
})
@meshtastic_bp.route('/store-forward/status')
def get_store_forward_status():
"""
Check if Store & Forward router is available.
Returns:
JSON with availability status and router info.
"""
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device',
'available': False
}), 400
sf_status = client.check_store_forward_available()
return jsonify({
'status': 'ok',
**sf_status
})
@meshtastic_bp.route('/store-forward/request', methods=['POST'])
def request_store_forward():
"""
Request missed messages from Store & Forward router.
JSON body:
{
"window_minutes": 60 // Minutes of history to request (default 60)
}
Returns:
JSON with request status.
"""
if not is_meshtastic_available():
return jsonify({
'status': 'error',
'message': 'Meshtastic SDK not installed'
}), 400
client = get_meshtastic_client()
if not client or not client.is_running:
return jsonify({
'status': 'error',
'message': 'Not connected to Meshtastic device'
}), 400
data = request.get_json(silent=True) or {}
window_minutes = data.get('window_minutes', 60)
if not isinstance(window_minutes, int) or window_minutes < 1 or window_minutes > 1440:
window_minutes = 60
success, error = client.request_store_forward(window_minutes=window_minutes)
if success:
return jsonify({
'status': 'sent',
'window_minutes': window_minutes
})
else:
return jsonify({
'status': 'error',
'message': error or 'Failed to request S&F history'
}), 500
+225 -1
View File
@@ -447,20 +447,27 @@
background: var(--accent-cyan); background: var(--accent-cyan);
border: 2px solid #fff; border: 2px solid #fff;
border-radius: 50%; border-radius: 50%;
box-shadow: 0 2px 6px rgba(0, 0, 0, 0.4); box-shadow:
0 2px 6px rgba(0, 0, 0, 0.4),
0 0 12px 4px rgba(0, 212, 255, 0.5); /* Cyan glow */
color: #000; color: #000;
font-family: 'JetBrains Mono', monospace; font-family: 'JetBrains Mono', monospace;
font-size: 10px; font-size: 10px;
font-weight: bold; font-weight: bold;
text-shadow: 0 0 2px rgba(255, 255, 255, 0.8);
} }
.mesh-node-marker.local { .mesh-node-marker.local {
background: var(--accent-green); background: var(--accent-green);
box-shadow:
0 2px 6px rgba(0, 0, 0, 0.4),
0 0 12px 4px rgba(34, 197, 94, 0.5); /* Green glow */
} }
.mesh-node-marker.stale { .mesh-node-marker.stale {
background: var(--text-dim); background: var(--text-dim);
opacity: 0.7; opacity: 0.7;
box-shadow: 0 2px 6px rgba(0, 0, 0, 0.4); /* No glow for stale */
} }
/* ============================================ /* ============================================
@@ -1352,3 +1359,220 @@
transform: rotate(90deg); transform: rotate(90deg);
} }
} }
/* ============================================
NODE POPUP ACTION BUTTONS
============================================ */
.mesh-position-btn,
.mesh-telemetry-btn {
padding: 6px 10px;
background: var(--bg-secondary);
border: 1px solid var(--border-color);
border-radius: 4px;
color: var(--text-primary);
font-family: 'JetBrains Mono', monospace;
font-size: 9px;
font-weight: 600;
text-transform: uppercase;
cursor: pointer;
transition: all 0.15s ease;
}
.mesh-position-btn:hover,
.mesh-telemetry-btn:hover {
background: var(--accent-cyan);
color: #000;
border-color: var(--accent-cyan);
}
/* ============================================
QR CODE BUTTON
============================================ */
.mesh-qr-btn {
padding: 4px 8px;
background: var(--bg-secondary);
border: 1px solid var(--border-color);
border-radius: 4px;
color: var(--text-primary);
font-family: 'JetBrains Mono', monospace;
font-size: 9px;
font-weight: 600;
cursor: pointer;
transition: all 0.15s ease;
}
.mesh-qr-btn:hover {
background: var(--accent-cyan);
color: #000;
border-color: var(--accent-cyan);
}
/* ============================================
TELEMETRY CHARTS
============================================ */
.mesh-telemetry-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 16px;
padding-bottom: 10px;
border-bottom: 1px solid var(--border-color);
}
.mesh-telemetry-chart {
margin-bottom: 20px;
}
.mesh-telemetry-chart-title {
display: flex;
justify-content: space-between;
align-items: center;
font-family: 'JetBrains Mono', monospace;
font-size: 11px;
font-weight: 600;
color: var(--text-secondary);
text-transform: uppercase;
letter-spacing: 0.5px;
margin-bottom: 8px;
}
.mesh-telemetry-current {
font-size: 14px;
color: var(--accent-cyan);
}
.mesh-telemetry-svg {
width: 100%;
height: 100px;
background: var(--bg-secondary);
border: 1px solid var(--border-color);
border-radius: 6px;
}
.mesh-chart-line {
stroke: var(--accent-cyan);
stroke-width: 2;
stroke-linecap: round;
stroke-linejoin: round;
}
.mesh-chart-grid {
stroke: var(--border-color);
stroke-width: 1;
}
.mesh-chart-label {
font-family: 'JetBrains Mono', monospace;
font-size: 9px;
fill: var(--text-dim);
}
/* ============================================
NETWORK TOPOLOGY
============================================ */
.mesh-network-list {
display: flex;
flex-direction: column;
gap: 12px;
}
.mesh-network-node {
background: var(--bg-secondary);
border: 1px solid var(--border-color);
border-radius: 6px;
padding: 12px;
}
.mesh-network-node-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 10px;
padding-bottom: 8px;
border-bottom: 1px solid var(--border-color);
}
.mesh-network-node-id {
font-family: 'JetBrains Mono', monospace;
font-size: 12px;
font-weight: 600;
color: var(--accent-cyan);
}
.mesh-network-node-count {
font-size: 11px;
color: var(--text-dim);
}
.mesh-network-neighbors {
display: flex;
flex-wrap: wrap;
gap: 8px;
}
.mesh-network-neighbor {
display: flex;
align-items: center;
gap: 8px;
padding: 6px 10px;
background: var(--bg-card);
border: 1px solid var(--border-color);
border-radius: 4px;
}
.mesh-network-neighbor-id {
font-family: 'JetBrains Mono', monospace;
font-size: 10px;
color: var(--text-secondary);
}
.mesh-network-neighbor-snr {
font-family: 'JetBrains Mono', monospace;
font-size: 9px;
font-weight: 600;
padding: 2px 6px;
border-radius: 10px;
}
.mesh-network-neighbor-snr.snr-good {
background: rgba(34, 197, 94, 0.15);
color: var(--accent-green);
}
.mesh-network-neighbor-snr.snr-ok {
background: rgba(74, 158, 255, 0.15);
color: var(--accent-cyan);
}
.mesh-network-neighbor-snr.snr-poor {
background: rgba(255, 193, 7, 0.15);
color: var(--accent-orange);
}
.mesh-network-neighbor-snr.snr-bad {
background: rgba(255, 51, 102, 0.15);
color: var(--accent-red, #ff3366);
}
/* ============================================
FIRMWARE BADGES
============================================ */
.mesh-badge {
display: inline-block;
padding: 3px 8px;
font-family: 'JetBrains Mono', monospace;
font-size: 10px;
font-weight: 600;
border-radius: 10px;
text-transform: uppercase;
}
.mesh-badge-success {
background: rgba(34, 197, 94, 0.15);
color: var(--accent-green);
}
.mesh-badge-warning {
background: rgba(255, 193, 7, 0.15);
color: var(--accent-orange);
}
+766 -4
View File
@@ -33,7 +33,7 @@ const Meshtastic = (function() {
* Setup event delegation for dynamically created elements * Setup event delegation for dynamically created elements
*/ */
function setupEventDelegation() { function setupEventDelegation() {
// Handle traceroute button clicks in Leaflet popups // Handle button clicks in Leaflet popups and elsewhere
document.addEventListener('click', function(e) { document.addEventListener('click', function(e) {
const tracerouteBtn = e.target.closest('.mesh-traceroute-btn'); const tracerouteBtn = e.target.closest('.mesh-traceroute-btn');
if (tracerouteBtn) { if (tracerouteBtn) {
@@ -42,6 +42,22 @@ const Meshtastic = (function() {
sendTraceroute(nodeId); sendTraceroute(nodeId);
} }
} }
const positionBtn = e.target.closest('.mesh-position-btn');
if (positionBtn) {
const nodeId = positionBtn.dataset.nodeId;
if (nodeId) {
requestPosition(nodeId);
}
}
const qrBtn = e.target.closest('.mesh-qr-btn');
if (qrBtn) {
const channelIndex = qrBtn.dataset.channelIndex;
if (channelIndex !== undefined) {
showChannelQR(parseInt(channelIndex, 10));
}
}
}); });
} }
@@ -380,6 +396,7 @@ const Meshtastic = (function() {
<div class="mesh-channel-badges"> <div class="mesh-channel-badges">
<span class="mesh-channel-badge ${roleBadge}">${ch.role || 'SECONDARY'}</span> <span class="mesh-channel-badge ${roleBadge}">${ch.role || 'SECONDARY'}</span>
<span class="mesh-channel-badge ${encBadge}">${encText}</span> <span class="mesh-channel-badge ${encBadge}">${encText}</span>
<button class="mesh-qr-btn" data-channel-index="${ch.index}" title="Generate QR Code">QR</button>
<button class="mesh-channel-configure" onclick="Meshtastic.openChannelModal(${ch.index})">Configure</button> <button class="mesh-channel-configure" onclick="Meshtastic.openChannelModal(${ch.index})">Configure</button>
</div> </div>
</div> </div>
@@ -647,7 +664,18 @@ const Meshtastic = (function() {
envHtml += '</div>'; envHtml += '</div>';
} }
// Build popup content // Build popup content with action buttons
let actionButtons = '';
if (!isLocal) {
actionButtons = `
<div style="margin-top: 8px; display: flex; gap: 4px; flex-wrap: wrap;">
<button class="mesh-traceroute-btn" data-node-id="${nodeId}">Traceroute</button>
<button class="mesh-position-btn" data-node-id="${nodeId}">Request Position</button>
<button class="mesh-telemetry-btn" onclick="Meshtastic.showTelemetryChart('${nodeId}')">Telemetry</button>
</div>
`;
}
const popupContent = ` const popupContent = `
<div style="min-width: 150px;"> <div style="min-width: 150px;">
<strong style="color: var(--accent-cyan);">${node.long_name || shortName}</strong><br> <strong style="color: var(--accent-cyan);">${node.long_name || shortName}</strong><br>
@@ -660,7 +688,7 @@ const Meshtastic = (function() {
${node.last_heard ? `<span style="color: var(--text-dim);">Last heard:</span> ${new Date(node.last_heard).toLocaleTimeString()}<br>` : ''} ${node.last_heard ? `<span style="color: var(--text-dim);">Last heard:</span> ${new Date(node.last_heard).toLocaleTimeString()}<br>` : ''}
${telemetryHtml} ${telemetryHtml}
${envHtml} ${envHtml}
${!isLocal ? `<button class="mesh-traceroute-btn" data-node-id="${nodeId}">Traceroute</button>` : ''} ${actionButtons}
</div> </div>
`; `;
@@ -1453,6 +1481,721 @@ const Meshtastic = (function() {
return null; return null;
} }
/**
* Request position from a specific node
*/
async function requestPosition(nodeId) {
if (!nodeId) return;
try {
const response = await fetch('/meshtastic/position/request', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ node_id: nodeId })
});
const data = await response.json();
if (data.status === 'sent') {
showNotification('Meshtastic', `Position requested from ${nodeId}`);
// Refresh nodes after a delay to get updated position
setTimeout(loadNodes, 5000);
} else {
showStatusMessage(data.message || 'Failed to request position', 'error');
}
} catch (err) {
console.error('Position request error:', err);
showStatusMessage('Error requesting position: ' + err.message, 'error');
}
}
/**
* Check firmware version and show update status
*/
async function checkFirmware() {
try {
const response = await fetch('/meshtastic/firmware/check');
const data = await response.json();
if (data.status === 'ok') {
showFirmwareModal(data);
} else {
showStatusMessage(data.message || 'Failed to check firmware', 'error');
}
} catch (err) {
console.error('Firmware check error:', err);
showStatusMessage('Error checking firmware: ' + err.message, 'error');
}
}
/**
* Show firmware information modal
*/
function showFirmwareModal(info) {
let modal = document.getElementById('meshFirmwareModal');
if (!modal) {
modal = document.createElement('div');
modal.id = 'meshFirmwareModal';
modal.className = 'signal-details-modal';
document.body.appendChild(modal);
}
const updateBadge = info.update_available
? '<span class="mesh-badge mesh-badge-warning">Update Available</span>'
: '<span class="mesh-badge mesh-badge-success">Up to Date</span>';
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeFirmwareModal()"></div>
<div class="signal-details-modal-content">
<div class="signal-details-modal-header">
<h3>Firmware Information</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeFirmwareModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<div class="signal-details-section">
<div class="signal-details-title">Current Version</div>
<p style="color: var(--text-secondary); font-size: 14px;">
${info.current_version || 'Unknown'}
</p>
</div>
<div class="signal-details-section">
<div class="signal-details-title">Latest Version</div>
<p style="color: var(--text-secondary); font-size: 14px;">
${info.latest_version || 'Unknown'} ${updateBadge}
</p>
</div>
${info.release_url ? `
<div class="signal-details-section">
<a href="${info.release_url}" target="_blank" rel="noopener" class="preset-btn" style="display: inline-block; text-decoration: none;">
View Release Notes
</a>
</div>
` : ''}
${info.error ? `
<div class="signal-details-section">
<p style="color: var(--status-error); font-size: 12px;">
Note: ${info.error}
</p>
</div>
` : ''}
</div>
</div>
`;
modal.classList.add('show');
}
/**
* Close firmware modal
*/
function closeFirmwareModal() {
const modal = document.getElementById('meshFirmwareModal');
if (modal) modal.classList.remove('show');
}
/**
* Show QR code for a channel
*/
async function showChannelQR(channelIndex) {
let modal = document.getElementById('meshQRModal');
if (!modal) {
modal = document.createElement('div');
modal.id = 'meshQRModal';
modal.className = 'signal-details-modal';
document.body.appendChild(modal);
}
const channel = channels.find(ch => ch.index === channelIndex);
const channelName = channel?.name || `Channel ${channelIndex}`;
// Show loading state
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeQRModal()"></div>
<div class="signal-details-modal-content">
<div class="signal-details-modal-header">
<h3>Channel QR Code</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeQRModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<div style="text-align: center; padding: 20px;">
<div class="mesh-traceroute-spinner"></div>
<p style="color: var(--text-dim); margin-top: 10px;">Generating QR code...</p>
</div>
</div>
</div>
`;
modal.classList.add('show');
try {
const response = await fetch(`/meshtastic/channels/${channelIndex}/qr`);
if (response.ok) {
const blob = await response.blob();
const imageUrl = URL.createObjectURL(blob);
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeQRModal()"></div>
<div class="signal-details-modal-content">
<div class="signal-details-modal-header">
<h3>Channel QR Code</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeQRModal()">&times;</button>
</div>
<div class="signal-details-modal-body" style="text-align: center;">
<p style="color: var(--text-secondary); font-size: 14px; margin-bottom: 15px;">
${escapeHtml(channelName)}
</p>
<img src="${imageUrl}" alt="Channel QR Code" style="max-width: 256px; background: white; padding: 10px; border-radius: 8px;">
<p style="color: var(--text-dim); font-size: 11px; margin-top: 15px;">
Scan with the Meshtastic app to join this channel
</p>
</div>
</div>
`;
} else {
const data = await response.json();
throw new Error(data.message || 'Failed to generate QR code');
}
} catch (err) {
console.error('QR generation error:', err);
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeQRModal()"></div>
<div class="signal-details-modal-content">
<div class="signal-details-modal-header">
<h3>Channel QR Code</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeQRModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<p style="color: var(--status-error);">
Error: ${escapeHtml(err.message)}
</p>
<p style="color: var(--text-dim); font-size: 11px; margin-top: 10px;">
Make sure the qrcode library is installed: pip install qrcode[pil]
</p>
</div>
</div>
`;
}
}
/**
* Close QR modal
*/
function closeQRModal() {
const modal = document.getElementById('meshQRModal');
if (modal) modal.classList.remove('show');
}
/**
* Load and display telemetry history for a node
*/
async function showTelemetryChart(nodeId, hours = 24) {
let modal = document.getElementById('meshTelemetryModal');
if (!modal) {
modal = document.createElement('div');
modal.id = 'meshTelemetryModal';
modal.className = 'signal-details-modal';
document.body.appendChild(modal);
}
// Show loading
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeTelemetryModal()"></div>
<div class="signal-details-modal-content" style="max-width: 600px;">
<div class="signal-details-modal-header">
<h3>Telemetry History</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeTelemetryModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<div style="text-align: center; padding: 20px;">
<div class="mesh-traceroute-spinner"></div>
<p style="color: var(--text-dim); margin-top: 10px;">Loading telemetry data...</p>
</div>
</div>
</div>
`;
modal.classList.add('show');
try {
const response = await fetch(`/meshtastic/telemetry/history?node_id=${encodeURIComponent(nodeId)}&hours=${hours}`);
const data = await response.json();
if (data.status === 'ok') {
renderTelemetryCharts(modal, nodeId, data.data, hours);
} else {
throw new Error(data.message || 'Failed to load telemetry');
}
} catch (err) {
console.error('Telemetry load error:', err);
modal.querySelector('.signal-details-modal-body').innerHTML = `
<p style="color: var(--status-error);">Error: ${escapeHtml(err.message)}</p>
`;
}
}
/**
* Render telemetry charts
*/
function renderTelemetryCharts(modal, nodeId, data, hours) {
if (!data || data.length === 0) {
modal.querySelector('.signal-details-modal-body').innerHTML = `
<p style="color: var(--text-dim); text-align: center; padding: 20px;">
No telemetry data available for this node in the last ${hours} hours.
</p>
`;
return;
}
// Build charts for available metrics
let chartsHtml = `
<div class="mesh-telemetry-header">
<span>Node: ${escapeHtml(nodeId)}</span>
<span style="color: var(--text-dim);">${data.length} data points</span>
</div>
`;
// Battery chart
const batteryData = data.filter(p => p.battery_level !== null);
if (batteryData.length > 0) {
chartsHtml += renderSimpleChart('Battery Level', batteryData, 'battery_level', '%', 0, 100);
}
// Voltage chart
const voltageData = data.filter(p => p.voltage !== null);
if (voltageData.length > 0) {
chartsHtml += renderSimpleChart('Voltage', voltageData, 'voltage', 'V', null, null);
}
// Temperature chart
const tempData = data.filter(p => p.temperature !== null);
if (tempData.length > 0) {
chartsHtml += renderSimpleChart('Temperature', tempData, 'temperature', '°C', null, null);
}
// Humidity chart
const humidityData = data.filter(p => p.humidity !== null);
if (humidityData.length > 0) {
chartsHtml += renderSimpleChart('Humidity', humidityData, 'humidity', '%', 0, 100);
}
modal.querySelector('.signal-details-modal-body').innerHTML = chartsHtml;
}
/**
* Render a simple SVG line chart
*/
function renderSimpleChart(title, data, field, unit, minY, maxY) {
if (data.length < 2) {
return `
<div class="mesh-telemetry-chart">
<div class="mesh-telemetry-chart-title">${title}</div>
<p style="color: var(--text-dim); font-size: 11px;">Not enough data points</p>
</div>
`;
}
// Extract values
const values = data.map(p => p[field]);
const timestamps = data.map(p => new Date(p.timestamp));
// Calculate bounds
const min = minY !== null ? minY : Math.min(...values) * 0.95;
const max = maxY !== null ? maxY : Math.max(...values) * 1.05;
const range = max - min || 1;
// Chart dimensions
const width = 500;
const height = 100;
const padding = { left: 40, right: 10, top: 10, bottom: 20 };
const chartWidth = width - padding.left - padding.right;
const chartHeight = height - padding.top - padding.bottom;
// Build path
const points = values.map((v, i) => {
const x = padding.left + (i / (values.length - 1)) * chartWidth;
const y = padding.top + chartHeight - ((v - min) / range) * chartHeight;
return `${x},${y}`;
});
const pathD = 'M' + points.join(' L');
// Current value
const currentValue = values[values.length - 1];
return `
<div class="mesh-telemetry-chart">
<div class="mesh-telemetry-chart-title">
${title}
<span class="mesh-telemetry-current">${currentValue.toFixed(1)}${unit}</span>
</div>
<svg viewBox="0 0 ${width} ${height}" class="mesh-telemetry-svg">
<!-- Y axis labels -->
<text x="${padding.left - 5}" y="${padding.top + 5}" class="mesh-chart-label" text-anchor="end">${max.toFixed(0)}</text>
<text x="${padding.left - 5}" y="${height - padding.bottom}" class="mesh-chart-label" text-anchor="end">${min.toFixed(0)}</text>
<!-- Grid lines -->
<line x1="${padding.left}" y1="${padding.top}" x2="${width - padding.right}" y2="${padding.top}" class="mesh-chart-grid"/>
<line x1="${padding.left}" y1="${height - padding.bottom}" x2="${width - padding.right}" y2="${height - padding.bottom}" class="mesh-chart-grid"/>
<!-- Data line -->
<path d="${pathD}" class="mesh-chart-line" fill="none"/>
</svg>
</div>
`;
}
/**
* Close telemetry modal
*/
function closeTelemetryModal() {
const modal = document.getElementById('meshTelemetryModal');
if (modal) modal.classList.remove('show');
}
/**
* Show network topology (neighbors)
*/
async function showNetworkTopology() {
let modal = document.getElementById('meshNetworkModal');
if (!modal) {
modal = document.createElement('div');
modal.id = 'meshNetworkModal';
modal.className = 'signal-details-modal';
document.body.appendChild(modal);
}
// Show loading
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeNetworkModal()"></div>
<div class="signal-details-modal-content" style="max-width: 700px;">
<div class="signal-details-modal-header">
<h3>Network Topology</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeNetworkModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<div style="text-align: center; padding: 20px;">
<div class="mesh-traceroute-spinner"></div>
<p style="color: var(--text-dim); margin-top: 10px;">Loading neighbor data...</p>
</div>
</div>
</div>
`;
modal.classList.add('show');
try {
const response = await fetch('/meshtastic/neighbors');
const data = await response.json();
if (data.status === 'ok') {
renderNetworkTopology(modal, data.neighbors);
} else {
throw new Error(data.message || 'Failed to load neighbors');
}
} catch (err) {
console.error('Network topology error:', err);
modal.querySelector('.signal-details-modal-body').innerHTML = `
<p style="color: var(--status-error);">Error: ${escapeHtml(err.message)}</p>
`;
}
}
/**
* Render network topology visualization
*/
function renderNetworkTopology(modal, neighbors) {
if (!neighbors || Object.keys(neighbors).length === 0) {
modal.querySelector('.signal-details-modal-body').innerHTML = `
<p style="color: var(--text-dim); text-align: center; padding: 20px;">
No neighbor information available yet.<br>
<span style="font-size: 11px;">Neighbor data is collected from NEIGHBOR_INFO_APP packets.</span>
</p>
`;
return;
}
// Build a simple list view of neighbors
let html = '<div class="mesh-network-list">';
for (const [nodeId, neighborList] of Object.entries(neighbors)) {
html += `
<div class="mesh-network-node">
<div class="mesh-network-node-header">
<span class="mesh-network-node-id">${escapeHtml(nodeId)}</span>
<span class="mesh-network-node-count">${neighborList.length} neighbors</span>
</div>
<div class="mesh-network-neighbors">
`;
neighborList.forEach(neighbor => {
const snrClass = getSnrClass(neighbor.snr);
html += `
<div class="mesh-network-neighbor">
<span class="mesh-network-neighbor-id">${escapeHtml(neighbor.neighbor_id)}</span>
<span class="mesh-network-neighbor-snr ${snrClass}">${neighbor.snr.toFixed(1)} dB</span>
</div>
`;
});
html += '</div></div>';
}
html += '</div>';
modal.querySelector('.signal-details-modal-body').innerHTML = html;
}
/**
* Close network modal
*/
function closeNetworkModal() {
const modal = document.getElementById('meshNetworkModal');
if (modal) modal.classList.remove('show');
}
/**
* Show range test modal
*/
function showRangeTestModal() {
let modal = document.getElementById('meshRangeTestModal');
if (!modal) {
modal = document.createElement('div');
modal.id = 'meshRangeTestModal';
modal.className = 'signal-details-modal';
document.body.appendChild(modal);
}
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeRangeTestModal()"></div>
<div class="signal-details-modal-content">
<div class="signal-details-modal-header">
<h3>Range Test</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeRangeTestModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<div class="form-group">
<label>Number of Packets</label>
<input type="number" id="rangeTestCount" value="10" min="1" max="100" style="width: 100%; padding: 8px; background: var(--bg-secondary); border: 1px solid var(--border-color); border-radius: 4px; color: var(--text-primary);">
</div>
<div class="form-group" style="margin-top: 12px;">
<label>Interval (seconds)</label>
<input type="number" id="rangeTestInterval" value="5" min="1" max="60" style="width: 100%; padding: 8px; background: var(--bg-secondary); border: 1px solid var(--border-color); border-radius: 4px; color: var(--text-primary);">
</div>
<div style="margin-top: 16px; display: flex; gap: 8px;">
<button id="rangeTestStartBtn" class="run-btn" onclick="Meshtastic.startRangeTest()">Start Test</button>
<button id="rangeTestStopBtn" class="run-btn" style="display: none; background: var(--accent-red);" onclick="Meshtastic.stopRangeTest()">Stop Test</button>
</div>
<div id="rangeTestStatus" style="margin-top: 16px; display: none;">
<div class="mesh-traceroute-spinner" style="margin: 0 auto;"></div>
<p style="color: var(--text-dim); text-align: center; margin-top: 10px;">Sending packets...</p>
</div>
</div>
</div>
`;
modal.classList.add('show');
}
/**
* Start range test
*/
async function startRangeTest() {
const countInput = document.getElementById('rangeTestCount');
const intervalInput = document.getElementById('rangeTestInterval');
const startBtn = document.getElementById('rangeTestStartBtn');
const stopBtn = document.getElementById('rangeTestStopBtn');
const statusDiv = document.getElementById('rangeTestStatus');
const count = parseInt(countInput?.value || '10', 10);
const interval = parseInt(intervalInput?.value || '5', 10);
try {
const response = await fetch('/meshtastic/range-test/start', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ count, interval })
});
const data = await response.json();
if (data.status === 'started') {
if (startBtn) startBtn.style.display = 'none';
if (stopBtn) stopBtn.style.display = 'inline-block';
if (statusDiv) statusDiv.style.display = 'block';
showNotification('Meshtastic', `Range test started: ${count} packets`);
// Poll for completion
pollRangeTestStatus();
} else {
showStatusMessage(data.message || 'Failed to start range test', 'error');
}
} catch (err) {
console.error('Range test error:', err);
showStatusMessage('Error starting range test: ' + err.message, 'error');
}
}
/**
* Stop range test
*/
async function stopRangeTest() {
try {
await fetch('/meshtastic/range-test/stop', { method: 'POST' });
resetRangeTestUI();
showNotification('Meshtastic', 'Range test stopped');
} catch (err) {
console.error('Error stopping range test:', err);
}
}
/**
* Poll range test status
*/
async function pollRangeTestStatus() {
try {
const response = await fetch('/meshtastic/range-test/status');
const data = await response.json();
if (data.running) {
setTimeout(pollRangeTestStatus, 1000);
} else {
resetRangeTestUI();
showNotification('Meshtastic', 'Range test complete');
}
} catch (err) {
console.error('Error polling range test:', err);
resetRangeTestUI();
}
}
/**
* Reset range test UI
*/
function resetRangeTestUI() {
const startBtn = document.getElementById('rangeTestStartBtn');
const stopBtn = document.getElementById('rangeTestStopBtn');
const statusDiv = document.getElementById('rangeTestStatus');
if (startBtn) startBtn.style.display = 'inline-block';
if (stopBtn) stopBtn.style.display = 'none';
if (statusDiv) statusDiv.style.display = 'none';
}
/**
* Close range test modal
*/
function closeRangeTestModal() {
const modal = document.getElementById('meshRangeTestModal');
if (modal) modal.classList.remove('show');
}
/**
* Show Store & Forward modal
*/
async function showStoreForwardModal() {
let modal = document.getElementById('meshStoreForwardModal');
if (!modal) {
modal = document.createElement('div');
modal.id = 'meshStoreForwardModal';
modal.className = 'signal-details-modal';
document.body.appendChild(modal);
}
// Show loading state
modal.innerHTML = `
<div class="signal-details-modal-backdrop" onclick="Meshtastic.closeStoreForwardModal()"></div>
<div class="signal-details-modal-content">
<div class="signal-details-modal-header">
<h3>Store & Forward</h3>
<button class="signal-details-modal-close" onclick="Meshtastic.closeStoreForwardModal()">&times;</button>
</div>
<div class="signal-details-modal-body">
<div style="text-align: center; padding: 20px;">
<div class="mesh-traceroute-spinner"></div>
<p style="color: var(--text-dim); margin-top: 10px;">Checking for S&F router...</p>
</div>
</div>
</div>
`;
modal.classList.add('show');
try {
const response = await fetch('/meshtastic/store-forward/status');
const data = await response.json();
if (data.available) {
modal.querySelector('.signal-details-modal-body').innerHTML = `
<div class="mesh-sf-info">
<p style="color: var(--accent-green); margin-bottom: 12px;">
✓ Store & Forward router found
</p>
<p style="color: var(--text-secondary); font-size: 12px;">
Router: ${escapeHtml(data.router_name || data.router_id || 'Unknown')}
</p>
</div>
<div class="form-group" style="margin-top: 16px;">
<label>Request history for:</label>
<select id="sfWindowMinutes" style="width: 100%; padding: 8px; background: var(--bg-secondary); border: 1px solid var(--border-color); border-radius: 4px; color: var(--text-primary);">
<option value="15">Last 15 minutes</option>
<option value="60" selected>Last hour</option>
<option value="240">Last 4 hours</option>
<option value="1440">Last 24 hours</option>
</select>
</div>
<button class="run-btn" style="margin-top: 16px; width: 100%;" onclick="Meshtastic.requestStoreForward()">
Fetch Missed Messages
</button>
`;
} else {
modal.querySelector('.signal-details-modal-body').innerHTML = `
<p style="color: var(--text-dim); text-align: center; padding: 20px;">
No Store & Forward router found on the mesh.<br><br>
<span style="font-size: 11px;">
S&F requires a node with ROUTER role running the<br>
Store & Forward module with history enabled.
</span>
</p>
`;
}
} catch (err) {
console.error('S&F status error:', err);
modal.querySelector('.signal-details-modal-body').innerHTML = `
<p style="color: var(--status-error);">Error: ${escapeHtml(err.message)}</p>
`;
}
}
/**
* Request Store & Forward history
*/
async function requestStoreForward() {
const select = document.getElementById('sfWindowMinutes');
const windowMinutes = parseInt(select?.value || '60', 10);
try {
const response = await fetch('/meshtastic/store-forward/request', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ window_minutes: windowMinutes })
});
const data = await response.json();
if (data.status === 'sent') {
showNotification('Meshtastic', `Requested ${windowMinutes} minutes of history`);
closeStoreForwardModal();
} else {
showStatusMessage(data.message || 'Failed to request S&F history', 'error');
}
} catch (err) {
console.error('S&F request error:', err);
showStatusMessage('Error: ' + err.message, 'error');
}
}
/**
* Close Store & Forward modal
*/
function closeStoreForwardModal() {
const modal = document.getElementById('meshStoreForwardModal');
if (modal) modal.classList.remove('show');
}
return { return {
init, init,
start, start,
@@ -1473,7 +2216,26 @@ const Meshtastic = (function() {
toggleSidebar, toggleSidebar,
toggleOptionsPanel, toggleOptionsPanel,
sendTraceroute, sendTraceroute,
closeTracerouteModal closeTracerouteModal,
// New features
requestPosition,
checkFirmware,
closeFirmwareModal,
showChannelQR,
closeQRModal,
showTelemetryChart,
closeTelemetryModal,
showNetworkTopology,
closeNetworkModal,
// Range test
showRangeTestModal,
startRangeTest,
stopRangeTest,
closeRangeTestModal,
// Store & Forward
showStoreForwardModal,
requestStoreForward,
closeStoreForwardModal
}; };
/** /**
+571 -1
View File
@@ -13,6 +13,9 @@ import base64
import hashlib import hashlib
import secrets import secrets
import threading import threading
import urllib.request
import json
from collections import deque
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Callable from typing import Callable
@@ -202,6 +205,69 @@ class TracerouteResult:
} }
@dataclass
class TelemetryPoint:
"""Single telemetry data point for graphing."""
timestamp: datetime
battery_level: int | None = None
voltage: float | None = None
temperature: float | None = None
humidity: float | None = None
pressure: float | None = None
channel_utilization: float | None = None
air_util_tx: float | None = None
def to_dict(self) -> dict:
return {
'timestamp': self.timestamp.isoformat(),
'battery_level': self.battery_level,
'voltage': self.voltage,
'temperature': self.temperature,
'humidity': self.humidity,
'pressure': self.pressure,
'channel_utilization': self.channel_utilization,
'air_util_tx': self.air_util_tx,
}
@dataclass
class PendingMessage:
"""Message waiting for ACK/NAK."""
packet_id: int
destination: int
text: str
channel: int
timestamp: datetime
status: str = 'pending' # pending, acked, failed
def to_dict(self) -> dict:
return {
'packet_id': self.packet_id,
'destination': self.destination,
'text': self.text,
'channel': self.channel,
'timestamp': self.timestamp.isoformat(),
'status': self.status,
}
@dataclass
class NeighborInfo:
"""Neighbor information from NEIGHBOR_INFO_APP."""
neighbor_num: int
neighbor_id: str
snr: float
timestamp: datetime
def to_dict(self) -> dict:
return {
'neighbor_num': self.neighbor_num,
'neighbor_id': self.neighbor_id,
'snr': self.snr,
'timestamp': self.timestamp.isoformat(),
}
class MeshtasticClient: class MeshtasticClient:
"""Client for connecting to Meshtastic devices.""" """Client for connecting to Meshtastic devices."""
@@ -216,6 +282,25 @@ class MeshtasticClient:
self._traceroute_results: list[TracerouteResult] = [] self._traceroute_results: list[TracerouteResult] = []
self._max_traceroute_results = 50 self._max_traceroute_results = 50
# Telemetry history for graphing (node_num -> deque of TelemetryPoints)
self._telemetry_history: dict[int, deque] = {}
self._max_telemetry_points = 1000
# Pending messages for ACK tracking (packet_id -> PendingMessage)
self._pending_messages: dict[int, PendingMessage] = {}
# Neighbor info (node_num -> list of NeighborInfo)
self._neighbors: dict[int, list[NeighborInfo]] = {}
# Firmware version cache
self._firmware_version: str | None = None
self._latest_firmware: dict | None = None
self._firmware_check_time: datetime | None = None
# Range test state
self._range_test_running: bool = False
self._range_test_results: list[dict] = []
@property @property
def is_running(self) -> bool: def is_running(self) -> bool:
return self._running return self._running
@@ -357,13 +442,21 @@ class MeshtasticClient:
if portnum == 'TRACEROUTE_APP': if portnum == 'TRACEROUTE_APP':
self._handle_traceroute_response(packet, decoded) self._handle_traceroute_response(packet, decoded)
# Handle ACK/NAK for message delivery tracking
if portnum == 'ROUTING_APP':
self._handle_routing_packet(packet, decoded)
# Handle neighbor info for mesh topology
if portnum == 'NEIGHBOR_INFO_APP':
self._handle_neighbor_info(packet, decoded)
# Skip callback if none set # Skip callback if none set
if not self._callback: if not self._callback:
return return
# Filter out internal protocol messages that aren't useful to users # Filter out internal protocol messages that aren't useful to users
ignored_portnums = { ignored_portnums = {
'ROUTING_APP', # Mesh routing/acknowledgments 'ROUTING_APP', # Mesh routing/acknowledgments - handled above
'ADMIN_APP', # Admin commands 'ADMIN_APP', # Admin commands
'REPLY_APP', # Internal replies 'REPLY_APP', # Internal replies
'STORE_FORWARD_APP', # Store and forward protocol 'STORE_FORWARD_APP', # Store and forward protocol
@@ -375,6 +468,7 @@ class MeshtasticClient:
'TELEMETRY_APP', # Device telemetry (battery, etc.) - too noisy 'TELEMETRY_APP', # Device telemetry (battery, etc.) - too noisy
'POSITION_APP', # Position updates - used for map, not messages 'POSITION_APP', # Position updates - used for map, not messages
'NODEINFO_APP', # Node info - used for tracking, not messages 'NODEINFO_APP', # Node info - used for tracking, not messages
'NEIGHBOR_INFO_APP', # Neighbor info - handled above
} }
if portnum in ignored_portnums: if portnum in ignored_portnums:
logger.debug(f"Ignoring {portnum} message from {from_num}") logger.debug(f"Ignoring {portnum} message from {from_num}")
@@ -499,6 +593,32 @@ class MeshtasticClient:
if pressure is not None: if pressure is not None:
node.barometric_pressure = pressure node.barometric_pressure = pressure
# Store telemetry point for historical graphing
self._store_telemetry_point(from_num, device_metrics, env_metrics)
def _store_telemetry_point(self, node_num: int, device_metrics: dict, env_metrics: dict) -> None:
"""Store a telemetry data point for historical graphing."""
# Skip if no actual data
if not device_metrics and not env_metrics:
return
point = TelemetryPoint(
timestamp=datetime.now(timezone.utc),
battery_level=device_metrics.get('batteryLevel'),
voltage=device_metrics.get('voltage'),
temperature=env_metrics.get('temperature'),
humidity=env_metrics.get('relativeHumidity'),
pressure=env_metrics.get('barometricPressure'),
channel_utilization=device_metrics.get('channelUtilization'),
air_util_tx=device_metrics.get('airUtilTx'),
)
# Initialize deque for this node if needed
if node_num not in self._telemetry_history:
self._telemetry_history[node_num] = deque(maxlen=self._max_telemetry_points)
self._telemetry_history[node_num].append(point)
def _lookup_node_name(self, node_num: int) -> str | None: def _lookup_node_name(self, node_num: int) -> str | None:
"""Look up a node's name by its number.""" """Look up a node's name by its number."""
if node_num == 0 or node_num == BROADCAST_ADDR: if node_num == 0 or node_num == BROADCAST_ADDR:
@@ -921,6 +1041,456 @@ class MeshtasticClient:
results = results[:limit] results = results[:limit]
return results return results
def _handle_routing_packet(self, packet: dict, decoded: dict) -> None:
"""Handle ROUTING_APP packets for ACK/NAK tracking."""
try:
routing = decoded.get('routing', {})
error_reason = routing.get('errorReason')
request_id = packet.get('requestId', 0)
if request_id and request_id in self._pending_messages:
msg = self._pending_messages[request_id]
if error_reason and error_reason != 'NONE':
msg.status = 'failed'
logger.debug(f"Message {request_id} failed: {error_reason}")
else:
msg.status = 'acked'
logger.debug(f"Message {request_id} acknowledged")
except Exception as e:
logger.error(f"Error handling routing packet: {e}")
def _handle_neighbor_info(self, packet: dict, decoded: dict) -> None:
"""Handle NEIGHBOR_INFO_APP packets for mesh topology."""
try:
from_num = packet.get('from', 0)
if from_num == 0:
return
neighbor_info = decoded.get('neighborinfo', {})
neighbors = neighbor_info.get('neighbors', [])
now = datetime.now(timezone.utc)
neighbor_list = []
for neighbor in neighbors:
neighbor_num = neighbor.get('nodeId', 0)
if neighbor_num:
neighbor_list.append(NeighborInfo(
neighbor_num=neighbor_num,
neighbor_id=self._format_node_id(neighbor_num),
snr=neighbor.get('snr', 0.0),
timestamp=now,
))
if neighbor_list:
self._neighbors[from_num] = neighbor_list
logger.debug(f"Updated neighbors for {self._format_node_id(from_num)}: {len(neighbor_list)} neighbors")
except Exception as e:
logger.error(f"Error handling neighbor info: {e}")
def get_neighbors(self, node_num: int | None = None) -> dict[int, list[NeighborInfo]]:
"""
Get neighbor information for mesh topology visualization.
Args:
node_num: Specific node number, or None for all nodes
Returns:
Dict mapping node_num to list of NeighborInfo
"""
if node_num is not None:
return {node_num: self._neighbors.get(node_num, [])}
return dict(self._neighbors)
def get_telemetry_history(self, node_num: int, hours: int = 24) -> list[TelemetryPoint]:
"""
Get telemetry history for a node.
Args:
node_num: Node number to get history for
hours: Number of hours of history to return
Returns:
List of TelemetryPoint objects
"""
if node_num not in self._telemetry_history:
return []
cutoff = datetime.now(timezone.utc).timestamp() - (hours * 3600)
return [
p for p in self._telemetry_history[node_num]
if p.timestamp.timestamp() > cutoff
]
def get_pending_messages(self) -> dict[int, PendingMessage]:
"""Get all pending messages waiting for ACK."""
return dict(self._pending_messages)
def request_position(self, destination: str | int) -> tuple[bool, str]:
"""
Request position from a specific node.
Args:
destination: Target node ID (string like "!a1b2c3d4" or int)
Returns:
Tuple of (success, error_message)
"""
if not self._interface:
return False, "Not connected to device"
if not HAS_MESHTASTIC:
return False, "Meshtastic SDK not installed"
try:
# Parse destination
if isinstance(destination, int):
dest_id = destination
elif destination.startswith('!'):
dest_id = int(destination[1:], 16)
else:
try:
dest_id = int(destination)
except ValueError:
return False, f"Invalid destination: {destination}"
if dest_id == BROADCAST_ADDR:
return False, "Cannot request position from broadcast address"
# Send position request using admin message
# The Meshtastic SDK's localNode.requestPosition works for the local node
# For remote nodes, we send a POSITION_APP request
from meshtastic import portnums_pb2
# Request position by sending an empty position request packet
self._interface.sendData(
b'', # Empty payload triggers position response
destinationId=dest_id,
portNum=portnums_pb2.PortNum.POSITION_APP,
wantAck=True,
wantResponse=True,
)
logger.info(f"Sent position request to {self._format_node_id(dest_id)}")
return True, None
except Exception as e:
logger.error(f"Error requesting position: {e}")
return False, str(e)
def check_firmware(self) -> dict:
"""
Check current firmware version and compare to latest release.
Returns:
Dict with current_version, latest_version, update_available, release_url
"""
result = {
'current_version': None,
'latest_version': None,
'update_available': False,
'release_url': None,
'error': None,
}
# Get current firmware version from device
if self._interface:
try:
my_info = self._interface.getMyNodeInfo()
if my_info:
metadata = my_info.get('deviceMetrics', {})
# Firmware version is in the user section or metadata
if 'firmware_version' in my_info:
self._firmware_version = my_info['firmware_version']
elif hasattr(self._interface, 'myInfo') and self._interface.myInfo:
self._firmware_version = getattr(self._interface.myInfo, 'firmware_version', None)
result['current_version'] = self._firmware_version
except Exception as e:
logger.warning(f"Could not get device firmware version: {e}")
# Check GitHub for latest release (cache for 15 minutes)
now = datetime.now(timezone.utc)
cache_valid = (
self._firmware_check_time and
self._latest_firmware and
(now - self._firmware_check_time).total_seconds() < 900
)
if not cache_valid:
try:
url = 'https://api.github.com/repos/meshtastic/firmware/releases/latest'
req = urllib.request.Request(url, headers={'User-Agent': 'INTERCEPT'})
with urllib.request.urlopen(req, timeout=10) as response:
data = json.loads(response.read().decode('utf-8'))
self._latest_firmware = {
'version': data.get('tag_name', '').lstrip('v'),
'url': data.get('html_url'),
'name': data.get('name'),
}
self._firmware_check_time = now
except Exception as e:
logger.warning(f"Could not check latest firmware: {e}")
result['error'] = str(e)
if self._latest_firmware:
result['latest_version'] = self._latest_firmware.get('version')
result['release_url'] = self._latest_firmware.get('url')
# Compare versions
if result['current_version'] and result['latest_version']:
result['update_available'] = self._compare_versions(
result['current_version'],
result['latest_version']
)
return result
def _compare_versions(self, current: str, latest: str) -> bool:
"""Compare semver versions, return True if update available."""
try:
def parse_version(v: str) -> tuple:
# Strip any leading 'v' and split by dots
v = v.lstrip('v').split('-')[0] # Remove pre-release suffix
parts = v.split('.')
return tuple(int(p) for p in parts[:3])
current_parts = parse_version(current)
latest_parts = parse_version(latest)
return latest_parts > current_parts
except Exception:
return False
def generate_channel_qr(self, channel_index: int) -> bytes | None:
"""
Generate QR code for a channel configuration.
Args:
channel_index: Channel index (0-7)
Returns:
PNG image bytes, or None on error
"""
try:
import qrcode
from io import BytesIO
except ImportError:
logger.error("qrcode library not installed. Install with: pip install qrcode[pil]")
return None
if not self._interface:
return None
try:
channels = self.get_channels()
channel = None
for ch in channels:
if ch.index == channel_index:
channel = ch
break
if not channel:
logger.error(f"Channel {channel_index} not found")
return None
# Build Meshtastic URL
# Format: https://meshtastic.org/e/#CgMSAQ... (base64 channel config)
# The URL encodes the channel settings protobuf
# For simplicity, we'll create a URL with the channel name and key info
# The official format requires protobuf serialization
channel_data = {
'name': channel.name,
'index': channel.index,
'psk': base64.b64encode(channel.psk).decode('utf-8') if channel.psk else '',
}
# Encode as base64 JSON (simplified format)
encoded = base64.urlsafe_b64encode(
json.dumps(channel_data).encode('utf-8')
).decode('utf-8')
url = f"https://meshtastic.org/e/#{encoded}"
# Generate QR code
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=10,
border=4,
)
qr.add_data(url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
# Convert to PNG bytes
buffer = BytesIO()
img.save(buffer, format='PNG')
return buffer.getvalue()
except Exception as e:
logger.error(f"Error generating QR code: {e}")
return None
def start_range_test(self, count: int = 10, interval: int = 5) -> tuple[bool, str]:
"""
Start a range test by sending test packets.
Args:
count: Number of test packets to send
interval: Seconds between packets
Returns:
Tuple of (success, error_message)
"""
if not self._interface:
return False, "Not connected to device"
if not HAS_MESHTASTIC:
return False, "Meshtastic SDK not installed"
if self._range_test_running:
return False, "Range test already running"
try:
from meshtastic import portnums_pb2
self._range_test_running = True
self._range_test_results = []
# Send range test packets in a background thread
import threading
def send_packets():
import time
for i in range(count):
if not self._range_test_running:
break
try:
# Send range test packet with sequence number
payload = f"RangeTest #{i+1}".encode('utf-8')
self._interface.sendData(
payload,
destinationId=BROADCAST_ADDR,
portNum=portnums_pb2.PortNum.RANGE_TEST_APP,
)
logger.info(f"Range test packet {i+1}/{count} sent")
except Exception as e:
logger.error(f"Error sending range test packet: {e}")
if i < count - 1 and self._range_test_running:
time.sleep(interval)
self._range_test_running = False
logger.info("Range test complete")
thread = threading.Thread(target=send_packets, daemon=True)
thread.start()
return True, None
except Exception as e:
self._range_test_running = False
logger.error(f"Error starting range test: {e}")
return False, str(e)
def stop_range_test(self) -> None:
"""Stop an ongoing range test."""
self._range_test_running = False
def get_range_test_status(self) -> dict:
"""Get range test status."""
return {
'running': self._range_test_running,
'results': self._range_test_results,
}
def request_store_forward(self, window_minutes: int = 60) -> tuple[bool, str]:
"""
Request missed messages from a Store & Forward router.
Args:
window_minutes: Minutes of history to request
Returns:
Tuple of (success, error_message)
"""
if not self._interface:
return False, "Not connected to device"
if not HAS_MESHTASTIC:
return False, "Meshtastic SDK not installed"
try:
from meshtastic import portnums_pb2, storeforward_pb2
# Find S&F router (look for nodes with router role)
router_num = None
if self._interface.nodes:
for node_id, node_data in self._interface.nodes.items():
# Check for router role
role = node_data.get('user', {}).get('role')
if role in ('ROUTER', 'ROUTER_CLIENT'):
if isinstance(node_id, str) and node_id.startswith('!'):
router_num = int(node_id[1:], 16)
elif isinstance(node_id, int):
router_num = node_id
break
if not router_num:
return False, "No Store & Forward router found on mesh"
# Build S&F history request
sf_request = storeforward_pb2.StoreAndForward()
sf_request.rr = storeforward_pb2.StoreAndForward.RequestResponse.CLIENT_HISTORY
sf_request.history.window = window_minutes * 60 # Convert to seconds
self._interface.sendData(
sf_request.SerializeToString(),
destinationId=router_num,
portNum=portnums_pb2.PortNum.STORE_FORWARD_APP,
)
logger.info(f"Requested S&F history from {self._format_node_id(router_num)} for {window_minutes} minutes")
return True, None
except ImportError:
return False, "Store & Forward protobuf not available"
except Exception as e:
logger.error(f"Error requesting S&F history: {e}")
return False, str(e)
def check_store_forward_available(self) -> dict:
"""
Check if a Store & Forward router is available.
Returns:
Dict with available status and router info
"""
result = {
'available': False,
'router_id': None,
'router_name': None,
}
if not self._interface or not self._interface.nodes:
return result
for node_id, node_data in self._interface.nodes.items():
role = node_data.get('user', {}).get('role')
if role in ('ROUTER', 'ROUTER_CLIENT'):
result['available'] = True
if isinstance(node_id, str):
result['router_id'] = node_id
else:
result['router_id'] = self._format_node_id(node_id)
result['router_name'] = node_data.get('user', {}).get('shortName')
break
return result
# Global client instance # Global client instance
_client: MeshtasticClient | None = None _client: MeshtasticClient | None = None