Arminta
/
watermanagement-backend-prod
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

changes

Browse Source
master^2
Varun 7 months ago
parent 5d54fce8f7
commit a3480c5d67
1 changed files with 84 additions and 145 deletions
Show Stats Download Patch File Download Diff File

229
src/controllers/tanksController.js
Unescape View File

@ -5855,146 +5855,109 @@ exports.getBlockData = async (req, reply) => {
const mqtt = require('mqtt'); const mqtt = require('mqtt');
const brokerUrl = 'mqtt://35.207.198.4:1883'; const brokerUrl = 'mqtt://35.207.198.4:1883';
const clients = new Map(); // Store client instances dynamically per hw_Id const deviceLastSeen = new Map(); // Track last seen timestamps for devices
const deviceLastSeen = new Map(); // Track last seen timestamps for devices
function createClient(hw_Id) { // Connect a single global MQTT client
if (clients.has(hw_Id)) return clients.get(hw_Id); // Return existing client const mqttClient = mqtt.connect(brokerUrl, {
clientId: 'global_subscriber',
const client = mqtt.connect(brokerUrl, { clean: true,
clientId: `client_${hw_Id}`, reconnectPeriod: 5000, // Reconnect every 5 seconds
clean: false, // Ensures session persistence });
reconnectPeriod: 5000, // Attempt reconnect every 5 seconds
});
client.on('connect', () => { mqttClient.on('connect', () => {
console.log(`✅ Client for ${hw_Id} connected to MQTT broker`); console.log('🌎 Global MQTT client connected');
const topic = `water/iot-data/${hw_Id}`; // Subscribe to all IoT data topics
client.subscribe(topic, { qos: 1 }, (err) => { mqttClient.subscribe('water/iot-data/+', { qos: 1 }, (err) => {
if (err) { if (err) {
console.error(`❌ Error subscribing to topic ${topic}:`, err); console.error('❌ Error subscribing to wildcard topic:', err);
} else { } else {
console.log(`📡 Subscribed to topic: ${topic}`); console.log('📡 Subscribed to wildcard topic: water/iot-data/+');
} }
});
}); });
});
client.on('message', async (topic, message) => { mqttClient.on('message', async (topic, message) => {
try {
console.log(`📩 Message received on topic ${topic}:`, message.toString()); console.log(`📩 Message received on topic ${topic}:`, message.toString());
try { const data = JSON.parse(message.toString());
const data = JSON.parse(message.toString()); const { hw_Id, Motor_status, tanks } = data.objects;
const { hw_Id, Motor_status, tanks } = data.objects;
const currentTime = moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm');
const currentTime = moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm'); deviceLastSeen.set(hw_Id, new Date().toISOString());
deviceLastSeen.set(hw_Id, new Date().toISOString());
// Save IoT data
// Save IoT data const iotTankData = new IotData({
const iotTankData = new IotData({ hardwareId: hw_Id,
hardwareId: hw_Id, Motor_status,
Motor_status, tanks: tanks.map((tank) => ({
tanks: tanks.map((tank) => ({ tankhardwareId: tank.Id,
tankhardwareId: tank.Id, tankHeight: tank.level,
tankHeight: tank.level,
date: currentTime,
time: moment().tz('Asia/Kolkata').format('HH:mm'),
})),
date: currentTime, date: currentTime,
time: moment().tz('Asia/Kolkata').format('HH:mm'), time: moment().tz('Asia/Kolkata').format('HH:mm'),
}); })),
date: currentTime,
await iotTankData.save(); time: moment().tz('Asia/Kolkata').format('HH:mm'),
console.log(`✅ Data saved for device: ${hw_Id}`); });
// Keep only the latest 3 records
const records = await IotData.find({ hardwareId: hw_Id }).sort({ date: -1 });
if (records.length > 3) {
const recordsToDelete = records.slice(3); // Get older records
await Promise.all(recordsToDelete.map((record) => record.remove()));
}
// Process tanks and update status await iotTankData.save();
await Promise.all( console.log(`✅ Data saved for device: ${hw_Id}`);
tanks.map(async (tank) => {
const existingTank = await Tank.findOne({ hardwareId: hw_Id, tankhardwareId: tank.Id });
if (!existingTank) return;
const tankHeightInCm = parseInt(existingTank.height.replace(/,/g, ''), 10) * 30.48;
const waterLevelHeight = tankHeightInCm - tank.level;
const waterCapacityPerCm = parseInt(existingTank.waterCapacityPerCm.replace(/,/g, ''), 10);
const waterLevel = parseInt(waterLevelHeight * waterCapacityPerCm, 10);
existingTank.waterlevel = waterLevel;
await existingTank.save();
console.log(`✅ Tank data saved for ${hw_Id}, tank: ${tank.Id}`);
})
);
// Update motor status // Keep only the latest 3 records
const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hw_Id }); const records = await IotData.find({ hardwareId: hw_Id }).sort({ date: -1 });
if (motorTank) { if (records.length > 3) {
const inputConnection = motorTank.connections.inputConnections.find((conn) => conn.motor_id === hw_Id); const recordsToDelete = records.slice(3); // Get older records
if (inputConnection) { await Promise.all(recordsToDelete.map((record) => record.remove()));
inputConnection.motor_status = Motor_status;
if (inputConnection.motor_stop_status === "1" && Motor_status === 2) {
inputConnection.motor_stop_status = "2";
inputConnection.motor_on_type = "forced_manual";
inputConnection.startTime = moment().tz('Asia/Kolkata').format('HH:mm');
}
if (inputConnection.motor_stop_status === "2" && Motor_status === 1) {
inputConnection.motor_stop_status = "1";
inputConnection.stopTime = moment().tz('Asia/Kolkata').format('HH:mm');
}
await motorTank.save();
console.log(`✅ Motor status updated for device: ${hw_Id}`);
}
}
} catch (err) {
console.error(`❌ Error processing message from ${hw_Id}:`, err.message);
} }
});
client.on('error', (err) => { // Process tanks and update status
console.error(`❌ MQTT Error for ${hw_Id}:`, err); await Promise.all(
}); tanks.map(async (tank) => {
const existingTank = await Tank.findOne({ hardwareId: hw_Id, tankhardwareId: tank.Id });
if (!existingTank) return;
client.on('disconnect', () => { const tankHeightInCm = parseInt(existingTank.height.replace(/,/g, ''), 10) * 30.48;
console.log(`⚠️ Client for ${hw_Id} disconnected`); const waterLevelHeight = tankHeightInCm - tank.level;
}); const waterCapacityPerCm = parseInt(existingTank.waterCapacityPerCm.replace(/,/g, ''), 10);
const waterLevel = parseInt(waterLevelHeight * waterCapacityPerCm, 10);
clients.set(hw_Id, client); existingTank.waterlevel = waterLevel;
return client; await existingTank.save();
} console.log(`✅ Tank data updated for ${hw_Id}, tank: ${tank.Id}`);
})
);
// Subscribe to wildcard topic to detect new devices // Update motor status
const globalClient = mqtt.connect(brokerUrl, { const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hw_Id });
clientId: 'global_subscriber', if (motorTank) {
clean: false, 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_stop_status = "2";
inputConnection.motor_on_type = "forced_manual";
inputConnection.startTime = moment().tz('Asia/Kolkata').format('HH:mm');
}
if (inputConnection.motor_stop_status === "2" && Motor_status === 1) {
inputConnection.motor_stop_status = "1";
inputConnection.stopTime = moment().tz('Asia/Kolkata').format('HH:mm');
}
await motorTank.save();
console.log(`✅ Motor status updated for device: ${hw_Id}`);
}
}
} catch (err) {
console.error('❌ Error processing message:', err.message);
}
}); });
globalClient.on('connect', () => { mqttClient.on('error', (err) => {
console.log('🌎 Global client connected to MQTT broker'); console.error('❌ MQTT Client Error:', err);
globalClient.subscribe('water/iot-data/+', { qos: 1 }, (err) => {
if (err) {
console.error('❌ Error subscribing to wildcard topic:', err);
} else {
console.log('📡 Subscribed to wildcard topic: water/iot-data/+');
}
});
}); });
globalClient.on('message', (topic, message) => { mqttClient.on('disconnect', () => {
try { console.log('⚠️ Global MQTT client disconnected');
const data = JSON.parse(message.toString());
const { hw_Id } = data.objects;
if (!clients.has(hw_Id)) {
console.log(`➕ Creating new MQTT client for device: ${hw_Id}`);
createClient(hw_Id);
}
} catch (err) {
console.error('❌ Error parsing global message:', err.message);
}
}); });
// Periodically check for offline devices // Periodically check for offline devices
@ -6012,30 +5975,6 @@ setInterval(() => {
exports.getPendingAndCompletedsurveyOfparticularInstaller = async (request, reply) => {
try {
const { installationId } = request.params;
const survey_status = request.body;
const surveydata = await User.find({
installationId,
survey_status,
});
// Send the response, including both total consumption and filtered consumption records
reply.send({
status_code: 200,
surveydata,
});
} catch (err) {
throw boom.boomify(err);
}
};

Write Preview
Loading…
Cancel
Save