244 lines
7.4 KiB
JavaScript
244 lines
7.4 KiB
JavaScript
/*
|
||
* 智慧停车设备模拟服务
|
||
* - 通过 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); // 错开连接，避免同时冲击
|
||
});
|