mqtt changes

src/controllers/tanksController.js

@@ -1369,7 +1369,18 @@ const sendNotification = async (fcmTokens, title, body) => {
 //   }
 // };
 
-
+exports.publishMotorStopStatus = async (motor_id, motor_stop_status) => {
+  const payload = {
+    topic: 'operation',
+    object: {
+      'motor-id': motor_id,
+      control: motor_stop_status
+    }
+  };
+  console.log("enetred publish")
+  console.log(payload)
+  client.publish('water/operation', JSON.stringify(payload));
+};
 
 const stat_stop_intervals = {};
 exports.motorAction = async (req, reply) => {
@@ -1413,7 +1424,7 @@ exports.motorAction = async (req, reply) => {
           }
         }
       );
-
+      this.publishMotorStopStatus(motorId, motorStopStatus);
       // Send immediate response to the client
       reply.code(200).send({ message: "Motor stopped successfully." });
 
@@ -1480,7 +1491,7 @@ exports.motorAction = async (req, reply) => {
           receiverInitialwaterlevel:parseInt(receiver_tank_info7.waterlevel, 10)
         });
         await newMotorData.save();
-
+        this.publishMotorStopStatus(motorId, motorStopStatus);
         for await (const tank of Tank.find({  "connections.inputConnections.motor_id": motorId })) {
           const index = tank.connections.inputConnections.findIndex(connection => connection.motor_id === motorId);
           if (index !== -1) {
@@ -1517,8 +1528,9 @@ exports.motorAction = async (req, reply) => {
                 }
               }
             );
+            
             clearInterval(intervalId);
-
+            this.publishMotorStopStatus(motorId, "1");
             await delay(300000);
 
             const motorData = await MotorData.findOne({ customerId, motor_id: motorId, start_instance_id: start_instance_id });
@@ -3929,11 +3941,11 @@ client.on('connect', () => {
 // Handling incoming MQTT messages
 client.on('message', async (topic, message) => {
   console.log(`Message received on topic ${topic}:`, message.toString());
-  
+
   if (topic === 'water/iot-data') {
     try {
       const data = JSON.parse(message.toString());
-      const { hardwareId, Motor_status, tanks } = data.objects;
+      const { hw_Id, Motor_status, tanks } = data.objects; // Updated variable names according to new format
 
       // Get the current date and time in the required format
       const currentDate = new Date();
@@ -3942,15 +3954,15 @@ client.on('message', async (topic, message) => {
 
       // Create array of tank documents with current date and time
       const tankDocuments = tanks.map(tank => ({
-        tankhardwareId: tank.tankhardwareId,
+        tankhardwareId: tank.Id, // Updated to match the new format
-        tankHeight: tank.tankHeight,
+        tankHeight: tank.level,   // Updated to match the new format
         date,
         time
       }));
 
       // Save IoT data for the received tanks
       const iotTankData = new IotData({
-        hardwareId,
+        hardwareId: hw_Id,       // Updated variable name
         Motor_status,
         tanks: tankDocuments,
         date,
@@ -3960,7 +3972,7 @@ client.on('message', async (topic, message) => {
 
       // Delete excess records (keep only the latest three records)
       const recordsToKeep = 3;
-      const recordsToDelete = await IotData.find({ hardwareId })
+      const recordsToDelete = await IotData.find({ hardwareId: hw_Id }) // Updated variable name
         .sort({ date: -1, time: -1 })
         .skip(recordsToKeep);
 
@@ -3970,9 +3982,9 @@ client.on('message', async (topic, message) => {
 
       // Process each tank to update water level and connections
       for (const tank of tanks) {
-        const { tankhardwareId, tankHeight } = tank;
+        const { Id: tankhardwareId, level: tankHeight } = tank; // Updated to match the new format
         // Find the corresponding tank in the Tank schema using hardwareId and tankhardwareId
-        const existingTank = await Tank.findOne({ hardwareId, tankhardwareId });
+        const existingTank = await Tank.findOne({ hardwareId: hw_Id, tankhardwareId }); // Updated variable name
         if (!existingTank) continue;
 
         const customerId = existingTank.customerId;
@@ -4008,7 +4020,7 @@ client.on('message', async (topic, message) => {
 
       // Update motor status
       const status = Motor_status;
-      const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hardwareId });
+      const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hw_Id }); // Updated variable name
 
       if (!motorTank) {
         console.log('Motor not found for the specified motor_id');
@@ -4016,13 +4028,13 @@ client.on('message', async (topic, message) => {
       }
 
       // Find the inputConnection for the motor and update motor status
-      const inputConnection = motorTank.connections.inputConnections.find(conn => conn.motor_id === hardwareId);
+      const inputConnection = motorTank.connections.inputConnections.find(conn => conn.motor_id === hw_Id); // Updated variable name
       if (inputConnection) {
         inputConnection.motor_status = status; // Update motor status
         await motorTank.save(); // Save the updated tank
       }
 
-      console.log('Data processed successfully for hardwareId:', hardwareId);
+      console.log('Data processed successfully for hardwareId:', hw_Id); // Updated variable name
 
     } catch (err) {
       console.error('Error processing message:', err.message);
@@ -4030,18 +4042,19 @@ client.on('message', async (topic, message) => {
   }
 });
 
+
 // Function to publish motor stop status
-exports.publishMotorStopStatus = async (motor_id, motor_stop_status) => {
+// exports.publishMotorStopStatus = async (motor_id, motor_stop_status) => {
-  const payload = {
+//   const payload = {
-    topic: 'operation',
+//     topic: 'operation',
-    object: {
+//     object: {
-      'motor-id': motor_id,
+//       'motor-id': motor_id,
-      control: motor_stop_status
+//       control: motor_stop_status
-    }
+//     }
-  };
+//   };
 
-  client.publish('water/operation', JSON.stringify(payload));
+//   client.publish('water/operation', JSON.stringify(payload));
-};
+// };