Is it possible to have the machine call all required transitions to achieve desired state #679

Whytey · 2024-06-28T13:22:04Z

Whytey
Jun 28, 2024

I have read the document and I don't think this is possible, but I thought I would check.

I have a simple model with three states and four transitions.

states = ['STANDBY', 'MOWING', 'PAUSED']

transitions = [
    {'trigger': 'StartMowing', 'source': 'STANDBY', 'dest': 'MOWING', 'prepare': 'start'},
    {'trigger': 'PauseWork', 'source': 'MOWING', 'dest': 'PAUSED', 'prepare': 'pause'},
    {'trigger': 'ContinueWork', 'source': 'PAUSED', 'dest': 'MOWING', 'prepare': 'start'},
    {'trigger': 'CancelWork', 'source': 'PAUSED', 'dest': 'STANDBY', 'prepare': 'cancel'},
]

If my model is in the 'MOWING' state, I would like to be able to request it to go to the 'STANDBY' state but have the machine (?) calculate the transition steps required (i.e. trigger 'PauseWork' and then 'CancelWork') and call the intermediate states.

Is this possible?

Answered by aleneum

Jun 30, 2024

I just checked the pydot documentation and found a way to process the dotstring directly:

import networkx as nx
import graphviz
import pydot
from transitions.extensions import GraphMachine


print(f"Graphviz: {graphviz.__version__}")  # >> Graphviz: 0.20.3
print(f"PyDot: {pydot.__version__}")  # >> PyDot: 2.0.0


states = ["STANDBY", "MOWING", "PAUSED"]


transitions = [
    {"trigger": "StartMowing", "source": "STANDBY", "dest": "MOWING"},
    {"trigger": "PauseWork", "source": "MOWING", "dest": "PAUSED"},
    {"trigger": "ContinueWork", "source": "PAUSED", "dest": "MOWING"},
    {"trigger": "CancelWork", "source": "PAUSED", "dest": "STANDBY"},
]


class NetworkXMachine(GraphMachine):
    …

View full answer

aleneum · 2024-06-28T13:59:42Z

aleneum
Jun 28, 2024
Maintainer

Hello @Whytey,

unfortunately, transitions does not have this feature built-in. This is a common graph theory task where networkx can be used. Since transitions can generate pygraphviz.AGraph which can be parsed by networkx, the data exchange is rather easy:

import networkx as nx

from transitions.extensions import GraphMachine

states = ['STANDBY', 'MOWING', 'PAUSED']

transitions = [
    {'trigger': 'StartMowing', 'source': 'STANDBY', 'dest': 'MOWING'},
    {'trigger': 'PauseWork', 'source': 'MOWING', 'dest': 'PAUSED'},
    {'trigger': 'ContinueWork', 'source': 'PAUSED', 'dest': 'MOWING'},
    {'trigger': 'CancelWork', 'source': 'PAUSED', 'dest': 'STANDBY'},
]


class NetworkXMachine(GraphMachine):

    def shortest_path(self, target):
        graph = nx.drawing.nx_agraph.from_agraph(self.get_graph())
        path = nx.shortest_path(graph, self.state, target)
        # The shortest path will only contain nodes/states ['STANDBY', 'MOWING', ...] and thus we have to 
        # get the appropriate edges/transitions in an extra step. Luckily, networkx
        # keeps the edge label which is equivalent to the trigger called to get from
        # one state to the other 
        for trigger in (graph[u][v].get("label") for u, v in nx.utils.pairwise(path)):
            print(f"Executing {trigger}...")
            self.trigger(trigger)


machine = NetworkXMachine(states=states, transitions=transitions, initial="STANDBY")
machine.shortest_path("PAUSED")
assert machine.state == "PAUSED"

This is a very simple example where there is only one transition between each state. This approach does not consider conditions either but assumes that all transitions will be possible. You might want to think about how to make use of 'weights' in case your path could look different depending on certain conditions.

0 replies

Whytey · 2024-06-29T11:39:56Z

Whytey
Jun 29, 2024
Author

Thanks, that seems very clever.

I tried running a slightly modified version of your sample code but there seems to be an issue.

/home/djwhyte/workspace/moebot-hass-integration/venv2/bin/python /home/djwhyte/.config/JetBrains/PyCharmCE2024.1/scratches/state_machine_2.py 
Traceback (most recent call last):
  File "/home/djwhyte/.config/JetBrains/PyCharmCE2024.1/scratches/state_machine_2.py", line 55, in <module>
    machine.shortest_path("PAUSED")
  File "/home/djwhyte/.config/JetBrains/PyCharmCE2024.1/scratches/state_machine_2.py", line 43, in shortest_path
    graph = nx.drawing.nx_agraph.from_agraph(self.get_graph())
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/djwhyte/workspace/moebot-hass-integration/venv2/lib/python3.12/site-packages/networkx/utils/decorators.py", line 789, in func
    return argmap._lazy_compile(__wrapper)(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<class 'networkx.utils.decorators.argmap'> compilation 4", line 3, in argmap_from_agraph_1
  File "/home/djwhyte/workspace/moebot-hass-integration/venv2/lib/python3.12/site-packages/networkx/utils/backends.py", line 633, in __call__
    return self.orig_func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/djwhyte/workspace/moebot-hass-integration/venv2/lib/python3.12/site-packages/networkx/drawing/nx_agraph.py", line 69, in from_agraph
    if A.is_directed():
       ^^^^^^^^^^^^^
AttributeError: 'Digraph' object has no attribute 'is_directed'. Did you mean: 'directed'?

I have just installed the following two packages:
networkx v3.3
graphviz v0.20.3

Are these the appropriate versions to use for your module?

Note, this is the digraph that was produced from the NetworkXMachine:

digraph "State Machine" {
	graph [color=black directed=true fillcolor=white label="State Machine" rankdir=LR "strict"=false style=solid]
	node [color=black fillcolor=white peripheries=1 shape=rectangle style="rounded, filled"]
	edge [color=black]
	STANDBY [label="STANDBY\l" color=red fillcolor=darksalmon peripheries=2]
	MOWING [label="MOWING\l"]
	PAUSED [label="PAUSED\l"]
	FIXED_MOWING [label="FIXED_MOWING\l"]
	PARK [label="PARK\l"]
	STANDBY -> MOWING [label=StartMowing]
	MOWING -> PAUSED [label=PauseWork]
	PAUSED -> MOWING [label=ContinueWork]
	PAUSED -> STANDBY [label=CancelWork]
}

0 replies

aleneum · 2024-06-29T18:33:04Z

aleneum
Jun 29, 2024
Maintainer

From what I can tell it's the wrong combination of Graphviz output. I have used pygraphviz which produces an pygraphviz.AGraph that can be read by nx.drawing.nx_agraph.from_agraph. transitions also supports the graphviz 1 package that you have installed. This will, however, return a graphviz.Digraph that is different from a pygraphviz.AGraph. To use graphviz or dot directly, you need to use another networkx function. There is more than one parse_dot function in networkx: One uses pygraphviz, the other pydot. Both have the disadvantage that they do not read dot directly but from a path. That means that you have to write dot to a tmp file first which makes things slightly more complex. graphviz (and probably pydot too) are easier to install as they don't require to compile Graphviz bindings like pygraphviz. Pygraphviz does feature more sophisticated graph manipulation methods though.

Variant A (using pygraphviz):

import networkx as nx
import pygraphviz
import sys
import transitions
from transitions.extensions import GraphMachine


print(f"Python: {sys.version}")  # >> Python: 3.12.3 | packaged by Anaconda, Inc. | (main, May  6 2024, 14:46:42) [Clang 14.0.6 ]
print(f"Networkx: {nx.__version__}")  # >> Networkx: 3.2.1
print(f"PyGraphviz: {pygraphviz.__version__}")  # >> PyGraphviz: 1.13
print(f"Transitions: {transitions.__version__}")  # >> Transitions: 0.9.1


states = ['STANDBY', 'MOWING', 'PAUSED']

transitions = [
    {'trigger': 'StartMowing', 'source': 'STANDBY', 'dest': 'MOWING'},
    {'trigger': 'PauseWork', 'source': 'MOWING', 'dest': 'PAUSED'},
    {'trigger': 'ContinueWork', 'source': 'PAUSED', 'dest': 'MOWING'},
    {'trigger': 'CancelWork', 'source': 'PAUSED', 'dest': 'STANDBY'},
]


class NetworkXMachine(GraphMachine):

    def shortest_path(self, target):
        graph = nx.drawing.nx_agraph.from_agraph(self.get_graph())
        path = nx.shortest_path(graph, self.state, target)
        for trigger in (graph[u][v].get("label") for u, v in nx.utils.pairwise(path)):
            print(f"Executing {trigger}...")  # >> Executing StartMowing... >> Executing PauseWork... 
            self.trigger(trigger)


machine = NetworkXMachine(states=states, transitions=transitions, initial="STANDBY")
machine.shortest_path("PAUSED")
assert machine.state == "PAUSED"

Variant B (using graphviz and pydot):

import tempfile
import networkx as nx
import graphviz
from transitions.extensions import GraphMachine


print(f"Graphviz: {graphviz.__version__}")  # >> Graphviz: 0.20.3


states = ['STANDBY', 'MOWING', 'PAUSED']


transitions = [
    {'trigger': 'StartMowing', 'source': 'STANDBY', 'dest': 'MOWING'},
    {'trigger': 'PauseWork', 'source': 'MOWING', 'dest': 'PAUSED'},
    {'trigger': 'ContinueWork', 'source': 'PAUSED', 'dest': 'MOWING'},
    {'trigger': 'CancelWork', 'source': 'PAUSED', 'dest': 'STANDBY'},
]

class NetworkXMachine(GraphMachine):

    def shortest_path(self, target):
        with tempfile.NamedTemporaryFile(mode="wt", delete_on_close=False) as fp:
            fp.write(self.get_graph().source)
            fp.close()
            # using pydot here because if you use pygraphviz,
            # you could also use variant A and 
            # nx.drawing.nx_agraph.from_agraph
            graph = nx.drawing.nx_pydot.read_dot(fp.name)

        path = nx.shortest_path(graph, self.state, target)
        # the pydot graph seems to be slightly different organized and returns a dict of edges.
        for trigger in (graph[u][v][0].get("label") for u, v in nx.utils.pairwise(path)):
            print(f"Executing {trigger}...")
            self.trigger(trigger)


# if you have `pygraphviz` and `graphviz` installed, you must tell transitions not to use `pygraphviz`
# by passing `use_pygraphviz=False`. Note that this will change in transitions 0.9.2 (current master)
# where `use_pygraphviz` is replaced with `graph_engine="pygraphviz|graphviz|mermaid"`
machine = NetworkXMachine(states=states, transitions=transitions, use_pygraphviz=False, initial="STANDBY")
machine.shortest_path("PAUSED")
assert machine.state == "PAUSED"

0 replies

Whytey · 2024-06-30T08:40:36Z

Whytey
Jun 30, 2024
Author

Thank you kindly. That all makes sense.

I chose to utilise method B - I don't want to force my users to require the dependencies of graphviz (e.g. need to have the graphviz-dev package installed on Ubuntu) since I will be utilising the code in Home Assistant which tends to limit dependencies to python packages as much as possible.

Once I have the integration up and running, I will link it here - for full completeness - but I think this question is resolved now.

Thankyou.

4 replies

aleneum Jun 30, 2024
Maintainer

I just checked the pydot documentation and found a way to process the dotstring directly:

import networkx as nx
import graphviz
import pydot
from transitions.extensions import GraphMachine


print(f"Graphviz: {graphviz.__version__}")  # >> Graphviz: 0.20.3
print(f"PyDot: {pydot.__version__}")  # >> PyDot: 2.0.0


states = ["STANDBY", "MOWING", "PAUSED"]


transitions = [
    {"trigger": "StartMowing", "source": "STANDBY", "dest": "MOWING"},
    {"trigger": "PauseWork", "source": "MOWING", "dest": "PAUSED"},
    {"trigger": "ContinueWork", "source": "PAUSED", "dest": "MOWING"},
    {"trigger": "CancelWork", "source": "PAUSED", "dest": "STANDBY"},
]


class NetworkXMachine(GraphMachine):
    def shortest_path(self, target):
        graph = nx.drawing.nx_pydot.from_pydot(
            pydot.graph_from_dot_data(self.get_graph().source)[0]
        )
        path = nx.shortest_path(graph, self.state, target)
        # the pydot graph seems to be slightly different organized and returns a list of edges.
        for trigger in (
            graph[u][v][0].get("label") for u, v in nx.utils.pairwise(path)
        ):
            print(f"Executing {trigger}...")
            self.trigger(trigger)


# if you have `pygraphviz` and `graphviz` installed, you must tell transitions not to use `pygraphviz`
# by passing `use_pygraphviz=False`. Note that this will change in transitions 0.9.2 (current master)
# where `use_pygraphviz` is replaced with `graph_engine="pygraphviz|graphviz|mermaid"`
machine = NetworkXMachine(
    states=states, transitions=transitions, use_pygraphviz=False, initial="STANDBY"
)
machine.shortest_path("PAUSED")
assert machine.state == "PAUSED"

Answer selected by Whytey

aleneum Jun 30, 2024
Maintainer

If dependencies are an issue, you could use path finding algos like dijkstra directly with transitions. The core transitions feature here is machine.get_triggers(state) that returns all possible triggers from a certain state. But again, this does not check for conditions:

from collections import deque

from transitions import Machine


states = ["STANDBY", "MOWING", "PAUSED"]


transitions = [
    {"trigger": "StartMowing", "source": "STANDBY", "dest": "MOWING"},
    {"trigger": "PauseWork", "source": "MOWING", "dest": "PAUSED"},
    {"trigger": "ContinueWork", "source": "PAUSED", "dest": "MOWING"},
    {"trigger": "CancelWork", "source": "PAUSED", "dest": "STANDBY"},
]

# https://en.wikipedia.org/wiki/Dijkstra%27s_algorithm
def dijkstra_path(machine: Machine, target: str) -> list[str]:
    seen = set()
    queue = deque([(machine.state, [])])
    while queue:
        state, path = queue.popleft()
        if state == target:
            return path
        if state in seen:
            continue
        seen.add(state)
        for trigger in machine.get_triggers(state):
            new_path = path + [trigger]
            # since it's always the first transition that is taken
            # we can ignore the rest. If there are conditions to be checked
            # this needs to be reworked.
            new_state = machine.events[trigger].transitions[state][0].dest
            queue.append((new_state, new_path))
    raise ValueError(f"Could not find a path to {target}")



# don't forget to disable auto transitions with auto_transitions=False
machine = Machine(
    states=states, transitions=transitions, initial="STANDBY", auto_transitions=False
)
print(dijkstra_path(machine, "PAUSED")) # ['StartMowing', 'PauseWork']

Whytey Jul 16, 2024
Author

Once I have the integration up and running, I will link it here - for full completeness - but I think this question is resolved now.

Hi,

The implementation is in use and can be reviewed at https://github.com/Whytey/moebot-hass-integration/blob/34a77d40d43791104bdb53ad3a0f3311d57fe732/custom_components/moebot/lawn_mower.py#L45

Thanks again for the help - you were very prompt and descriptive in your replies 👍

aleneum Jul 17, 2024
Maintainer

Glad I could help and I appreciate that you took the time to leave some feedback and link your results.

Cheers!

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Is it possible to have the machine call all required transitions to achieve desired state #679

{{title}}

Replies: 4 comments 4 replies

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

Select a reply

Is it possible to have the machine call all required transitions to achieve desired state #679

Whytey Jun 28, 2024

Replies: 4 comments · 4 replies

aleneum Jun 28, 2024 Maintainer

Whytey Jun 29, 2024 Author

aleneum Jun 29, 2024 Maintainer

Variant A (using pygraphviz):

Variant B (using graphviz and pydot):

Whytey Jun 30, 2024 Author

aleneum Jun 30, 2024 Maintainer

aleneum Jun 30, 2024 Maintainer

Whytey Jul 16, 2024 Author

aleneum Jul 17, 2024 Maintainer

Whytey
Jun 28, 2024

Replies: 4 comments 4 replies

aleneum
Jun 28, 2024
Maintainer

Whytey
Jun 29, 2024
Author

aleneum
Jun 29, 2024
Maintainer

Whytey
Jun 30, 2024
Author

aleneum Jun 30, 2024
Maintainer

aleneum Jun 30, 2024
Maintainer

Whytey Jul 16, 2024
Author

aleneum Jul 17, 2024
Maintainer