How to delegate listening to entering states of a `transitions` state machine?

Question

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?

Answer 1

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.