pytransitions三层嵌套分层状态机queued=True时状态跳转异常问题

问题根因 #

你遇到的是pytransitions 0.8.x版本分层状态机（HSM）队列模式下的模型实例隔离问题：

• 分层状态机的状态重映射逻辑默认要求所有层级共享同一个模型实例，状态变更会在同一个实例上同步
• 当前实现中NestedMachine没有传入父级模型，DeeperMachine硬绑定自身为模型，三层状态机各用各的模型实例，状态变更无法跨层级同步
• queued=True模式下事件队列绑定在模型实例上，子状态机的状态变更事件无法传递到父级模型，导致重映射逻辑无法触发，最终停留在子状态机的中间状态
• queued=False是同步递归执行跳转，事件会直接穿透层级，所以刚好能得到预期结果，但递归调用会导致调用栈过深

解决方案（保留queued=True） #

仅需要调整子状态机的初始化逻辑，让所有层级复用父级的模型实例即可，改动量极小：

1. 修改NestedMachine，支持接收外部传入的模型 #

class NestedMachine(GenericMachine):
    # 新增model参数，接收父级传入的模型
    def __init__(self, model=None):
        deeper = DeeperMachine(model=model)
        remap = {"done": "done"}
        states = [
            {"name": "deeper", "children": deeper, "remap": remap},
        ]
        transitions = [
            ["go", "initial", "deeper"],
        ]
        # 把外部model传给父类初始化
        super().__init__(states, transitions, model=model)

    def entry_initial(self, event_data):
        event_data.model.go()

2. 修改DeeperMachine，支持接收外部传入的模型 #

class DeeperMachine(GenericMachine):
    # 新增model参数，默认优先使用外部传入的模型
    def __init__(self, model=None):
        states = [
            {"name": "working", "on_enter": self.entry_working},
        ]
        transitions = [
            ["go", "initial", "working"],
            ["go", "working", "done"],
        ]
        # 没有外部模型传入时才用自身作为模型
        super().__init__(states, transitions, model=model if model is not None else self)

    def entry_initial(self, event_data):
        event_data.model.go()

    def entry_working(self, event_data):
        event_data.model.go()

3. 修改MainMachine，把自身作为模型传给子状态机 #

class MainMachine(GenericMachine):
    def __init__(self):
        # 把自身作为模型传入NestedMachine
        nested = NestedMachine(model=self)
        remap = {"done": "done"}
        states = [
            {"name": "nested", "children": nested, "remap": remap},
        ]
        transitions = [
            ["go", "initial", "nested"],
        ]
        super().__init__(states, transitions, model=self)

    def entry_done(self, event_data):
        print("job finished")

修改完成后运行测试代码，queued=True配置保留，状态可以正常重映射回顶层的done状态，断言通过，同时事件队列是平层处理不会产生递归调用栈，错误回溯长度正常。

内容的提问来源于stack exchange，提问作者Jan Krejci