更深的嵌套过渡

Question

我正在尝试使用具有三个嵌套级别的分层机器，称为main -> nested -> deeper。我希望状态机一个接一个地执行，然后将状态重新映射回第一台机器。所以我希望最终状态是done，但它是nested_deeper_working，所以显然我错过了一些东西。

这里的解决方法是使用queued=False，然后它按预期工作。但缺点是调用栈真的很长，而且出现一些错误的回溯也很长。

抱歉，我无法将其缩短。在现实生活中，我使用MainMachine 作为整体生产控制，它启动更小的机器来擦除、闪存、校准或测试设备。这些由NestedMachine 表示。这些机器内部是使用的最小的机器，例如。对于硬重置，一个测试序列，左右。在这种情况下，这是DeeperMachine。

pytransitions 0.8.10
python 3.7.3

GenericMachine 类只是一个抽象类。这里我定义了默认状态 initial 和 done 以及基本配置。

from transitions.extensions import HierarchicalMachine

class GenericMachine(HierarchicalMachine):
    def __init__(self, states, transitions, model=None):
        generic_states = [
            {"name": "initial", "on_enter": self.entry_initial},
            {"name": "done", "on_enter": self.entry_done},
        ]
        states += generic_states
        super().__init__(
            states=states,
            transitions=transitions,
            model=model,
            send_event=True,
            queued=True,
        )

    def entry_initial(self, event_data):
        raise NotImplementedError

    def entry_done(self, event_data):
        raise NotImplementedError

MainMachine 是层次结构中最高的机器，它启动NestedMachine。预计在所有嵌套机器完成后，执行done状态。

class MainMachine(GenericMachine):
    def __init__(self):
        nested = NestedMachine()
        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")

NestedMachine 充当第二级嵌套。它启动DeeperMachine 并重新映射done 状态。

class NestedMachine(GenericMachine):
    def __init__(self):
        deeper = DeeperMachine()
        remap = {"done": "done"}
        states = [
            {"name": "deeper", "children": deeper, "remap": remap},
        ]
        transitions = [
            ["go", "initial", "deeper"],
        ]
        super().__init__(states, transitions)

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

第三层嵌套由DeeperMachine实现。工作完成后，它触发go 事件转换到done 状态并通过NestedMachine 跳回到MainMachine

class DeeperMachine(GenericMachine):
    def __init__(self):
        states = [
            {"name": "working", "on_enter": self.entry_working},
        ]
        transitions = [
            ["go", "initial", "working"],
            ["go", "working", "done"],
        ]
        super().__init__(states, transitions, model=self)

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

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

Test 实例化MainMachine 并触发第一个事件。预计嵌套机器将被调用，并且在作业完成后，它将通过done 状态重新映射，回到MainMachine。

import logging as log

def main():
    log.basicConfig(level=log.DEBUG)
    log.getLogger("transitions").setLevel(log.INFO)

    machine = MainMachine()
    machine.go()

    assert machine.state == "done"

if __name__ == "__main__":
    main()

Answer 1

确认为错误

https://github.com/pytransitions/transitions/issues/554

在 dev-0.9 中已解决，示例和生产代码对我来说效果很好。