Setting initial to list causes NestedGraphMachine to raise an exception

[thedrow]

tests/test_tasks.py:12 (test_start_stop_tasks[trio])
self = <jumpstarter.tasks.Task object at 0x7fcda7d1c0a0>
owner = <class 'tests.test_tasks.test_start_stop_tasks.<locals>.FakeActor'>
name = 'foo'

    def __set_name__(self, owner, name):
        @wraps(self._task_callback)
        async def task_starter(event_data) -> None:
            self_ = event_data.model
    
            async def task_runner():
                retried = False
    
                while not self_._cancel_scope.cancel_called:
                    try:
                        await self._task_callback(self_)
                    except (Success, Failure):
                        # TODO: Change task state
                        return
                    except Retry as retry:
                        # TODO: Use a retry library
                        # TODO: Change task state
                        if retry.countdown:
                            await anyio.sleep(retry.countdown)
    
                        retried = True
    
                    if not retried:
                        # Yield so the scheduler may run another coroutine.
                        await anyio.sleep(0)
                    else:
                        # Retry occurred so next time we will yield to schedule the task more fairly.
                        retried = False
    
            await self_.spawn_task(task_runner, name)
    
        state_machine: ActorStateMachine = owner._state_machine
        transition: NestedAsyncTransition = state_machine.get_transitions(
            "start",
            ActorStartingState.resources_acquired,
            ActorStartingState.tasks_started,
        )[0]
        transition.before.append(task_starter)
    
>       self._initialize_task_state(name, state_machine)

../jumpstarter/tasks.py:91: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <jumpstarter.tasks.Task object at 0x7fcda7d1c0a0>, name = 'foo'
state_machine = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>

    def _initialize_task_state(self, name, state_machine):
        self.task_state = NestedAsyncState(f"{name}_task")
        self.task_state.add_substates(
            (
                self.initialized_state,
                self.running_state,
                self.succeeded_state,
                self.failed_state,
                self.retrying_state,
                self.crashed_state,
            )
        )
        self.task_state.initial = _get_task_state_name(name, TaskState.initialized)
    
        transition = state_machine.get_transitions(
            "start",
            state_machine.actor_state.starting.value.tasks_started,
            state_machine.actor_state.started,
        )[0]
        transition.after.append(f"run_{name}")
    
>       state_machine.add_transition(
            f"run_{name}",
            state_machine.actor_state.started,
            [_get_task_state_name(name, TaskState.initialized), state_machine.actor_state.started.name],
            after=[f"run_{name}"],
        )

../jumpstarter/tasks.py:121: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>
trigger = 'run_foo', source = <ActorState.started: 3>
dest = ['foo_task↦initialized', 'started'], conditions = None, unless = None
before = None, after = ['run_foo'], prepare = None, kwargs = {}
model = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>

    def add_transition(self, trigger, source, dest, conditions=None,
                       unless=None, before=None, after=None, prepare=None, **kwargs):
        """ Calls the base method and regenerates all models's graphs. """
        _super(GraphMachine, self).add_transition(trigger, source, dest, conditions=conditions, unless=unless,
                                                  before=before, after=after, prepare=prepare, **kwargs)
        for model in self.models:
>           model.get_graph(force_new=True)

/home/thedrow/.cache/pypoetry/virtualenvs/jumpstarter-q-gDbjwh-py3.9/lib/python3.9/site-packages/transitions/extensions/diagrams.py:246: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>
model = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>
title = None, force_new = True, show_roi = False

    def _get_graph(self, model, title=None, force_new=False, show_roi=False):
        if force_new:
>           grph = self.graph_cls(self, title=title if title is not None else self.title)

/home/thedrow/.cache/pypoetry/virtualenvs/jumpstarter-q-gDbjwh-py3.9/lib/python3.9/site-packages/transitions/extensions/diagrams.py:192: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <transitions.extensions.diagrams_pygraphviz.NestedGraph object at 0x7fcda7ac0df0>
args = (<jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>,)
kwargs = {'title': 'State Machine'}

    def __init__(self, *args, **kwargs):
        self.seen_transitions = []
>       _super(NestedGraph, self).__init__(*args, **kwargs)

/home/thedrow/.cache/pypoetry/virtualenvs/jumpstarter-q-gDbjwh-py3.9/lib/python3.9/site-packages/transitions/extensions/diagrams_pygraphviz.py:140: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <transitions.extensions.diagrams_pygraphviz.NestedGraph object at 0x7fcda7ac0df0>
machine = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>
title = 'State Machine'

    def __init__(self, machine, title=None):
        self.machine = machine
        self.fsm_graph = None
        self.roi_state = None
>       self.generate(title)

/home/thedrow/.cache/pypoetry/virtualenvs/jumpstarter-q-gDbjwh-py3.9/lib/python3.9/site-packages/transitions/extensions/diagrams.py:255: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <transitions.extensions.diagrams_pygraphviz.NestedGraph object at 0x7fcda7ac0df0>
title = 'State Machine'

    def generate(self, title=None):
        """ Generate a DOT graph with pygraphviz, returns an AGraph object """
        if not pgv:  # pragma: no cover
            raise Exception('AGraph diagram requires pygraphviz')
    
        title = '' if not title else title
    
        self.fsm_graph = pgv.AGraph(label=title, **self.machine.machine_attributes)
        self.fsm_graph.node_attr.update(self.machine.style_attributes['node']['default'])
        self.fsm_graph.edge_attr.update(self.machine.style_attributes['edge']['default'])
        states, transitions = self._get_elements()
        self._add_nodes(states, self.fsm_graph)
>       self._add_edges(transitions, self.fsm_graph)

/home/thedrow/.cache/pypoetry/virtualenvs/jumpstarter-q-gDbjwh-py3.9/lib/python3.9/site-packages/transitions/extensions/diagrams_pygraphviz.py:66: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <transitions.extensions.diagrams_pygraphviz.NestedGraph object at 0x7fcda7ac0df0>
transitions = [{'after': ['start'], 'dest': 'initialized', 'source': 'initializing', 'trigger': 'start'}, {'after': ['start'], 'dest...cquired', ...}, {'after': ['run_foo'], 'dest': 'started', 'source': 'starting↦tasks_started', 'trigger': 'start'}, ...]
container = <AGraph <Swig Object of type 'Agraph_t *' at 0x7fcda7ac0db0>>

    def _add_edges(self, transitions, container):
    
        for transition in transitions:
            # enable customizable labels
            label_pos = 'label'
            src = transition['source']
            try:
                dst = transition['dest']
            except KeyError:
                dst = src
            edge_attr = {}
            if _get_subgraph(container, 'cluster_' + src) is not None:
                edge_attr['ltail'] = 'cluster_' + src
                src_name = src + "_anchor"
                label_pos = 'headlabel'
            else:
                src_name = src
    
>           dst_graph = _get_subgraph(container, 'cluster_' + dst)
E           TypeError: can only concatenate str (not "list") to str

/home/thedrow/.cache/pypoetry/virtualenvs/jumpstarter-q-gDbjwh-py3.9/lib/python3.9/site-packages/transitions/extensions/diagrams_pygraphviz.py:179: TypeError

[pkonnov]

@thedrow
{'after': ['start'] - 'after' cannot be a list, only str or callable object
TypeError: can only concatenate str (not "list") to str

[aleneum]

There seems to be an issue with the resolving the names as @pkonnov points out. I maybe found another bug which is triggered when a state is added during on_enter/exit callbacks. I fixed that one in b8540bc. This works now:

from transitions.extensions.factory import HierarchicalGraphMachine

class Model:

    def __init__(self):
        self.machine = HierarchicalGraphMachine(self, states=['A', 'B'], initial='A')

    def on_enter_B(self):
        child_1 = HierarchicalGraphMachine.state_cls(name='a')
        child_2 = HierarchicalGraphMachine.state_cls(name='b')
        child_3 = HierarchicalGraphMachine.state_cls(name='c')
        state = HierarchicalGraphMachine.state_cls(name='C', initial=['a', 'b', 'c'])
        state.add_substate([child_1, child_2, child_3])
        self.machine.add_state(state)
        self.to_C()

model = Model()
model.to_B()
assert model.state == ['C_a', 'C_b', 'C_c']
model.get_graph().draw('test.png', prog='dot')

However, this might be unrelated to the bug you found.

[aleneum]

Ah wait, I misread @pkonnov answer. I think it should be possible to use list of strings/callables in any callback property.

[pkonnov]

@aleneum yes, i also think that this is not related

[aleneum]

So either dst=transition['dest'] or src=transition['source'] seems to be a list even though it shouldnt. I cannot reproduce this though. Could you boil it down to a minimal example?

[aleneum]

The parameter of this look weird:

>       state_machine.add_transition(
            f"run_{name}",
            state_machine.actor_state.started,
            [_get_task_state_name(name, TaskState.initialized), state_machine.actor_state.started.name],
            after=[f"run_{name}"],
        )

dest seems to be a list:

self = <jumpstarter.states.ActorStateMachine object at 0x7fcda7d1c190>
trigger = 'run_foo', source = <ActorState.started: 3>
dest = ['foo_task↦initialized', 'started'], conditions = None, unless = None

A transition's dest cannot be a list.

[thedrow]

So maybe I'm getting something wrong. Can't I enter two parallel states?
Do I need two different state machines?

[thedrow]

Let's clarify the question: I can transition from multiple states to one state but I can't transition from one state to multiple states?

[aleneum]

So maybe I'm getting something wrong. Can't I enter two parallel states?

if you refering to two different states then the answer is no. You can only use one destination right now. This state, however, can of course enter it's children in parallel where children's themselves might enter their children in parallel. When you set the destination to be a child, its siblings will not be entered. When I reworked the HSM, I actually tried to go that route (multiple dests mean parallel states) but hit a dead end somewhere.

Let's clarify the question: I can transition from multiple states to one state but I can't transition from one state to multiple states?

When you specifiy multiple sources, transitions will actually create one transition for each source. A wildcard (*) will be converted into N transitions where N is the number of root states your machine is currently aware off. The same procedure would not work with dest since creating multiple dests for one source would always result in a set of transitions where only the first once can theoretically be executed. E.g dict(trigger=go, source='A', dest=['B', 'C']) will always transition to B (if you have some conditions like 'is_C' you should consider splitting your definitions.) There have been some discussions about conditional destinations in the issue tracker (for instance here).

Furthermore, the machine does not transition from multiple states to one state but only choses the transition with the currently most fitting source (e.g. dict(trigger='go', source=['A', 'B'], dest='C') will only exit the state, the model is currently in). It is then evaluated which substates need to be exited based on the current model state and the transition's destination. If that transition cannot be executed for some reason, the trigger is delegated. I wrote something about the execution order here.

[thedrow]

But a state machine should be able to enter multiple parallel states even if they are not the initial states, no?

I need to be able to model transitions from multiple states to multiple states. I now understand that's currently not supported and possibly hard to implement but I didn't understand it from the documentation at all.
Should I open an issue about it?

[aleneum]

But a state machine should be able to enter multiple parallel states even if they are not the initial states, no?

I agree, that this would be quite nice to have. That's why I tried to implement it. Without success though :(. Since parallel state support is quite rare across most FSM implementations I know, I opted to have limited parallel state support in transitions rather than None. Unfortunately, I cannot give a reliable ETA as I have no time resources to tackle that issue. I created an issue for that (#507) though just in case someone has the resources and nerves to approach that issue.

I now understand that's currently not supported and possibly hard to implement but I didn't understand it from the documentation at all.

I am sorry to hear that there has been some confusion. Some features are not documented as they should be but if something like dict(trigger='go', source='A', dest=['B', 'C']) would be possible, I'd totally brag about this in the README. Well, one day I/we might can...

Closing this as it currently does not qualify as a bug. A separate feature request has been created.