diff --git a/src/controllers/tanksController.js b/src/controllers/tanksController.js index c48818d6..9ea2c3bb 100644 --- a/src/controllers/tanksController.js +++ b/src/controllers/tanksController.js @@ -2343,7 +2343,8 @@ cron.schedule("* * * * *", async () => { // } // }; -exports. publishMotorStopStatus = async (motor_id, motor_stop_status) => { +exports.publishMotorStopStatus = async (motor_id, motor_stop_status) => { + const deviceTopic = `water/operation/${motor_id}`; // Target specific IoT const payload = { topic: 'operation', object: { @@ -2351,9 +2352,12 @@ exports. publishMotorStopStatus = async (motor_id, motor_stop_status) => { control: motor_stop_status } }; - console.log("enetred publish") - console.log(payload) - client.publish('water/operation', JSON.stringify(payload)); + + console.log(`📡 Publishing to ${deviceTopic}`); + console.log(payload); + + // Publish to the specific device's control topic + client.publish(deviceTopic, JSON.stringify(payload)); }; const stat_stop_intervals = {}; @@ -5881,73 +5885,132 @@ exports.getBlockData = async (req, reply) => { // } // } // }); + + + + + const mqtt = require('mqtt'); -// Connect to MQTT broker -const client = mqtt.connect('mqtt://35.207.198.4:1883'); +require('dotenv').config(); + +// **Persistent MQTT Connection** +const client = mqtt.connect('mqtt://35.207.198.4:1883', { + clientId: `mqtt_server_${Math.random().toString(16).substr(2, 8)}`, + clean: false, // Ensures MQTT retains subscriptions + reconnectPeriod: 2000, // Reconnect every 2 seconds +}); + +const subscribedTopics = new Set(); +const activeDevices = new Set(); // Keep track of active devices client.on('connect', () => { - console.log('Connected to MQTT broker'); - client.subscribe('water/iot-data/+', (err) => { + console.log('✅ Connected to MQTT broker'); + + // **Ensure re-subscriptions after reconnect** + subscribedTopics.forEach(topic => { + client.subscribe(topic, { qos: 1 }, (err) => { + if (err) { + console.error(`❌ Error resubscribing to ${topic}:`, err); + } else { + console.log(`🔄 Resubscribed to ${topic}`); + } + }); + }); + + // **Subscribe to new device announcements** + client.subscribe('water/iot-data/announce', { qos: 1 }, (err) => { if (err) { - console.error('Error subscribing to topic:', err); + console.error('❌ Error subscribing to announcement topic:', err); } else { - console.log('Subscribed to all IoT devices under water/iot-data/#'); + console.log('📡 Subscribed to water/iot-data/announce'); } }); }); client.on('message', async (topic, message) => { - console.log(`Message received on topic ${topic}:`, message.toString()); + console.log(`📩 Message received on topic ${topic}:`, message.toString()); try { - const topicParts = topic.split('/'); // ['water', 'iot-data', '140924'] - const hw_Id = topicParts[2]; // Extract hardwareId from topic + const data = JSON.parse(message.toString()); - if (!hw_Id) { - console.error('Invalid topic format: Missing hw_Id'); + // **Handle device announcements** + if (topic === 'water/iot-data/announce') { + if (!data.objects || !data.objects.hw_Id) { + console.error("❌ Invalid announcement format. Missing hw_Id."); + return; + } + + const hw_Id = data.objects.hw_Id; + const deviceTopic = `water/iot-data/${hw_Id}`; + + if (!subscribedTopics.has(deviceTopic)) { + client.subscribe(deviceTopic, { qos: 1 }, (err) => { + if (err) { + console.error(`❌ Error subscribing to ${deviceTopic}:`, err); + } else { + console.log(`✅ Subscribed to ${deviceTopic}`); + subscribedTopics.add(deviceTopic); + activeDevices.add(hw_Id); + console.log('📡 Active Devices:', Array.from(activeDevices)); + } + }); + } return; } - const data = JSON.parse(message.toString()); - const { Motor_status, tanks } = data.objects; + // **Process data messages asynchronously to avoid blocking** + if (topic.startsWith('water/iot-data/')) { + setImmediate(() => { + const hw_Id = topic.split('/')[2]; + processIotData(hw_Id, data); + }); + } + } catch (err) { + console.error('❌ Error processing message:', err.message); + } +}); - console.log(`Processing data for hardwareId: ${hw_Id}`); +client.on('error', (err) => console.error('❌ MQTT Error:', err)); +client.on('close', () => console.log('⚠️ MQTT Connection Closed.')); +client.on('offline', () => console.log('⚠️ MQTT Broker Offline.')); - // Get the current date and time in the required format - const currentTime = moment().tz('Asia/Kolkata'); - const date = currentTime.format('YYYY-MM-DD'); // Adjust format if needed - const time = currentTime.format('HH:mm:ss'); +async function processIotData(hw_Id, data) { + try { + const { Motor_status, tanks } = data.objects; + console.log(`📡 Processing data for hw_Id: ${hw_Id}`); + + const currentDate = new Date(); + const date = currentDate.toISOString(); + const time = currentDate.toLocaleTimeString('en-IN', { hour12: false, timeZone: 'Asia/Kolkata' }); - // Create tank documents const tankDocuments = tanks.map(tank => ({ tankhardwareId: tank.Id, tankHeight: tank.level, date, - time, + time })); - // Save IoT data const iotTankData = new IotData({ hardwareId: hw_Id, Motor_status, tanks: tankDocuments, date, - time, + time }); await iotTankData.save(); - // Delete excess records (keep only the latest three records) const recordsToKeep = 3; const recordsToDelete = await IotData.find({ hardwareId: hw_Id }) .sort({ date: -1, time: -1 }) .skip(recordsToKeep); - await Promise.all(recordsToDelete.map(record => record.remove())); + for (const record of recordsToDelete) { + await record.remove(); + } - // Process tanks and update water level for (const tank of tanks) { const { Id: tankhardwareId, level: tankHeight } = tank; const existingTank = await Tank.findOne({ hardwareId: hw_Id, tankhardwareId }); @@ -5956,68 +6019,53 @@ client.on('message', async (topic, message) => { const customerId = existingTank.customerId; const tank_name = existingTank.tankName; - // Convert height and calculate water level - const tankHeightInCm = parseInt(existingTank.height.replace(/,/g, ''), 10) * 30.48; - const tank_height = Math.round(tankHeightInCm); + const tankHeightInCm = (parseInt(existingTank.height.replace(/,/g, ''), 10)) * 30.48; + const tank_height = parseInt(tankHeightInCm.toFixed(0), 10); const waterLevelHeight = tank_height - tankHeight; const waterCapacityPerCm = parseInt(existingTank.waterCapacityPerCm.replace(/,/g, ''), 10); - const waterLevel = Math.max(0, waterLevelHeight * waterCapacityPerCm); - console.log(`${tankHeight} - Processed in MQTT subscriber`); + const waterLevel = parseInt(waterLevelHeight * waterCapacityPerCm, 10); - if (tankHeight > 0) { + if (tankHeight > 0 && waterLevel >= 0) { existingTank.waterlevel = waterLevel; await existingTank.save(); - - for (const outputConnection of existingTank.connections.outputConnections) { - const linkedTank = await Tank.findOne({ - customerId, - tankName: outputConnection.outputConnections, - tankLocation: outputConnection.output_type - }); - - if (linkedTank) { - for (const inputConnection of linkedTank.connections.inputConnections) { - if (inputConnection.inputConnections === tank_name) { - inputConnection.water_level = waterLevel; - await linkedTank.save(); - } - } - } - } } } - // Update motor status + const status = Motor_status; const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hw_Id }); + if (!motorTank) { - console.log(`Motor not found for motor_id: ${hw_Id}`); + console.log('Motor not found for the specified motor_id'); return; } const inputConnection = motorTank.connections.inputConnections.find(conn => conn.motor_id === hw_Id); if (inputConnection) { - inputConnection.motor_status = Motor_status; - - if (inputConnection.motor_stop_status === "1" && Motor_status === 2 && inputConnection.motor_on_type !== "forced_manual") { + inputConnection.motor_status = status; + if (inputConnection.motor_stop_status === "1" && status === 2 && inputConnection.motor_on_type !== "forced_manual") { + const currentTime = moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm'); inputConnection.motor_stop_status = "2"; inputConnection.motor_on_type = "forced_manual"; - inputConnection.startTime = currentTime.format('DD-MMM-YYYY - HH:mm'); + inputConnection.startTime = currentTime; } - if (inputConnection.motor_stop_status === "2" && Motor_status === 1) { + if (inputConnection.motor_stop_status === "2" && status === 1) { + const currentTime = moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm'); inputConnection.motor_stop_status = "1"; - inputConnection.stopTime = currentTime.format('DD-MMM-YYYY - HH:mm'); + inputConnection.stopTime = currentTime; } await motorTank.save(); } - console.log(`Processed successfully for hardwareId: ${hw_Id}`); + console.log('✅ Data processed successfully for hardwareId:', hw_Id); } catch (err) { - console.error('Error processing message:', err.message); + console.error('❌ Error processing IoT data:', err.message); } -}); +} + +