bird_monitoring/script/mqtt/mqtt-simulate.js

#!/usr/bin/env node
/**
 * mqtt-simulate.js —— 设备侧模拟器（联调用，无需硬件）
 *
 * 把《设备通信协议-一期.md》附录 A 的手工 mosquitto 命令固化成一个可重复执行的进程，
 * 让**下行闭环（§8.2）**与**前置拦截（§8.3）**在真机到位之前就能端到端验证。
 *
 * 它模拟的是「设备侧」行为，因此刻意做两件服务端做不到的事：
 *   1. 收到 `command/hood` 后按需回执（或不回）——没有它，「只有 ack 才算成功」这条不变量无法被验证；
 *   2. 自己维护本地 `hoodCoverOn`，ack 成功后才切换——用真机语义约束模拟器，避免模拟器比真机更宽松。
 *
 * 依赖：node >= 18 + mqtt 包（在本目录执行 `npm install`）
 *
 * 用法：
 *   node mqtt-simulate.js --device-no BM202609160001
 *   node mqtt-simulate.js --device-no BM202609160001 --scenario lowbat
 *   node mqtt-simulate.js --device-no BM202609160001 --reply none        # 用例 5：5s 超时
 *   node mqtt-simulate.js --device-no BM202609160001 --reply reject      # 失败分支
 *   node mqtt-simulate.js --scenario register --imei 861234567890123    # §6.2 注册
 *   node mqtt-simulate.js --dry-run                                      # 只看报文，不连 broker
 *
 * @author Kou
 * @date 2026-09-16
 */
'use strict';

const mqtt = require('mqtt');

/** 设备编号格式：BM + 8 位日期 + 4 位序号（与 MqttTopics.DEVICE_NO_REGEX 同一口径） */
const DEVICE_NO_REGEX = /^BM\d{12}$/;

/** IMEI 格式：15 位数字（与 MqttTopics.IMEI_REGEX 同一口径） */
const IMEI_REGEX = /^\d{15}$/;

// 凭据不落明文：通过环境变量注入（export MQTT_BROKER / MQTT_USERNAME / MQTT_PASSWORD）
const DEFAULTS = {
  broker: process.env.MQTT_BROKER || 'mqtt://47.97.217.123:1883',
  username: process.env.MQTT_USERNAME || '',
  password: process.env.MQTT_PASSWORD || '',
  scenario: 'online',
  reply: 'auto',
  interval: 60000,
  count: 0,
  deviceNo: '',
  imei: '',
  hoodCover: 0,
  offlineAfter: 5000,
  otaStepMs: 2000,
  dryRun: false,
};

const USAGE = `
设备侧 MQTT 模拟器

必填（register 场景为 --imei）：
  --device-no <no>       设备编号，形如 BM202609160001

可选：
  --broker <url>         broker 地址            默认 ${DEFAULTS.broker}
  --username <u>         用户名                 默认 ${DEFAULTS.username}
  --password <p>         密码                   默认 ${DEFAULTS.password}
  --scenario <name>      设备状态场景           默认 online
                           online   周期遥测 + 自动回执（用例 1/4/6）
                           lowbat   低电遥测，验 §8.3 低电拦截与告警
                           offline  发一条 offline 事件后退出（用例 2 · 主动下线路径）
                           silent   静默不再上报（用例 2 · 心跳超时路径，需等 120s）
                           register 走 §6.2 注册链路（配合 --imei）
  --reply <mode>         对下行指令的回执方式   默认 auto
                           auto     正常回 ack{ok:true}（OTA 含进度事件）
                           none     不回执 —— 验 5s 超时（用例 5）
                           reject   回 ack{ok:false,code:3} —— 验失败分支
  --interval <ms>        遥测上报周期           默认 60000（协议 Q2 默认 60s）
  --count <n>            发 n 包遥测后退出      默认 0（不限）
  --hood-cover <0|1>     本地眼罩初始状态       默认 0（打开）
  --imei <15位>          register 场景必填
  --offline-after <ms>   offline 场景：上线后多久发 offline 事件   默认 5000
  --ota-step-ms <ms>     OTA 进度事件间隔（真机为 10s，联调默认加速）默认 2000
  --dry-run              只打印将要发送的报文，不连接 broker
  --help                 显示本帮助
`;

// ---------------------------------------------------------------------------
// 参数解析
// ---------------------------------------------------------------------------

/**
 * 解析命令行参数。
 *
 * @param {string[]} argv process.argv.slice(2)
 * @return {object} 合并默认值后的配置
 */
function parseArgs(argv) {
  const opts = { ...DEFAULTS };
  const map = {
    '--broker': 'broker',
    '--username': 'username',
    '--password': 'password',
    '--device-no': 'deviceNo',
    '--scenario': 'scenario',
    '--reply': 'reply',
    '--imei': 'imei',
  };
  const numMap = {
    '--interval': 'interval',
    '--count': 'count',
    '--hood-cover': 'hoodCover',
    '--offline-after': 'offlineAfter',
    '--ota-step-ms': 'otaStepMs',
  };
  for (let i = 0; i < argv.length; i++) {
    const arg = argv[i];
    if (arg === '--help' || arg === '-h') {
      opts.help = true;
      continue;
    }
    if (arg === '--dry-run') {
      opts.dryRun = true;
      continue;
    }
    if (map[arg]) {
      opts[map[arg]] = requireValue(argv, ++i, arg);
      continue;
    }
    if (numMap[arg]) {
      const raw = requireValue(argv, ++i, arg);
      const num = Number(raw);
      if (!Number.isFinite(num) || num < 0) {
        fail(`参数 ${arg} 需要非负数字，收到: ${raw}`);
      }
      opts[numMap[arg]] = num;
      continue;
    }
    fail(`未知参数: ${arg}（用 --help 查看用法）`);
  }
  return opts;
}

/**
 * 取参数值，缺失时终止。
 *
 * @param {string[]} argv 参数数组
 * @param {number} index 值所在下标
 * @param {string} name 参数名（报错用）
 * @return {string} 参数值
 */
function requireValue(argv, index, name) {
  const value = argv[index];
  if (value === undefined || value.startsWith('--')) {
    fail(`参数 ${name} 缺少取值`);
  }
  return value;
}

/**
 * 打印错误并退出。
 *
 * @param {string} msg 错误信息
 */
function fail(msg) {
  console.error(`[错误] ${msg}`);
  process.exit(2);
}

// ---------------------------------------------------------------------------
// 报文构造
// ---------------------------------------------------------------------------

/**
 * 构造上行 Topic 集合。
 *
 * @param {string} deviceNo 设备编号
 * @return {object} Topic 集合
 */
function topicsOf(deviceNo) {
  return {
    telemetry: `bm/${deviceNo}/telemetry`,
    event: `bm/${deviceNo}/event`,
    ack: `bm/${deviceNo}/ack`,
    commandWildcard: `bm/${deviceNo}/command/#`,
    provisionRegister: 'bm/provision/register',
    provisionAck: (imei) => `bm/provision/ack/${imei}`,
  };
}

/**
 * 设备本地状态与遥测生成器。
 *
 * 这里的「本地状态」是模拟器的核心：它只在**回执成功**后才切换眼罩状态，
 * 从而使服务端 `hood_cover_on` 的回写路径能被真实地验证（而不是模拟器自己抢先改口）。
 */
class SimulatedDevice {

  /**
   * @param {string} scenario 场景名
   * @param {number} initialCover 眼罩初始状态
   */
  constructor(scenario, initialCover) {
    this.scenario = scenario;
    this.hoodCoverOn = initialCover;
    this.hoodConnected = 1;
    this.signalLevel = 3;
    this.hoodBattery = scenario === 'lowbat' ? 12 : 86;
    this.ringBattery = scenario === 'lowbat' ? 15 : 74;
    this.ticks = 0;
  }

  /**
   * 生成一包遥测（§5.1：6 个必填字段全带，整份快照）。
   *
   * @return {object} 遥测报文
   */
  telemetry() {
    this.ticks += 1;
    // lowbat 场景保持低电，用于稳定复现 §8.3 拦截与低电告警（不被滞回恢复掉）
    if (this.scenario !== 'lowbat') {
      this.drain();
    }
    return {
      v: 1,
      ts: Date.now(),
      hoodBattery: this.hoodBattery,
      ringBattery: this.ringBattery,
      signalLevel: this.signalLevel,
      hoodConnected: this.hoodConnected,
      hoodCoverOn: this.hoodCoverOn,
      lng: 116.397128,
      lat: 39.916527,
      address: '北京市东城区东长安街',
    };
  }

  /**
   * 电量与信号缓慢衰减，让联调中的数值看起来像真实设备。
   */
  drain() {
    if (this.ticks % 5 === 0 && this.hoodBattery > 30) {
      this.hoodBattery -= 1;
    }
    if (this.ticks % 7 === 0 && this.ringBattery > 30) {
      this.ringBattery -= 1;
    }
  }

}

/**
 * 构造事件报文（§5.2）。
 *
 * @param {string} eventType 事件类型
 * @param {object} extra 附加字段
 * @return {object} 事件报文
 */
function eventPayload(eventType, extra) {
  return { v: 1, ts: Date.now(), eventType, ...extra };
}

/**
 * 构造回执报文（§5.3）。
 *
 * @param {string} msgId 下行指令的 msgId，必须原样回传
 * @param {string} cmd 指令类型
 * @param {object} extra 附加字段（ok/code/msg/progress）
 * @return {object} 回执报文
 */
function ackPayload(msgId, cmd, extra) {
  return { v: 1, msgId, cmd, ...extra };
}

// ---------------------------------------------------------------------------
// 日志
// ---------------------------------------------------------------------------

/**
 * 带时间戳的日志。
 *
 * @param {string} msg 日志内容
 */
function log(msg) {
  const ts = new Date().toISOString().replace('T', ' ').slice(0, 19);
  console.log(`[${ts}] ${msg}`);
}

/**
 * 参数校验：先校验、后打印横幅，避免「先输出一段正常日志再报错」的观感。
 *
 * @param {object} opts 配置
 */
function validate(opts) {
  if (opts.scenario === 'register') {
    if (!IMEI_REGEX.test(opts.imei)) {
      fail(`register 场景需要 --imei（15 位数字），收到: ${opts.imei || '(空)'}`);
    }
    return;
  }
  if (!DEVICE_NO_REGEX.test(opts.deviceNo)) {
    fail(`--device-no 必须是 BM + 8 位日期 + 4 位序号（如 BM202609160001），收到: ${opts.deviceNo || '(空)'}`);
  }
}

/**
 * 打印启动横幅：告诉联调的人「这次跑的是哪条用例」。
 *
 * @param {object} opts 配置
 */
function banner(opts) {
  const cases = {
    online: '用例 1（遥测落库）/ 4（指令+ack 成功）/ 6（重复 msgId）',
    lowbat: '§8.3 低电拦截（遮盖被拒、打开放行）+ 低电告警',
    offline: '用例 2 · 主动下线路径（offline 事件）',
    silent: '用例 2 · 心跳超时路径（120s 后被扫描判离线）',
    register: '§6.2 注册链路（含 code=404 未建档分支）',
  };
  const identity = opts.scenario === 'register' ? `provision (imei=${opts.imei})` : `bm/${opts.deviceNo}`;
  log('='.repeat(64));
  log(`设备模拟器启动 · 场景=${opts.scenario} · 回执=${opts.reply}`);
  log(`覆盖：${cases[opts.scenario] || '(自定义)'}`);
  log(`broker=${opts.broker} · topic 前缀=${identity}`);
  log('='.repeat(64));
}

// ---------------------------------------------------------------------------
// register 场景（§6.2）
// ---------------------------------------------------------------------------

/**
 * 跑一次注册流程：上行 register，等 ack 后退出。
 *
 * @param {object} opts 配置
 */
function runRegister(opts) {
  const imei = opts.imei;
  const ackTopic = topicsOf('').provisionAck(imei);
  const registerPayload = {
    ts: Date.now(),
    imei,
    sn: `SN${imei.slice(-8)}`,
    model: 'BM-H1',
    hwVersion: '1.0',
    fwVersion: '1.0.3',
    mac: 'AC:1F:6B:00:11:22',
  };
  if (opts.dryRun) {
    log('[dry-run] 将订阅: ' + ackTopic);
    log('[dry-run] 将上行: bm/provision/register ' + JSON.stringify(registerPayload));
    return;
  }
  const client = mqtt.connect(opts.broker, clientOptions(opts, `reg-${imei}`));
  const timer = setTimeout(() => {
    log('注册应答超时（10s），请确认服务端已订阅 bm/provision/register');
    finish(client, 1);
  }, 10000);
  client.on('error', (err) => {
    log(`MQTT 错误: ${err.message}`);
  });
  client.on('connect', () => {
    log('已连接 broker');
    client.subscribe(ackTopic, { qos: 1 }, (err) => {
      if (err) {
        log(`订阅失败: ${err.message}`);
        finish(client, 1);
        return;
      }
      log(`<< 已订阅 ${ackTopic}`);
      client.publish('bm/provision/register', JSON.stringify(registerPayload), { qos: 1 });
      log(`>> ${JSON.stringify(registerPayload)}`);
    });
  });
  client.on('message', (topic, buf) => {
    const text = buf.toString('utf8');
    log(`<< ${topic} ${text}`);
    const ack = safeParse(text);
    if (ack && ack.ok === true) {
      log(`注册成功，deviceNo=${ack.deviceNo}（设备端需持久化，重启不再注册）`);
    } else if (ack && ack.code === 404) {
      log('注册被拒：IMEI 未建档 —— 后台需先在 bm_device 录入该 IMEI（这是私拆/串货的线索）');
    }
    clearTimeout(timer);
    finish(client, 0);
  });
}

// ---------------------------------------------------------------------------
// 设备场景
// ---------------------------------------------------------------------------

/**
 * 跑设备场景。
 *
 * @param {object} opts 配置
 */
function runDevice(opts) {
  const deviceNo = opts.deviceNo;
  const device = new SimulatedDevice(opts.scenario, opts.hoodCover);
  const t = topicsOf(deviceNo);

  if (opts.dryRun) {
    log('[dry-run] 将订阅: ' + t.commandWildcard);
    log('[dry-run] 将上行(connect 后): ' + t.event + ' ' + JSON.stringify(eventPayload('online', {})));
    log('[dry-run] 将上行(周期): ' + t.telemetry + ' ' + JSON.stringify(device.telemetry()));
    log('[dry-run] 收到 hood 指令按 --reply=' + opts.reply + ' 处理：'
      + JSON.stringify(ackPayload('<下行 msgId>', 'hood', { ok: true, code: 0, msg: '' })));
    log('[dry-run] 将上行(退出前): ' + t.event + ' ' + JSON.stringify(eventPayload('offline', { reason: 'normal' })));
    return;
  }

  const client = mqtt.connect(opts.broker, clientOptions(opts, deviceNo));
  const state = { telemetryTimer: null, sentCount: 0, otaRunning: false, shuttingDown: false };

  client.on('error', (err) => log(`MQTT 错误: ${err.message}`));
  client.on('reconnect', () => log('重连中…'));

  client.on('connect', () => {
    log('已连接 broker');
    client.subscribe(t.commandWildcard, { qos: 1 }, (err) => {
      if (err) {
        log(`订阅失败: ${err.message}`);
        return;
      }
      // 协议 §5.2：连接成功且完成订阅后才发 online
      publishJson(client, t.event, eventPayload('online', { reason: 'normal' }));
      log(`<< 已订阅 ${t.commandWildcard}`);
      if (opts.scenario === 'silent' || opts.scenario === 'offline') {
        // 先落一包遥测把 status 推到在线，再进入静默/下线，避免「从未有过心跳」的设备不受心跳约束
        publishJson(client, t.telemetry, device.telemetry());
      }
      if (opts.scenario === 'offline') {
        setTimeout(() => shutdown(client, opts, 'normal'), opts.offlineAfter);
        return;
      }
      if (opts.scenario === 'silent') {
        log(`进入静默：不再上报遥测。等待服务端在 offline-threshold(120s) 后判离线，勿按 Ctrl+C。`);
        return;
      }
      state.telemetryTimer = setInterval(() => {
        publishJson(client, t.telemetry, device.telemetry());
        state.sentCount += 1;
        if (opts.count > 0 && state.sentCount >= opts.count) {
          shutdown(client, opts, 'normal');
        }
      }, opts.interval);
    });
  });

  client.on('message', (topic, buf) => {
    const text = buf.toString('utf8');
    log(`<< ${topic} ${text}`);
    const payload = safeParse(text);
    if (!payload || !payload.msgId) {
      log('下行报文无 msgId，按设备侧策略忽略');
      return;
    }
    if (topic.endsWith('/command/hood')) {
      handleHoodCommand(client, opts, device, t, payload);
    } else if (topic.endsWith('/command/ota')) {
      handleOtaCommand(client, opts, t, payload, state);
    } else {
      log('未知下行指令，按前向兼容忽略（真实设备应记日志不动作）');
    }
  });

  const onSignal = () => {
    if (state.shuttingDown) {
      return;
    }
    state.shuttingDown = true;
    log('收到中断信号，按「正常下线」流程退出');
    shutdown(client, opts, 'normal');
  };
  process.on('SIGINT', onSignal);
  process.on('SIGTERM', onSignal);
}

/**
 * 处理眼罩指令：按 `--reply` 决定回执行为，并只在成功后切换本地状态。
 *
 * @param {object} client MQTT 客户端
 * @param {object} opts 配置
 * @param {SimulatedDevice} device 模拟设备
 * @param {object} t Topic 集合
 * @param {object} cmd 下行指令报文
 */
function handleHoodCommand(client, opts, device, t, cmd) {
  const target = cmd.coverOn === true;
  if (opts.reply === 'none') {
    log(`[用例 5] 收到遮盖=${target} 的指令但**故意不回执** → 服务端应 5s 超时并返回「指令下发失败」，hood_cover_on 不变`);
    return;
  }
  if (opts.reply === 'reject') {
    publishJson(client, t.ack, ackPayload(cmd.msgId, 'hood', { ok: false, code: 3, msg: 'device busy' }));
    log('已回失败回执（ok=false, code=3）→ 服务端应返回「指令下发失败」且不写 hood_cover_on');
    return;
  }
  // auto：设备真实执行后才改本地状态，再用本地状态回执——顺序反了会掩盖服务端的回写缺陷
  device.hoodCoverOn = target ? 1 : 0;
  publishJson(client, t.ack, ackPayload(cmd.msgId, 'hood', { ok: true, code: 0, msg: '' }));
  log(`眼罩已${target ? '遮盖' : '打开'}，已回执 ok=true（本地 hoodCoverOn=${device.hoodCoverOn}）`);
}

/**
 * 处理 OTA 指令：先回执「已受理」，再周期上报进度，最后回终态（协议 §7.2）。
 *
 * @param {object} client MQTT 客户端
 * @param {object} opts 配置
 * @param {object} t Topic 集合
 * @param {object} cmd 下行指令报文
 * @param {object} state 运行状态（含 otaRunning 幂等标志）
 */
function handleOtaCommand(client, opts, t, cmd, state) {
  if (opts.reply === 'none') {
    log('收到 OTA 指令但不回执 → 服务端应 5s 超时');
    return;
  }
  if (opts.reply === 'reject') {
    publishJson(client, t.ack, ackPayload(cmd.msgId, 'ota', { ok: false, code: 3, msg: 'md5 mismatch' }));
    log('已回失败回执（code=3）→ 任务应落 FAILED，result_msg 记 md5 mismatch');
    return;
  }
  if (state.otaRunning) {
    log('已有 OTA 在跑，按设备侧幂等策略忽略该次指令');
    return;
  }
  state.otaRunning = true;
  // 受理 ≠ 完成：先回 ok=true 表示「已收到」，最终结果用第二条 ack
  publishJson(client, t.ack, ackPayload(cmd.msgId, 'ota', { ok: true, code: 0, progress: 0 }));
  log(`OTA 已受理 version=${cmd.version}，开始上报进度（间隔 ${opts.otaStepMs}ms，真机为 10s）`);
  let progress = 0;
  const timer = setInterval(() => {
    progress += 20;
    if (progress >= 100) {
      clearInterval(timer);
      publishJson(client, t.event, eventPayload('ota_progress', { version: cmd.version, value: 100 }));
      publishJson(client, t.ack, ackPayload(cmd.msgId, 'ota', { ok: true, code: 0, progress: 100 }));
      state.otaRunning = false;
      log('OTA 完成，已回终态 ack{ok:true, progress:100}');
      return;
    }
    publishJson(client, t.event, eventPayload('ota_progress', { version: cmd.version, value: progress }));
  }, opts.otaStepMs);
}

// ---------------------------------------------------------------------------
// 工具
// ---------------------------------------------------------------------------

/**
 * 组装 MQTT 连接选项。
 *
 * @param {object} opts 配置
 * @param {string} identity 身份标识（用于 clientId，便于在 broker 侧区分连接）
 * @return {object} mqtt.connect 选项
 */
function clientOptions(opts, identity) {
  return {
    username: opts.username,
    password: opts.password,
    clientId: `bm-sim-${identity}-${Math.random().toString(16).slice(2, 8)}`,
    protocolVersion: 4,
    clean: true,
    keepalive: 60,
    reconnectPeriod: 5000,
    connectTimeout: 10000,
  };
}

/**
 * 发布 JSON 报文并打印。
 *
 * @param {object} client MQTT 客户端
 * @param {string} topic 目标 Topic
 * @param {object} payload 报文对象
 */
function publishJson(client, topic, payload) {
  const text = JSON.stringify(payload);
  client.publish(topic, text, { qos: topic.endsWith('/telemetry') ? 0 : 1 });
  log(`>> ${topic} ${text}`);
}

/**
 * 正常下线：发 offline 事件 → 断开。
 *
 * @param {object} client MQTT 客户端
 * @param {object} opts 配置
 * @param {string} reason 下线原因
 */
function shutdown(client, opts, reason) {
  if (!opts.deviceNo) {
    finish(client, 0);
    return;
  }
  const t = topicsOf(opts.deviceNo);
  publishJson(client, t.event, eventPayload('offline', { reason }));
  log('已发 offline 事件（服务端应置 status=2 并写一条 offline 告警）');
  setTimeout(() => finish(client, 0), 500);
}

/**
 * 结束进程。
 *
 * @param {object} client MQTT 客户端
 * @param {number} code 退出码
 */
function finish(client, code) {
  client.end(true, () => process.exit(code));
}

/**
 * 容错解析 JSON。
 *
 * @param {string} text 原始报文
 * @return {object|null} 解析结果；失败返回 null
 */
function safeParse(text) {
  try {
    return JSON.parse(text);
  } catch (e) {
    return null;
  }
}

// ---------------------------------------------------------------------------
// 入口
// ---------------------------------------------------------------------------

function main() {
  const opts = parseArgs(process.argv.slice(2));
  if (opts.help) {
    console.log(USAGE);
    return;
  }
  validate(opts);
  banner(opts);
  if (opts.scenario === 'register') {
    runRegister(opts);
    return;
  }
  runDevice(opts);
}

main();