Files
smart-parking-platform/simulator/simulator.js
T

244 lines
7.4 KiB
JavaScript
Raw Normal View History

/*
* 智慧停车设备模拟服务
* - 通过 WebSocket 伪装算法端(ai-end)注册,使摄像头在平台上线
* - 调用 ffmpeg 将本地模拟视频循环推送到 nginx-rtmp,生成 HLS 供监控页播放
* - 按摄像头报警规则随机上报报警事件(抓拍图转 base64)
*/
const { spawn } = require('child_process');
const crypto = require('crypto');
const fs = require('fs');
const path = require('path');
const { io } = require('socket.io-client');
// 发送前对明文密码做 HMAC-SHA256(base64),与库中哈希比对
const HMAC_KEY = 'smart-parking-platform';
const hashPassword = (plain) =>
crypto.createHmac('sha256', HMAC_KEY).update(plain).digest('base64');
const HTTP_SERVER_URL = (process.env.HTTP_SERVER_URL || 'http://app').replace(/\/+$/, '');
const RTMP_SERVER_URL = process.env.RTMP_SERVER_URL || 'rtmp://app:1515/live';
const USERNAME = process.env.ADMIN_USERNAME || 'admin';
const PASSWORD = process.env.ADMIN_PASSWORD || 'admin123';
const ASSETS_DIR = process.env.ASSETS_DIR || '/assets';
const FFMPEG_BIN = process.env.FFMPEG_BIN || 'ffmpeg';
// 要模拟的摄像头 ID 列表;逗号分隔,默认 1..12(与 seeder 的 12 路对应)
const CAMERA_IDS = (process.env.CAMERA_IDS || '1,2,3,4,5,6,7,8,9,10,11,12')
.split(',')
.map((s) => parseInt(s.trim(), 10))
.filter((n) => Number.isInteger(n) && n > 0);
// 摄像头 ID 与素材文件一一对应(cam-<两位ID>.mp4)
const videoFileFor = (cameraID) =>
`cam-${String(cameraID).padStart(2, '0')}.mp4`;
// 报警抓拍图(与 seeder 中 algorithmType 对应:body / vehicle)
const SHOTS = {
body: ['alarm-crowd.jpg', 'alarm-bicycle.jpg'],
vehicle: ['alarm-parking.jpg', 'alarm-peak.jpg'],
};
const log = (...args) => console.log(`[${new Date().toISOString()}]`, ...args);
const devices = new Map();
function startFfmpeg(cameraID, videoFile) {
const videoPath = path.join(ASSETS_DIR, videoFile);
const target = `${RTMP_SERVER_URL}/${cameraID}`;
const args = [
'-re',
'-stream_loop', '-1',
'-i', videoPath,
'-an',
'-c:v', 'copy',
'-f', 'flv',
target,
];
return spawn(FFMPEG_BIN, args, { stdio: ['ignore', 'ignore', 'pipe'] });
}
/**
* 每台设备的推流管理器:
* - 仅在 WS 在线时保持推流;
* - ffmpeg 异常退出后自动重启,退避 3s→6s→…封顶 30s;
* - app 重建导致 RTMP 断连时也能自愈。
*/
class StreamManager {
constructor(device, videoFile) {
this.device = device;
this.videoFile = videoFile;
this.proc = null;
this.restartTimer = null;
this.backoff = 3000;
this.stopped = false;
}
start() {
this.stopped = false;
if (this.proc || this.restartTimer) return;
if (!fs.existsSync(path.join(ASSETS_DIR, this.videoFile))) {
log(`camera ${this.device.cameraID} 素材缺失:${this.videoFile}`);
return;
}
const proc = startFfmpeg(this.device.cameraID, this.videoFile);
this.proc = proc;
proc.stderr.on('data', (d) => {
const line = d.toString();
if (/conversion failed|broken pipe|connection refused|could not|error/i.test(line)) {
log(`camera ${this.device.cameraID} ffmpeg: ${line.trim().slice(-160)}`);
}
});
proc.on('exit', (code) => {
this.proc = null;
log(`camera ${this.device.cameraID} ffmpeg 退出 code=${code},准备重启`);
if (!this.stopped) this.scheduleRestart();
});
}
scheduleRestart() {
if (this.restartTimer || this.stopped) return;
// app 尚未上线时,等待 WS 重新连接后再推
if (!this.device.connected) {
this.restartTimer = setTimeout(() => {
this.restartTimer = null;
this.scheduleRestart();
}, 2000);
return;
}
const delay = this.backoff;
this.backoff = Math.min(this.backoff * 2, 30000);
this.restartTimer = setTimeout(() => {
this.restartTimer = null;
this.start();
}, delay);
}
// 成功稳定推流一段时间后可由外部复位退避;这里在 start 成功连接后保守复位
resetBackoff() {
this.backoff = 3000;
}
stop() {
this.stopped = true;
if (this.restartTimer) {
clearTimeout(this.restartTimer);
this.restartTimer = null;
}
if (this.proc) {
try {
this.proc.kill('SIGTERM');
} catch {}
this.proc = null;
}
}
}
function readShotBase64(file) {
const buf = fs.readFileSync(path.join(ASSETS_DIR, file));
return `data:image/jpeg;base64,${buf.toString('base64')}`;
}
function scheduleAlarm(device) {
// 每台设备 180~480 秒尝试一次报警,70% 概率真正触发
const delay = 180000 + Math.random() * 300000;
device.alarmTimer = setTimeout(() => {
const rules = device.rules || [];
if (device.connected && rules.length > 0 && Math.random() < 0.7) {
const rule = rules[Math.floor(Math.random() * rules.length)];
const pool = SHOTS[rule.algorithmType] || SHOTS.vehicle;
const shot = pool[Math.floor(Math.random() * pool.length)];
try {
const picBase64 = readShotBase64(shot);
device.socket.emit('alarm', { picBase64, alarmRuleID: rule.id });
log(`camera ${device.cameraID} 上报报警 rule=${rule.id}(${rule.name}) shot=${shot}`);
} catch (e) {
log(`camera ${device.cameraID} 报警失败: ${e.message}`);
}
}
scheduleAlarm(device);
}, delay);
}
function startDevice(cameraID) {
const videoFile = videoFileFor(cameraID);
const socket = io(HTTP_SERVER_URL, {
path: '/ws/ai/',
extraHeaders: {
data: JSON.stringify({
username: USERNAME,
password: hashPassword(PASSWORD),
cameraID: String(cameraID),
}),
},
transports: ['websocket'],
reconnection: true,
reconnectionDelay: 3000,
});
const device = {
cameraID,
socket,
stream: null,
rules: [],
connected: false,
alarmTimer: null,
};
device.stream = new StreamManager(device, videoFile);
socket.on('connect', () => {
device.connected = true;
log(`camera ${cameraID} 已注册上线 (${videoFile})`);
device.stream.resetBackoff();
device.stream.start();
scheduleAlarm(device);
});
socket.on('cameraConfigChange', (config) => {
device.rules = Array.isArray(config.alarmRules) ? config.alarmRules : [];
log(`camera ${cameraID} 收到规则 ${device.rules.map((r) => r.id).join(',') || '无'}`);
});
socket.on('disconnect', (reason) => {
device.connected = false;
log(`camera ${cameraID} 断开: ${reason},暂停推流等待重连`);
device.stream.stop();
});
socket.io.on('reconnect', () => {
// connect 事件中已处理 start,这里仅记录
log(`camera ${cameraID} 触发重连`);
});
socket.on('connect_error', (err) => {
log(`camera ${cameraID} 连接错误: ${err.message}`);
});
devices.set(cameraID, device);
}
function shutdown() {
log('正在停止模拟服务…');
for (const device of devices.values()) {
clearTimeout(device.alarmTimer);
try {
device.stream.stop();
} catch {}
try {
device.socket.close();
} catch {}
}
setTimeout(() => process.exit(0), 1000);
}
process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);
log(`将模拟 ${CAMERA_IDS.length} 路摄像头:${CAMERA_IDS.join(',')}`);
log(`平台=${HTTP_SERVER_URL} RTMP=${RTMP_SERVER_URL}`);
CAMERA_IDS.forEach((id, idx) => {
setTimeout(() => startDevice(id), idx * 1500); // 错开连接,避免同时冲击
});