#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
鹤仆云 · 设备模拟器（演示用）

它扮演的角色是「网关 + 设备」，不是业务层。数据路径和真设备完全一致：

    sim_device.py ──HTTP──→ M 层 /internal/iot/v1/**  ──→ iot_event
                                                          │
                                                  AlarmEngine（业务层）
                                                          ↓
                                                    alarm + notify_task
                                                          ↓
                                                    家属端 H5

所以模拟器只调三个内部接口，一个业务接口都不碰：
    I-01  POST /internal/iot/v1/events/batch          事件
    I-02  POST /internal/iot/v1/device-status/batch   状态快照（电量/信号/在线）
    I-12  POST /internal/iot/v1/raw-frames            原文帧（Base64）

用法：
    python sim_device.py            正常节奏（门磁、水振、电流随机冒出）
    python sim_device.py --fast     快节奏，演示时用
    python sim_device.py --seed     启动先补几条历史，页面一打开就有内容
    python sim_device.py --quiet    只开控制台，不自动产生事件

控制台：http://127.0.0.1:9901
    —— 按钮按设备类型出现（门磁只有开门/关门，水浸只有漏水/已止），
    选错设备发不出对应的告警——真实设备也是这样。
"""

import argparse
import base64
import hashlib
import json
import os
import random
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer

BASE = os.environ.get("SIM_BASE", "http://127.0.0.1:8080")
PORT = int(os.environ.get("SIM_PORT", "9901"))
CST = timezone(timedelta(hours=8))

OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))

# ---------------------------------------------------------------- 设备台账
# sensor_id / source_channel 必须和 01_初始化_设备台账.sql 对得上，
# 对不上就会被 M 层判成「未建档」丢进 unregistered_device，家属端看不见。
#
# ★ type 必须和库里 device.type_code 一致（2026-11-09 修）：
#   之前 3F7A9C23 在这里写成「厨房水浸」，库里其实是「床头SOS」，
#   结果控制台选门磁点 SOS 也能发出去、还生成告警——纯属模拟器在骗自己。
#   通道名必须是后端 Channel 枚举的四个之一：OWN_GW / HORN_HOST / HORN_CELL / MIR_CELL。
#   （曾经写成 H_MQTT，后端直接 400 "sourceChannel 非法"，断电/来电永远发不出去）
DEVICES = [
    dict(sn="3F7A9C21", dev="T-G-01", name="入户门磁", type="DOOR", ch="OWN_GW", hh=9001, loc="入户门"),
    dict(sn="3F7A9C22", dev="T-G-02", name="卧室门磁", type="DOOR", ch="OWN_GW", hh=9003, loc="卧室"),
    dict(sn="3F7A9C23", dev="T-G-03", name="床头SOS", type="SOS", ch="OWN_GW", hh=9001, loc="卧室床头"),
    dict(sn="3F7A9C24", dev="T-G-04", name="厨房水浸", type="WATER", ch="OWN_GW", hh=9003, loc="厨房水槽下"),
    dict(sn="MIR:861234567890123", dev="T-S-01", name="卫生间水浸", type="WATER", ch="MIR_CELL", hh=9001, loc="卫生间"),
    dict(sn="HORN:SN12345678", dev="T-H-01", name="客厅烟感", type="SMOKE", ch="HORN_HOST", hh=9001, loc="客厅"),
    # 水振（麦乐克那台 HORN:999111 也在库里，型号 VIB）。它报的是"用水活动"，
    # 不是告警——放进来是为了能演示"这类事件只进基线、不弹窗"。
    dict(sn="HORN:999111", dev="D1794213326404263", name="水振", type="VIB", ch="HORN_HOST", hh=None, loc="厨房水管"),
    # 验收专用：一台"已入库、还没人绑"的门磁。账号 A 手填 A1B2C3D4 绑上之后，
    # 在控制台把设备选成它再点「门打开了」，就能看见完整闭环。
    dict(sn="A1B2C3D4", dev="T-G-09", name="验收门磁", type="DOOR", ch="OWN_GW", hh=9001, loc="验收"),
]
DEV_BY_SN = {d["sn"]: d for d in DEVICES}

# 设备类型码（J-00 §1.3 payload.deviceTypeRaw）
TYPE_RAW = {"DOOR": 1, "VIB": 2, "CURRENT": 3, "WATER": 4, "SOS": 5, "SMOKE": 6}

# 每种设备能干什么。列表里没有的动作，控制台不会显示，后端也会拒。
# 这不是"模拟器偷懒"——真实设备也一样：门磁没有水浸探针，物理上就发不出漏水。
ACTIONS = {
    "DOOR":  [("door", "🚪 门打开了", "warn"), ("door_close", "✅ 门关上了", "ok")],
    "WATER": [("leak", "💧 漏水了", "hot"), ("leak_clear", "✅ 漏水已止", "ok")],
    "SOS":   [("sos", "🆘 按下 SOS", "hot")],
    # 烟感：J-00 §1.5.1 是 SMOKE / SMOKE_CLEAR，规则表里 SECOND + URGENT，与漏水同一套
    "SMOKE": [("smoke", "🔥 烟感报警", "hot"), ("smoke_clear", "✅ 烟雾已散", "ok")],
    # 水振：J-00 §1.5.3 里它报的是"用水活动" WATER_ACTIVITY / _STOP，不是告警事件。
    # 它喂给基线和「水振＋电流」组合研判，本身不弹窗——所以按钮上直接写清楚。
    "VIB":   [("vib", "💧 用水活动（不告警）", ""), ("vib_stop", "用水停止", "")],
}
# 任何设备都能做的：断电/来电（网关级事件）、心跳
COMMON_ACTIONS = [("power_off", "⚡ 设备断电", "warn"),
                  ("power_on", "🔌 设备来电", "ok"),
                  ("status", "📊 上报心跳", "")]


# 每台设备的运行态。seq 必须单调递增，否则 M 层会判成重传或断档。
STATE = {}
LOCK = threading.Lock()
COUNTER = {"n": 0}
LOG = []          # 最近 60 条动作，控制台轮询显示
LOG_CAP = 60


def now_iso(dt=None):
    dt = dt or datetime.now(CST)
    return dt.strftime("%Y-%m-%dT%H:%M:%S+08:00")


def log(msg):
    line = time.strftime("%H:%M:%S ") + msg
    with LOCK:
        LOG.append(line)
        del LOG[:-LOG_CAP]
    print(line, flush=True)


# ---------------------------------------------------------------- HTTP
def post(path, body):
    data = json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(BASE + path, data=data, method="POST",
                                 headers={"Content-Type": "application/json"})
    try:
        with OPENER.open(req, timeout=10) as r:
            return json.loads(r.read().decode("utf-8") or "{}")
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "ignore")[:200]
        return {"code": e.code, "msg": body}
    except Exception as e:                       # noqa: BLE001
        return {"code": -1, "msg": str(e)}


def next_seq(dev):
    STATE[dev]["seq"] += 1
    return STATE[dev]["seq"]


def raw_ref():
    return hashlib.md5(str(random.random()).encode()).hexdigest()


# ---------------------------------------------------------------- 发事件
def emit(dev_sn, event_type, payload=None, when=None, channel=None, with_raw=True):
    """发一条事件。when 可传 datetime，用来补历史和制造断档场景。"""
    d = DEV_BY_SN[dev_sn]
    st = STATE[d["dev"]]
    with LOCK:
        COUNTER["n"] += 1
        n = COUNTER["n"]
    seq = next_seq(d["dev"])
    ref = raw_ref()
    ev = {
        "eventId": "EVT-%s-%05d" % (time.strftime("%m%d%H%M%S"), n),
        "sensorId": dev_sn,
        "sourceChannel": channel or d["ch"],
        "eventType": event_type,
        "deviceTime": now_iso(when),
        "receivedAt": now_iso(),
        "tsSource": 3,                     # 3 = 网关打戳（G 线恒定）
        "protocolVersion": "gw-1.2",
        "seq": str(seq),
        "payload": payload or {},
        "rawRef": ref,
    }
    r = post("/internal/iot/v1/events/batch", {"sourceChannel": channel or d["ch"], "events": [ev]})
    ok = r.get("code") == 0
    acc = (r.get("data") or {}).get("accepted", 0) if ok else 0
    if ok:
        if acc:
            log("✓  %s（%s）→ %s  seq=%s" % (d["name"], dev_sn, event_type, seq))
        else:
            reason = json.dumps(r.get("data", {}), ensure_ascii=False)[:300]
            log("○  %s（%s）→ %s  未被收下：%s" % (d["name"], dev_sn, event_type, reason))
    else:
        log("✗  %s（%s）→ %s 被 M 层拒绝：%s" % (d["name"], dev_sn, event_type, r.get("msg")))

    if with_raw:
        frame = base64.b64encode(b"\xAA\xBB" + bytes([seq % 256]) + dev_sn.encode()[:12]).decode()
        post("/internal/iot/v1/raw-frames", {"sourceChannel": channel or d["ch"], "frames": [{
            "rawRef": ref, "sensorId": dev_sn, "sourceChannel": channel or d["ch"],
            "receivedAt": now_iso(), "rawFrame": frame, "frameBytes": len(frame),
            "gatewayDevId": "GW-SIM-01", "protocolVersion": "gw-1.2",
        }]})
    return {"ok": bool(ok and acc), "accepted": acc,
            "msg": None if ok else r.get("msg")}


def emit_status():
    """状态快照：电量、信号、在线。走 I-02，覆盖写 device_tech_cache。"""
    statuses = []
    for d in DEVICES:
        st = STATE[d["dev"]]
        statuses.append({
            "sensorId": d["sn"],
            "sourceChannel": d["ch"],
            "online": st["online"],
            "lastSeenAt": now_iso(),
            "batteryMv": st["battery"],
            "csq": st["csq"],
            "lastSeq": st["seq"],
            "lostCnt": 0,
            "protocolVersion": "gw-1.2",
            "reportedAt": now_iso(),
        })
    post("/internal/iot/v1/device-status/batch", {"sourceChannel": "OWN_GW", "statuses": statuses})


# ---------------------------------------------------------------- 场景
def init_state():
    for d in DEVICES:
        STATE[d["dev"]] = dict(
            seq=random.randint(100, 400),
            battery=random.randint(3450, 3900),
            csq=random.randint(16, 28),
            online=True,
            door_open_at=None,
            leak_at=None,
            power_off_at=None,
        )


def tick(fast=False):
    """一轮随机行为。概率按 5 秒一轮算，故意压低——真设备的动静比演示脚本小得多。"""
    speed = 3.0 if fast else 1.0
    g = DEV_BY_SN["3F7A9C21"]
    st = STATE["T-G-01"]

    # 门磁：开 → 过一会儿关
    if st["door_open_at"] is None and random.random() < 0.10 * speed:
        st["door_open_at"] = time.time()
        st["battery"] -= random.randint(1, 3)
        emit("3F7A9C21", "DOOR_OPEN", {"deviceTypeRaw": 1, "rawEventCode": 1, "batteryMv": st["battery"]})
    elif st["door_open_at"] and time.time() - st["door_open_at"] > random.randint(20, 50):
        st["door_open_at"] = None
        emit("3F7A9C21", "DOOR_CLOSE", {"deviceTypeRaw": 1, "rawEventCode": 2, "batteryMv": st["battery"]})

    # 水浸：漏了就等它自己解除
    w = STATE["T-G-04"]
    if w["leak_at"] is None and random.random() < 0.02 * speed:
        w["leak_at"] = time.time()
        emit("3F7A9C24", "WATER_LEAK", {"deviceTypeRaw": 4, "rawEventCode": 1, "batteryMv": w["battery"]})
    elif w["leak_at"] and time.time() - w["leak_at"] > random.randint(40, 100):
        w["leak_at"] = None
        emit("3F7A9C24", "WATER_LEAK_CLEAR", {"deviceTypeRaw": 4, "rawEventCode": 2, "batteryMv": w["battery"]})

    # S 线水浸：裸 TCP 直连，没有网关打戳，tsSource 恒 2
    s = STATE["T-S-01"]
    if s["leak_at"] is None and random.random() < 0.012 * speed:
        s["leak_at"] = time.time()
        emit("MIR:861234567890123", "WATER_LEAK",
             {"deviceTypeRaw": 4, "rawEventCode": 1, "batteryMv": s["battery"]}, channel="MIR_CELL")
    elif s["leak_at"] and time.time() - s["leak_at"] > random.randint(40, 90):
        s["leak_at"] = None
        emit("MIR:861234567890123", "WATER_LEAK_CLEAR",
             {"deviceTypeRaw": 4, "rawEventCode": 2, "batteryMv": s["battery"]}, channel="MIR_CELL")

    # SOS：真设备极少误触，这里给 0.4%。必须由 SOS 那台发，门磁发不出来。
    if random.random() < 0.004 * speed:
        st_sos = STATE["T-G-03"]
        emit("3F7A9C23", "SOS_PRESS", {"deviceTypeRaw": 5, "rawEventCode": 1, "batteryMv": st_sos["battery"]})


def seed_history():
    """补几条历史，让页面一打开就有东西看。时间戳往回拨，走设备时间。"""
    now = datetime.now(CST)
    log("补历史：3 条（40 分钟前水浸、25 分钟前门磁、8 分钟前 SOS）")
    emit("3F7A9C24", "WATER_LEAK", {"deviceTypeRaw": 4, "rawEventCode": 1, "batteryMv": 3610},
         when=now - timedelta(minutes=40))
    emit("3F7A9C21", "DOOR_OPEN", {"deviceTypeRaw": 1, "rawEventCode": 1, "batteryMv": 3720},
         when=now - timedelta(minutes=25))
    emit("3F7A9C23", "SOS_PRESS", {"deviceTypeRaw": 5, "rawEventCode": 1, "batteryMv": 3720},
         when=now - timedelta(minutes=8))


# ---------------------------------------------------------------- 控制台
PAGE = """<!DOCTYPE html><html lang="zh-CN"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>设备模拟器 · 鹤仆云</title>
<style>
:root{--ink:#1b2430;--sub:#6b7784;--line:#e4eaef;--brand:#146b7a;--red:#d94b4b}
*{box-sizing:border-box}
body{margin:0;background:#eef2f5;color:var(--ink);font-family:-apple-system,"PingFang SC","Microsoft YaHei",sans-serif;font-size:14px}
.wrap{max-width:860px;margin:0 auto;padding:22px 18px 50px}
h1{font-size:20px;margin:0 0 4px}
.sub{color:var(--sub);font-size:12.5px;margin-bottom:18px}
.card{background:#fff;border:1px solid var(--line);border-radius:12px;padding:16px;margin-bottom:14px}
.card h2{font-size:14px;margin:0 0 12px;font-weight:600}
.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(120px,1fr));gap:9px}
button{border:1px solid var(--line);background:#fff;border-radius:9px;padding:11px 8px;font-size:13px;
  cursor:pointer;font-weight:500;transition:.15s}
button:hover{border-color:var(--brand);color:var(--brand)}
button.hot{background:#fdeaea;border-color:#f5c6c6;color:#c0392b}
button.warn{background:#fdf3e3;border-color:#f3ddb8;color:#a9741c}
button.ok{background:#e9f5ef;border-color:#c6e5d5;color:#2f9e68}
table{width:100%;border-collapse:collapse;font-size:13px}
th{text-align:left;color:var(--sub);font-weight:500;padding:6px 8px;border-bottom:1px solid var(--line);font-size:12px}
td{padding:7px 8px;border-bottom:1px solid #f0f4f6}
.mono{font-family:Consolas,monospace;font-size:12px}
.dot{display:inline-block;width:7px;height:7px;border-radius:50%;margin-right:5px}
.on{background:#2f9e68}.off{background:#b9c3cb}
#log{background:#16262f;color:#b8cdd6;border-radius:10px;padding:12px;font-family:Consolas,monospace;
  font-size:12px;line-height:1.75;height:210px;overflow-y:auto;white-space:pre-wrap}
.tip{font-size:12px;color:var(--sub);line-height:1.8}
</style></head><body><div class="wrap">
<h1>设备模拟器</h1>
<div class="sub">扮演网关和 7 台设备，往 M 层发真实格式的报文 · 控制台 <b>127.0.0.1:__PORT__</b></div>

<div class="card"><h2>按设备发事件</h2>
<div style="margin:0 0 10px;font-size:13px;color:#5a6b76">
  从哪台设备发：
  <select id="snsel" style="padding:6px 10px;border-radius:8px;border:1px solid #dfe8ec;
    font-size:13px;background:#fff;color:#1b2430;margin-left:6px"></select>
  <span style="font-size:12px;color:#8b9aa1">　验收新绑的设备时记得切到这里</span>
</div>
<div id="devhint" style="font-size:12.5px;color:#8b9aa1;line-height:1.7;margin-bottom:10px"></div>
<div class="grid" id="acts"></div>
<div id="tmsg" style="font-size:12.5px;line-height:1.7;margin-top:10px;min-height:18px"></div>
</div>

<div class="card"><h2>设备状态</h2>
<table id="devs"><thead><tr><th>设备</th><th>设备编号</th><th>类型</th><th>电量</th><th>信号</th><th>seq</th></tr></thead><tbody></tbody></table></div>

<div class="card"><h2>最近动作</h2><div id="log"></div></div>

<div class="card tip">
  数据流：本页 → <span class="mono">POST /internal/iot/v1/events/batch</span> → M 层入库 iot_event
  → 告警引擎（入库即触发，30 秒兜底扫一次）→ alarm → 家属端。<br>
  关掉这个窗口不影响已产生的数据；想看新数据得让它继续跑。
</div>
</div>
<script>
// 每种设备只显示它发得出的动作。门磁没有水浸探针，界面上就不该出现"漏水"按钮——
// 以前 8 个按钮常驻，选门磁也能点漏水，纯属误导（2026-11-09 改）。
var ACTIONS={
  DOOR:[['door','🚪 门打开了','warn'],['door_close','✅ 门关上了','ok']],
  WATER:[['leak','💧 漏水了','hot'],['leak_clear','✅ 漏水已止','ok']],
  SOS:[['sos','🆘 按下 SOS','hot']],
  SMOKE:[['smoke','🔥 烟感报警','hot'],['smoke_clear','✅ 烟雾已散','ok']],
  VIB:[['vib','💧 用水活动（不告警）',''],['vib_stop','用水停止','']]
};
var TYPE_NAME={DOOR:'门磁',WATER:'水浸',SOS:'紧急按钮',SMOKE:'烟感',VIB:'水振'};
var COMMON=[['power_off','⚡ 设备断电','warn'],['power_on','🔌 设备来电','ok'],['status','📊 上报心跳','']];
var DEVS=[];

function curDev(){
  var sel=document.getElementById('snsel');
  var sn=sel?sel.value:'';
  return DEVS.filter(function(x){return x.sn===sn;})[0]||DEVS[0]||null;
}

function renderBtns(){
  var d=curDev(); if(!d) return;
  var list=(ACTIONS[d.type]||[]).concat(COMMON);
  document.getElementById('acts').innerHTML=list.map(function(a){
    return '<button class="'+a[2]+'" onclick="t(\\''+a[0]+'\\')">'+a[1]+'</button>';
  }).join('');
  var hint='当前：<b>'+d.name+'</b> · '+d.sn+' · '+(TYPE_NAME[d.type]||d.type);
  if(d.type==='VIB'){
    hint+='　<span style="color:#a9741c">水振报的是"用水活动"，不是告警——只进基线和组合研判，不会弹窗</span>';
  }
  hint+='<br><span style="color:#8b9aa1">恢复类事件（门关上 / 漏水已止 / 来电）只关闭"还没恢复"的那条告警，不会新建告警。</span>';
  document.getElementById('devhint').innerHTML=hint;
}

function t(k){
  var sel=document.getElementById('snsel'), sn=sel?sel.value:'';
  var box=document.getElementById('tmsg');
  box.style.color='#5a6b76'; box.textContent='发送中…';
  fetch('/api/trigger?k='+k+(sn?'&sn='+encodeURIComponent(sn):''),{method:'POST'})
    .then(function(r){return r.json()})
    .then(function(d){
      box.style.color = d.code===0 ? '#2f9e68' : '#c0392b';
      box.textContent = (d.code===0?'✓ ':'✗ ') + (d.msg||'');
      poll();
    })
    .catch(function(e){box.style.color='#c0392b';box.textContent='✗ 请求失败：'+e});
}

var SN_FILLED=false;
function fillSn(list){
  if(SN_FILLED||!list) return;
  DEVS=list;
  document.getElementById('snsel').innerHTML=list.map(function(x){
    return '<option value="'+x.sn+'">'+x.name+' · '+x.sn+'</option>';
  }).join('');
  document.getElementById('snsel').onchange=renderBtns;
  SN_FILLED=true;
  renderBtns();
}
function poll(){
  fetch('/api/state').then(function(r){return r.json()}).then(function(d){
    fillSn(d.devices);
    document.getElementById('devs').querySelector('tbody').innerHTML = d.devices.map(function(x){
      return '<tr><td><span class="dot '+(x.online?'on':'off')+'"></span>'+x.name+'</td>'+
      '<td class="mono">'+x.sn+'</td><td>'+(TYPE_NAME[x.type]||x.type)+'</td>'+
      '<td>'+x.battery+' mV</td><td>CSQ '+x.csq+'</td>'+
      '<td class="mono">seq '+x.seq+'</td></tr>';
    }).join('');
    const box=document.getElementById('log');
    const atBottom = box.scrollTop + box.clientHeight >= box.scrollHeight - 10;
    box.textContent = d.log.slice().reverse().join('\\n');
    if(atBottom) box.scrollTop = 0;
  })
}
poll(); setInterval(poll, 3000);
</script></body></html>"""


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *a):        # 关掉默认访问日志，别刷屏
        pass

    def _send(self, code, ctype, body):
        b = body.encode("utf-8") if isinstance(body, str) else body
        self.send_response(code)
        self.send_header("Content-Type", ctype + "; charset=utf-8")
        self.send_header("Content-Length", str(len(b)))
        self.end_headers()
        self.wfile.write(b)

    def do_GET(self):
        if self.path in ("/", "/index.html"):
            self._send(200, "text/html", PAGE.replace("__PORT__", str(PORT)))
        elif self.path.startswith("/api/state"):
            with LOCK:
                devs = [dict(name=d["name"], sn=d["sn"], type=d["type"],
                             battery=STATE[d["dev"]]["battery"],
                             csq=STATE[d["dev"]]["csq"], seq=STATE[d["dev"]]["seq"],
                             online=STATE[d["dev"]]["online"]) for d in DEVICES]
                lg = list(LOG)
            self._send(200, "application/json",
                       json.dumps({"devices": devs, "log": lg}, ensure_ascii=False))
        else:
            self._send(404, "text/plain", "not found")

    def do_POST(self):
        if not self.path.startswith("/api/trigger"):
            return self._send(404, "text/plain", "not found")
        k = ""
        sn = None
        if "k=" in self.path:
            k = self.path.split("k=", 1)[1].split("&")[0]
        if "sn=" in self.path:
            sn = urllib.parse.unquote(self.path.split("sn=", 1)[1].split("&")[0])
        try:
            res = trigger(k, sn)
            body = json.dumps({"code": 0 if res.get("ok") else 1,
                               "msg": res.get("msg") or ""}, ensure_ascii=False)
            self._send(200, "application/json", body)
        except Exception as e:                   # noqa: BLE001
            self._send(500, "application/json", json.dumps({"code": 1, "msg": str(e)}, ensure_ascii=False))


def trigger(k, sn=None):
    """控制台按钮 → 一条事件。全部走真实接口，不直接改库。

    ★ <b>必须按设备类型把关</b>（2026-11-09 改）：门磁没有水浸探针，物理上就发不出
    WATER_LEAK；SOS 按钮也不该拿门磁的编号发。以前不校验，选「入户门磁」点 SOS
    照样生成 S1 告警——不是系统聪明，是模拟器在骗自己。

    返回 {"ok": bool, "msg": 给页面弹的话}
    """
    d = DEV_BY_SN.get(sn) if sn else DEVICES[0]
    if d is None:
        return {"ok": False, "msg": "未知设备编号 " + str(sn)}
    st = STATE[d["dev"]]
    raw = TYPE_RAW.get(d["type"], 1)

    allowed = [a[0] for a in ACTIONS.get(d["type"], [])]
    all_actions = {a[0]: a[1] for a in ACTIONS.get(d["type"], []) + COMMON_ACTIONS}
    if k not in allowed and k not in [a[0] for a in COMMON_ACTIONS]:
        return {"ok": False,
                "msg": "%s 是「%s」，没有这个动作。（%s 只能发：%s）" % (
                    d["sn"], d["name"], d["name"],
                    "、".join(all_actions.get(x, x) for x in allowed) or "只能断电/来电/心跳")}

    if k == "sos":
        r = emit(d["sn"], "SOS_PRESS", {"deviceTypeRaw": raw, "rawEventCode": 1, "batteryMv": st["battery"]})
    elif k == "leak":
        st["leak_at"] = time.time()
        r = emit(d["sn"], "WATER_LEAK", {"deviceTypeRaw": raw, "rawEventCode": 1, "batteryMv": st["battery"]})
    elif k == "leak_clear":
        st["leak_at"] = None
        r = emit(d["sn"], "WATER_LEAK_CLEAR", {"deviceTypeRaw": raw, "rawEventCode": 2, "batteryMv": st["battery"]})
    elif k == "door":
        st["door_open_at"] = time.time()
        r = emit(d["sn"], "DOOR_OPEN", {"deviceTypeRaw": raw, "rawEventCode": 1, "batteryMv": st["battery"]})
    elif k == "door_close":
        st["door_open_at"] = None
        r = emit(d["sn"], "DOOR_CLOSE", {"deviceTypeRaw": raw, "rawEventCode": 2, "batteryMv": st["battery"]})
    elif k == "power_off":
        st["online"] = False
        r = emit(d["sn"], "POWER_OFF", {"deviceTypeRaw": raw, "rawEventCode": 5})
    elif k == "power_on":
        st["online"] = True
        r = emit(d["sn"], "POWER_ON", {"deviceTypeRaw": raw, "rawEventCode": 6})
    elif k == "smoke":
        r = emit(d["sn"], "SMOKE", {"deviceTypeRaw": raw, "rawEventCode": 1, "batteryMv": st["battery"]})
    elif k == "smoke_clear":
        r = emit(d["sn"], "SMOKE_CLEAR", {"deviceTypeRaw": raw, "rawEventCode": 2, "batteryMv": st["battery"]})
    elif k == "vib":
        r = emit(d["sn"], "WATER_ACTIVITY", {"deviceTypeRaw": raw, "rawEventCode": 1, "batteryMv": st["battery"]})
    elif k == "vib_stop":
        r = emit(d["sn"], "WATER_ACTIVITY_STOP", {"deviceTypeRaw": raw, "rawEventCode": 2, "batteryMv": st["battery"]})
    elif k == "status":
        emit_status()
        log("○ 心跳已上报（%d 台）" % len(DEVICES))
        return {"ok": True, "msg": "已上报 " + str(len(DEVICES)) + " 台设备的心跳"}
    else:
        return {"ok": False, "msg": "未知指令 " + k}

    if not r.get("ok"):
        return {"ok": False, "msg": "M 层没接收：" + str(r.get("msg") or "看左侧日志")}
    # 恢复类事件不新建告警，只关闭"还没清过"的同类告警——没单子可关时看起来"没反应"，
    # 这里当场说清楚，省得以为是坏了。
    if k in ("door_close", "leak_clear", "power_on", "smoke_clear"):
        return {"ok": True, "msg": "已发送（%s）。恢复事件只关闭「%s」这一类、且还没恢复的告警，"
                                   "不新建告警，也不会碰别的类型" % (d["name"], d["name"])}
    if k in ("vib", "vib_stop"):
        return {"ok": True, "msg": "已记录为「用水活动」（%s）。这类事件不产生告警，"
                                   "只进基线和「水振＋电流」组合研判" % d["name"]}
    return {"ok": True, "msg": "已发送（%s · %s）" % (d["name"], d["sn"])}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--fast", action="store_true", help="快节奏（每轮 3 秒，概率翻倍）")
    ap.add_argument("--seed", action="store_true", help="启动先补历史告警")
    ap.add_argument("--quiet", action="store_true", help="只开控制台，不自动产生事件")
    args = ap.parse_args()

    init_state()
    print("=" * 62)
    print("  鹤仆云 · 设备模拟器")
    print("  目标 M 层：%s" % BASE)
    print("  控制台   ：http://127.0.0.1:%d" % PORT)
    print("  节奏     ：%s" % ("静默（只响应按钮）" if args.quiet else ("快" if args.fast else "正常")))
    print("=" * 62)

    # 先探活。不能用空 statuses 去试 I-02——后端会按"空批"返回 400，
    # 那是正确的校验行为，不是服务没起来。
    try:
        with OPENER.open(BASE + "/health", timeout=8) as h:
            h.read()
    except Exception as e:                       # noqa: BLE001
        print("!! M 层没响应（%s）。先确认 :8080 起来了。" % e)
        print("   启动命令：G:\\aomin-dev\\app\\aomin-gateway\\start-gateway.bat")
        return
    log("✓ M 层在线：" + BASE)

    if args.seed:
        seed_history()
    emit_status()

    srv = HTTPServer(("127.0.0.1", PORT), Handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()

    step = 3 if args.fast else 5
    n = 0
    try:
        while True:
            time.sleep(step)
            n += 1
            if not args.quiet:
                tick(args.fast)
            if n % 6 == 0:
                emit_status()
            # 电量缓慢下降，模拟真实电池
            for d in DEVICES:
                STATE[d["dev"]]["battery"] -= random.randint(0, 1)
                STATE[d["dev"]]["csq"] = max(12, min(31, STATE[d["dev"]]["csq"] + random.randint(-1, 1)))
    except KeyboardInterrupt:
        print("\n已停止。")


if __name__ == "__main__":
    main()