星纵电表数据接入

package.json

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

serverConfig.xml

@@ -36,4 +36,11 @@
         <!-- PID-TVOC 换算系数（ppb → μg/m³），入库时换算，默认 0.0013 -->
         <diqinTvocConvertFactor>0.0013</diqinTvocConvertFactor>
     </diqin>
+    <mqttsrv>
+        <mqttHost>49.235.185.26</mqttHost>
+        <mqttPort>1883</mqttPort>
+        <mqttTopic>/milesight/uplink/2026</mqttTopic>
+        <username>admin</username>
+        <password>admin123</password>
+    </mqttsrv>
 </config>

src/biz/dataIntegration.ts

@@ -727,8 +727,8 @@ async function integrateAcDeviceData() {
 
     // TODO: 考虑是否需要在插入前对数据按 device_time 排序，避免数据库存储顺序混乱
 
-    // 5. 补偿：当天无开关机记录的设备，拉取内机列表实时状态兜底写库（30 分钟节流）
-    await compensateAcRealtimeData(deviceModel, deviceDataModel, deviceIdSet, toInsert);
+    // 5. 补偿：最近 30 分钟无记录的设备，拉取内机列表实时状态兜底写库（30 分钟节流）
+    await compensateAcRealtimeData(deviceModel, deviceDataModel, toInsert);
 
     // 6. 批量插入
     if (toInsert.length > 0) {
@@ -744,14 +744,19 @@ const AC_REALTIME_COMPENSATE_INTERVAL_MS = 30 * 60 * 1000; // 30 分钟
 
 /**
  * 空调数据补偿机制：
- * 当天无开关机记录（或记录不全）的设备，拉取内机列表实时状态作为快照写入 device_data，
+ * 最近 30 分钟内没有开关机记录（或记录不全）的设备，拉取内机列表实时状态作为快照写入 device_data，
  * 避免"不调温不开关机时空调数据不更新"。
- * 只补 rows 未覆盖的设备；完全无记录时补所有空调设备。
+ *
+ * 状态判定以开关机记录（最近一条 power）为准，内机列表仅用于修正设备在线/离线/故障：
+ *  - 最近是开机 + 内机在线(0) → 记录开机快照
+ *  - 最近是开机 + 内机离线(1)/故障(2) → 记录关机（设备实际不可用）
+ *  - 最近是关机 → 记录关机
+ *  - 无历史记录（全新设备）→ 以内机列表 onOff 为准（onOff=1 开，0 关）
+ *  注意：内机列表 state=0 代表设备在线，不代表开机；onOff=0 也不代表用户主动关机。
  */
 async function compensateAcRealtimeData(
     deviceModel: any,
     deviceDataModel: any,
-    eventDeviceIds: Set<string>,
     toInsert: any[],
 ): Promise<void> {
     // 节流：距上次补偿不足 30 分钟则跳过
@@ -771,14 +776,20 @@ async function compensateAcRealtimeData(
         return;
     }
 
-    // 找出当天无开关机记录的设备（记录不全时只补这些）
-    const missingDeviceIds = allAcDeviceIds.filter((id: string) => !eventDeviceIds.has(id));
+    // 查询每台设备最新的 device_time，找出"最近 30 分钟没有记录"的设备（含全新无记录设备）
+    const maxTimeMap = await buildMaxTimeMap(deviceDataModel, allAcDeviceIds);
+    const thirtyMinAgo = new Date(now - AC_REALTIME_COMPENSATE_INTERVAL_MS);
+    const missingDeviceIds = allAcDeviceIds.filter((deviceId: string) => {
+        const maxTime = maxTimeMap.get(deviceId);
+        // 无历史记录，或最新记录已超过 30 分钟 → 需要补偿
+        return !maxTime || maxTime < thirtyMinAgo;
+    });
     if (missingDeviceIds.length === 0) {
-        // 全部设备今天都有开关机记录，无需补偿（仍更新节流时间，避免每次轮询都查设备表）
+        // 全部设备最近 30 分钟都有记录，无需补偿（仍更新节流时间，避免每次轮询都查设备表）
         lastAcRealtimeCompensateAt = now;
         return;
     }
-    console.log(`[空调数据补偿] ${missingDeviceIds.length} 台设备当天无开关机记录，拉取实时状态兜底...`);
+    console.log(`[空调数据补偿] ${missingDeviceIds.length} 台设备最近 30 分钟无记录，拉取实时状态兜底...`);
 
     // 分页拉取所有内机实时状态，建立 indoorUnitAddressFull → 内机映射
     const unitRows = await fetchAllPages((page, limit) => getIndoorUnitList({ page, limit }));
@@ -789,6 +800,22 @@ async function compensateAcRealtimeData(
         }
     }
 
+    // 查询缺失设备最近一条开关机记录的 power，用于判断最近是开机还是关机
+    const lastPowerMap = new Map<string, string>();
+    await Promise.all(
+        missingDeviceIds.map(async (deviceId) => {
+            const latest = await deviceDataModel.findOne({
+                where: { device_id: deviceId },
+                order: [['device_time', 'DESC']],
+                attributes: ['device_data'],
+                raw: true,
+            });
+            if (latest?.device_data?.power) {
+                lastPowerMap.set(deviceId, latest.device_data.power);
+            }
+        }),
+    );
+
     // 只对缺失设备构造实时快照（device_time 用当前轮询时间，source 标记区分）
     const compensateTime = new Date();
     let compensateCount = 0;
@@ -796,8 +823,22 @@ async function compensateAcRealtimeData(
         const unit = unitMap.get(deviceId);
         if (!unit) continue;
 
+        const lastPower = lastPowerMap.get(deviceId);
+        let power: string;
+        if (lastPower === 'on') {
+            // 最近是开机：内机在线则保持开机；离线/故障则记录关机
+            const unitState = unit.state ?? DEVICE_STATE.在线;
+            power = unitState === DEVICE_STATE.在线 ? 'on' : 'off';
+        } else if (lastPower === 'off') {
+            // 最近是关机：记录关机
+            power = 'off';
+        } else {
+            // 无历史记录（全新设备）：以内机列表 onOff 为准（1 开 0 关）
+            power = unit.onOff === 1 ? 'on' : 'off';
+        }
+
         const data = {
-            power: unit.onOff === 1 ? 'on' : 'off',
+            power,
             mode: WORK_MODE_TEXT_MAP[unit.workMode] || '未知',
             fanSpeed: FAN_SPEED_TEXT_MAP[unit.fanSpeed] || '自动',
             setTemp: unit.tempSet || 0,

src/biz/mqttClient.ts

+import { TABLENAME } from "../config/dbEnum";
+import { addData } from "../data/addData";
+import { selectOneDataByParam } from "../data/findData";
+
+/**
+ * 处理接收到的电表设备数据，并存入数据库
+ *
+ * 消息示例：
+ * { "CUB": 52.620000000000005, "EPEc": 0, "EPF": 0, "EPG": 0, "EPIc": 243.41, "EPJ": 0, "EPP": 0, "EPc": 243.41, "EQCc": 109.94, "EQLc": 0.01, "IL": 0, "IaTHD": 77.85000000000001, "IbTHD": 77.05, "IcTHD": 48.27, "MD": 14.722, "MDTimeStamp": "8-5 12:5", "RD": 10.763, "TempA": 0, "TempB": 0, "TempC": 0, "UaTHD": 3.47, "Uab": 0, "UbTHD": 3.11, "UcTHD": 3.09, "VUB": 0.53, "devEUI": "009569000005ad2b", "deviceName": "DB_ZDY", "gatewayTime": "2026-08-18T17:09:28+08:00" }
+ */
+export async function electricDeviceData(message) {
+    const msgStr = message.toString();
+
+    try {
+        // 解析JSON消息
+        let data: any;
+        try {
+            data = JSON.parse(msgStr);
+        } catch (e) {
+            console.error('消息不是合法JSON，跳过:', msgStr);
+            return;
+        }
+        console.log('收到电表设备数据:', data.length);
+
+        // 提取关键字段
+        const devEUI = data.devEUI ? data.devEUI.toUpperCase() : null;
+        if (!devEUI) {
+            console.warn('消息缺少devEUI字段，跳过');
+            return;
+        }
+        const gatewayTime = data.gatewayTime ? new Date(data.gatewayTime) : new Date();
+
+        // ===== 设备校验：设备不存在则自动注册到设备表 =====
+        const deviceRow = await selectOneDataByParam(
+            TABLENAME.设备表, { device_id: devEUI }
+        );
+        if (!deviceRow || !deviceRow.data) {
+            console.warn(`设备 ${devEUI} 不存在，自动注册到设备表...`);
+            try {
+                const now = new Date();
+                await addData(TABLENAME.设备表, {
+                    device_id: devEUI,
+                    device_name: data.deviceName || devEUI,
+                    device_type: '电表',
+                    device_state: 0,
+                    created_at: now,
+                    updated_at: now,
+                });
+                console.log(`设备 ${devEUI} 自动注册完成`);
+            } catch (e) {
+                // 并发上报时可能刚被注册（唯一键冲突），忽略并继续入库数据
+                console.warn(`设备 ${devEUI} 自动注册失败（可能已存在）:`, e);
+            }
+        }
+        const now = new Date();
+        await addData(
+            TABLENAME.设备数据表,
+            { device_id: devEUI, device_data: data, device_time: gatewayTime, created_at: now }
+        );
+
+        console.log(`数据入库成功: 设备 ${devEUI}`);
+
+    } catch (error) {
+        console.error('数据处理失败:', error);
+    }
+}

src/biz/running.ts

@@ -717,7 +717,9 @@ export async function getRunAnalysis() {
     ====== MOCK DATA END ====== */
 
     // ---- 真实查询 ----
-    // 趋势基于 device_data 表：每小时有数据上报的设备数代表在线数
+    // 设备在线率趋势：每个小时点 T 的在线率 = (设备总数 − 离线数) / 设备总数 × 100
+    // 离线判定：设备在 [趋势起点 ~ T] 内最近一条 device_data 记录的 power=off（含补偿快照），
+    //           状态保持直到出现更新的记录；无 power 字段（电表/IEQ）或无记录 → 默认在线
     const trend24hStart = getHoursAgo(24);
     const trendEnd = formatTime(new Date());
 
@@ -727,10 +729,11 @@ export async function getRunAnalysis() {
         ['device_id']
     );
     const allDeviceIds = (envDevicesResult.data || []).map((d: any) => d.device_id);
+    const totalDevices = allDeviceIds.length;
 
-    let trendHourly: { key: string; value: number }[] = [];
+    let deviceMonitorTrend: { key: string; value: number }[] = [];
 
-    if (allDeviceIds.length > 0) {
+    if (totalDevices > 0) {
         const trendDataResult = await selectDataListByParam(
             TABLENAME.设备数据表,
             {
@@ -738,24 +741,46 @@ export async function getRunAnalysis() {
                 device_time: { '%gte%': trend24hStart, '%lte%': trendEnd },
                 '%orderAsc%': 'device_time',
             },
-            ['device_id', 'device_time']
+            ['device_id', 'device_time', 'device_data']
         );
 
-        const hourlyDevices = new Map<string, Set<string>>();
+        // 生成完整小时 key 序列（起点向上取整到整点，到当前时刻），保证无数据小时也输出
+        const startHour = new Date(trend24hStart);
+        startHour.setMinutes(0, 0, 0);
+        if (startHour.getTime() < new Date(trend24hStart).getTime()) {
+            startHour.setHours(startHour.getHours() + 1);
+        }
+        const hourKeys: string[] = [];
+        for (let t = new Date(startHour); t <= new Date(trendEnd); t.setHours(t.getHours() + 1)) {
+            hourKeys.push(formatHourKey(t));
+        }
+
+        // 按小时分组记录
+        const recordsByHour = new Map<string, any[]>();
         (trendDataResult.data || []).forEach((r: any) => {
-            const t = new Date(r.device_time);
-            const key = formatHourKey(t);
-            if (!hourlyDevices.has(key)) hourlyDevices.set(key, new Set());
-            hourlyDevices.get(key)!.add(r.device_id);
+            const hk = formatHourKey(new Date(r.device_time));
+            if (!recordsByHour.has(hk)) recordsByHour.set(hk, []);
+            recordsByHour.get(hk)!.push(r);
         });
 
-        trendHourly = Array.from(hourlyDevices.entries())
-            .map(([key, devices]) => ({ key, value: devices.size }))
-            .sort((a, b) => a.key.localeCompare(b.key));
+        // 状态保持模型：逐小时推进，维护各设备最近一条记录的 power，统计离线数
+        const lastPowerMap = new Map<string, string>();
+        deviceMonitorTrend = hourKeys.map((hk) => {
+            for (const r of recordsByHour.get(hk) || []) {
+                const parsed = parseDeviceData(r.device_data);
+                if (parsed && typeof parsed.power === 'string') {
+                    lastPowerMap.set(r.device_id, parsed.power);
+                }
+            }
+            let offlineCount = 0;
+            lastPowerMap.forEach((p) => {
+                if (p === 'off') offlineCount++;
+            });
+            const onlineRate = Math.round(((totalDevices - offlineCount) / totalDevices) * 100);
+            return { key: hk, value: onlineRate };
+        });
     }
 
-    let deviceMonitorTrend = trendHourly;
-
     // ========================================================================
     // 四、预警工单处理
     // 1、预警处理状态
@@ -1311,6 +1336,7 @@ export async function getAnalysisPopup(regionKey: string) {
  * @param selectedRegionKey 已选择区域
  */
 export async function getAnalysisPopupSubTrend(currRegionKey: string, selectedRegionKey: string) {
+    /* ====== MOCK DATA（保留，勿删） ======
     // 能耗监测（该区域电表24小时趋势）
     let nhjcdb: { key: string; dqqy: string, dbqy: string }[] = [];
     // 环境监测（该区域电表24小时趋势）
@@ -1326,6 +1352,124 @@ export async function getAnalysisPopupSubTrend(currRegionKey: string, selectedRe
         "nhjc": nhjcdb,
         "hjjc": hjjcdb
     };
+    ====== MOCK DATA END ====== */
+
+    // ---- 真实查询 ----
+    // 时间范围：过去 24 小时 ~ 当前时刻（只输出到当前小时，不查未来）
+    const trendStart = getHoursAgo(24);
+    const trendEnd = formatTime(new Date());
+
+    // 两区域分别聚合：能耗（电表按小时求和）、环境（IEQ 按小时取最新一条）
+    const [currMeterHourly, selMeterHourly] = await Promise.all([
+        queryRegionMeterHourly(currRegionKey, trendStart, trendEnd),
+        queryRegionMeterHourly(selectedRegionKey, trendStart, trendEnd),
+    ]);
+    const [currIeqHourly, selIeqHourly] = await Promise.all([
+        queryRegionIeqHourly(currRegionKey, trendStart, trendEnd),
+        queryRegionIeqHourly(selectedRegionKey, trendStart, trendEnd),
+    ]);
+
+    // 能耗对比：hourKey 取两区域并集，缺数据侧输出 null（不补零）
+    const nhjcHourKeys = Array.from(new Set([...currMeterHourly.keys(), ...selMeterHourly.keys()])).sort();
+    let nhjcdb: { key: string; dqqy: number | null, dbqy: number | null }[] = nhjcHourKeys.map((key) => ({
+        key,
+        dqqy: currMeterHourly.has(key) ? Math.round(currMeterHourly.get(key)!) : null,
+        dbqy: selMeterHourly.has(key) ? Math.round(selMeterHourly.get(key)!) : null,
+    }));
+
+    // 环境对比：六项指标分别取两区域并集小时，缺数据侧输出 null（不补零）
+    const ENV_COMPARE_MAP: Record<string, string> = {
+        wdhjjc: 'temperature',
+        sdhjjc: 'humidity',
+        pm25hjjc: 'pm25',
+        co2hjjc: 'co2',
+        pm10hjjc: 'pm10',
+        tvochjjc: 'tvoc',
+    };
+    let hjjcdb: any = {};
+    Object.entries(ENV_COMPARE_MAP).forEach(([outKey, indicator]) => {
+        const hourKeys = Array.from(new Set([...currIeqHourly.keys(), ...selIeqHourly.keys()])).sort();
+        hjjcdb[outKey] = hourKeys.map((key) => ({
+            key,
+            dqqy: currIeqHourly.get(key)?.get(indicator) ?? null,
+            dbqy: selIeqHourly.get(key)?.get(indicator) ?? null,
+        }));
+    });
+
+    return {
+        "nhjc": nhjcdb,
+        "hjjc": hjjcdb
+    };
+}
+
+/**
+ * 查询某区域电表设备在指定时间范围内的耗电量，按小时聚合（求和）
+ * @returns Map<hourKey, 该小时耗电量总和>
+ */
+async function queryRegionMeterHourly(regionKey: string, startTime: string, endTime: string): Promise<Map<string, number>> {
+    const devices = await getDevicesByRegion(regionKey, ['device_id', 'device_type']);
+    const meterIds = devices.filter((d: any) => d.device_type === '电表').map((d: any) => d.device_id);
+    if (meterIds.length === 0) return new Map();
+
+    const dataResult = await selectDataListByParam(
+        TABLENAME.设备数据表,
+        {
+            device_id: { '%in%': meterIds },
+            device_time: { '%gte%': startTime, '%lte%': endTime },
+            '%orderAsc%': 'device_time',
+        },
+        ['device_data', 'device_time']
+    );
+
+    const hourlyMap = new Map<string, number>();
+    (dataResult.data || []).forEach((r: any) => {
+        const t = new Date(r.device_time);
+        const hourKey = `${t.getFullYear()}-${String(t.getMonth() + 1).padStart(2, '0')}-${String(t.getDate()).padStart(2, '0')} ${String(t.getHours()).padStart(2, '0')}:00`;
+        const parsed = parseDeviceData(r.device_data);
+        const val = parsed && parsed[ELECTRICITY_KEY];
+        if (typeof val === 'number') {
+            hourlyMap.set(hourKey, (hourlyMap.get(hourKey) || 0) + val);
+        }
+    });
+    return hourlyMap;
+}
+
+/**
+ * 查询某区域 IEQ 传感器在指定时间范围内的环境数据，按小时聚合（每小时取最新一条）
+ * @returns Map<hourKey, Map<indicatorKey, 最新值>>
+ */
+async function queryRegionIeqHourly(regionKey: string, startTime: string, endTime: string): Promise<Map<string, Map<string, number>>> {
+    const devices = await getDevicesByRegion(regionKey, ['device_id', 'device_type']);
+    const ieqIds = devices.filter((d: any) => d.device_type === 'IEQ传感器').map((d: any) => d.device_id);
+    if (ieqIds.length === 0) return new Map();
+
+    const dataResult = await selectDataListByParam(
+        TABLENAME.设备数据表,
+        {
+            device_id: { '%in%': ieqIds },
+            device_time: { '%gte%': startTime, '%lte%': endTime },
+            '%orderAsc%': 'device_time',
+        },
+        ['device_data', 'device_time']
+    );
+
+    const hourlyMap = new Map<string, Map<string, number>>();
+    (dataResult.data || []).forEach((r: any) => {
+        const t = new Date(r.device_time);
+        const hourKey = `${t.getFullYear()}-${String(t.getMonth() + 1).padStart(2, '0')}-${String(t.getDate()).padStart(2, '0')} ${String(t.getHours()).padStart(2, '0')}:00`;
+        const parsed = parseDeviceData(r.device_data);
+        if (!parsed) return;
+        if (!hourlyMap.has(hourKey)) hourlyMap.set(hourKey, new Map());
+        const hourData = hourlyMap.get(hourKey)!;
+        ENV_INDICATOR_KEYS.forEach((key) => {
+            const val = parsed[key];
+            if (val !== undefined && val !== null && typeof val === 'number') {
+                // device_time 升序遍历，后写覆盖 → 保留该小时最新一条
+                hourData.set(key, val);
+            }
+        });
+    });
+    return hourlyMap;
 }
 
 /**

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, feiyi, diqin } = configInfo.config;
+        let { port, sign, img, mysqldb, feiyi, diqin, mqttsrv } = configInfo.config;
         
         // 基本配置
         systemConfig.port = parseInt(port[0]);
@@ -56,6 +56,20 @@ export async function initConfig() {
             if (diqinConfig.diqinTvocConvertFactor) systemConfig.diqin.tvocConvertFactor = parseFloat(diqinConfig.diqinTvocConvertFactor[0]);
         }
 
+        // mqtt
+        if (mqttsrv) {
+            let configInfo = mqttsrv[0];
+            systemConfig.mqttsrv = { host: '', port: 0, topic_env: '', 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_env = 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

@@ -29,4 +29,12 @@ export class ServerConfig {
         /** PID-TVOC 换算系数（ppb → μg/m³），入库时换算，默认 0.0013 */
         tvocConvertFactor: number;
     }
+    /** 星纵MQTT服务配置 */
+    mqttsrv: {
+        host:string,
+        port:number,
+        topic_env: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, startAlertSchedule, startFaultStatusSchedule } from "./config/schedule";
@@ -15,10 +16,12 @@ async function lanuch() {
     httpServer.createServer(systemConfig.port);
     /**启动定时任务 */
     startSchedule();
+    /**启动MQTT订阅服务 */
+    await mqttSrv.startMqttClient();
     /**启动预警工单定时任务（10分钟） */
-    // startAlertSchedule();
+    startAlertSchedule();
     /**启动故障状态更新定时任务（5分钟） */
-    // startFaultStatusSchedule();
+    startFaultStatusSchedule();
 
     console.log('This indicates that the server is started successfully.');
 

src/routers/admin.ts

@@ -43,10 +43,14 @@ export function setRouter(httpServer) {
     httpServer.post("/api/dq/admin/device/edit", asyncHandler(editDevice));
     /** 设备管理 - 批量编辑设备 */
     httpServer.post("/api/dq/admin/device/edit_batch", asyncHandler(batchEditDevices));
+    /** 设备管理 - 新增设备信息 */
+    httpServer.post("/api/dq/admin/device/addDevice", asyncHandler(addLocalDevice));
     /** 设备管理 - 编辑设备信息 */
     httpServer.post("/api/dq/admin/device/editDevice", asyncHandler(editLocalDevice));
     /** 设备管理 - 批量编辑设备信息 */
     httpServer.post("/api/dq/admin/device/editDevice_batch", asyncHandler(batchEditLocalDevices));
+    /** 设备管理 - 删除设备信息 */
+    httpServer.post("/api/dq/admin/device/deleteDevice", asyncHandler(deleteLocalDevice));
 
     /** 用户管理 - 列表 */
     httpServer.get("/api/dq/admin/users/list", asyncHandler(getUsersList));
@@ -362,6 +366,31 @@ async function batchEditDevices(req, res) {
 }
 
 /**
+ * POST /api/dq/admin/device/addDevice
+ * 设备管理 - 新增单台设备表信息
+ * body: { deviceId, deviceName?, regionKey?, deviceType?, controlParams?, deviceState? }
+ */
+async function addLocalDevice(req, res) {
+    let reqConf = {
+        deviceId: 'String',
+        deviceName: 'String',
+        regionKey: 'Number',
+        deviceType: 'String',
+        deviceState: 'String',
+        linkageStart: 'String',
+    };
+    const NotMustHaveKeys = ["deviceName", "regionKey", "deviceType", "deviceState", "linkageStart"];
+    let { deviceId, deviceName, regionKey, deviceType, deviceState, linkageStart } =
+        eccReqParamater(reqConf, req.body, NotMustHaveKeys);
+    if (!deviceId) {
+        res.success({success: false, msg: "缺少必填参数：deviceId"});
+        return;
+    }
+    const data = await deviceBiz.addLocalDevice({ deviceId, deviceName, regionKey, deviceType, deviceState, linkageStart });
+    res.success(data);
+}
+
+/**
  * POST /api/dq/admin/device/editDevice
  * 设备管理 - 编辑单台设备表信息
  * body: { deviceId, deviceName?, regionKey?, deviceType?, controlParams?, deviceState? }
@@ -423,6 +452,21 @@ async function batchEditLocalDevices(req, res) {
     res.success(data);
 }
 
+/**
+ * POST /api/dq/admin/device/deleteDevice
+ * 设备管理 - 删除设备表信息
+ * body: { deviceId }
+ */
+async function deleteLocalDevice(req, res) {
+    const deviceId: string = req.body?.deviceId as string;
+    if (!deviceId) {
+        res.success({success: false, msg: "缺少参数 deviceId"});
+        return;
+    }
+    const data = await deviceBiz.deleteLocalDevices(deviceId);
+    res.success(data);
+}
+
 // ======================= 用户管理 =======================
 
 /**

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_env, (err) => {
+            if (err) {
+                console.error('总电表订阅失败:', err);
+            } else {
+                console.log(`成功订阅总电表主题: ${systemConfig.mqttsrv.topic_env}`);
+                console.log('等待总电表数据...');
+            }
+        });
+    });
+
+    // 当收到消息时触发处理
+    client.on('message', (topic, message) => {
+        // 只处理配置的上行主题
+        if (topic === systemConfig.mqttsrv.topic_env) {
+            processDeviceData(message);
+        }
+    });
+
+    client.on('error', (err) => {
+        console.error('MQTT连接错误:', err);
+    });
+}
\ No newline at end of file