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.
