【问题标题】:Is it possible to block execution of a trigger if the transition isn't valid?如果转换无效,是否可以阻止触发器的执行?
【发布时间】:2023-03-09 13:51:01
【问题描述】:

当转换从当前状态无效时,触发方法似乎仍在运行,然后引发 MachineError 异常。有没有办法阻止触发器的执行,以便在模型上调用触发器只会引发异常而不执行触发器?

抱歉,忘记提及使用常见问题解答中覆盖的 _checked_assignment 可能是导致此行为的原因。

from transitions import State, Machine

class StateMachine(Machine):

    def _checked_assignment(self, model, name, func):
        if hasattr(model, name):
            predefined_func = getattr(model, name)
            def nested_func(*args, **kwargs):
                predefined_func()
                func(*args, **kwargs)
            setattr(model, name, nested_func)
        else:
            setattr(model, name, func)

class Rocket(StateMachine):
    def __init__():
        StateMachine.__init__(
            self,
            states=["on_pad", "fueling", "ready", "launched", "meco", "second_stage", "orbit"],
            transitions=[
                {'trigger': 'fuel', 'source': 'on_pad', 'dest': 'fueling'},
                {'trigger': 'power_on', 'source': 'fueling', 'dest': 'ready'},
                {'trigger': 'launch', 'source': 'ready', 'dest': 'launched'}
            ],
            initial='on_pad'
        )

    def fuel():
        print("cryos loading...")

    def launch():
        print("launching")

def main():
    rocket = Rocket()
    rocket.launch()  # prints "launching" then throws Machine Error, need to block actual method execution

【问题讨论】:

  • 您能否为您的问题提供一个代码示例? “触发器”包括从当前状态检查转换是否有效,但不会处理实际转换。
  • 添加示例。抱歉,忘了提及使用常见问题解答中的覆盖 _checked_assignment,我认为这是导致该行为的原因

标签: python state-machine pytransitions


【解决方案1】:

虽然您可以按照this 答案中的描述使用覆盖Machine._checked_assignment 来包装您的回调,但我建议您将应该在转换上下文中调用的方法绑定到其回调。如文档的章节Callback execution order 中所述,可以在转换期间多次调用回调。需要注意的是,回调不能与预期的触发器具有相同的名称,但这通常是一个小挫折,并且还允许您向同一事件添加多个回调。我修改了你的例子。 Rocket 充当有状态模型,但机器本身已分离。如果您计划使用多个实例,您还可以完全独立于 Rocket 来管理状态机。一台机器可以处理多个有状态的对象。此外,我稍微重命名了您的回调并将它们传递给转换的before 关键字。如前所述,这也可以是一个列表({'before': ['on_launch']} 也是有效的)。这样,它们将在转换发生之前被调用,并且在 a) Rocket 未处于正确状态或 b) 对相关转换的条件检查失败时不会被调用。

from transitions import Machine, MachineError


class Rocket:
    
    def __init__(self):
        self.machine = Machine(
            self,
            states=["on_pad", "fueling", "ready", "launched", "meco", "second_stage", "orbit"],
            transitions=[
                {'trigger': 'fuel', 'source': 'on_pad', 'dest': 'fueling', 'before': 'on_fueling'},
                {'trigger': 'power_on', 'source': 'fueling', 'dest': 'ready'},
                {'trigger': 'launch', 'source': 'ready', 'dest': 'launched', 'before': 'on_launch'}
            ],
            initial='on_pad'
        )

    def on_fueling(self):
        print("cryos loading...")

    def on_launch(self):
        print("launching")


rocket = Rocket()
try:
    rocket.launch()
    assert False
except MachineError:
    pass

rocket.fuel()  # >>> cryos loading...
rocket.power_on() 
rocket.launch()  # >>> launching

【讨论】:

    【解决方案2】:

    在我输入示例时发现问题,需要在_checked_assignment 中的predefined_func() 之前调用传入的func()。此方法应在FAQ 上更新以保留原始转换功能。

        def _checked_assignment(self, model, name, func):
            if hasattr(model, name):
                predefined_func = getattr(model, name)
                def nested_func(*args, **kwargs):
                    func(*args, **kwargs)  # need to call before pre-defined
                    predefined_func()
                setattr(model, name, nested_func)
            else:
                setattr(model, name, func)
    

    【讨论】:

      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2020-05-20
      • 2016-08-02
      • 2019-05-13
      • 1970-01-01
      相关资源
      最近更新 更多