Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

get_triggers does not get child states in HierarchicalAsyncMachine #646

Open
jorritsmit opened this issue Feb 27, 2024 · 3 comments
Open
Assignees
Labels

Comments

@jorritsmit
Copy link

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
Copy link

spearsear commented Feb 27, 2024

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
Copy link
Member

aleneum commented May 24, 2024

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
Copy link
Author

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
Projects
None yet
Development

No branches or pull requests

3 participants