mqtt服务

package.json

@@ -19,6 +19,7 @@
     "md5": "^2.2.1",
     "moment": "^2.24.0",
     "mongoose": "^7.6.0",
+    "mqtt": "^5.15.1",
     "mysql": "^2.18.1",
     "mysql2": "^3.6.0",
     "node-xlsx": "^0.16.1",

serverConfig.xml

@@ -16,4 +16,11 @@
         <mysqlPwd>qaz123456</mysqlPwd>
         <dataBase>beermuseumplatform</dataBase> -->
     </mysqldb>
+    <mqttsrv>
+        <mqttHost>10.100.48.16</mqttHost>
+        <mqttPort>1883</mqttPort>
+        <mqttTopic>/milesight/uplink/2026</mqttTopic>
+        <username>admin</username>
+        <password>admin123</password>
+    </mqttsrv>
 </config>

src/biz/mqttClient.ts

+import { TABLENAME } from "../config/dbEnum";
+import { addData } from "../data/addData";
+import { selectOneDataByParam } from "../data/findData";
+
+/**
+ * 处理接收到的单条设备数据，并存入数据库
+ *
+ * 消息示例：
+ * {
+ *   "applicationID": 1,
+ *   "cellularIP": "0.0.0.0",
+ *   "co2": 1649,
+ *   "devEUI": "24e124710e466786",
+ *   "deviceName": "cgq3",
+ *   "gatewayTime": "2026-06-17T09:53:55+08:00",
+ *   "hcho": 0.01,
+ *   "humidity": 66.5,
+ *   "light_level": 0,
+ *   "pir": 1,
+ *   "pm10": 32,
+ *   "pm2_5": 32,
+ *   "pressure": 1009.7,
+ *   "temperature": 22.5,
+ *   "tvoc": 1.56
+ * }
+ */
+export async function processDeviceData(message) {
+    try {
+        // 解析JSON消息
+        const msgStr = message.toString();
+        let data: any;
+        try {
+            data = JSON.parse(msgStr);
+        } catch (e) {
+            console.error('消息不是合法JSON，跳过:', msgStr);
+            return;
+        }
+        console.log('收到数据:', data);
+
+        // 提取关键字段
+        const devEUI = data.devEUI;
+        if (!devEUI) {
+            console.warn('消息缺少devEUI字段，跳过');
+            return;
+        }
+        const gatewayTime = data.gatewayTime ? new Date(data.gatewayTime) : new Date();
+        
+        // 解码data字段（Base64 -> 十六进制）
+        // const rawHex = decodePayload(data.data);
+        // if (!rawHex) {
+        //     console.warn('data字段为空或无效');
+        //     return;
+        // }
+
+        // ===== 设备校验（保留） =====
+        const deviceRow = await selectOneDataByParam(
+            TABLENAME.设备表, { device_id: devEUI}
+        );
+        if (!deviceRow) {
+            console.warn(`设备不存在，跳过入库: ${devEUI}`);
+            return;
+        }
+
+        // ===== 新设备入库（注释掉，设备由定时任务从设备商接口同步） =====
+        // let now = formatLocalTime(new Date());
+        // if (!deviceRow) {
+        //     // 设备不存在，插入新设备（区域等后续可手动关联）
+        //     await addData(
+        //         TABLENAME.设备表, {
+        //             device_id: devEUI,
+        //             region_key: 0,
+        //             device_type: '环境监测',
+        //             device_name: `Device_${devEUI.slice(-6)}`,
+        //             control_params: message.toString(),
+        //             created_at: now,
+        //             updated_at: now,
+        //         }
+        //     );
+        //     console.log(`新设备已创建: ${devEUI}`);
+        // }
+
+        // ===== 插入设备数据（优化） =====
+        const now = new Date();
+        await addData(
+            TABLENAME.设备数据表,
+            {device_id: devEUI, data: data, received_time: now, device_time: gatewayTime, created_at: now}
+        );
+        
+        console.log(`数据入库成功: 设备 ${devEUI}`);
+
+    } catch (error) {
+        console.error('数据处理失败:', error);
+    }
+}
+
+// ==================== 数据解析与入库函数 ====================
+/**
+ * 将Base64编码的payload解码为十六进制字符串
+ * 如果已知具体传感器协议，可以在这里进一步解析为具体数值
+ */
+function decodePayload(base64Data) {
+    if (!base64Data) return null;
+    // 方法1: 转为十六进制字符串（便于查看和存储）
+    const buffer = Buffer.from(base64Data, 'base64');
+    return buffer.toString('hex').toUpperCase();
+    
+    // 方法2: 如果知道协议格式，可以直接解析数值，例如：
+    // return { temperature: buffer.readInt16BE(0) / 10.0, humidity: buffer.readUInt8(2) };
+}
\ No newline at end of file

src/biz/running.ts

@@ -301,11 +301,16 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
             regionCustomerMap.set(regionName, totalNum);
         }
     }
+    // 当日客流
+    let todayCustomer = 0;
+    
+    let yesterdayCustomer = 0;
     // 计算每个区域的饱和度（客流量/区域最大承载量），并转换成表格
     for (let [regionName, customerNum] of regionCustomerMap.entries()) {
         let regionMaxCustomer = regionMaxCustomerMap.get(regionName) || 10000; // 默认最大承载量为10000
         let saturation = `${((customerNum / regionMaxCustomer) * 100).toFixed(2)}%`;
         regionCustomerTable.push({ regionName, customerNum, saturation });
+        todayCustomer += customerNum;
     }
     
     // 6.3 客流量趋势分析：根据查询到的客流传感器数据，统计所选区域过去24小时客流量变化趋势
@@ -342,9 +347,23 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
             let val = hourlyCustomerMap.get(key) ?? 0;
             customerTrend.push({ time: `${i}:00`, value: val });
         }
+
+        // 获取昨日客流量数据
+        let yesterdayDatas = await selectDataListByParam(TABLENAME.设备数据表, {
+            device_id: { "%in%": customerDeviceIds },
+            device_time: { "%gte%": formatLocalTime(yesterdayStart), "%lte%": formatLocalTime(yesterdayEnd) },
+            "%orderAsc%": "device_time"
+        }, ["device_id", "data", "device_time"]);
+        for (let rec of yesterdayDatas.data) {
+            let data = rec.data;
+            let customerNum = data.presence === '进' ? 1 : 0;
+            yesterdayCustomer += customerNum;
+        }
     }
     // 6.4. 人流监控实时数据和趋势图数据
     resultAny.customerOnlineTrend = {
+        todayCustomer: todayCustomer,
+        customerCompare: (todayCustomer - yesterdayCustomer) / yesterdayCustomer * 100, // 同比昨日(今日人流量-昨日人流量)÷昨日人流量x100%
         passengerFlowTable: regionCustomerTable,
         customerTrend: customerTrend
     };

src/config/serverConfig.ts

@@ -16,7 +16,7 @@ export async function initConfig() {
         
         if (!configInfo || !configInfo.config) throw new BizError('xml中无配置');
         
-        let { port, sign, img, mysqldb } = configInfo.config;
+        let { port, sign, img, mysqldb, mqttsrv } = configInfo.config;
         
         // 基本配置
         systemConfig.port = parseInt(port[0]);
@@ -37,6 +37,20 @@ export async function initConfig() {
             }
         }
 
+        // mqtt
+        if (mqttsrv) {
+            let configInfo = mqttsrv[0];
+            systemConfig.mqttsrv = { host: '', port: 0, topic: '', username: '', password: '' };
+            
+            if (configInfo.mqttHost && configInfo.mqttPort && configInfo.mqttTopic) {
+                systemConfig.mqttsrv.host = configInfo.mqttHost[0];
+                systemConfig.mqttsrv.port = parseInt(configInfo.mqttPort[0]);
+                systemConfig.mqttsrv.topic = configInfo.mqttTopic[0];
+                systemConfig.mqttsrv.username = configInfo.username[0];
+                systemConfig.mqttsrv.password = configInfo.password[0];
+            }
+        }
+
     } catch(err) {
         console.log('ERROR => 服务器配置解析错误 请检查根目录下 serverConfig.xml 文件是否正确');
         console.log(err);

src/config/systemClass.ts

@@ -15,5 +15,11 @@ export class ServerConfig {
         pwd:string,
         dataBase:string,
     }
-    
+    mqttsrv: {
+        host:string,
+        port:number,
+        topic:string,
+        username:string,
+        password:string
+    }
 }
\ No newline at end of file

src/main.ts

 import { initConfig, systemConfig} from "./config/serverConfig";
 import * as mysqlDB from "./db/mysqlInit";
+import * as mqttSrv from "./service/mqttInit";
 import { initMysqlModel } from "./model/sqlModelBind";
 import { httpServer } from "./net/http_server";
 import { startSchedule } from "./config/schedule";
@@ -13,6 +14,8 @@ async function lanuch() {
     await initMysqlModel();
     /**创建http服务 */
     httpServer.createServer(systemConfig.port);
+    /**启动MQTT订阅服务 */
+    await mqttSrv.startMqttClient();
     /**启动定时任务 */
     startSchedule();
 

src/service/mqttInit.ts

+import { processDeviceData } from "../biz/mqttClient";
+import { systemConfig } from "../config/serverConfig";
+
+const mqtt = require('mqtt');
+
+// ==================== 启动MQTT客户端并订阅 ====================
+export async function startMqttClient() {
+    // 在连接选项中添加用户名和密码
+    const mqttOptions = {
+        username: systemConfig.mqttsrv.username,
+        password: systemConfig.mqttsrv.password,
+        // 可选：如果遇到连接问题，可以添加以下参数
+        // clientId: 'node_subscriber_' + Math.random().toString(16).substring(2, 8),
+        // clean: true,
+        // connectTimeout: 4000,
+        // reconnectPeriod: 1000,
+    };
+    // 连接MQTT Broker
+    const client = mqtt.connect(`mqtt://${systemConfig.mqttsrv.host}:${systemConfig.mqttsrv.port}`, mqttOptions);
+    
+    client.on('connect', () => {
+        console.log('MQTT客户端已连接，正在订阅主题...');
+        // 订阅网关的上行主题
+        client.subscribe(systemConfig.mqttsrv.topic, (err) => {
+            if (err) {
+                console.error('订阅失败:', err);
+            } else {
+                console.log(`成功订阅主题: ${systemConfig.mqttsrv.topic}`);
+                console.log('等待设备数据...');
+            }
+        });
+    });
+
+    // 当收到消息时触发处理
+    client.on('message', (topic, message) => {
+        // 只处理配置的上行主题
+        if (topic === systemConfig.mqttsrv.topic) {
+            processDeviceData(message);
+        }
+    });
+
+    client.on('error', (err) => {
+        console.error('MQTT连接错误:', err);
+    });
+}
\ No newline at end of file