如何委托监听“转换"状态机的进入状态? [英] How to delegate listening to entering states of a `transitions` state machine?

问题描述

我正在尝试使用 transitions 库.
这个问题遵循 这个，相当松散.

我想将 on_enter 事件的侦听委托给所有状态，并创建几个这样的侦听器，这些侦听器可以订阅并在进入状态时收到通知.
就我而言，我想通知外部事件系统根据状态订阅不同的事件配置.

---

在这个例子中，我将使用一个状态机(比如带有事件 [heat,cool] 的solid<->fluid<->gas).

使用像这样的库可以很容易地做到这一点

from transitions import Machine从转换导入 EventData类物质(对象):def __init__(self):过渡 = [{'trigger': 'heat', 'source': 'solid', 'dest': 'liquid'},{'trigger': 'heat', 'source': 'liquid', 'dest': 'gas'},{'trigger': 'cool', 'source': 'gas', 'dest': 'liquid'},{'trigger': 'cool', 'source': 'liquid', 'dest': 'solid'}]self.machine = 机器(模型=自我，状态=['固体'，'液体'，'气体']，过渡=过渡，初始 =固体"，发送事件=真)def on_enter_gas(self, event: EventData):打印(f从{event.transition.source}进入气体")def on_enter_liquid(self, event: EventData):打印(f从{event.transition.source}进入液体")def on_enter_solid(self, event: EventData):打印(f从{event.transition.source}进入实体")物质 = 物质()material.heat() # 从固体进入液体material.heat() # 从液体进入气体matter.cool() # 从气体进入液体matter.cool() # 从液体进入固体

---

太好了！现在，我想通过订阅向外部通知有关 on_enter 事件的信息.
我想以一种最少将外部世界与机器内部耦合的方式来做到这一点，这样如果我要更改状态名称，或者添加或删除状态，我就不会担心破坏任何用户机器.

我可以实现的一种方法如下，缺点是与机器内部耦合，并迫使我自己实现库的许多功能.

from transitions import Machine从转换导入 EventDatafrom 输入 import Callable类物质(对象):状态 = ['固体'，'液体'，'气体']def __init__(self):过渡 = [{'trigger': 'heat', 'source': 'solid', 'dest': 'liquid'},{'trigger': 'heat', 'source': 'liquid', 'dest': 'gas'},{'trigger': 'cool', 'source': 'gas', 'dest': 'liquid'},{'trigger': 'cool', 'source': 'liquid', 'dest': 'solid'}]self.machine = 机器(模型=自我，状态=self.states，过渡=过渡，初始 =固体"，发送事件=真)self._subscriptions = {}def on_enter_gas(self, event: EventData):打印(f从{event.transition.source}进入气体")如果on_enter_gas"在 self._subscriptions 中:self._subscriptions[on_enter_solid"]()def on_enter_liquid(self, event: EventData):打印(f从{event.transition.source}进入液体")如果on_enter_liquid"在 self._subscriptions 中:self._subscriptions[on_enter_solid"]()def on_enter_solid(self, event: EventData):打印(f从{event.transition.source}进入实体")如果on_enter_solid"在 self._subscriptions 中:self._subscriptions[on_enter_solid"]()def subscribe(self, state: str, trigger: str, callback: Callable):在 self.states 中断言状态machine_event = 触发器 + _"+ 状态如果 machine_event 不在 self._subscriptions 中:self._subscriptions[machine_event] = 回调

这允许为任何状态添加外部回调.

根据评论此处，上面应该有一些更好的 API 来动态添加每个状态的订阅，但是我在 文档.

---

即使图书馆确实可以做到这一点，我相信这还不够.
任何订阅者都必须知道机器的状态才能订阅 <on_enter>它们，而不是简单地成为机器上的侦听器，并实现任何事件来通知它发生了，就像可以轻松添加on_enter_solid 仅仅通过状态solid"的存在.

我最理想的做法是拥有一些我可以继承(或以其他方式)的监听器类，并且只实现我需要监听的方法，外部.

使用库实现此目的或类似方法的最佳方法是什么?

解决方案

我想通过订阅 on_enter 事件向外部通知.我想以一种最少将外部世界与机器内部耦合的方式来做到这一点，这样如果我要更改状态名称，或者添加或删除状态，我就不会担心破坏任何用户机器.

最小耦合是转发事件并让订阅者决定如何处理它:

from transitions import Machine从转换导入 EventDatafrom 输入 import Callable班级观察员:def state_changed(self, event_data: EventData):打印(f状态现在是'{event_data.state.name}'")类 SubscribableMachine(Machine):状态 = ['固体'，'液体'，'气体']过渡 = [{'trigger': 'heat', 'source': 'solid', 'dest': 'liquid'},{'trigger': 'heat', 'source': 'liquid', 'dest': 'gas'},{'trigger': 'cool', 'source': 'gas', 'dest': 'liquid'},{'trigger': 'cool', 'source': 'liquid', 'dest': 'solid'}]def __init__(self):super().__init__(states=self.states, transitions=self.transitions,initial='solid', after_state_change="notify",发送事件=真)self._subscriptions = []def notify(self, event_data: EventData):对于 self._subscriptions 中的 func:功能(事件数据)def subscribe(self, func: Callable):self._subscriptions.append(func)machine = SubscribableMachine()观察者 = 观察者()machine.subscribe(observer.state_changed)machine.heat() # >>>状态现在是液体"

如果您让观察者订阅特定的转换和/或状态事件，当您稍后重命名这些事件时，这显然会破坏它们的代码.然而，在我看来，仅仅传递事件会大大降低状态机和状态模式的实用性，因为它是状态模式中最好的部分之一，它摆脱了 if-elif-else-cascades.

<块引用>

理想情况下，我想做的是拥有一些我可以继承(或以其他方式)的侦听器类，并且只在外部实现我需要侦听的方法.

我想说您不需要特定的侦听器类.您可以直接将可调用对象添加到状态进入/退出回调.此外，您可以将字符串替换为 (string) Enums 作为状态标识符.这样，您可以更改 Enum 的值而不会对观察者产生任何影响.这可以防止在订阅特定状态时出现拼写错误:

from transitions import Machine从转换导入 EventDatafrom 输入 import Callable从枚举导入枚举，自动班级观察员:def state_changed(self, event_data: EventData):打印(f状态现在是'{event_data.state.name}'")类状态(枚举):固体 = 自动()液体 = 自动()GAS = 自动()类 SubscribableMachine(Machine):过渡 = [{'trigger': 'heat', 'source': State.SOLID, 'dest': State.LIQUID},{'trigger': 'heat', 'source': State.LIQUID, 'dest': State.GAS},{'trigger': 'cool', 'source': State.GAS, 'dest': State.LIQUID},{'trigger': 'cool', 'source': State.LIQUID, 'dest': State.SOLID}]def __init__(self):super().__init__(states=State, transitions=self.transitions,初始=State.SOLID，send_event=True)def subscribe(self, func: Callable, state: State):self.get_state(state).on_enter.append(func)def unsubscribe(self, func: Callable, state: State):self.get_state(state).on_enter.remove(func)machine = SubscribableMachine()观察者 = 观察者()machine.subscribe(observer.state_changed, State.LIQUID)machine.heat() # >>>状态现在是液体"机器.热()断言 machine.state == State.GASmachine.unsubscribe(observer.state_changed，State.LIQUID)machine.cool() # 无输出断言 machine.state == State.LIQUID

<块引用>

以相同方式订阅特定转换的语法是什么?

对于过渡，您可以使用 machine.get_transitions(trigger, source, dest) 来获取一组过渡.正如文档中提到的(例如 回调执行顺序)，转换具有两个回调事件:before 和 after.如果您想在转换发生后(也在 State.enter 被调用后)收到通知，您的订阅/取消订阅方法可能如下所示:

 def subscribe(self, func, trigger="", source="*", dest="*"):对于 self.get_transitions(trigger, source, dest) 中的转换:transition.after.append(func)def unsubscribe(self, func, trigger="", source="*", dest="*"):对于 self.get_transitions(trigger, source, dest) 中的转换:transition.after.remove(func)# ...machine.subscribe(observer.state_changed，热")machine.heat() >>>状态现在是液体"machine.heat() >>>状态现在是GAS"

你可以使用 before 代替，看看 state_changed 的输出是如何变化的.此外，您可以通过 source 或 destination 进一步缩小范围:

machine.subscribe(observer.state_changed, heat", source=State.LIQUID)# ...machine.heat() >>><什么都没有>machine.heat() >>>状态现在是GAS"

要取消订阅，您需要记住过滤器设置或在 list.remove 尝试删除不在回调数组中的元素时捕获错误.

I am trying to use the transitions library.
This question follows this one, quite loosely.

I would like to delegate listening to on_enter events to all states and create several such listeners that can subscribe and be notified when entering a state.
In my case, I want to notify an external event system to subscribe to a different configuration of events depending on the state.

---

For this example I will use a state machine (say solid<->fluid<->gas with events [heat, cool]).

This can be done quite easily using the library like so

from transitions import Machine
from transitions import EventData


class Matter(object):
    def __init__(self):
        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'}
        ]
        self.machine = Machine(
                model=self,
                states=['solid', 'liquid', 'gas'],
                transitions=transitions,
                initial='solid',
                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}")


matter = Matter()
matter.heat()  # entering liquid from solid
matter.heat()  # entering gas from liquid
matter.cool()  # entering liquid from gas
matter.cool()  # entering solid from liquid

---

Great! Now, I want to notify externally, via subscriptions about on_enter events.
I want to do that in a way that would least couple the outside world to the insides of the machine, so that if I were to change a state name, or add or remove a state, I wouldn't worry about breaking any users of the machine.

One way I could accomplish that would be the following, with drawbacks of being coupled to the insides of the machine, and forcing me to implement a lot of the functionality of the library myself.

from transitions import Machine
from transitions import EventData
from typing import Callable


class Matter(object):
    states = ['solid', 'liquid', 'gas']
    
    def __init__(self):
        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'}
        ]
        self.machine = Machine(
                model=self,
                states=self.states,
                transitions=transitions,
                initial='solid',
                send_event=True
        )

        self._subscriptions = {}

    def on_enter_gas(self, event: EventData):
        print(f"entering gas from {event.transition.source}")
        if "on_enter_gas" in self._subscriptions:
            self._subscriptions["on_enter_solid"]()

    def on_enter_liquid(self, event: EventData):
        print(f"entering liquid from {event.transition.source}")
        if "on_enter_liquid" in self._subscriptions:
            self._subscriptions["on_enter_solid"]()

    def on_enter_solid(self, event: EventData):
        print(f"entering solid from {event.transition.source}")
        if "on_enter_solid" in self._subscriptions:
            self._subscriptions["on_enter_solid"]()
        
    def subscribe(self, state: str, trigger: str, callback: Callable):
        assert state in self.states
        machine_event = trigger + "_" + state
        if machine_event not in self._subscriptions:
            self._subscriptions[machine_event] = callback            

This allows to add external callbacks for any state.

According to a comment here, the above should have some better API to dynamically add subscriptions per state, but I was not able to find it in the doc.

---

Even if this is indeed possible with the library, I believe that is not enough.
Any subscriber would have to know the states of the machine in order to subscribe to <on_enter>them, instead of simply being a listener on the machine, and implementing just any event to be notified it occured, just like one can easily add a on_enter_solid just through the existance of a state "solid".

What I would ideally like to do is have some listener class I can inherit (or otherwise) and only implement the methods I need to listen to, externally.

What is the best way to accomplish this, or similar using the library?

解决方案

I want to notify externally, via subscriptions about on_enter events. I want to do that in a way that would least couple the outside world to the insides of the machine, so that if I were to change a state name, or add or remove a state, I wouldn't worry about breaking any users of the machine.

The least coupling would be to just forward the event and let the subscriber decide what to do with it:

from transitions import Machine
from transitions import EventData
from typing import Callable


class Observer:

    def state_changed(self, event_data: EventData):
        print(f"state is now '{event_data.state.name}'")


class SubscribableMachine(Machine):
    states = ['solid', 'liquid', 'gas']
    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'}
    ]

    def __init__(self):
        super().__init__(states=self.states, transitions=self.transitions,
                         initial='solid', after_state_change="notify",
                         send_event=True)
        self._subscriptions = []

    def notify(self, event_data: EventData):
        for func in self._subscriptions:
            func(event_data)

    def subscribe(self, func: Callable):
        self._subscriptions.append(func)


machine = SubscribableMachine()
observer = Observer()
machine.subscribe(observer.state_changed)
machine.heat()  # >>> state is now 'LIQUID'

If you let observer subscribe to particular transition and/or state events this would obviously break their code when you rename these events later on. However, in my opinion, just passing events drastically reduces the usefulness of a state machine and the state pattern in general since it's one of the best parts of the state pattern that it gets rid of if-elif-else-cascades.

What I would ideally like to do is have some listener class I can inherit (or otherwise) and only implement the methods I need to listen to, externally.

I'd say you don't need a particular listener class. You can add callables to the state enter/exit callbacks directly. Furthermore, you can replace strings with (string) Enums as state identifier. This way, you could change the Enum's value without any influence on the observers. This prevents typos when subscribing to a particular state:

from transitions import Machine
from transitions import EventData
from typing import Callable
from enum import Enum, auto


class Observer:

    def state_changed(self, event_data: EventData):
        print(f"state is now '{event_data.state.name}'")


class State(Enum):
    SOLID = auto()
    LIQUID = auto()
    GAS = auto()


class SubscribableMachine(Machine):

    transitions = [
        {'trigger': 'heat', 'source': State.SOLID, 'dest': State.LIQUID},
        {'trigger': 'heat', 'source': State.LIQUID, 'dest': State.GAS},
        {'trigger': 'cool', 'source': State.GAS, 'dest': State.LIQUID},
        {'trigger': 'cool', 'source': State.LIQUID, 'dest': State.SOLID}
    ]

    def __init__(self):
        super().__init__(states=State, transitions=self.transitions,
                         initial=State.SOLID, send_event=True)

    def subscribe(self, func: Callable, state: State):
        self.get_state(state).on_enter.append(func)

    def unsubscribe(self, func: Callable, state: State):
        self.get_state(state).on_enter.remove(func)


machine = SubscribableMachine()
observer = Observer()
machine.subscribe(observer.state_changed, State.LIQUID)
machine.heat()  # >>> state is now 'LIQUID'
machine.heat()
assert machine.state == State.GAS
machine.unsubscribe(observer.state_changed, State.LIQUID)
machine.cool()  # no output
assert machine.state == State.LIQUID

What is the syntax to subscribe in the same way to specific transitions?

For transitions, you can use machine.get_transitions(trigger, source, dest) to get a set of transitions. As mentions in the documentation (for instance Callback execution order), transitions feature two callback events: before and after. If you want to be informed after a transition has taken place (also after State.enter has been called), your subscribe/unsubscribe methods could look like this:

    def subscribe(self, func, trigger="", source="*", dest="*"):
        for transition in self.get_transitions(trigger, source, dest):
            transition.after.append(func)
            
    def unsubscribe(self, func, trigger="", source="*", dest="*"):
        for transition in self.get_transitions(trigger, source, dest):
            transition.after.remove(func)
# ...
machine.subscribe(observer.state_changed, "heat")
machine.heat()  >>> state is now 'LIQUID'
machine.heat()  >>> state is now 'GAS'

You could yest before instead and have a look how the output of state_changed changes. Furthermore, you can pass source or destination to narrow it further down:

machine.subscribe(observer.state_changed, "heat", source=State.LIQUID)
# ...
machine.heat()  >>> <nothing>
machine.heat()  >>> state is now 'GAS'

For unsubscribe you need to remember the filter settings or catch errors when list.remove tries to remove an element that is not in the callback array.

这篇关于如何委托监听“转换"状态机的进入状态?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！