2. HSMS 单会话连接建立与心跳维持

任务说明

本案例实现一个 HSMS 主动模式（Active）客户端：向设备（Passive）发起 TCP 连接，完成 Select 握手进入 Selected 状态，随后周期性发送 Linktest 心跳维持连接，并在连接断开时自动重连。HSMS 报文头固定 10 字节，关键字段为 SessionID(2)、HeaderByte2/3(2)、PType(1)、SType(1)、SystemBytes(4)。

环境与依赖

• Python 3.8+，仅用标准库 socket、struct、threading、time。
• 设备侧需开启 HSMS Passive 监听端口（示例用 5000）。

核心实现

SType 约定：Select.req=1、Select.rsp=2、Deselect.req=3、Linktest.req=5、Linktest.rsp=6。握手流程为：连接后发 Select.req，收到 Select.rsp 且状态码为 0 即进入 Selected；之后每 5 秒发一次 Linktest.req，收到 Linktest.rsp 视为心跳成功；若发送或接收超时/异常，则关闭套接字并重连。

import socket, struct, threading, time

HOST, PORT = "127.0.0.1", 5000
LINKTEST_INTERVAL = 5.0
RECV_TIMEOUT = 3.0

def build_header(s_type, sys_bytes, session=0xFFFF):
    # SessionID, HeaderByte2, HeaderByte3, PType, SType, SystemBytes
    return struct.pack(">HBBBI", session, 0, 0, 0, s_type, sys_bytes)

def send_msg(sock, s_type, sys_bytes):
    sock.sendall(build_header(s_type, sys_bytes))

def recv_msg(sock):
    head = b""
    while len(head) < 10:
        chunk = sock.recv(10 - len(head))
        if not chunk:
            raise ConnectionError("closed")
        head += chunk
    session, _, _, ptype, stype, sysb = struct.unpack(">HBBBI", head)
    return stype, sysb

def run_session(sock):
    sock.settimeout(RECV_TIMEOUT)
    send_msg(sock, 1, 1)                 # Select.req
    stype, _ = recv_msg(sock)
    if stype != 2:
        raise ConnectionError("no Select.rsp")
    print("Selected")
    last = time.time()
    while True:
        if time.time() - last >= LINKTEST_INTERVAL:
            send_msg(sock, 5, int(time.time()) & 0xFFFFFFFF)  # Linktest.req
            stype, _ = recv_msg(sock)
            if stype != 6:
                raise ConnectionError("no Linktest.rsp")
            print("Linktest ok")
            last = time.time()
        else:
            time.sleep(0.2)

def main():
    while True:
        try:
            s = socket.create_connection((HOST, PORT), timeout=5)
            run_session(s)
        except Exception as e:
            print("disconnect:", e, "-> reconnect in 3s")
            time.sleep(3)

if __name__ == "__main__":
    main()

运行与验证

1. 在设备侧启动 HSMS Passive 监听（端口 5000），或先用任意能回 Select.rsp/Linktest.rsp 的模拟器。
2. 运行 python hsms_active.py。
3. 通过条件：控制台先打印 Selected，随后每约 5 秒打印一次 Linktest ok。
4. 断线重连验证：手动关闭设备侧连接，客户端应打印 disconnect: ... -> reconnect in 3s，并在设备恢复后重新打印 Selected。