Skip to content

Q
qingdaoMuseumPlatform
Commit 37602267 by PC-20251223ZVQQ\Administrator
Options

客流数据接入

parent 8874fba1
Showing with 424 additions and 16 deletions
serverConfig.xml
......@@ -19,7 +19,8 @@
<mqttsrv>
<mqttHost>10.100.48.16</mqttHost>
<mqttPort>1883</mqttPort>
<mqttTopic>/milesight/uplink/2026</mqttTopic>
<mqttTopic1>/milesight/uplink/2026</mqttTopic1>
<mqttTopic2>/milesight/uplink/2026_crowd</mqttTopic2>
<username>admin</username>
<password>admin123</password>
</mqttsrv>
......
src/biz/device.ts
import { selectOneDataByParam } from "../data/findData";
import { selectOneDataByParam, selectDataListByParam } from "../data/findData";
import { TABLENAME } from "../config/dbEnum";
import { getMySqlMs } from "../tools/systemTools";
import { BizError } from "../util/bizError";
......@@ -138,10 +138,273 @@ let deviceTypeKeyMap: { [key: string]: string } = {
"空调": "hvac",
"新风": "freshAir",
"人体传感": "occupancySensor",
"人体传感器": "occupancySensor",
"空气质量传感": "airSensor",
"客流传感器": "occupancySensor",
"环境监测": "airSensor",
"三相电表": "electricityMeter",
"电表": "electricityMeter",
"电能监测": "electricityMeter"
};
/**
* 指标定义配置
*/
const INDICATOR_CONFIG = [
{ key: "co2", name: "二氧化碳", unit: "ppm", min: 0, max: 1000 },
{ key: "pm10", name: "PM10", unit: "μg/m³", min: 0, max: 150 },
{ key: "pm2_5", name: "PM2.5", unit: "μg/m³", min: 0, max: 75 },
{ key: "hcho", name: "甲醛", unit: "mg/m³", min: 0, max: 0.08 },
{ key: "tvoc", name: "TVOC", unit: "mg/m³", min: 0, max: 0.5 },
{ key: "temperature", name: "温度", unit: "℃", min: 18, max: 26 },
{ key: "humidity", name: "湿度", unit: "%", min: 40, max: 60 },
{ key: "pressure", name: "气压", unit: "hPa", min: null, max: null },
{ key: "light_level", name: "光照度", unit: "lux", min: null, max: null },
{ key: "pir", name: "人体感应", unit: "", min: null, max: null },
{ key: "o3", name: "臭氧", unit: "mg/m³", min: 0, max: 0.1 },
];
/**
* 根据数据值判断质量等级
*/
function getQuality(key: string, value: number): string | null {
if (value === null || value === undefined) return null;
if (key === "pir") return value === 1 ? "有人" : "无感应";
const config = INDICATOR_CONFIG.find(c => c.key === key);
if (!config || config.min === null || config.max === null) return null;
if (value >= config.min && value <= config.max) return "优秀";
return "超标";
}
/**
* 格式化时间为 yyyy-MM-dd HH:mm:ss
*/
function formatTime(date: Date): string {
if (!date) return null;
return new Date(date).toISOString().slice(0, 19).replace("T", " ");
}
/**
* 获取当天零点的时间字符串
*/
function getTodayStart(): string {
const now = new Date();
now.setHours(0, 0, 0, 0);
return formatTime(now);
}
/**
* 获取当天的 device_data 记录
*/
async function getTodayDeviceData(deviceId: string): Promise<any[]> {
const todayStart = getTodayStart();
const result = await selectDataListByParam(TABLENAME.设备数据表, {
device_id: deviceId,
created_at: { "%gte%": todayStart },
"%orderAsc%": "created_at",
});
return result.data || [];
}
/**
* 构建 indicators 数组（基于当天最新数据）
*/
function buildIndicators(allLatestData: { [deviceId: string]: any }): any[] {
const sampleData = Object.values(allLatestData).find(d => d && d.data) as any;
const raw = sampleData?.data || {};
return INDICATOR_CONFIG.map(config => {
const currentValue = raw[config.key] !== undefined ? raw[config.key] : null;
return {
key: config.key,
name: config.name,
unit: config.unit,
current_value: currentValue,
min: config.min,
max: config.max,
quality: getQuality(config.key, currentValue),
};
});
}
/**
* 构建 records 数据
*/
function buildRecords(deviceDataMap: { [deviceId: string]: any[] }, isMultiple: boolean) {
const allRows: any[] = [];
for (const deviceId in deviceDataMap) {
for (const record of deviceDataMap[deviceId]) {
const rawData = record.data || {};
const row: any = {
device_time: record.device_time ? formatTime(record.device_time) : formatTime(record.created_at),
};
if (isMultiple) {
row.device_id = deviceId;
row.device_name = record.device_name || deviceId;
}
INDICATOR_CONFIG.forEach(c => {
row[c.key] = rawData[c.key] !== undefined ? rawData[c.key] : null;
});
allRows.push(row);
}
}
// 按时间升序
allRows.sort((a, b) => (a.device_time || "").localeCompare(b.device_time || ""));
// titleList
const titleList = isMultiple
? ["设备编号", "设备名", "时间", ...INDICATOR_CONFIG.map(c => c.name)]
: ["时间", ...INDICATOR_CONFIG.map(c => c.name)];
// dataList：将 row 转为数组按 titleList 顺序
const dataList = allRows.map(row => {
if (isMultiple) {
return [
row.device_id,
row.device_name,
row.device_time,
...INDICATOR_CONFIG.map(c => row[c.key]),
];
} else {
return [
row.device_time,
...INDICATOR_CONFIG.map(c => row[c.key]),
];
}
});
return { titleList, dataList };
}
/**
* 获取环境监测数据
* - 传入 deviceId 时：查询指定设备的环境数据
* - 传入 regionKey 时：查询该区域下所有环境监测设备的数据
* - 传入 regionGroup + regionType 时：查询该楼栋+楼层下所有环境监测设备的数据
* @param regionGroup 楼栋（如 "A馆"）
* @param regionType 楼层（如 "1F"）
* @param regionKey 区域ID
* @param deviceId 设备ID，精确查询单个设备
* @returns
*/
export async function getEnvironmentData(regionGroup: string, regionType: string, regionKey: number, deviceId: string) {
// ====== 假数据（调试用） ======
// return {
// indicators: [
// { key: "co2", name: "二氧化碳", unit: "ppm", current_value: 450, min: 0, max: 1000, quality: "优秀" },
// { key: "pm10", name: "PM10", unit: "μg/m³", current_value: 30, min: 0, max: 150, quality: "优秀" },
// { key: "pm2_5", name: "PM2.5", unit: "μg/m³", current_value: 15, min: 0, max: 75, quality: "优秀" },
// { key: "hcho", name: "甲醛", unit: "mg/m³", current_value: 0.02, min: 0, max: 0.08, quality: "优秀" },
// { key: "tvoc", name: "TVOC", unit: "mg/m³", current_value: 0.3, min: 0, max: 0.5, quality: "优秀" },
// { key: "temperature", name: "温度", unit: "℃", current_value: 25, min: 18, max: 26, quality: "优秀" },
// { key: "humidity", name: "湿度", unit: "%", current_value: 60, min: 40, max: 60, quality: "优秀" },
// { key: "pressure", name: "气压", unit: "hPa", current_value: 1013, min: null, max: null, quality: null },
// { key: "light_level", name: "光照度", unit: "lux", current_value: 300, min: null, max: null, quality: null },
// { key: "pir", name: "人体感应", unit: "", current_value: 0, min: null, max: null, quality: "无感应" },
// { key: "o3", name: "臭氧", unit: "mg/m³", current_value: 0.04, min: 0, max: 0.1, quality: "优秀" },
// ],
// records: {
// titleList: ["时间", "二氧化碳", "PM10", "PM2.5", "甲醛", "TVOC", "温度", "湿度", "气压", "光照度", "PIR", "臭氧"],
// dataList: [
// ["2025-06-18 10:00:00", 450, 30, 15, 0.02, 0.3, 25, 60, 1013, 300, 0, 0.04],
// ["2025-06-18 10:30:00", 460, 32, 16, 0.02, 0.28, 25.5, 58, 1012, 320, 1, 0.04],
// ],
// },
// areaInfo: {
// building: "A馆",
// floor: "1F",
// room: "101室",
// devices: [{ device_id: "ENV-001", device_name: "环境监测仪1号" }],
// },
// };
let devices: any[] = [];
// 1. 如果传了 deviceId，直接查该设备
if (deviceId) {
const deviceResult = await selectDataListByParam(TABLENAME.设备表, {
device_id: deviceId,
device_type: "环境监测",
});
devices = deviceResult.data || [];
} else {
// 2. 确定区域查询条件
const regionWhere: any = {};
if (regionKey) {
regionWhere.id = regionKey;
} else {
if (regionGroup) regionWhere.groups = regionGroup;
if (regionType) regionWhere.type = regionType;
}
// 3. 查询区域
const regionResult = await selectDataListByParam(TABLENAME.区域表, regionWhere);
const regions = regionResult.data || [];
if (regions.length === 0) {
return { indicators: [], records: { titleList: [], dataList: [] }, areaInfo: null };
}
// 4. 查询区域下的环境监测设备
const regionKeys = regions.map((r: any) => r.id);
const deviceResult = await selectDataListByParam(TABLENAME.设备表, {
region_key: { "%in%": regionKeys },
device_type: "环境监测",
});
devices = deviceResult.data || [];
}
if (devices.length === 0) {
return { indicators: [], records: { titleList: [], dataList: [] }, areaInfo: null };
}
const isMultiple = devices.length > 1;
// 5. 获取每个设备当天的数据
const deviceDataMap: { [deviceId: string]: any[] } = {};
const allLatestData: { [deviceId: string]: any } = {};
for (const device of devices) {
const devId = device.device_id;
const records = await getTodayDeviceData(devId);
records.forEach((r: any) => {
r.device_name = device.device_name || devId;
});
deviceDataMap[devId] = records;
if (records.length > 0) {
allLatestData[devId] = records[records.length - 1];
}
}
// 6. 构建 indicators
const indicators = buildIndicators(allLatestData);
// 7. 构建 records
const records = buildRecords(deviceDataMap, isMultiple);
// 8. 查询设备所属区域，构建 areaInfo
const deviceRegionKeys = [...new Set(devices.map((d: any) => d.region_key).filter(Boolean))];
let building = "";
let floor = "";
let room = "";
if (deviceRegionKeys.length > 0) {
const regionInfoResult = await selectDataListByParam(TABLENAME.区域表, {
id: { "%in%": deviceRegionKeys },
}, ["id", "name", "type", "groups"]);
const regionList = regionInfoResult.data || [];
if (regionList.length > 0) {
// 取第一个区域的信息填充
building = regionList[0].groups || "";
floor = regionList[0].type || "";
room = regionList[0].name || "";
}
}
const areaInfo = {
building,
floor,
room,
devices: devices.map((d: any) => ({
device_id: d.device_id,
device_name: d.device_name || d.device_id,
})),
};
return { indicators, records, areaInfo };
}
\ No newline at end of file
src/biz/mqttClient.ts
......@@ -38,7 +38,7 @@ export async function processDeviceData(message) {
console.log('收到数据:', data);
// 提取关键字段
const devEUI = data.devEUI;
const devEUI = data.devEUI ? data.devEUI.toUpperCase() : null;
if (!devEUI) {
console.warn('消息缺少devEUI字段，跳过');
return;
......@@ -93,6 +93,122 @@ export async function processDeviceData(message) {
}
}
/**
* 处理接收到的客流设备数据，并存入数据库
*
* 消息示例：
* {
* device_info: {
* device_mac: 'C0:BA:1F:02:98:F6',
* device_name: 'People Counter',
* device_sn: '6537F53875140008',
* firmware_version: 'V_125-LW.1.0.3-r4',
* hardware_version: 'V1.1',
* ip_address: '192.168.125.114',
* running_time: 4764787,
* wlan_mac: 'C0:BA:1F:02:98:F7'
* },
* region_trigger_data: { region_count_data: [ { region: 1, region_name: 'Region1', region_uuid: 'b63ffafc-857f-4d12-9dc7-d7fddfeb5f0f', total: { current_total: 9 } } ] },
* time_info: {
* dst_status: false,
* enable_dst: false,
* time: '2026-06-18T05:19:23-00:00',
* time_zone: 'UTC-0:00 Western European Time (WET), Greenwich Mean Time (GMT)'
* }
* }
*/
export async function customerDeviceData(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 deviceInfo = data.device_info;
if (!deviceInfo) {
console.warn('消息缺少device_info字段，跳过');
return;
}
const deviceSn = deviceInfo.device_sn ? deviceInfo.device_sn.toUpperCase() : null; // 设备序列号，转大写作为设备标识
if (!deviceSn) {
console.warn('消息缺少device_info.device_sn字段，跳过');
return;
}
const deviceMac = deviceInfo.device_mac;
const deviceName = deviceInfo.device_name || `客流设备_${deviceSn.slice(-6)}`;
// 提取设备时间
const timeInfo = data.time_info;
const deviceTime = timeInfo && timeInfo.time ? new Date(timeInfo.time) : new Date();
// ===== 提取客流数据 =====
const regionTriggerData = data.region_trigger_data;
if (!regionTriggerData || !regionTriggerData.region_count_data || regionTriggerData.region_count_data.length === 0) {
console.warn('消息缺少客流区域数据，跳过');
return;
}
const regionCountData = regionTriggerData.region_count_data;
const now = new Date();
// ===== 设备校验/新增 =====
const deviceRow = await selectOneDataByParam(
TABLENAME.设备表, { device_id: deviceSn }
);
if (!deviceRow) {
// 设备不存在，为每个区域创建一条设备记录
for (const regionData of regionCountData) {
await addData(
TABLENAME.设备表,
{
device_id: deviceSn,
region_key: regionData.region,
device_type: '客流传感器',
device_name: deviceName,
control_params: data,
created_at: now,
updated_at: now,
}
);
console.log(`客流设备已创建: ${deviceSn}, 区域 ${regionData.region}`);
}
}
// 遍历每个区域的客流数据，逐条入库
for (const regionData of regionCountData) {
const regionUuid = regionData.region_uuid;
const regionName = regionData.region_name || '';
const regionNumber = regionData.region;
const currentTotal = regionData.total?.current_total ?? 0;
await addData(
TABLENAME.设备数据表,
{
device_id: deviceSn,
data: regionData.total,
received_time: now,
device_time: deviceTime,
created_at: now,
}
);
console.log(`客流数据入库成功: 设备 ${deviceSn}, 区域 ${regionName || regionUuid}, 当前人数 ${currentTotal}`);
}
console.log(`客流数据处理完成: 设备 ${deviceSn}, 共入库 ${regionCountData.length} 条区域数据`);
} catch (error) {
console.error('客流数据处理失败:', error);
}
}
// ==================== 数据解析与入库函数 ====================
/**
* 将Base64编码的payload解码为十六进制字符串
......
src/biz/running.ts
......@@ -659,7 +659,7 @@ export async function controlAcRunning(params: {}) {
let device = await selectOneDataByParam(
TABLENAME.设备表,
{ device_id: deviceId },
["device_id", "region_key", "device_name", "device_ad", "control_params"]
["id", "device_id", "region_key", "device_name", "device_ad", "control_params"]
);
if (!device.data || !device.data.id) {
throw new BizError(ERRORENUM.未找到数据, `设备 ${deviceId} 未注册`);
......
src/config/serverConfig.ts
......@@ -40,12 +40,13 @@ export async function initConfig() {
// mqtt
if (mqttsrv) {
let configInfo = mqttsrv[0];
systemConfig.mqttsrv = { host: '', port: 0, topic: '', username: '', password: '' };
systemConfig.mqttsrv = { host: '', port: 0, topic_env: '', topic_crowd: '', username: '', password: '' };
if (configInfo.mqttHost && configInfo.mqttPort && configInfo.mqttTopic) {
if (configInfo.mqttHost && configInfo.mqttPort && (configInfo.mqttTopic1 || configInfo.mqttTopic2)) {
systemConfig.mqttsrv.host = configInfo.mqttHost[0];
systemConfig.mqttsrv.port = parseInt(configInfo.mqttPort[0]);
systemConfig.mqttsrv.topic = configInfo.mqttTopic[0];
systemConfig.mqttsrv.topic_env = configInfo.mqttTopic1[0];
systemConfig.mqttsrv.topic_crowd = configInfo.mqttTopic2[0];
systemConfig.mqttsrv.username = configInfo.username[0];
systemConfig.mqttsrv.password = configInfo.password[0];
}
......
src/config/systemClass.ts
......@@ -18,7 +18,8 @@ export class ServerConfig {
mqttsrv: {
host:string,
port:number,
topic:string,
topic_env:string,
topic_crowd:string,
username:string,
password:string
}
......
src/routers/device.ts
......@@ -38,6 +38,8 @@ export function setRouter(httpServer) {
/** 停止数据集成 */
httpServer.post('/api/qdm/run/stop/schedule', asyncHandler(stopSchedule));
/** 获取环境监测数据 */
httpServer.post('/api/qdm/device/environment', asyncHandler(getEnvironmentData));
}
/**
......@@ -157,6 +159,18 @@ async function stopSchedule(req, res) {
res.success(result);
}
/**
* 获取环境监测数据
*/
async function getEnvironmentData(req, res) {
let reqConf = {regionGroup:'String', regionType:'String', regionKey:'Number', deviceId:'String'};
const NotMustHaveKeys = ["regionGroup", "regionType", "regionKey", "deviceId"];
let { regionGroup, regionType, regionKey, deviceId } = eccReqParamater(reqConf, req.body, NotMustHaveKeys);
const result = await deviceBiz.getEnvironmentData(regionGroup, regionType, regionKey, deviceId);
res.success(result);
}
......
src/service/mqttInit.ts
import { processDeviceData } from "../biz/mqttClient";
import { processDeviceData, customerDeviceData} from "../biz/mqttClient";
import { systemConfig } from "../config/serverConfig";
const mqtt = require('mqtt');
......@@ -21,12 +21,20 @@ export async function startMqttClient() {
client.on('connect', () => {
console.log('MQTT客户端已连接，正在订阅主题...');
// 订阅网关的上行主题
client.subscribe(systemConfig.mqttsrv.topic, (err) => {
client.subscribe(systemConfig.mqttsrv.topic_env, (err) => {
if (err) {
console.error('订阅失败:', err);
console.error('环境订阅失败:', err);
} else {
console.log(`成功订阅主题: ${systemConfig.mqttsrv.topic}`);
console.log('等待设备数据...');
console.log(`成功订阅环境主题: ${systemConfig.mqttsrv.topic_env}`);
console.log('等待环境设备数据...');
}
});
client.subscribe(systemConfig.mqttsrv.topic_crowd, (err) => {
if (err) {
console.error('客流订阅失败:', err);
} else {
console.log(`成功订阅客流主题: ${systemConfig.mqttsrv.topic_crowd}`);
console.log('等待客流设备数据...');
}
});
});
......@@ -34,9 +42,12 @@ export async function startMqttClient() {
// 当收到消息时触发处理
client.on('message', (topic, message) => {
// 只处理配置的上行主题
if (topic === systemConfig.mqttsrv.topic) {
if (topic === systemConfig.mqttsrv.topic_env) {
processDeviceData(message);
}
if (topic === systemConfig.mqttsrv.topic_crowd) {
customerDeviceData(message);
}
});
client.on('error', (err) => {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment