离线联调利器:用 Python secsgem 搭建 GEM300 模拟器,不依赖设备也能跑通 E87 / E90
摘要:FAT 现场设备放鸽子、硬件还没到货、PLC 还在调试——EAP 工程师最痛苦的莫过于“巧妇难为无米之炊”。本文手把手教你用 Python + secsgem 搭建一个 GEM300 模拟器,在没有任何真实设备的情况下,离线跑通 HSMS 握手、E87 Carrier 装卸、E90 Substrate 追踪全流程。代码可直接复制运行,是EAP 开发者的居家旅行必备良药。
一、为什么你需要一个模拟器?
在 Fab 项目里,EAP 工程师的进度往往被设备卡住:
- 设备商延迟交付,FAT 排期压到你头上
- 硬件到了但 PLC 还没联调,状态机跑不起来
- 想测试异常场景(断网、脏数据),但设备商不让你随便折腾
- 联调时出了问题,分不清是设备的问题还是你代码的问题
有一个模拟器的好处:
二、环境准备
2.1 安装 secsgem
pip install secsgem
secsgem 是一个纯 Python 实现的 SECS/GEM 通信库,支持 HSMS-SS、SECS-I 等协议,GEM300 的 E87/E90 对象模型都有现成封装。
2.2 确认 Python 版本
python –version
三、第一步:HSMS 握手(先让灯亮起来)
先写一个最小的 HSMS 客户端,连上模拟设备,确认通信通了。
import secsgem.hsms
创建 HSMS 连接
参数说明:
address:设备IP
port:HSMS端口(通常5000)
session_id:会话ID(通常1)
connect_mode:主动连接(ACTIVE)还是被动监听(PASSIVE)
client = secsgem.hsms.HsmsClient(
address = "127.0.0.1", #模拟器地址
port = 5000,
session_id = 1,
connect_mode = secsgem.hsms.ConnectMode.ACTIVE
)
连接
client.enable()
发送S1F13(Are You There?)
S1F13是GEM标准里的“心跳”消息,用来确认设备在线
response = client.send_and_waitfor_response(
secsgem.hsms.HsmsStreamFunction(1,13,[])
)
print(f"S1F13 Response:{response}")
client.disable()
预期结果:
- S1F13 返回 S1F14,设备回复COMM_ACK = 0(通信已建立)
- S1F1 返回 S1F2,设备状态变为ONLINE
⚠️ 如果你没有现成的模拟器,可以用 secsgem 自带的 HsmsServer搭一个最简单的回环服务,或者用开源项目gem300-simulator之类的工具。
四、第二步:模拟 E87 Carrier 装卸
E87 的核心是 LoadPort 状态机和 Carrier 对象。下面模拟一个完整的 Carrier 装卸流程。
4.1 定义 LoadPort 状态
from enum import Enum
class LoadPortState(Enum):
NOT_ACCESSED = "NOT_ACCESSED"
READY_TO_LOAD = "READY_TO_LOAD"
CARRIER_PRESENT = "CARRIER_PRESENT"
CLAMPED = "CLAMPED"
IN_ACCESS = "IN_ACCESS"
CARRIER_COMPLETE = "CARRIER_COMPLETE"
READY_TO_UNLOAD = "READY_TO_UNLOAD"
class LoadPort:
def init(self,port_id:str):
self.port_id = port_id
self.state = LoadPortState.NOT_ACCESSED
self.carrier_id = None
self.slot_map = {}
def place_carrier(self, carrier_id: str) –> bool:
"""模拟放盒"""
if self.state != LoadPortState.READY_TO_LOAD:
return False
self.carrier_id = carrier_id
self.state = LoadPortState.CARRIER_PRESENT
return True
def clamp(self) –> bool:
"""模拟夹紧"""
if self.state != LoadPortState.CARRIER_PRESENT:
return False
self.state = LoadPortState.CLAMPED
return True
def start_access(self) –> bool:
"""模拟开盖"""
if self.state != LoadPortState.CLAMPED:
return False
self.state = LoadPortState.IN_ACCESS
return True
def get_state(self) –> str:
return self.state.value
4.2 模拟 CIV (Carrier ID Verification)
def civ_check(carrier_id:str,expected_prefix:str = "FOUP")–>boll:
""""
模拟 Carrier ID 校验
真实场景:读 RFID 标签,和 Host 下发的 ID 对比。
""""
if not carrier_id.startwith(expected_prefix):
return False
if len(carrier_id)<8:
return False
return True
测试 CIV
test_id = "FOUP_001"
if civ_check(test_id):
print(f"✅ CIV Pass: {test_id}")
else:
print(f"❌ CIV Fail: {test_id}")
4.3 模拟 SlotMap 上报
def generate_slot_map(wafer_count:int = 25) –> dict:
""""
模拟 SlotMap 生成
真实场景:读 Slot 传感器,确认每个槽位是否有晶圆。
""""
slot_map = {}
for slot in range(1,26):
if slot <= wafer_count:
slot_map[slot] = "Occupied"
else:
slot_map[slot] = "Empty"
return slot_map
模拟一个只有12片晶圆的FOUP
slot_map = generate_slot_map(12)
for slot,status in slot_map.items():
if status == "Occupied":
print(f" Slot {slot:2d}: 🔵 Wafer")
else:
print(f" Slot {slot:2d}: ⚪ Empty")
4.4 完整 E87 流程演示
def simulate_e87_flow():
""""模拟完整的E97 Carrier 装卸流程""""
print("===E87 Carrier Management Simulation ===\\n")
# 创建 LoadPort
lp = LoadPort("LP1")
# 步骤1:准备放盒
lp.state = LoadPortState.READY_TO_LOAD
print(f"[1]LoadPort state:{lp.get_state()}")
# 步骤2:放盒
lp.place_carrier("FOUP_001")
print(f"[2] Carrier placed:{lp.carrier_id}")
print(f"LaodPort state:{lp.get_state()}")
# 步骤3:CIV 校验
if civ_check(lp.carrier_id):
print(f"[3] CIV:✅ Pass")
else:
print(f"[3] CIV: ❌ Fail")
return
# 步骤4:夹紧
lp.clamp()
print(f"[4] Clamped.State:{lp.get_state()}")
# 步骤5:开盖
lp.start_access()
print(f"[5]Access started.State:{lp.get_state()}")
# 步骤6:生成 SlotMao
lp.slot_map = generate_slot_map(12)
print(f"[6] SlotMap generated:12 Wafers")
# 步骤7:模拟取片完成
lp.state = LoadPortState.Carrier_COMPLETE
print(f"[7] All wafers extracted.State:{lp.get_state()}")
print("\\n✅ E87 Flow Complete!")
simulate_e87_flow()
五、第三步:模拟 E90 Substrate 追踪
E90 的核心是 Substrate 的位置和状态变化。下面模拟一片晶圆从 FOUP 到 PM 的完整流转。
5.1 定义 Substrate 状态
class SubstrateState(Enum):
NO_STATE = "NoState"
NEEDS_PROCESSING = "NeedsProcessing"
IN_MOTION = "InMotion"
AT_WORK = "AtWork"
PROCESSING_COMPLETE = "ProcessingComplete"
REJECTED = "Rejected"
LOST = "Lost"
class Substrate:
def init(self,substrate_id:str,slot:int):
self.id = substrate_id
self.slot = slot
self.state = SubstrateState.NO_STATE
self.location = None
self.history = []
def move(self, new_location: str, timestamp: str):
"""记录移动事件"""
self.location = new_location
if new_location in ["ROBOT_ARM1", "ROBOT_ARM2"]:
self.state = SubstrateState.IN_MOTION
self.history.append({
"timestamp": timestamp,
"event": "MOVE",
"from": self.location,
"to": new_location,
"state": self.state.value
})
def start_process(self, chamber: str, timestamp: str):
"""开始工艺"""
self.location = chamber
self.state = SubstrateState.AT_WORK
self.history.append({
"timestamp": timestamp,
"event": "PROCESS_START",
"location": chamber,
"state": self.state.value
})
def complete_process(self, timestamp: str):
"""工艺完成"""
self.state = SubstrateState.PROCESSING_COMPLETE
self.history.append({
"timestamp": timestamp,
"event": "PROCESS_COMPLETE",
"location": self.location,
"state": self.state.value
})
def print_history(self):
"""打印历史轨迹"""
print(f"\\n{'='*60}")
print(f"SubstHistory for {self.id} (Slot {self.slot})")
print(f"{'='*60}")
for i, record in enumerate(self.history, 1):
print(f" [{i}] {record['timestamp']}")
print(f" Event: {record['event']}")
if 'from' in record:
print(f" From: {record['from']} → To: {record['to']}")
if 'location' in record:
print(f" Location: {record['location']}")
print(f" State: {record['state']}")
print()
5.2 模拟完整流转
form datetime import datetime
def get_timestamp() –> str:
""""模拟硬件锁存时间戳(E148格式)""""
return datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")[:–3] + "Z"
def simulate_e90_flow():
"""模拟一片晶圆从 FOUP 到 PM 的完整流转"""
print("=== E90 Substrate Tracking Simulation ===\\n")
# 创建晶圆对象
wafer = Substrate("WAFER_001",slot = 5)
# 步骤1:从 FOUP 取出
ts = get_timestamp()
wafer.move("ROBOT_ARM1",ts)
print(f"[1] Pick from FOUP Slot 5 → Robot")
print(f"Time:{ts}")
print(f"State:{Wafer.state.value}")
# 步骤2:放入 LoadLock
ts = get_timestamp()
wafer.move("LOADLOCK",ts)
print(f"[2] Place to LoadLock")
print(f"Time:{ts}")
# 步骤3:从 LoadLock 取出
ts = get_timestamp()
wafer.move("ROBOT_ARM1",ts)
print(f"[3] Pick from LoadLock → Robot")
# 步骤4:放入 PM1
ts = get_timestamp()
wafer.move("PM1",ts)
print(f"[4] Place to PM1")
# 步骤5:开始工艺
ts = get_timestamp()
wafer.start_process("PM1",ts)
print(f"[5] Process Start in PM1")
print(f"Time:{ts}")
#模拟工艺耗时
import time
time.sleep(1)
# 步骤6:工艺完成
ts = get_timestamp()
wafer.complete_process(ts)
print(f"[6] Process Complete")
print(f"Time:{ts}")
#打印完整历史
wafer.print_history()
print("✅ E90 Flow Complete!")
simulate_e90_flow()
六、第四步:异常注入(模拟器的真正价值)
模拟器最大的优势是可以随意制造异常,验证 Host 的容错能力。
6.1 模拟 CIV 失败
def test_civ_failure():
"""测试 CIV 失败场景"""
print("=== Test: CIV Failure ===\\n")
lp = LoadPort("LP1")
lp.state = LoadPortState.READY_TO_LOAD
# 放一个 ID 不匹配的 FOUP
bad_id = "BAD_ID_999"
lp.place_carrier(bad_id)
if civ_check(bad_id):
print("❌ CIV should have failed!")
else:
print(f"✅ CIV correctly rejected: {bad_id}")
print(" Expected Alarm: IDVerificationFailed")
test_civ_failure()
6.2 模拟 Robot Pick 失败
def test_pick_failure():
"""测试 Robot Pick 失败场景"""
print("\\n=== Test: Robot Pick Failure ===\\n")
wafer = Substrate("WAFER_002", slot=10)
# 模拟真空传感器没信号
vacuum_ok = False
if not vacuum_ok:
wafer.state = SubstrateState.REJECTED
wafer.history.append({
"timestamp": get_timestamp(),
"event": "PICK_FAILED",
"reason": "Vacuum sensor not detected",
"state": wafer.state.value
})
print("✅ Substrate correctly marked as REJECTED")
print(" Expected S6F11: SubstrateStateChange → REJECTED")
print(f" Reason: Vacuum sensor not detected")
else:
wafer.move("ROBOT_ARM1", get_timestamp())
test_pick_failure()
6.3 模拟时间戳倒流
def test_time_reversal():
"""测试时间戳倒流检测"""
print("\\n=== Test: Time Reversal Detection ===\\n")
wafer = Substrate("WAFER_003", slot=3)
# 模拟 NTP 未同步时的错误时间
bad_timestamp = "2026-08-04T09:00:00.000Z" # 比上一条早
good_timestamp = "2026-08-04T10:05:12.345Z"
wafer.history.append({
"timestamp": good_timestamp,
"event": "MOVE",
"to": "ROBOT_ARM1",
"state": "InMotion"
})
# 检查时间戳单调性
last_ts = wafer.history[–1]["timestamp"]
if bad_timestamp <= last_ts:
print("✅ Time reversal detected!")
print(f" Last: {last_ts}")
print(f" New: {bad_timestamp}")
print(" Expected: Report TimeChange event or mark UNSYNCHRONIZED")
else:
print("❌ Time check passed (no reversal)")
test_time_reversal()
七、第五步:用 secsgem 搭建完整模拟器
下面是一个更完整的模拟器框架,可以作为你项目的基础模板:
import secsgem.hsms
import threading
import time
from datetime import datetime
class Gem300Simulator:
"""GEM300 模拟器(简化版)"""
def __init__(self, address: str = "0.0.0.0", port: int = 5000):
self.address = address
self.port = port
self.server = None
self.connected = False
self.state = "OFFLINE"
self.loadports = {f"LP{i}": LoadPort(f"LP{i}") for i in range(1, 3)}
self.substrates = {}
def start(self):
"""启动模拟器"""
print(f"Starting GEM300 Simulator on {self.address}:{self.port}…")
# 这里应该启动 HSMS Server
# 为简化演示,用 print 代替
print("✅ Simulator started. Waiting for Host connection…")
self.state = "ONLINE_LOCAL"
def handle_s1f13(self, message):
"""处理 S1F13(Are You There)"""
# 返回 S1F14
return {"COMM_ACK": 0, "COMM_TEXT": "GEM300 Simulator v1.0"}
def handle_s1f1(self, message):
"""处理 S1F1(Request ONLINE)"""
self.state = "ONLINE_REMOTE"
return {"ONL_ACK": 0, "MDLN": "Simulator", "SOFTREV": "1.0"}
def handle_s3f17(self, message):
"""处理 S3F17(Proceed With Carrier)"""
# 解析 CarrierID 和 LoadPort
carrier_id = message.get("CARRIERID", "")
port_id = message.get("PORTID", "")
if not civ_check(carrier_id):
return {"ACK": 1, "REASON": "IDVerificationFailed"}
lp = self.loadports.get(port_id)
if not lp:
return {"ACK": 1, "REASON": "Invalid Port"}
lp.place_carrier(carrier_id)
lp.clamp()
return {"ACK": 0, "REASON": "OK"}
def handle_s12f3(self, message):
"""处理 S12F3(Substrate History Request)"""
substrate_id = message.get("SUBSTRATEID", "")
if substrate_id in self.substrates:
return self.substrates[substrate_id].history
return {"ERROR": "Substrate not found"}
def stop(self):
"""停止模拟器"""
self.connected = False
self.state = "OFFLINE"
print("Simulator stopped.")
启动模拟器
if name == "main":
sim = Gem300Simulator()
sim.start()
# 模拟一些初始状态
print("\\n— Simulator Ready —")
print(f"State: {sim.state}")
print(f"LoadPorts: {list(sim.loadports.keys())}")
print("\\nWaiting for Host commands…")
八、联调技巧:怎么用模拟器加速开发
技巧 1:先跑通 Happy Path,再注入异常
阶段 1: 正常流程
Host → S1F13 → Simulator 回 S1F14
Host → S1F1 → Simulator 回 S1F2
Host → S3F17 → Simulator 回 ACK=0
Host → S12F3 → Simulator 回 History
阶段 2: 异常注入
Host → S3F17 (Bad ID) → Simulator 回 ACK=1
Host → S2F41 (Invalid Recipe) → Simulator 回 REJECTED
阶段 3: 压力测试
Host 并发发指令 → Simulator 正确排队/拒绝
技巧 2:用 WireShark 抓包对比
模拟器跑通后,用 WireShark 抓 HSMS 报文,和真实设备的报文做对比:
- 消息格式是否一致
- 状态码是否正确
- 时序是否合理
技巧 3:把模拟器当单元测试框架
def test_host_code_with_simulator():
"""用模拟器测试 Host 端代码"""
sim = Gem300Simulator()
sim.start()
# Host 端代码
client = secsgem.hsms.HsmsClient("127.0.0.1", 5000, 1)
client.enable()
# 测试各种场景
assert client.send_s1f13().comm_ack == 0
assert client.send_s3f17("BAD_ID").ack == 1 # CIV 失败
client.disable()
sim.stop()
print("✅ All tests passed!")
九、常见坑(血泪史)
坑 1:secsgem 版本不兼容
检查版本
pip show secsgem
推荐版本
secsgem>=0.5.0
不同版本的 API 差异很大,特别是 HsmsClient 的初始化参数。
坑 2:HSMS 超时参数不匹配
设备端和 Host 端的 T3/T5/T6/T7 必须一致
否则会出现"连上了但收不到回复"的假死现象
T3 = 4.0 # Reply Timeout
T5 = 10.0 # Connection Separation Timeout
T6 = 5.0 # Control Transaction Timeout
T7 = 10.0 # Not Selected Timeout
坑 3:SML 文件没加载
GEM300 要求设备提供 SML(State Machine Logic)文件描述状态机。模拟器虽然没有真实状态机,但应该返回合理的状态列表:
def get_state_model():
"""返回设备状态模型"""
return {
"EQUIPMENT_STATE": ["CONFIGURED", "IDLE", "PROCESSING", "ALARM"],
"PROCESS_STATE": ["SETUP", "READY", "ACTIVE", "COMPLETE"],
"LOADPORT_STATE": [s.value for s in LoadPortState]
}
十、总结
有了这个模拟器,你的 EAP 开发流程变成:
模拟器不是"替代品",是"加速器"。 它让你在设备到来之前就把 80% 的代码写完、测完,FAT 现场只需要验证那 20% 的硬件交互。
相关阅读:
- GEM300 联调避坑指南:E87/E90/E40 到底该先实现哪个?
- E90 SubstHistory 可信度验证:别让晶圆"穿越时空"
- FAT 攻坚战:别让"Demo 级接口"骗了你——EAP 验收的 4 个必杀技
你用 secsgem 搭过模拟器吗?遇到过什么坑?或者你有其他更好用的工具推荐?评论区聊聊。



