使用 PyTransitions 的非确定性状态机?

Non deterministic state machine using PyTransitions?

我正在使用 pytransitions and have come across the need to have several states which are unrelated with others, and would make much sense to model using a non deterministic state machine,这在数学上是等价的。

我想要像下面这样的东西

from transitions import Machine
from transitions import EventData


class Matter(object):
    def __init__(self):
        transitions1 = [
            {'trigger': 'heat', 'source': 'solid', 'dest': 'liquid'},
            {'trigger': 'heat', 'source': 'liquid', 'dest': 'gas'},
            {'trigger': 'cool', 'source': 'gas', 'dest': 'liquid'},
            {'trigger': 'cool', 'source': 'liquid', 'dest': 'solid'}
        ]

        transitions2 = [
            {'trigger': 'turn_on', 'source': 'off', 'dest': 'on'},
            {'trigger': 'turn_off', 'source': 'on', 'dest': 'off'},
        ]
        self.machine = Machine(
                model=self,
                states=[['solid', 'liquid', 'gas'], ['on', 'off']],
                transitions=[transitions1, transitions2],
                initial=['solid', 'off'],
                send_event=True
        )

    def on_enter_gas(self, event: EventData):
        print(f"entering gas from {event.transition.source}")

    def on_enter_liquid(self, event: EventData):
        print(f"entering liquid from {event.transition.source}")

    def on_enter_solid(self, event: EventData):
        print(f"entering solid from {event.transition.source}")

    def on_enter_on(self, event: EventData):
        print(f"entering on from {event.transition.source}")

    def on_enter_off(self, event: EventData):
        print(f"entering off from {event.transition.source}")

我可以将一组新状态定义为 states=itertools.product(states1, states2),然后定义所有转换,如等价定理所示。

我想知道库是否支持这种行为,如果支持,如何实现。

我有不止 2 组(大部分)独立状态。真的,我有一堆偶尔有交互的切换,但大多数是独立的。

... to have several states which are unrelated with others, and would make much sense to model using a non deterministic state machine

对我来说,这听起来像是您正在寻找的不一定是非确定性的,而是 hierarchical/compound states and concurrency/parallelism

您可以使用还具有并发性的转换 Hierarchical State Machine 扩展:

from transitions.extensions import HierarchicalMachine

states1 = ['solid', 'liquid', 'gas']
states2 = ['on', 'off']

transitions1 = [
    {'trigger': 'heat', 'source': 'solid', 'dest': 'liquid'},
    {'trigger': 'heat', 'source': 'liquid', 'dest': 'gas'},
    {'trigger': 'cool', 'source': 'gas', 'dest': 'liquid'},
    {'trigger': 'cool', 'source': 'liquid', 'dest': 'solid'}
]

transitions2 = [
    {'trigger': 'turn_on', 'source': 'off', 'dest': 'on'},
    {'trigger': 'turn_off', 'source': 'on', 'dest': 'off'},
]

combined_states = [
    {"name": "running", "parallel":
        [
            dict(name="component1", states=states1, transitions=transitions1, initial=states1[0]),
            dict(name="component2", states=states2, transitions=transitions2, initial=states2[0])
        ]
    }
]

m = HierarchicalMachine(states=combined_states, auto_transitions=False, initial="running")
print(m.state)  # >>> ['running_component1_solid', 'running_component2_on']
m.turn_off()
print(m.state)  # >>> ['running_component1_solid', 'running_component2_off']

但是,HSM 比简单 Machines 复杂得多。考虑到需要遵循的命名约定和 nesting/initialization 配置,该文档提到了一些限制。

这就是为什么我通常会尝试为我的 FSM 架构寻找最简单的解决方案。现在您的嵌套相当平坦,也可以通过一组模型和 Machines 来实现。 'rulebook' 转换方法使得仅用一台机器及其 'dispatch' 方法管理处于不同状态的多个模型变得相当容易:

from transitions import Machine


class Model:
    pass


class MultiMachine(Machine):

    def __init__(self, configurations):
        # Initialize the machine blank, no states, no transitions and
        # no initial states. Disable auto_transitions since there shouldn't
        # be the possibility to transition e.g. from 'on' to 'liquid'.
        # Furthermore, ignore_invalid_triggers so that events not considered
        # by a model will not throw an exception.
        super().__init__(model=None, states=[], transitions=[], initial=None, auto_transitions=False,
                         ignore_invalid_triggers=True)
        # create a model for each configuration
        for states, transitions, initial in configurations:
            self.add_states(states)
            self.add_transitions(transitions)
            self.add_model(Model(), initial=initial)

    @property
    def state(self):
        return [model.state for model in self.models]


m = MultiMachine([(states1, transitions1, 'solid'), (states2, transitions2, 'off')])
assert m.state == ['solid', 'off']
m.dispatch("turn_on")
assert m.state == ['solid', 'on']
m.dispatch("heat")
assert m.state == ['liquid', 'on']

来自您的评论:

How can I add a conditional transition in one sub-machine, based on the state in another? For example, heat should only make solid into gas in case of on? [...] HSMs, maybe it is better in this case.

这可以通过仅在源状态 on_* 上定义 heat 事件来使用 HSM 来解决。但是,如果您有许多这样的因变量,嵌套可能会变得相当复杂。相反,您可以将对另一台机器的 is_<state> 便利函数的引用添加到所有相关转换的条件列表中。这可以在初始化后完成,以防引导程序出现问题:

from transitions import Machine
from transitions.core import Condition

states1 = ['solid', 'liquid', 'gas']
states2 = ['off', 'on']

m1 = Machine(states=states1, initial=states1[0],
             transitions=[{'trigger': 'heat', 'source': 'solid', 'dest': 'liquid'},
                          {'trigger': 'heat', 'source': 'liquid', 'dest': 'gas'},
                          {'trigger': 'cool', 'source': 'gas', 'dest': 'liquid'},
                          {'trigger': 'cool', 'source': 'liquid', 'dest': 'solid'}])
m2 = Machine(states=states2, initial=states2[0],
             transitions=[{'trigger': 'turn_on', 'source': 'off', 'dest': 'on'},
                          {'trigger': 'turn_off', 'source': 'on', 'dest': 'off'}])

# get all heat transitions and add the condition that they may only be valid when m2.is_on returns True
for trans in m1.get_transitions("heat"):
    trans.conditions.append(Condition(func=m2.is_on))
    # if you want to add an 'unless' statement pass `target=False`
    # to the condition. e.g. "heat unless m2 is off"
    # trans.conditions.append(Condition(func=m2.is_off, target=False))

assert m1.is_solid()
assert m2.is_off()
assert not m1.heat()
assert m1.is_solid()
assert m2.turn_on()
assert m1.heat()
assert m1.is_liquid()