简介:这是以Python语言实现的SECS/GEM半导体通信协议开源项目,面向设备自动化工程师、协议研究与工业上位机开发者,重点展示SECS I与SECS II层次下的数据编解码、消息交互、文件传输及事件通知机制,并涵盖了同步与定时处理、异常与错误检测等工程细节。压缩包共含99个文件,包含46个Python源代码、38个RST格式说明文档,以及若干YAML、Shell、配置文件与许可证等辅助内容,整体大小约160KB。目前已有1298人参与学习,适合希望从源代码层面掌握SECS协议实现思路的读者。通过梳理源码可以清晰看到common、secs、hsms、gem等模块的分工,包括HSMS连接管理、SECS消息打包解包、GEM设备与主机Handler的协作模式;tests目录下还提供了针对数据项、回调、连接状态机等场景的测试用例,可直接运行验证协议行为。目录结构清晰且包含示例脚本,便于按模块检索和二次开发,为自主实现或调试SECS通信系统提供工程范本。
1. 读懂 secs 协议,从 secsgem-master 落地开始
如果你做半导体或面板行业的设备联网集成,多半见过这样的场景:现场设备明明支持 SECS,但 MES/EAP 一直连不上,厂商售后丢来一份几百页的协议文档,光看报文流就晕了。secs 协议其实不是单个标准,而是 SECS-I/HSMS 传输层、SECS-II 消息层和 GEM 行为模型的集合,自己从头实现一遍,光时序磋商就要折腾几周。而 secsgem-master 这个基于 Python 的开源实现,把 HSMS 通信、SECS-II 编解码、GEM 状态机全部封装好了,适合快速搭出设备端或主机端原型。这篇按我平时接项目的顺序写:先立协议骨架,再写消息收发,最后把参数调优和验证手段讲透。
2. SECS 协议的层次模型与 secsgem-master 的通信骨架
2.1 SECS 协议家族:传输层、消息层和应用层各管什么
SECS 标准拆开是 SEMI E4(SECS-I,RS-232 串口传输)、E37(HSMS,TCP/IP 传输)、E5(SECS-II 消息标准)和 E30(GEM 通用设备模型)。实际项目里,新设备几乎全部走 HSMS,即设备端作为 TCP 服务端监听 5000~50000 端口的连接,主机端(Host)主动连入后,双方按长度前缀 + 消息体的帧格式交换数据。secsgem-master 对这种层次做了很干净的映射:HsmsHandler管传输,SecsMessage/DataItem管编码,GemEquipmentHandler把 GEM 行为约束成可执行的流程。
SECS 标准族与 secsgem 的对应关系大致如下:
| 标准号 | 层级 | 传输载体 | secsgem 对应 |
|---|---|---|---|
| E4 | SECS-I | RS-232 | 主推 HSMS,串口仅兼容 |
| E37 | HSMS | TCP/IP | HsmsSettings / HsmsHandler |
| E5 | SECS-II | 消息格式 | SecsMessage / Item |
| E30 | GEM | 设备行为 | GemEquipmentHandler |
值得强调一个反直觉的点:SECS 的消息体里,数据项(Data Item)的二进制布局并不像 JSON 那样能直接看字段名。比如A表示 ASCII 字符串,B是二进制字节,L是列表,U1/U2/U4/U8表示无符号整数宽度。secsgem 内部用一套 Item 类来维护这些类型,收到字节流后先按 SECS-II 规则拆出每个 Item,再按 SML(SECS Message Language)文本或 Python 字典转成可读结构。你在代码里看到{'L': [{'A': 'secsgem'}, ...]}这种字典,实际就是 SECS 消息的可视化形态,底层与报文格式一一对应。
2.2 secsgem-master 里的三个关键类:设备端、主机端和消息体
secsgem 库对外的入口很集中,分设备端和主机端两条链路。设备端用secsgem.gem.GemEquipmentHandler,负责应答主机下发的 SxFy 请求,并主动上报报警(S5F1)、数据(S6F11)等消息;主机端用GemHostHandler,从设备侧拉取数据。两个 handler 都继承自HsmsGemHandler,所以 HSMS 的建立链路(主动连接、被动监听、链路测试 T5/T6/T8)是共享的。
核心类里还要认识SecsMessage:它由一个header(10 字节,含 Stream/Function、W 位、Session ID)和一个data(Item 树)组成。收发消息时,不需要手工拼字节:secsgem 暴露send_message(message, wait_for_response=True)方法,返回值就是响应消息,编码解码全部自动完成。一个小技巧是:在调试环境里先打印message.sml,得到的就是可直接人工读的 SML 字符串,这条字符串可以反向喂给解析器,非常利于写测试用例时校对报文内容。
2.2.1 最小可跑的 HSMS 连接代码(设备端)
下面这段代码是设备端最小骨架,作用是启动一个 HSMS 服务端,等待主机连入并答复链路磋商:
from secsgem.hsms import HsmsSettings, HsmsConnectionMode from secsgem.gem import GemEquipmentHandler settings = HsmsSettings( device_ip="0.0.0.0", device_port=5000, active=False, # 被动模式,设备作为服务端 session_id=1, # 设备会话 ID,主机靠它识别设备 connect_mode=HsmsConnectionMode.PASSIVE, name="equip-a", ) def on_connected(): print("host connected!") eq = GemEquipmentHandler(settings) eq._on_connected = on_connected eq.enable()active=False就是被动方,等价于设备开放 5000 端口;session_id在 HSMS 里相当于设备地址,多设备接入时用于路由。eq.enable()内部会把GemServer和HsmsServer两个组件拉起来,并启动 GEM 初始化流程。运行后,从任意主机端工具连接IP:5000就能看到握手。注意,这段代码只提供传输通道,还没注册任何 Stream Function,所以主机发S1F1(查询设备)时会收到默认的异常响应,这正好说明协议先立骨架、再补业务的路子是通的。
2.3 为什么选 secsgem-master 而不是自己手写报文
手写 SECS 报文不是不能做,但在生产环境中不划算。第一个原因是位级细节太多:长度前缀用 4 字节,实际只取低 24 位,超过 16MB 的要分段;消息头的 10 字节里,Session ID 与 Stream/Function 的位域划分在不同标准里还有差异。第二个原因是 GEM 状态机:设备上电后要先完成 Communication State 从 DISABLED 到 ENABLED、再到等待主机接入的迁移,在 S1F13 磋商模式、S1F14 返回设备信息这些环节上,接管流程比想象中繁琐。第三方库的价值在于把这些“无业务价值但有协议价值”的部分稳定实现,项目组只需要关心 Stream/Function 和数据处理。
当然,选 secsgem 也有代价:Python 的性能上限低,高频 FDC(Fault Detection and Classification)数据上报场景,单消息几千个数据点的话,建议只用于消息转发或原型验证,生产环境的数据面仍然交给 C++/Java 模块。不过对大多数 MES/EAP 集成项目,secsgem-master 的吞吐已经足够,而且它的事件回调模型和send_and_waitfor_message阻塞接口都非常直观,维护成本比自研低很多。
3. 用 secsgem-master 实现 S1F13 磋商与数据上报的完整流程
3.1 消息收发机制:handler 与回调
secsgem 的消息处理机制是事件注册加自动路由。每个设备端 class 里都有_message_handlers字典,按(stream, function)键(如(1, 13))把回调函数挂上去。收到消息后,框架解析 Stream/Function,若找到对应回调就执行,找不到则依据标准走默认响应。这意味着业务代码不需要写一个大 if-else 分派器,只需要为需要的功能注册 handler。
这种设计的另一个好处是职责分离:传输层(连接、心跳)和业务层(S1F13、S6F11、S2F41)互不干扰。和设备厂商对接时,可以先只注册 S1F13,跑通握手再接其他功能;也可以一次注册多个 handler,然后逐一调试。调试时优先级最高的手段是:在 handler 里打印message.sml,它会把二进制内容变成S1F13 W这样的文本。
3.2 设备端代码:S1F13/S1F14 握手应答
SECS 连接真正意义上的第一个业务消息是 S1F13(Establish Communications Request),设备必须回 S1F14。下面把设备端补全,让它能正常磋商:
from secsgem.secs import SecsMessage from secsgem.hsms import HsmsSettings, HsmsConnectionMode from secsgem.gem import GemEquipmentHandler settings = HsmsSettings( device_ip="0.0.0.0", device_port=5000, active=False, session_id=1, connect_mode=HsmsConnectionMode.PASSIVE, ) eq = GemEquipmentHandler(settings) def on_s1f13(message, context): print("from host:", message.sml) eq.send_message( SecsMessage(stream=1, function=14, reply_expected=False, data={ "MDLN": "secsgem", "SOFTWARE_VERSION": "1.0.0", }) ) eq._message_handlers[(1, 13)] = on_s1f13 eq.enable()关键点在reply_expected和data的写法。reply_expected=False表示不期待对方再回应,而S1F14的 data 部分是一个设备信息列表,SECS-II 规定为L[3-9]:MDLN(设备型号)、SOFTWARE_VERSION、VERSION_ID 等,用字典展开即可;字段少一个都能通过类型校验,但厂商设备管理端往往要求完整。回复时 Stream/Function 会自动设为 1/14,不需要手工指定。
3.3 主机端代码:主动连接并读取设备上报
与设备端相对,主机端常做成轮询加订阅的混合模式。主动上报的场景,比如设备发送异常报警S5F1,或 FDC 数据S6F11,主机端注册(5, 1)/(6, 11)回调即可。下面是从主机端发起 S1F13 并等待响应的最小实现:
from secsgem.hsms import HsmsSettings, HsmsConnectionMode from secsgem.gem import GemHostHandler from secsgem.secs import SecsMessage host_settings = HsmsSettings( device_ip="192.168.1.20", # 设备侧 IP device_port=5000, active=True, session_id=1, connect_mode=HsmsConnectionMode.ACTIVE, name="eap-host", ) host = GemHostHandler(host_settings) host.connect() def on_s6f11(message, context): print(message.sml) host.send_message(SecsMessage(stream=6, function=12, reply_expected=False, data={})) host._message_handlers[(6, 11)] = on_s6f11 resp = host.send_and_waitfor_message( SecsMessage(stream=1, function=13, reply_expected=True), 120 ) print("S1F14:", resp.sml)主机端最常踩的坑是reply_expected:默认True时,若设备 60 秒内不回消息会抛TimeoutError,若不期望回复却用了send_and_waitfor_message,会一直等下去。代码里 120 秒是协商超时,比 HSMS 默认 T3(响应超时)大,但协议栈内部还有 T3 控制,两边参数要一致。resp.sml打印出的 S1F14 内容通常形如L,3: [A: "secsgem", A: "1.0.0", L,0: EMPTY],比对数据手册时一目了然。
3.4 GEM 状态机:从 DISABLED 到 COMMUNICATING 的迁移条件
只靠消息应答还不够,GEM 标准要求设备端维护一个状态:DISABLED→INIT→COMMUNICATING。上电初始化时,假设默认disable(),那么设备不会响应任何消息;enable()之后进入INIT,会自动发送S1F13向主机请求磋商(如果配置了 active 模式)。secsgem 已经内置这一套:eq.enable()会触发状态流转,主动模式从主机侧发送 S1F13,被动模式则等主机来磋商。
常用的控制接口就三个:eq.enable()启动通信、eq.disable()停止并断开链路、eq.is_communicating查询连接状态。实际设备中,很多协议调试问题的根因不是消息内容错,而是状态机没跑起来,比如设备端忘了 enable,主机侧永远收不到任何数据。调试时先用print(eq.is_communicating)确认链路状态,再去看 handler 有没有生效,能省去大量排查时间。
4. SML 解析、超时参数与设备接入中的常见坑
4.1 SML 语法:测试与调试的最佳伴侣
SML(SECS Message Language)是 SEMI E5 附录里定义的消息描述文本,比如S1F13 W或带 data 的S1F14 L,3: [A: "设备", A: "1.0", L,0: EMPTY]。secsgem 里所有 SecsMessage 都能用sml()方法转成此格式,也可以直接用 SML 字符串构造消息,这对写单测极其方便:
from secsgem.secs import SecsMessage msg = SecsMessage.sml_load("S1F14 W L,3: [A: 'ML', A: '1.0', L,0: EMPTY]")上面构造出来的msg的 stream/function 为 1/14,回复时把msg交给eq.send_message(msg)即可。sml_load的好处是:当你需要模拟一个具体报文时,不去数二进制位数,直接从日志里复制一行文本就能复现。这个能力用于复现现场问题很有价值,设备厂商往往只给你一段十六进制日志,你可以先转成 SML,再调 secsgem 构造一个等价的SecsMessage发出去演练。
4.2 HSMS 连接参数与 T3/T5/T6/T8 的设定建议
HSMS 连接超时参数比较多,很多协议对接问题都出在参数不一致。下表是 secsgem 中HsmsSettings里常用的参数及参考值:
| 参数 | 参考值(秒) | 作用 | 建议 |
|---|---|---|---|
| t3 | 45 | 等待响应消息的超时 | 与主机侧一致,不小于 45 |
| t5 | 10 | 建立连接后,等待远端消息的最大间隔 | 不小于 10,防止误判断链 |
| t6 | 5 | 控制消息(如 S1F13)等待响应的超时 | 5~10 |
| t7 | 10 | 链路空闲后发送 Linktest 的间隔 | 10~30 |
| t8 | 5 | 单条消息接收超时 | 5,防止丢字节挂死 |
secsgem 里 T 参数命名方式与 SEMI 标准一致,直接以关键字传进HsmsSettings即可(如t3=30, t7=15)。要注意的是,把 t7 设为 0 表示禁用超时,很多协议栈对心跳有独立处理,乱禁会导致连接假死。调试主机端连不上设备时,第一看t5、t7;设备端收不到消息时,看t8。这个排查顺序在半导体行业比较通用,建议直接刻进团队 check-list。
4.3 三个高频坑:消息长度、数据类型与 reply_expected
第一个坑是消息长度超限。SECS-II 消息体(含 header)不能超过 16,777,215 字节(24 bit)。FDC 上报动辄几百字节还好,但若用B类型塞大量原始数据,比如 20 MB 的 trace 数据直接封装成一条消息,必然编码失败。常规做法是拆包:按max_message_length切分,或者用多条S6F1流式传输。
第二个坑是数据类型不匹配。SECS 的数据项有严格位宽,U8与U4、I1与I2不能混。secsgem 在send_message时会做类型校验,校验不过抛TypeError。常见错误是想用 Python 原生int表示 64 位数据,然后 SECS 端用U8,直接用字典{'U8': 123}是没问题的;但如果后端解析的期望是U4,就会出现长度不匹配。解决办法是先用sml文本把期望结构打出来,再按字段逐个对。
第三个坑是回包时机不对。在 handler 里收到请求(比如 S6F11)后,必须在一个会话上下文里回 ACK(S6F12),否则对端可能认为控制超时并重复发送。secsgem 的 handler 是同步执行的,若在回调里做耗时 IO(如写数据库),建议先回 ACK、再异步处理数据。还有一个细节:如果reply_expected=False,你主动回包后对端并不会再给你任何响应,别期待返回消息。
5. 进阶技巧:无硬件验证协议栈,自写回环与报文检查
最后给一个实战技巧:在没有实体设备或主机时,怎么验证整套 secsgem 通信栈?方法就是自写回环,一台机器上同时跑设备端和主机端,设备端作为服务端监听本机端口,主机端作为客户端连到127.0.0.1,然后通信。
import time import threading from secsgem.secs import SecsMessage from secsgem.gem import GemEquipmentHandler, GemHostHandler from secsgem.hsms import HsmsSettings, HsmsConnectionMode DEVICE_PORT = 5000 def run_equipment(): eq_settings = HsmsSettings( device_ip="0.0.0.0", device_port=DEVICE_PORT, active=False, session_id=1, connect_mode=HsmsConnectionMode.PASSIVE, name="loop-equip", ) eq = GemEquipmentHandler(eq_settings) def on_s1f13(msg, ctx): eq.send_message(msg.response({"L": [{"A": "loop-back"}, {"A": "v1.0"}, {"L": []}]})) eq._message_handlers[(1, 13)] = on_s1f13 eq.enable() def host_probe(): host_settings = HsmsSettings( device_ip="127.0.0.1", device_port=DEVICE_PORT, active=True, session_id=1, connect_mode=HsmsConnectionMode.ACTIVE, name="loop-host", ) host = GemHostHandler(host_settings) host.connect() resp = host.send_and_waitfor_message( SecsMessage(stream=1, function=13, reply_expected=True), timeout=10 ) assert resp.function == 14 print(resp.sml) t = threading.Thread(target=run_equipment, daemon=True) t.start() time.sleep(0.5) host_probe()这个回环最大的价值是隔离问题:如果连本机回环都跑不通,问题必然出在配置参数;如果回环通了但真机不通,那多半是设备厂商的协议兼容性问题。细节上要注意设备端 handler 里用msg.response({...})自动生成回复消息,它会让 Stream/Function 自动匹配(S1F13 回 S1F14),reply_expected也会自动设置为 False。
推开到进阶级用法:生产环境调试时,不必在全链路上打断点,直接启一个GemHostHandler连向可疑设备,然后用on_message_received插桩打所有消息的 SML,就能判断是设备不回、回错格式,还是主机端线程卡死。再往前一步,可以把SecsMessage的sml字符串写入日志审计系统。SECS 协议带时间戳的 SML 日志,比抓包更贴近业务语义,适合做 EAP 的离线回放测试。回放时用sml_load逐条喂给 handler,不必等待真实设备响应。这条无硬件回环链路建议写进 pytest 用例,设备厂商更新固件版本之后自动回归一遍,抓到的第一个失败往往不是参数,而是厂商悄悄改掉的某个报文字段。
本文还有配套的精品资源,点击获取