欢迎光临
我们一直在努力

离线联调利器:用 Python secsgem 搭建 GEM300 模拟器,不依赖设备也能跑通 E87/E90

离线联调利器:用 Python secsgem 搭建 GEM300 模拟器,不依赖设备也能跑通 E87 / E90

摘要:FAT 现场设备放鸽子、硬件还没到货、PLC 还在调试——EAP 工程师最痛苦的莫过于“巧妇难为无米之炊”。本文手把手教你用 Python + secsgem 搭建一个 GEM300 模拟器,在没有任何真实设备的情况下,离线跑通 HSMS 握手、E87 Carrier 装卸、E90 Substrate 追踪全流程。代码可直接复制运行,是EAP 开发者的居家旅行必备良药。


一、为什么你需要一个模拟器?

在 Fab 项目里,EAP 工程师的进度往往被设备卡住:

  • 设备商延迟交付,FAT 排期压到你头上
  • 硬件到了但 PLC 还没联调,状态机跑不起来
  • 想测试异常场景(断网、脏数据),但设备商不让你随便折腾
  • 联调时出了问题,分不清是设备的问题还是你代码的问题

有一个模拟器的好处:

  • 不依赖设备:设备没到,你先把 Host 端代码写完、测完。
  • 可重复测试:异常场景想怎么造怎么造,不用求设备商配合。
  • 隔离问题:Host 连模拟器没问题但连设备有问题→设备的问题:连模拟器就有问题→你代码的问题。
  • 新人培训:让新同事在模拟器上练手,不用冒着报废晶圆的风险上机。

  • 二、环境准备

    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 开发流程变成:

  • 设备没到:用模拟器写完 Host 代码,跑通所有 Happy Path + Sad Path。
  • 设备到了:直接连真机,只调通物理信号部分。
  • 出了问题:连模拟器复现,确认是你的代码还是设备的问题。
  • 新人入职:在模拟器上培训,不用冒风险上真机。
  • 模拟器不是"替代品",是"加速器"。 它让你在设备到来之前就把 80% 的代码写完、测完,FAT 现场只需要验证那 20% 的硬件交互。


    相关阅读:

    • GEM300 联调避坑指南:E87/E90/E40 到底该先实现哪个?
    • E90 SubstHistory 可信度验证:别让晶圆"穿越时空"
    • FAT 攻坚战:别让"Demo 级接口"骗了你——EAP 验收的 4 个必杀技

    你用 secsgem 搭过模拟器吗?遇到过什么坑?或者你有其他更好用的工具推荐?评论区聊聊。


    赞(0)
    未经允许不得转载:171主机测评 » 离线联调利器:用 Python secsgem 搭建 GEM300 模拟器,不依赖设备也能跑通 E87/E90
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址