我正在尝试使用具有三个嵌套级别(称为main -> nested -> deeper )的分层机器。我认为,状态机将被一个接一个地执行,然后状态被重新映射到第一台机器上。所以我希望最后的状态是done,但是它是nested_deeper_working,所以很明显我遗漏了一些东西。
这里的解决方法是使用queued=False,然后按预期工作。但缺点是调用堆栈确实很长,如果出现错误,跟踪的时间会很长。
抱歉,我的例子太长了,我没能把它缩短。在现实生活中,我使用MainMachine作为整体生产控制,它启动较小的机器来擦除、闪光、校准或测试设备。它们由NestedMachine表示。这些机器里面是使用的最小的机器。对于硬复位,一个测试序列左右。这是本例中的DeeperMachine。
pytransitions 0.8.10
python 3.7.3GenericMachine类只是一个抽象类。在这里,我定义了默认状态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 NotImplementedErrorMainMachine是层次结构中最高的机器,它启动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()测试实例化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()发布于 2021-11-24 09:45:52
https://stackoverflow.com/questions/69710551
复制相似问题