initial commit
This commit is contained in:
Will Bradley
@@ -0,0 +1,401 @@
|
||||
const mqtt = require('mqtt');
|
||||
const config = require('../config/config');
|
||||
const logger = require('../utils/logger');
|
||||
const { nodeQueries, positionQueries, messageQueries, telemetryQueries } = require('../database/queries');
|
||||
|
||||
class MeshtasticMQTTClient {
|
||||
constructor() {
|
||||
this.client = null;
|
||||
this.connected = false;
|
||||
this.messageCallbacks = [];
|
||||
}
|
||||
|
||||
connect() {
|
||||
const { broker, username, password, topic, options } = config.mqtt;
|
||||
|
||||
logger.info(`Connecting to MQTT broker: ${broker}`);
|
||||
|
||||
// Create MQTT client
|
||||
this.client = mqtt.connect(broker, {
|
||||
...options,
|
||||
username,
|
||||
password
|
||||
});
|
||||
|
||||
// Connection event handlers
|
||||
this.client.on('connect', () => {
|
||||
this.connected = true;
|
||||
logger.info('Connected to MQTT broker');
|
||||
|
||||
// Subscribe to Meshtastic topics
|
||||
this.client.subscribe(topic, (err) => {
|
||||
if (err) {
|
||||
logger.error('Failed to subscribe to topic:', err);
|
||||
} else {
|
||||
logger.info(`Subscribed to topic: ${topic}`);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
this.client.on('error', (error) => {
|
||||
logger.error('MQTT connection error:', error);
|
||||
this.connected = false;
|
||||
});
|
||||
|
||||
this.client.on('close', () => {
|
||||
this.connected = false;
|
||||
logger.warn('MQTT connection closed');
|
||||
});
|
||||
|
||||
this.client.on('reconnect', () => {
|
||||
logger.info('Reconnecting to MQTT broker...');
|
||||
});
|
||||
|
||||
// Message handler
|
||||
this.client.on('message', (topic, message) => {
|
||||
this.handleMessage(topic, message);
|
||||
});
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
handleMessage(topic, message) {
|
||||
try {
|
||||
// Parse the topic to extract information
|
||||
const topicParts = topic.split('/');
|
||||
|
||||
// Log raw message for debugging (changed to info for testing)
|
||||
logger.info(`Received MQTT message on topic: ${topic} (${message.length} bytes)`);
|
||||
|
||||
// Try to parse as JSON first (some messages might be JSON)
|
||||
let payload;
|
||||
try {
|
||||
payload = JSON.parse(message.toString());
|
||||
this.processJsonMessage(topic, payload);
|
||||
} catch (e) {
|
||||
// If not JSON, treat as protobuf or binary data
|
||||
this.processRawMessage(topic, message);
|
||||
}
|
||||
|
||||
// Notify subscribers
|
||||
this.notifyCallbacks(topic, message);
|
||||
} catch (error) {
|
||||
logger.error('Error handling MQTT message:', error);
|
||||
}
|
||||
}
|
||||
|
||||
processJsonMessage(topic, payload) {
|
||||
try {
|
||||
const { from, to, channel, id, sender } = payload;
|
||||
|
||||
// Extract node ID (convert to hex string if it's a number)
|
||||
const fromNode = sender || (from ? `!${from.toString(16).padStart(8, '0')}` : null);
|
||||
const toNode = to ? `!${to.toString(16).padStart(8, '0')}` : null;
|
||||
|
||||
// Handle different message types based on actual JSON structure
|
||||
if (payload.type === 'sendtext' && payload.payload) {
|
||||
this.handleTextMessage(fromNode, toNode, {
|
||||
...payload,
|
||||
text: typeof payload.payload === 'string' ? payload.payload : payload.payload.text,
|
||||
id: payload.id,
|
||||
channel: payload.channel,
|
||||
rxSnr: payload.snr,
|
||||
rxRssi: payload.rssi
|
||||
});
|
||||
}
|
||||
|
||||
if (payload.type === 'position' && payload.payload) {
|
||||
const pos = payload.payload;
|
||||
this.handlePositionUpdate(fromNode, {
|
||||
latitude: pos.latitude_i / 10000000,
|
||||
longitude: pos.longitude_i / 10000000,
|
||||
altitude: pos.altitude,
|
||||
precisionBits: pos.precision_bits
|
||||
});
|
||||
}
|
||||
|
||||
if (payload.type === 'nodeinfo' && payload.payload) {
|
||||
this.handleNodeInfo(fromNode, {
|
||||
shortName: payload.payload.shortname,
|
||||
longName: payload.payload.longname,
|
||||
hardwareModel: payload.payload.hardware,
|
||||
role: payload.payload.role
|
||||
});
|
||||
}
|
||||
|
||||
if (payload.type === 'telemetry' && payload.payload) {
|
||||
this.handleTelemetry(fromNode, {
|
||||
deviceMetrics: {
|
||||
batteryLevel: payload.payload.battery_level,
|
||||
voltage: payload.payload.voltage,
|
||||
channelUtilization: payload.payload.channel_utilization,
|
||||
airUtilTx: payload.payload.air_util_tx,
|
||||
uptimeSeconds: payload.payload.uptime_seconds
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
if (payload.type === 'neighborinfo' && payload.payload) {
|
||||
// Just update last_seen for the node
|
||||
nodeQueries.upsertNode.run(
|
||||
fromNode,
|
||||
null, null, null, null, null,
|
||||
new Date().toISOString(),
|
||||
null, null, null, null, null
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Error processing JSON message:', error);
|
||||
}
|
||||
}
|
||||
|
||||
processRawMessage(topic, message) {
|
||||
// For protobuf messages, we'll need to decode them
|
||||
// This is a simplified version - real implementation would use protobuf definitions
|
||||
try {
|
||||
// Extract information from topic
|
||||
const topicParts = topic.split('/');
|
||||
|
||||
// Basic message logging for debugging
|
||||
logger.debug(`Raw message length: ${message.length} bytes`);
|
||||
} catch (error) {
|
||||
logger.error('Error processing raw message:', error);
|
||||
}
|
||||
}
|
||||
|
||||
handleTextMessage(fromNode, toNode, data) {
|
||||
try {
|
||||
if (!fromNode) return;
|
||||
|
||||
logger.info(`Text message from ${fromNode}: ${data.text || '(empty)'}`);
|
||||
|
||||
// Ensure node exists first (upsert)
|
||||
nodeQueries.upsertNode.run(
|
||||
fromNode,
|
||||
null, null, null, null, null,
|
||||
new Date().toISOString(),
|
||||
null, null, null, null, null
|
||||
);
|
||||
|
||||
// Insert message - ensure all values are proper types
|
||||
const messageId = data.id ? String(data.id) : null;
|
||||
const channel = typeof data.channel === 'number' ? data.channel : 0;
|
||||
const text = data.text || '';
|
||||
const rxSnr = typeof data.rxSnr === 'number' ? data.rxSnr : null;
|
||||
const rxRssi = typeof data.rxRssi === 'number' ? data.rxRssi : null;
|
||||
const hopLimit = typeof data.hopLimit === 'number' ? data.hopLimit : null;
|
||||
const wantAck = data.wantAck === true ? 1 : 0;
|
||||
|
||||
messageQueries.insertMessage.run(
|
||||
messageId,
|
||||
fromNode,
|
||||
toNode,
|
||||
channel,
|
||||
text,
|
||||
new Date().toISOString(),
|
||||
rxSnr,
|
||||
rxRssi,
|
||||
hopLimit,
|
||||
wantAck
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('Error handling text message:', error);
|
||||
}
|
||||
}
|
||||
|
||||
handlePositionUpdate(nodeId, data) {
|
||||
try {
|
||||
if (!nodeId || !data.latitude || !data.longitude) return;
|
||||
|
||||
logger.info(`Position update from ${nodeId}: ${data.latitude}, ${data.longitude}`);
|
||||
|
||||
// Update node
|
||||
nodeQueries.upsertNode.run(
|
||||
nodeId,
|
||||
null, // short_name
|
||||
null, // long_name
|
||||
null, // hardware_model
|
||||
null, // role
|
||||
null, // firmware_version
|
||||
new Date().toISOString(),
|
||||
null, // battery_level
|
||||
null, // voltage
|
||||
null, // channel_utilization
|
||||
null, // air_util_tx
|
||||
null // uptime_seconds
|
||||
);
|
||||
|
||||
// Insert position
|
||||
positionQueries.insertPosition.run(
|
||||
nodeId,
|
||||
data.latitude,
|
||||
data.longitude,
|
||||
data.altitude || null,
|
||||
data.precisionBits || null
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('Error handling position update:', error);
|
||||
}
|
||||
}
|
||||
|
||||
handleNodeInfo(nodeId, data) {
|
||||
try {
|
||||
if (!nodeId) return;
|
||||
|
||||
logger.info(`Node info update for ${nodeId}`);
|
||||
|
||||
// Update node information
|
||||
nodeQueries.upsertNode.run(
|
||||
nodeId,
|
||||
data.shortName || data.user?.shortName || null,
|
||||
data.longName || data.user?.longName || null,
|
||||
data.hardwareModel || data.user?.hwModel || null,
|
||||
data.role || null,
|
||||
data.firmwareVersion || null,
|
||||
new Date().toISOString(),
|
||||
null, // battery_level
|
||||
null, // voltage
|
||||
null, // channel_utilization
|
||||
null, // air_util_tx
|
||||
null // uptime_seconds
|
||||
);
|
||||
} catch (error) {
|
||||
logger.error('Error handling node info:', error);
|
||||
}
|
||||
}
|
||||
|
||||
handleTelemetry(nodeId, data) {
|
||||
try {
|
||||
if (!nodeId) return;
|
||||
|
||||
logger.info(`Telemetry update from ${nodeId}`);
|
||||
|
||||
// Update node with telemetry data
|
||||
if (data.deviceMetrics) {
|
||||
const metrics = data.deviceMetrics;
|
||||
|
||||
nodeQueries.upsertNode.run(
|
||||
nodeId,
|
||||
null, // short_name
|
||||
null, // long_name
|
||||
null, // hardware_model
|
||||
null, // role
|
||||
null, // firmware_version
|
||||
new Date().toISOString(),
|
||||
metrics.batteryLevel || null,
|
||||
metrics.voltage || null,
|
||||
metrics.channelUtilization || null,
|
||||
metrics.airUtilTx || null,
|
||||
metrics.uptimeSeconds || null
|
||||
);
|
||||
|
||||
// Insert telemetry record
|
||||
telemetryQueries.insertTelemetry.run(
|
||||
nodeId,
|
||||
metrics.batteryLevel || null,
|
||||
metrics.voltage || null,
|
||||
metrics.channelUtilization || null,
|
||||
metrics.airUtilTx || null,
|
||||
metrics.uptimeSeconds || null,
|
||||
null, // temperature
|
||||
null, // relative_humidity
|
||||
null // barometric_pressure
|
||||
);
|
||||
}
|
||||
|
||||
if (data.environmentMetrics) {
|
||||
const env = data.environmentMetrics;
|
||||
|
||||
telemetryQueries.insertTelemetry.run(
|
||||
nodeId,
|
||||
null, // battery_level
|
||||
null, // voltage
|
||||
null, // channel_utilization
|
||||
null, // air_util_tx
|
||||
null, // uptime_seconds
|
||||
env.temperature || null,
|
||||
env.relativeHumidity || null,
|
||||
env.barometricPressure || null
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error('Error handling telemetry:', error);
|
||||
}
|
||||
}
|
||||
|
||||
// Publish a message to MQTT
|
||||
publish(topic, message) {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (!this.connected) {
|
||||
reject(new Error('MQTT client not connected'));
|
||||
return;
|
||||
}
|
||||
|
||||
this.client.publish(topic, message, (error) => {
|
||||
if (error) {
|
||||
logger.error('Error publishing message:', error);
|
||||
reject(error);
|
||||
} else {
|
||||
logger.info(`Published message to topic: ${topic}`);
|
||||
resolve();
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
// Send a text message
|
||||
async sendTextMessage(text, channel = 0) {
|
||||
try {
|
||||
const message = JSON.stringify({
|
||||
type: 'text',
|
||||
text,
|
||||
channel,
|
||||
timestamp: Date.now()
|
||||
});
|
||||
|
||||
// Publish to the appropriate topic
|
||||
const baseTopic = config.mqtt.topic.replace('/#', '');
|
||||
await this.publish(`${baseTopic}/2/json/mqtt`, message);
|
||||
|
||||
return true;
|
||||
} catch (error) {
|
||||
logger.error('Error sending text message:', error);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Register a callback for messages
|
||||
onMessage(callback) {
|
||||
this.messageCallbacks.push(callback);
|
||||
}
|
||||
|
||||
// Notify all callbacks
|
||||
notifyCallbacks(topic, message) {
|
||||
this.messageCallbacks.forEach(callback => {
|
||||
try {
|
||||
callback(topic, message);
|
||||
} catch (error) {
|
||||
logger.error('Error in message callback:', error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Get connection status
|
||||
isConnected() {
|
||||
return this.connected;
|
||||
}
|
||||
|
||||
// Disconnect
|
||||
disconnect() {
|
||||
if (this.client) {
|
||||
this.client.end();
|
||||
this.connected = false;
|
||||
logger.info('Disconnected from MQTT broker');
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Create singleton instance
|
||||
const mqttClient = new MeshtasticMQTTClient();
|
||||
|
||||
module.exports = mqttClient;
|
||||
Reference in New Issue
Block a user