pytransitions / transitions

A lightweight, object-oriented finite state machine implementation in Python with many extensions
MIT License
5.48k stars 525 forks source link

get_triggers does not get child states in HierarchicalAsyncMachine #646

Open jorritsmit opened 4 months ago

jorritsmit commented 4 months ago

I am using get_triggers to check for possible triggers from a certain state (to auto transfer to them if conditions are met). I noticed that when in the my initial child state i cannot find any triggers even though they are defined


from transitions.extensions.asyncio import HierarchicalAsyncMachine as Machine
from transitions.extensions.asyncio import NestedAsyncState as State
State.separator = ">"

    # Pressure calibration states
    _states_pc: ClassVar[list[dict[str, Any]]] = [
        {
            "name": "a",
            "initial": "b",
            "children": [
                {"name": "b", "on_enter": ["entered_pc_b"]},
                {"name": "c", "on_enter": ["entered_pc_c"]},
                {
                    "name": "d",
                    "on_enter": ["entered_pc_d"],
                },
            ],
        }
    ]

    _trans_pc: list[dict[str, Any]] = [
        {
            "trigger": "go_a",
            "source": "Z",
            "dest": "a",
        },
        {
            "trigger": "go_b",
            "source": "a>b",
            "dest": "a>c",
        },
]

 @log_time
    def __init__(self) -> None:
        """Initializes the pytransitions statemachine with all states, transitions and parameters."""
        super(StateMachineBase, self).__init__(
            self,
            states=self._states_main + self._states_pc,
            initial=self.initial,
            transitions=self._trans_main + self._trans_pc,
            auto_transitions=True,
            ignore_invalid_triggers=True,
            queued=True,
            # https://github.com/pytransitions/transitions#queued-transitions
        )

a = self.get_triggers(self.state)
triggers = [x for x in a if not x.startswith("to_")]

triggers is empty when i am in a>b

spearsear commented 4 months ago

I might face the similar issue although not exactly the same, I find "may_" function returns False in one of the parallel state but the trigger function can really be executed. See this:

https://stackoverflow.com/questions/78069426/pytransitions-nested-state-may-have-a-bug

aleneum commented 1 month ago

Hello @jorritsmit and @spearsear ,

unfortunately, I cannot reproduce the error with the information you have provided. I reduced the example down to this

from transitions import MachineError
from transitions.extensions.asyncio import HierarchicalAsyncMachine as Machine
from transitions.extensions.asyncio import NestedAsyncState as State
import asyncio

states = [
    "A",
    {"name": "B", "initial": "b", "children": ["a", "b", "c", "d"]},
    {"name": "C", "parallel": ["x", "y", {"name": "z", "children": ["1", "2"], "initial": "1",
                                          "transitions": [["inner", "1", "2"]]}]}
]

transitions = [
    ["go_B", "A", "B"],
    ["go_a", "B>b", "B>a"],
    ["go_C", "B", "C"],
    ["go_A", "C>z>2", "A"]
]

State.separator = ">"

async def run():
    s = Machine(states=states, transitions=transitions, auto_transitions=False, initial="A")
    assert s.get_triggers(s.state) == ['go_B']
    assert await s.go_B()
    assert not await s.may_go_A()  # we cannot go to A from A
    assert s.get_triggers(s.state) == ['go_a', 'go_C']
    assert not await s.may_inner()  # inner cannot be called when not in C>z>1
    try:
        await s.inner()
        raise RuntimeError("This should not happen!")
    except MachineError:
        pass
    assert await s.go_C()
    assert s.state == ["C>x", "C>y", "C>z>1"]
    assert s.get_triggers(*[state for state in s.state]) == ["inner"]
    assert await s.may_inner()  # now we should be able to call inner
    assert not await s.may_go_A()  # .. but not go_A since we are in C>z>1 not C>z>2
    assert await s.inner()
    assert s.get_triggers(*[state for state in s.state]) == ["go_A"]
    assert "C>z>2" in s.state
    assert await s.may_go_A()
    assert await s.go_A()

asyncio.run(run())

But this works as expected. So if this issue is still bothering you, I would appreciate if you could provide a functioning code snippet.

jorritsmit commented 3 weeks ago

Thanks for looking into it. I am currently travelling so it will take me some time to test what is different between your example and mine, but I will do that and report back