changes

src/config/swagger.js

@@ -14,7 +14,7 @@ exports.options = {
     },
     // We have to change this beacause this is running on local 
     // we have to run on this on swagger
-     //host: "localhost:3000",  //Devlopemnt host on lcoal
+    //host: "localhost:3000",  //Devlopemnt host on lcoal
     
      host: "35.207.198.4:3000", //Production for swagger
     schemes: ["http"],

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 });
@@ -2858,7 +2870,7 @@ exports.IotDeviceforstandalonedevice = async (req, reply) => {
 };
 
 
-
+    
 
 
 
@@ -3907,3 +3919,143 @@ exports.getBlockData = async (req, reply) => {
     reply.code(500).send({ error: err.message });
   }
 };
+
+
+
+
+
+const mqtt = require('mqtt');
+const client = mqtt.connect('mqtt://35.207.198.4:1883'); // Connect to MQTT broker
+
+client.on('connect', () => {
+  console.log('Connected to MQTT broker');
+  client.subscribe('water/iot-data', (err) => {
+    if (err) {
+      console.error('Error subscribing to topic:', err);
+    } else {
+      console.log('Subscribed to water/iot-data topic');
+    }
+  });
+});
+
+// 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 { 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();
+      const date = currentDate.toISOString(); // ISO string for date
+      const time = currentDate.toLocaleTimeString('en-IN', { hour12: false, timeZone: 'Asia/Kolkata' }); // Time in 'HH:MM:SS'
+
+      // Create array of tank documents with current date and time
+      const tankDocuments = tanks.map(tank => ({
+        tankhardwareId: tank.Id, // Updated to match the new format
+        tankHeight: tank.level,   // Updated to match the new format
+        date,
+        time
+      }));
+
+      // Save IoT data for the received tanks
+      const iotTankData = new IotData({
+        hardwareId: hw_Id,       // Updated variable name
+        Motor_status,
+        tanks: tankDocuments,
+        date,
+        time
+      });
+      await iotTankData.save();
+
+      // Delete excess records (keep only the latest three records)
+      const recordsToKeep = 3;
+      const recordsToDelete = await IotData.find({ hardwareId: hw_Id }) // Updated variable name
+        .sort({ date: -1, time: -1 })
+        .skip(recordsToKeep);
+
+      for (const record of recordsToDelete) {
+        await record.remove();
+      }
+
+      // Process each tank to update water level and connections
+      for (const tank of tanks) {
+        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: hw_Id, tankhardwareId }); // Updated variable name
+        if (!existingTank) continue;
+
+        const customerId = existingTank.customerId;
+        const tank_name = existingTank.tankName;
+
+        // Calculate water level using tank height and capacity
+        const tankHeightInCm = (parseInt(existingTank.height.replace(/,/g, ''), 10)) * 30.48; // Convert height to cm
+        const tank_height = parseInt(tankHeightInCm.toFixed(0), 10);
+        const waterLevelHeight = tank_height - tankHeight;
+        const waterCapacityPerCm = parseInt(existingTank.waterCapacityPerCm.replace(/,/g, ''), 10);
+
+        const waterLevel = parseInt(waterLevelHeight * waterCapacityPerCm, 10); // Calculated water level
+
+        // Update water level in the existing tank
+        if (waterLevel >= 0) {
+          existingTank.waterlevel = waterLevel;
+          await existingTank.save();
+
+          // Update linked tanks (input/output connections)
+          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; // Update water level for linked tank
+                  await linkedTank.save(); // Save updated linked tank
+                }
+              }
+            }
+          }
+        }
+      }
+
+      // Update motor status
+      const status = Motor_status;
+      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');
+        return;
+      }
+
+      // Find the inputConnection for the motor and update motor status
+      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:', hw_Id); // Updated variable name
+
+    } catch (err) {
+      console.error('Error processing message:', err.message);
+    }
+  }
+});
+
+
+// Function to publish motor stop status
+// exports.publishMotorStopStatus = async (motor_id, motor_stop_status) => {
+//   const payload = {
+//     topic: 'operation',
+//     object: {
+//       'motor-id': motor_id,
+//       control: motor_stop_status
+//     }
+//   };
+
+//   client.publish('water/operation', JSON.stringify(payload));
+// };
+
+
+
+