diff --git a/src/controllers/storeController.js b/src/controllers/storeController.js index 7d089e54..ad76f70b 100644 --- a/src/controllers/storeController.js +++ b/src/controllers/storeController.js @@ -2185,3 +2185,35 @@ exports.getOrdersByStoreId = async (req, reply) => { } }; + +exports.getallocatedsensorstouser= async (req, reply) => { + try { + const { customerId } = req.params; + + if (!customerId) { + return reply.status(400).send({ error: "customerId is required" }); + } + + // Fetch orders with the matching storeId + const allocated_iots = await Insensors.find({ customerId }); + + if (!allocated_iots.length) { + return reply.send({ + status_code: 200, + message: "No sensors found for this store", + data: [], + }); + } + + + + return reply.send({ + status_code: 200, + message: "iots fetched successfully", + data: allocated_iots, + }); + } catch (err) { + console.error("Error fetching iots:", err); + return reply.status(500).send({ error: "Internal server error" }); + } +}; \ No newline at end of file diff --git a/src/controllers/tanksController.js b/src/controllers/tanksController.js index 44f28967..91af2d0f 100644 --- a/src/controllers/tanksController.js +++ b/src/controllers/tanksController.js @@ -5850,148 +5850,174 @@ exports.getBlockData = async (req, reply) => { // } // }); - - const mqtt = require('mqtt'); +const client = mqtt.connect('mqtt://35.207.198.4:1883'); // Connect to MQTT broker - -const brokerUrl = 'mqtt://35.207.198.4:1883'; -const deviceLastSeen = new Map(); // Track last seen timestamps for devices - -// Connect a single global MQTT client -const mqttClient = mqtt.connect(brokerUrl, { - clientId: 'global_subscriber', - clean: false, // Retain session state - reconnectPeriod: 5000, // Reconnect every 5 seconds - keepalive: 60, // Send a ping every 60 seconds -}); - -// Event: When the MQTT client connects -mqttClient.on('connect', () => { - console.log('๐ŸŒŽ Global MQTT client connected'); - - // Subscribe to all IoT data topics - mqttClient.subscribe('water/iot-data/+', { qos: 1 }, (err) => { +client.on('connect', () => { + console.log('Connected to MQTT broker'); + client.subscribe('water/iot-data', (err) => { if (err) { - console.error('โŒ Error subscribing to wildcard topic:', err); + console.error('Error subscribing to topic:', err); } else { - console.log('๐Ÿ“ก Subscribed to wildcard topic: water/iot-data/+'); + console.log('Subscribed to water/iot-data topic'); } }); }); -// Event: When a message is received -mqttClient.on('message', async (topic, message) => { - try { - console.log(`๐Ÿ“ฉ Message received on topic ${topic}:`, message.toString()); - - const data = JSON.parse(message.toString()); - const { hw_Id, Motor_status, tanks } = data.objects; - - console.log(`๐Ÿ“… Device ${hw_Id} last seen at: ${moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm')}`); - - // Update the last seen timestamp for the device - deviceLastSeen.set(hw_Id, new Date().toISOString()); - - // Save IoT data to the database - const iotTankData = new IotData({ - hardwareId: hw_Id, - Motor_status, - tanks: tanks.map((tank) => ({ - tankhardwareId: tank.Id, - tankHeight: tank.level, - date: moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm'), - time: moment().tz('Asia/Kolkata').format('HH:mm'), - })), - date: moment().tz('Asia/Kolkata').format('DD-MMM-YYYY - HH:mm'), - time: moment().tz('Asia/Kolkata').format('HH:mm'), - }); - await iotTankData.save(); - console.log(`โœ… Data saved for device: ${hw_Id}`); - // Keep only the latest 3 records for the device - 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())); - console.log(`๐Ÿงน Deleted older records for device: ${hw_Id}`); - } +client.on('message', async (topic, message) => { + console.log(`Message received on topic ${topic}:`, message.toString()); - // Process tanks and update status - await Promise.all( - tanks.map(async (tank) => { - const existingTank = await Tank.findOne({ hardwareId: hw_Id, tankhardwareId: tank.Id }); - if (!existingTank) { - console.log(`โš ๏ธ Tank not found for device: ${hw_Id}, tank: ${tank.Id}`); - return; - } + 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); - const tankHeightInCm = parseInt(existingTank.height.replace(/,/g, ''), 10) * 30.48; - const waterLevelHeight = tankHeightInCm - tank.level; + 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); - existingTank.waterlevel = waterLevel; - await existingTank.save(); - console.log(`โœ… Tank data updated for ${hw_Id}, tank: ${tank.Id}`); - }) - ); + const waterLevel = parseInt(waterLevelHeight * waterCapacityPerCm, 10); // Calculated water level + + // Update water level in the existing tank + console.log(tankHeight,"this is located in tank controllers at iot-data mqtt sub ") + if (tankHeight>0 && 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 - // Update motor status - const motorTank = await Tank.findOne({ "connections.inputConnections.motor_id": hw_Id }); - if (motorTank) { - const inputConnection = motorTank.connections.inputConnections.find((conn) => conn.motor_id === hw_Id); + 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 = Motor_status; - if (inputConnection.motor_stop_status === "1" && Motor_status === 2) { + inputConnection.motor_status = status; // Update motor 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 = moment().tz('Asia/Kolkata').format('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 = moment().tz('Asia/Kolkata').format('HH:mm'); + inputConnection.motor_on_type = "manual"; + inputConnection.stopTime = currentTime; } - await motorTank.save(); - console.log(`โœ… Motor status updated for device: ${hw_Id}`); + + 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); } - } catch (err) { - console.error('โŒ Error processing message:', err.message); } }); -// Event: When the MQTT client encounters an error -mqttClient.on('error', (err) => { - console.error('โŒ MQTT Client Error:', err); -}); -// Event: When the MQTT client disconnects -mqttClient.on('disconnect', () => { - console.log('โš ๏ธ Global MQTT client disconnected'); -}); +// +// API function to get survey data for a particular installer +// +exports.getPendingAndCompletedsurveyOfparticularInstaller = async (request, reply) => { + try { + const { installationId } = request.params; + const { survey_status } = request.body; + + const surveyData = await User.find({ installationId, survey_status }); + + reply.send({ + status_code: 200, + surveyData, + }); + } catch (err) { + console.error('โŒ Error fetching survey data:', err); + throw boom.boomify(err); + } +}; + + + + + + + + + + + -// Periodically check for offline devices -setInterval(() => { - const now = new Date(); - deviceLastSeen.forEach((lastSeen, hw_Id) => { - const lastSeenDate = new Date(lastSeen); - const diffInSeconds = (now - lastSeenDate) / 1000; - if (diffInSeconds > 60) { - console.log(`๐Ÿšจ Device ${hw_Id} is offline. Last seen: ${lastSeen}`); - } - }); -}, 60000); // Check every 60 seconds - -// Handle application shutdown gracefully -process.on('SIGINT', () => { - console.log('๐Ÿ›‘ Application is shutting down...'); - mqttClient.end(() => { - console.log('๐Ÿšช MQTT client disconnected'); - process.exit(0); - }); -}); exports.getPendingAndCompletedsurveyOfparticularInstaller = async (request, reply) => { diff --git a/src/routes/storeRoute.js b/src/routes/storeRoute.js index f034f0f6..8329d829 100644 --- a/src/routes/storeRoute.js +++ b/src/routes/storeRoute.js @@ -1709,5 +1709,28 @@ fastify.get("/api/ordersofstore/:storeId", { }); + + +fastify.get("/api/getallocatedsensors/:customerId", { + schema: { + tags: ["Install"], + description: "Fetches sensors based on storeId", + summary: "Get sensors by customerId", + params: { + type: "object", + properties: { + customerId: { type: "string" }, + }, + required: ["customerId"], + }, + security: [ + { + basicAuth: [], + }, + ], + }, + handler: storeController.getallocatedsensorstouser, +}); + next(); };