/** * 机器人智慧服务系统 · 状态服务 Alpha (status-service) * 端口: 3020 * 职责: 设备状态上报接收、最新状态缓存、历史状态查询、SSE 实时订阅 * 数据: services/status/data/status.json(本地文件替代数据库) * 文档: 详见 docs/DOC-08-设备数据字典-V1.0草案.md */ const express = require('express'); const cors = require('cors'); const fs = require('fs'); const path = require('path'); const app = express(); const PORT = 3020; const DATA_FILE = path.join(__dirname, 'data', 'status.json'); const MAX_HISTORY_PER_DEVICE = 500; // 每设备历史条数上限(超出丢弃最旧) app.use(cors()); app.use(express.json()); // ─── 文件读写 ───────────────────────────────────────────── function readData() { try { const raw = fs.readFileSync(DATA_FILE, 'utf-8'); return JSON.parse(raw); } catch { return { latest: {}, history: {} }; } } function writeData(data) { fs.writeFileSync(DATA_FILE, JSON.stringify(data, null, 2), 'utf-8'); } // ─── SSE 订阅者 ─────────────────────────────────────────── const subscribers = new Set(); // res 集合 function broadcast(payload) { const msg = `data: ${JSON.stringify(payload)}\n\n`; for (const res of subscribers) { try { res.write(msg); } catch { subscribers.delete(res); } } } // ─── 路由:上报 ─────────────────────────────────────────── // POST /api/status/report 状态上报(设备 -> 平台) // body: { deviceId, position?, battery?, mode?, speed?, sensors?, alarms?, timestamp? } app.post('/api/status/report', (req, res) => { const { deviceId, position, battery, mode, speed, sensors, alarms, timestamp, meta = {}, } = req.body || {}; if (!deviceId) { return res.status(400).json({ code: 40001, msg: 'deviceId 为必填项' }); } const now = new Date().toISOString(); const record = { deviceId, position: position || null, // {x,y,z,yaw} battery: battery ?? null, // 0~100 整数百分比 mode: mode || null, // idle/patrol/charging/working/emergency speed: speed ?? null, // m/s sensors: sensors || {}, // 传感器键值 alarms: alarms || [], // 告警编码列表 meta, ts: timestamp || now, // 设备侧时间(ISO) receivedAt: now, // 平台接收时间(ISO) }; const data = readData(); data.latest[deviceId] = record; const history = data.history[deviceId] || (data.history[deviceId] = []); history.push(record); if (history.length > MAX_HISTORY_PER_DEVICE) history.splice(0, history.length - MAX_HISTORY_PER_DEVICE); writeData(data); // 实时广播给订阅者 broadcast({ type: 'status', deviceId, record }); res.json({ code: 0, msg: '状态已接收', data: { deviceId, receivedAt: now } }); }); // ─── 路由:查询 ─────────────────────────────────────────── // GET /api/status/summary 全设备最新状态汇总 app.get('/api/status/summary', (req, res) => { const data = readData(); res.json({ code: 0, data: Object.values(data.latest) }); }); // GET /api/status/:deviceId/latest 指定设备最新状态 app.get('/api/status/:deviceId/latest', (req, res) => { const data = readData(); const record = data.latest[req.params.deviceId]; if (!record) return res.status(404).json({ code: 40401, msg: '该设备暂无状态上报' }); res.json({ code: 0, data: record }); }); // GET /api/status/history/:deviceId?from=&to=&limit= 历史状态 app.get('/api/status/history/:deviceId', (req, res) => { const data = readData(); const history = data.history[req.params.deviceId] || []; const { from, to, limit } = req.query; let list = history; if (from) list = list.filter((r) => r.ts >= from); if (to) list = list.filter((r) => r.ts <= to); const n = Math.min(parseInt(limit, 10) || 100, 500); res.json({ code: 0, data: list.slice(-n) }); }); // GET /api/status/subscribe SSE 实时订阅 app.get('/api/status/subscribe', (req, res) => { res.setHeader('Content-Type', 'text/event-stream'); res.setHeader('Cache-Control', 'no-cache'); res.setHeader('Connection', 'keep-alive'); res.write(`data: ${JSON.stringify({ type: 'connected', msg: '订阅成功' })}\n\n`); subscribers.add(res); req.on('close', () => subscribers.delete(res)); }); app.listen(PORT, () => { console.log(`✅ 状态服务 Alpha 已启动 → http://localhost:${PORT}`); console.log(` API 示例: POST /api/status/report {deviceId, position, battery, mode}`); console.log(` GET /api/status/:deviceId/latest`); console.log(` GET /api/status/subscribe (SSE)`); });