fgmacedo / python-statemachine Goto Github PK
View Code? Open in Web Editor NEWPython Finite State Machines made easy.
License: MIT License
Python Finite State Machines made easy.
License: MIT License
Defining a transition called "run" overrides the run function of StateMachine in subclasses
I accidentally defined the transition run, which overwrote the StateMachine.run(<transition identifier>)
function.
I did notice it, but the program still ran, even though part of the interface to StateMachine had been altered implicitly through no explicit definition of a function on the class, which may lead to subtle bugs in some programs.
In my opinion, python-statemachine should complain if someone tries to define a transition this way. Or at least give a warning.
class TheMachine(StateMachine):
init = State("init", initial =True)
generated = State("generated")
running = State("running")
generate = init.to(generated)
run = generated.to(running)
cancel = running.to(generated)
delete = generated.to(init)
If I now use the .run function the transition run is triggered instead of the general run() function:
Traceback (most recent call last):
File "/home/oj/.local/bin/fsm-test", line 33, in <module>
sys.exit(load_entry_point('fsm-test', 'console_scripts', 'fsm-test')())
File "/home/oj/programming/python-statemachine-demos/fsm/fsm/cmd.py", line 33, in main
machine.run(t.identifier)
File "/home/oj/.local/lib/python3.8/site-packages/statemachine/statemachine.py", line 61, in __call__
return self.func(*args, **kwargs)
File "/home/oj/.local/lib/python3.8/site-packages/statemachine/statemachine.py", line 85, in transition_callback
return self._run(machine, *args, **kwargs)
File "/home/oj/.local/lib/python3.8/site-packages/statemachine/statemachine.py", line 112, in _run
self._verify_can_run(machine)
File "/home/oj/.local/lib/python3.8/site-packages/statemachine/statemachine.py", line 108, in _verify_can_run
raise TransitionNotAllowed(self, machine.current_state)
statemachine.exceptions.TransitionNotAllowed: Can't run when in init.
python-statemachine is joining the community of open source projects ending support for Python 2, and based on PyTest's and Pallets announcement, we'll also drop support for Python 3.6 and bellow, as their support windows have ended or will end around the same time.
Future releases of python-statemachine project will only support Python versions still supported upstream, which can be found in the Python Release Cycle.
The last version branch to support Python 2.7
and Python <=3.6
will be 1.0
.
The project will receive a major version bump to indicate support for only 3.7+, python-statemachine
2.0.
I will no longer backport patches for unsupported versions, but the branch will continue to exist. I will be happy to accept patches contributed by the community for any severe security and usability issues.
With the upcoming 1.0, I'm happy to provide the community with a stable and feature-complete version of the library in the long run. While also enabling the usage of new and modern syntax in the library internals.
Python State Machine version: 0.8.0
Python version: 3.7
Operating System: windows 10
I created a state machine for a system and now I would like to validate the state machine. Something like to proof that the state machine does not contain a block senario, if it's ambiguous? Is it deterministic?
Or even to test some LTL formule on it ?
Is there a way to do this ? with python, model cheker ?
The dynamic dispatcher is not filling default values for parameters on callbacks.
from statemachine import State
from statemachine import StateMachine
class XMachine(StateMachine):
initial = State(initial=True, enter="enter_with_default_value")
test = initial.to.itself()
def enter_with_default_value(self, value: int = 42):
assert value == 42, value
XMachine()
Raises:
assert value == 42, value
^^^^^^^^^^^
AssertionError: None
Development notes: Migrate dispatcher.py
to inspect.signature that has a nicer API instead of the currently inspect.getfullargspec.
Thanks for building this repo! I'm not 100% sure my use case is possible based on the documentation - if it is I'd love to add some documentation for it!
Let's say I have a state like the following for writing a plan:
Steps 3a and 3b are run in parallel - either inspector could approve the plan first. And step 4 is only achieved once both 3a and 3b have been achieved.
Is there an easy way to handle a "parallel state" like this (3a and 3b)? Thanks for any advice or help, and thanks for building this!
Trying to run copied example, throws type error, missing 1 required positional argument: 'name'
.
make_processed.py
from statemachine import StateMachine, State
class OrderControl(StateMachine):
waiting_for_payment = State(initial=True)
processing = State()
shipping = State()
completed = State(final=True)
# ...
python command line
>>> from src.data.make_processed import OrderControl
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/hdd/work/Research/.../make_processed.py", line 8, in <module>
class OrderControl(StateMachine):
File "/hdd/work/Research/...//make_processed.py", line 9, in OrderControl
waiting_for_payment = State(initial=True)
TypeError: __init__() missing 1 required positional argument: 'name'
Describe what you were trying to get done.
Tell us what happened, what went wrong, and what you expected to happen.
I'm writing a statemachine to track the state of a given entity and was hoping to use the mixin support for django models to make saving that state a little bit easier. However when I try to add something like this assuming my model is set up correct to use MyStatemachine
,
class MyStatemachine(Statemachine):
# defined states.....
def __init__(self, value: <a value passed to the class>):
self.value = value
super.__init__()
# conditional for a state
def is_true():
return self.value
__init__() got an unexpected keyword argument 'state_field'
It doesn't seem to work with how the Mixin
sets things up automatically. I'm wondering if you have any suggestions on the best way to do this?
I am not sure this was solved, if I create multiple instances of a state machine -- the newest statemachines class variables will serve as the single point of truth for all state machines.
from statemachine import State
from statemachine import StateMachine
class TestStateMachine(StateMachine):
# States
st_1 = State("One", initial=True)
st_2 = State("Two")
st_3 = State("Three")
# Transitions
tr_change = (
st_1.to(st_2, cond="two")
| st_2.to(st_3, cond="three")
| st_3.to(st_1, cond="one")
)
def __init__(self, name="unnamed"):
# variables
self.sm_name = name
self.one = False
self.two = True
self.three = False
super().__init__()
s1 = TestStateMachine()
s2 = TestStateMachine()
s1.current_state
State('One', id='st_1', value='st_1', initial=True, final=False)
s2.current_state
State('One', id='st_1', value='st_1', initial=True, final=False)
s1.tr_change()
s1.current_state
State('Two', id='st_2', value='st_2', initial=False, final=False)
s2.current_state
State('One', id='st_1', value='st_1', initial=True, final=False)
s2.two = False
s2
TestStateMachine(model=Model(state=st_1), state_field='state', current_state='st_1')
s2.tr_change()
statemachine.exceptions.TransitionNotAllowed: Can't tr_change when in One.
s1.three = True
s1.three
True
s2.three
False
s1.tr_change() # this shouldn't throw an error because s1.three is now true
statemachine.exceptions.TransitionNotAllowed: Can't tr_change when in Two.
s2.three = True # When I change the variable on s2, it then allows me to transition the state of s1
s1.tr_change()
s1.current_state
State('Two', id='st_2', value='st_2', initial=False, final=False)
Originally posted by @KellyP5 in #330 (comment)
All state machines events should execute on a run to completition (RTC) mode.
All state machine formalisms, including UML state machines, universally assume that a state machine completes processing of each event before it can start processing the next event. This model of execution is called run to completion, or RTC.
This introduces a breaking change on return values being passed from event handlers to the caller of an event.
Hi!
I'm trying to figure out how to include the machine's model or even the machine itself into validation parameters - let me illustrate it by an example.
Assume we have this machine with a single "run" transition and a class representing a runner. Of course running requires some energy, so a side effect of it is, naturally, a decrease in energy:
from statemachine import StateMachine, State
class RunningMachine(StateMachine):
start = State('start', initial=True)
end = State('end')
run = start.to(end)
def on_run(self):
self.model.energy -= 10
class Runner:
def __init__(self, energy):
self.energy = energy
self.state = 'start'
runner = Runner(15)
RunningMachine(runner).run()
print(runner.energy) # 5
print(runner.state) # end
So far so good!
Now what I'd like to do is to make sure that Runner
has enough energy to run in the first place, so I add a validation like this:
def has_enough_energy(runner):
assert runner.energy >= 10
class RunningMachine:
...
run = start.to(end)
run.validators = [has_enough_energy]
The problem is that in order to use it I now have to pass a Runner
instance as an argument for transition, otherwise it won't make it to the validation function:
class RunningMachine:
...
def on_run(self, runner):
runner.energy -= 10
machine = RunningMachine(runner)
machine.run(runner) # wait a second, don't you have reference to the 'runner' already?
It can get even worse since the runner I pass into it might in fact be a different runner! Which of course leads to an inconsistency:
fresh_runner = Runner(10)
tired_runner = Runner(0)
machine = RunningMachine(tired_runner)
machine.run(fresh_runner) # oops...
print(fresh_runner.energy) # 0
print(fresh_runner.state) # start
print(tired_runner.state) # end
Well there's not much I can do about it but make sure to pass the same Runner
instance to the transition itself. It doesn't come handy and raises some issues I've outlined above so it all comes down to the following questions:
validators
functionality could probably be improved by allowing str
arguments, e.g.:class RunningMachine:
run = a.to(b)
def can_run(self):
assert self.runner.energy >= 10
run.validators = ['can_run']
or by allowing class method (this would require some checks so that transition would know if it should pass model
to the validator in order to not break compatibility):
class RunningMachine:
def can_run(self):
assert self.model.energy >= 10
@start.to(end, validators=can_run)
def run(self):
self.model.energy -= 10
Of course I'd be happy to create a PR implementing either of these approaches - or any other if that would make more sense to you.
Thanks!
I followed example:
traffic_light.run('stop')
appear error: InvalidTransitionIdentifier
transition = getattr(self, transition_identifier, None)
This code will get the Transition's get function, so it will return CallableInstance, but in get_transition function, we get transition that is a CallableInstance type and it hasn't source.
finally, raise InvalidTransitionIdentifier
While exploring this really promising module, I coded up a state machine to put it through some paces. It has states (state1, state2, state3, state4), transitions (trans12, trans23, trans34), and cycle defined (trans12 | trans23 | trans34).
I noticed what might be either an incorrect assumption on my part, or possibly a bug: the first transition methods (before_/on_/after_ methods) were getting called after that transition has already passed, when later transitions' methods were also being called.
Demo code-
from statemachine import StateMachine, State
class TestSM(StateMachine):
state1 = State('state1', initial=True)
state2 = State('state2')
state3 = State('state3')
state4 = State('state4', final=True)
trans12 = state1.to(state2)
trans23 = state2.to(state3)
trans34 = state3.to(state4)
cycle = trans12 | trans23 | trans34
def before_cycle(self):
print("before cycle")
def on_cycle(self):
print("on cycle")
def after_cycle(self):
print("after cycle")
def on_enter_state1(self):
print('enter state1')
def on_exit_state1(self):
print('exit state1')
def on_enter_state2(self):
print('enter state2')
def on_exit_state2(self):
print('exit state2')
def on_enter_state3(self):
print('enter state3')
def on_exit_state3(self):
print('exit state3')
def on_enter_state4(self):
print('enter state4')
def on_exit_state4(self):
print('exit state4')
def before_trans12(self):
print('before trans12')
def on_trans12(self):
print('on trans12')
def after_trans12(self):
print('after trans12')
def before_trans23(self):
print('before trans23')
def on_trans23(self):
print('on trans23')
def after_trans23(self):
print('after trans23')
def before_trans34(self):
print('before trans34')
def on_trans34(self):
print('on trans34')
def after_trans34(self):
print('after trans34')
Paste the command(s) you ran and the output.
>>> m = ubr.TestSM()
enter state1
>>> m.is_state1, m.is_state2, m.is_state3, m.is_state4, m.current_state ; _ = m.cycle()
(True, False, False, False, State('state1', id='state1', value='state1', initial=True, final=False))
before cycle
before trans12
exit state1
on cycle
on trans12
enter state2
after cycle
after trans12
>>> m.is_state1, m.is_state2, m.is_state3, m.is_state4, m.current_state ; _ = m.cycle()
(False, True, False, False, State('state2', id='state2', value='state2', initial=False, final=False))
before cycle
before trans12
before trans23
exit state2
on cycle
on trans12
on trans23
enter state3
after cycle
after trans12
after trans23
>>> m.is_state1, m.is_state2, m.is_state3, m.is_state4, m.current_state ; _ = m.cycle()
(False, False, True, False, State('state3', id='state3', value='state3', initial=False, final=False))
before cycle
before trans12
before trans34
exit state3
on cycle
on trans12
on trans34
enter state4
after cycle
after trans12
after trans34
>>> m.is_state1, m.is_state2, m.is_state3, m.is_state4, m.current_state
(False, False, False, True, State('state4', id='state4', value='state4', initial=False, final=True))
As mentioned, before_trans12, on_trans12 and after_trans12 continue firing after their turn is finished. Did I misconfigure something in my StateMachine subclass?
I spent some time debug-stepping through metaclass code while importing TestSM. Nothing conclusive, except that the Transition instances, after the first one, always start out containing "cycle trans12"
before the current transition key is added to self._events.
For example, repr of trans23 after initialization, see how event='cycle trans12 trans23'
? That don't seem right.
Transition(State('state2', id='state2', value='state2', initial=False, final=False), State('state3', id='state3', value='state3', initial=False, final=False), event='cycle trans12 trans23')
I'll wait for a reply before continuing diagnosis.
-Kevin
I am running the "class TrafficLightMachine(StateMachine)" example from https://github.com/fgmacedo/python-statemachine.
It shows "module 'inspect' has no attribute 'getargspec'" when it is creating an instance(traffic_light = TrafficLightMachine())
Exception has occurred: AttributeError
module 'inspect' has no attribute 'getargspec'
File "..........\Projects\trafficLight.py", line 31, in <module>
traffic_light = TrafficLightMachine()
Add a verification that inspect if the resulting graph of a state machine definition and should raise an InvalidDefinition
exception if the graph is not a single component (has states that cannot be reached).
Is it possible to implement hierarchical/nested state machines like this explicit implementation?
Nothing tried yet.
Setup github actions as travis-ci started asking for credits to build open-source.
I have a decorator class for state machine child class's transitions methods.
The decorator has this function:
def __get__(self, instance, owner):
return partial(self.__call__, instance)
When partial is from functools package.
wrap function crashes on this line:
sig.__name__ = method.__name__
When method is functors.partial(decorator_class.call)
AttributeError: 'functools.partial' object has no attribute '__name__'
It would be nice, mainly for design docs and educational use, to have a way to plot the state machine. It can be something like using the mermaid library.
Or something simpler like an ascii plot (http://www.algorithm.co.il/blogs/ascii-plotter/)
Hi, I'd like to fix some typos of the documentation. I found out the default develop
branch is very different from main
regarding the doc, so which one should I start working on?
def on_enter_{state}:
self.view.state_change(state=self) #notifies an observer class of state change
The state reported is the state prior to the state change. This is because 'self.current_state=destination' happens after the functions are executed.
Placed the 'self.current_state=destination' below the 'result, destination = transition._get_destination(result)' to update the state before the functions are executed.
I am using statemachine
in a project, with Jupyter notebook and the IPython %autoreload
extension.
I have been using Jupyter and autoreload
for years, and I haven't met any serious problem until now. But it leads to a systematic exception in statemachine
. And unless it can be solved, I will have to give up statemachine
, since there is no way I give up Jupyter and autoreload
...
Create two files:
File 1: my_state_machine.py
from statemachine import StateMachine, State
class MyStateMachine(StateMachine):
state1 = State('State1', initial=True)
state2 = State('State2')
state3 = State('State3')
transition12 = state1.to(state2)
transition23 = state2.to(state3)
def do_something(self):
print('something')
File 2: Jupyter notebook
Create 2 cells:
Cell #1
%load_ext autoreload
%autoreload 2
from my_state_machine import MyStateMachine
Cell #2
state_machine = MyStateMachine()
state_machine.transition12()
Execute both cells (cell #1 then cell #2). It's working. Fine.
Execute cell #2 once again. It's working too. Fine.
Now modify my_state_machine.py
. No need to modify the states or transitions. Just modify the do_something()
method:
def do_something(self):
print('something else')
Now run cell #2 once again. Autoreload
automatically reloads the statemachine
module. But obviously something goes wrong, and systematically leads to an exception:
---------------------------------------------------------------------------
AttributeError Traceback (most recent call last)
<ipython-input-12-50d5b3013664> in <module>
1 state_machine = MyStateMachine()
----> 2 state_machine.transition12()
~/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/statemachine/statemachine.py in __call__(self, *args, **kwargs)
61
62 def __call__(self, *args, **kwargs):
---> 63 return self.func(*args, **kwargs)
64
65
~/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/statemachine/statemachine.py in transition_callback(*args, **kwargs)
89 def transition_callback(*args, **kwargs):
90 print('!@#$ Transition.__get__.transition_callback', 'machine', machine, 'self', id(self))
---> 91 return self._run(machine, *args, **kwargs)
92
93 return CallableInstance(self, func=transition_callback)
~/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/statemachine/statemachine.py in _run(self, machine, *args, **kwargs)
116
117 def _run(self, machine, *args, **kwargs):
--> 118 self._verify_can_run(machine)
119 self._validate(*args, **kwargs)
120 return machine._activate(self, *args, **kwargs)
~/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/statemachine/statemachine.py in _verify_can_run(self, machine)
110
111 def _verify_can_run(self, machine):
--> 112 transition = self._can_run(machine)
113 if not transition:
114 raise TransitionNotAllowed(self, machine.current_state)
~/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/statemachine/statemachine.py in _can_run(self, machine)
106
107 def _can_run(self, machine):
--> 108 if machine.current_state == self.source:
109 return self
110
AttributeError: 'NoneType' object has no attribute 'current_state'
Here's a (partial) callstack showing when autoreload is invoking Transition._ _ get _ _()
with a None machine
parameter, eventually leading to an exception, next time the transition is invoked. Unfortunately I am not familiar enough with autoreload or your own code to understand what is the expected workflow...
347, in update_generic
update(a, b)
File "/Users/fred/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/IPython/extensions/autoreload.py", line 287, in update_class
old_obj = getattr(old, key)
File "/Users/fred/opt/anaconda3/envs/tf20/lib/python3.7/site-packages/statemachine/statemachine.py", line 88, in __get__
traceback.print_stack()
Inside the Dockerfile: RUN pip3 install python-statemachine
Worked fine with 0.7.0, broke in 0.7.1
Step 6/21 : RUN pip3 install python-statemachine
---> Running in 95d3a40c6d15
Collecting python-statemachine
Downloading https://files.pythonhosted.org/packages/fd/ca/7bf947cef97789aed263fd142c4d3fa3ffccb5e7d5ebe52fd15b4435324d/python-statemachine-0.7.1.tar.gz
Complete output from command python setup.py egg_info:
Traceback (most recent call last):
File "<string>", line 1, in <module>
File "/tmp/pip-build-vq8ipedi/python-statemachine/setup.py", line 8, in <module>
readme = readme_file.read()
File "/usr/lib/python3.5/encodings/ascii.py", line 26, in decode
return codecs.ascii_decode(input, self.errors)[0]
UnicodeDecodeError: 'ascii' codec can't decode byte 0xc3 in position 4382: ordinal not in range(128)
I noticed that the on_enter_* callback is not executed on the state that is declared as initial state.
Defined an on_enter_ function callback on the state that is defined as initial state.
I have a problem where I want to transition from state A to B to C, but to do it in a single step, i.e. without calling my event twice. The only way I can figure out to do this currently is to add two separate events (for transitioning from A -> B and B -> C, respectively), which is a bit cumbersome.
How about overriding the "+" operator for TransitionList
s? Then you could do something like this:
class MySM(StateMachine):
a = State("A", initial=True)
b = State("B")
c = State("C")
my_event = a.to(b) + b.to(c)
When initializing a statemachine with an object which has a state field. A keyerror occurs when requesting the current_state property.
Describe what you were trying to get done.
Minimal example:
`from statemachine import StateMachine, State
class Model(object):
def __init__(self, state):
self.state = state
class ModelStateMachineOne(StateMachine):
state1 = State('State one', initial=True)
state2 = State('State two')
up = state1.to(state2)
down = state2.to(state1)
model1 = Model('State one')
msm1 = ModelStateMachineOne(model1)
print(msm1.current_state)`
This results in this traceback:
Traceback (most recent call last): File "/Users/jeroen/PycharmProjects/cdggitdev/cdgapi/tests/state_machine_bug.py", line 30, in <module> print(msm1.current_state) File "/Users/jeroen/.local/share/virtualenvs/cdggitdev-9pyfxZ8C/lib/python3.7/site-packages/statemachine/statemachine.py", line 389, in current_state return self.states_map[self.current_state_value] KeyError: 'State one'
When using a state machine which has the state name the same as the identifier, there is no error.
`from statemachine import StateMachine, State
class Model(object):
def __init__(self, state):
self.state = state
class ModelStateMachineTwo(StateMachine):
state1 = State('state1', initial=True)
state2 = State('state2')
up = state1.to(state2)
down = state2.to(state1)
model2 = Model('state1')
msm2 = ModelStateMachineTwo(model2)
print(msm2.current_state)`
This runs without issues.
I created a statemachine and instantiated it multiple times. Scenario is the control of my 14 thermostats for my home automation.
Here is the reduced "minimal" setup to show the error.
The failing test is
test_state(sm_list[2], "Three")
the statemachine remains in state "Two"
"""Example of python module statemachine: https://pypi.org/project/python-statemachine/"""
import sys
import os
import logging
import inspect
from statemachine import State, StateMachine
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)
# https://docs.python.org/3/howto/logging-cookbook.html
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter and add it to the handlers
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
# add the handlers to the logger
log.addHandler(ch)
class TestStateMachine(StateMachine):
# States
st_1 = State("One", initial=True)
st_2 = State("Two")
st_3 = State("Three")
one = False
two = True
three = True
def __init__(self, name="unnamed"):
# variables
self.sm_name = name
super(TestStateMachine, self).__init__()
# Transitions
tr_change = (st_1.to(st_1, cond="cond_one_is_active") |
st_1.to(st_2, cond="cond_two_is_active") |
st_1.to(st_3, cond="cond_three_is_active") |
st_2.to(st_1, cond="cond_one_is_active") |
st_2.to(st_2, cond="cond_two_is_active") |
st_2.to(st_3, cond="cond_three_is_active"))
# Conditions
def cond_one_is_active(self):
return self.one
def cond_two_is_active(self):
return self.two
def cond_three_is_active(self):
return self.three
def test_state(state_machine, state):
print(state_machine.current_state.name)
assert state_machine.current_state.name == state
def test_single_sm():
sm = TestStateMachine()
test_state(sm, "One")
sm.send("tr_change")
test_state(sm, "Two")
def test_multiple_sm():
sm_list = {}
for index in [1, 2, 3]:
sm_list[index] = TestStateMachine(str(index))
print(sm_list[index].sm_name)
test_state(sm_list[index], "One")
sm_list[index].send("tr_change")
test_state(sm_list[index], "Two")
sm_list[2].two = False
sm_list[2].send("tr_change")
test_state(sm_list[2], "Three")
test_single_sm()
test_multiple_sm()
Output:
python \\openhabian3\openHAB-conf\automation\test\statemachine_test.py
One
Two
1
One
Two
2
One
Two
3
One
Two
One
Two
3
One
Two
Two
Traceback (most recent call last):
File "\\openhabian3\openHAB-conf\automation\test\statemachine_test.py", line 91, in <module>
test_multiple_sm()
File "\\openhabian3\openHAB-conf\automation\test\statemachine_test.py", line 87, in test_multiple_sm
test_state(sm_list[2], "Three")
File "\\openhabian3\openHAB-conf\automation\test\statemachine_test.py", line 63, in test_state
assert state_machine.current_state.name == state
AssertionError
Hint:
when I change
sm_list[2].two = False
to change the LAST StateMachine in my list it works :-(
sm_list[3].two = False
I can reproduce the same error with index sm_list[1] the same way,
Assumed reason:
When I debug and set the breakpoints into the functions e.g. cond_one_is_active
the instance of TestStateMachine is always "3".
python-statemachine==0.8.0
Python 3.9.12
macOS Monterey 12.3.1
I am trying to setup a state machine with multiple destination states.
However, I can't seem to get it to work with one of the syntaxes.
Working transition:
pay = unpaid.to(paid, failed)
Nonworking transition:
pay = unpaid.to(paid) | unpaid.to(failed)
Can someone explain why the latter syntax won't work? If this is a bug, can we fix it?
Also, I would like to see more examples in the documentation, I spent so much time in digging through the unit test code before I can get the transition setup to work.
Thanks!
from statemachine import StateMachine, State
PAY_FAILED = True
class InvoiceStateMachine(StateMachine):
unpaid = State('unpaid', initial=True)
paid = State('paid')
failed = State('failed')
pay = unpaid.to(paid) | unpaid.to(failed)
def on_pay(self):
if PAY_FAILED:
return self.failed
else:
return self.paid
invoice_fsm = InvoiceStateMachine()
invoice_fsm.pay()
$ python test-invalid-destination-state.py
Traceback (most recent call last):
File "/Users/ye/test-invalid-destination-state.py", line 21, in <module>
invoice_fsm.pay()
File "/Users/ye/.venvs/test/lib/python3.9/site-packages/statemachine/statemachine.py", line 61, in __call__
return self.func(*args, **kwargs)
File "/Users/ye/.venvs/test/lib/python3.9/site-packages/statemachine/statemachine.py", line 85, in transition_callback
return self._run(machine, *args, **kwargs)
File "/Users/ye/.venvs/test/lib/python3.9/site-packages/statemachine/statemachine.py", line 193, in _run
return transition._run(machine, *args, **kwargs)
File "/Users/ye/.venvs/test/lib/python3.9/site-packages/statemachine/statemachine.py", line 114, in _run
return machine._activate(self, *args, **kwargs)
File "/Users/ye/.venvs/test/lib/python3.9/site-packages/statemachine/statemachine.py", line 416, in _activate
result, destination = transition._get_destination(result)
File "/Users/ye/.venvs/test/lib/python3.9/site-packages/statemachine/statemachine.py", line 161, in _get_destination
raise InvalidDestinationState(self, destination)
statemachine.exceptions.InvalidDestinationState: failed is not a possible destination state for pay transition.
Getting statemachine.exceptions.InvalidDefinition: There should be one and only one initial state. Your currently have these: []
error when initializing the state machine
../models/sales.py:398: in __init__
self.fsm = InvoiceStateMachine()
/Users/ye/.venvs/default/lib/python3.9/site-packages/statemachine/statemachine.py:323: in __init__
self.check()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <[KeyError(None) raised in repr()] InvoiceStateMachine object at 0x11a85c880>
def check(self):
if not self.states:
raise InvalidDefinition(_('There are no states.'))
if not self.transitions:
raise InvalidDefinition(_('There are no transitions.'))
initials = [s for s in self.states if s.initial]
if len(initials) != 1:
> raise InvalidDefinition(_('There should be one and only one initial state. '
'Your currently have these: {!r}'.format(initials)))
E statemachine.exceptions.InvalidDefinition: There should be one and only one initial state. Your currently have these: []
/Users/ye/.venvs/default/lib/python3.9/site-packages/statemachine/statemachine.py:357: InvalidDefinition
from statemachine import StateMachine, State
class InvoiceStateMachine(StateMachine):
unpaid = State(InvoiceStatus.UNPAID.value)
paid = State(InvoiceStatus.PAID.value)
failed = State(InvoiceStatus.FAILED.value)
pay = unpaid.to(paid) | unpaid.to(failed)
def on_pay(self, invoice):
pass
When using the on_enter_{state} callbacks I would assume that current_state has already be changed (because you have just 'entered' it). I am trying to get a transition to chain into another transition, which I can't quite manage with the current callback states, unless I am doing something silly?
from statemachine import StateMachine, State
class TestMachine(StateMachine):
state_1 = State('state_1', initial=True)
state_2 = State('state_2')
go_to_2 = state_1.to(state_2)
go_to_1 = state_2.to(state_1)
def on_enter_state_1(self):
print('on_enter_state_1, curr: {}'.format(self.current_state_value))
def on_enter_state_2(self):
print('on_enter_state_2, curr: {}'.format(self.current_state_value))
self.go_to_1()
if __name__ == '__main__':
sm = TestMachine()
sm.go_to_2()
prints on_enter_state_2, curr: state_1
and then errors because there is not transition from state_1 to state_1
Build status for master branch is missing and badge shows error in build status:
Twos possible solutions:
Created two instances of the same machine.
Process event on the first machine actually calls method on second machine
from statemachine import State
from statemachine import StateMachine
class MyStateMachine(StateMachine):
sInit = State("sInit", initial=True)
sMain = State("sMain")
eStart = sInit.to(sMain)
def __init__(self, name):
super().__init__()
print(f"{name} initializing")
self.name = name
def on_eStart(self, who):
print(f"{self.name}: {who} is starting")
sm1 = MyStateMachine("SM1") # SM1 initializing
sm2 = MyStateMachine("SM2") # SM1 initializing
sm1.eStart("sm1") # SM2: sm1 is starting <---Should be SM1 !!!
Add support for internacionalization, translating the exception messages as an example.
uname -a
== Linux belohorizonte 5.15.0-67-generic #74~20.04.1-Ubuntu SMP Wed Feb 22 14:52:34 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux
#365 breaks code that uses callbacks on transitions, when external cycling of states is done via TransitionList.__call__()
. (I have tested my system which relies upon 1.0.2, and it does not run.)
The usefulness of cycle = trans1 | trans2 | trans3 | ...
to move through state and transition callbacks cannot be overstated. This enables external code to "run" a machine without knowing anything about its states, transitions, or internal representation.
This patch removes a major behavioral feature, enabling external code to cycle a machine instance through states, transitions, and callbacks on both.
I added a condition on the callback to only run if the callback was associated with the expected event. So even sharing the same transition instance, the action will be only fired if the event name matches.
If you follow this thought all the way through, wouldn't you need to stop calling callbacks for states, too? Since they also do not share the name of the event, cycle
in my case.
I guess the only choice left is to convert my transition callbacks into state callbacks... but, will state callbacks also go away on TransitionList
too? Can conditions even be used on states?
What was your motivation for doing this? The "before" behavior seemed correct. The fix seems incorrect, given the prior intent of python-statemachine
to allow callbacks at various stages of a machine's lifecycle.
I should be able to get Transition
callbacks fired when cycling through a TransitionList
. That's why I started using this package, it provided a rich set of callbacks, plus the ability to cycle. I should not have to manually invoke a transition by name in the code that uses my StateMachine
subclass. Doing so would violate the principle of Dependency Inversion, which states that external code using an object must depend solely upon the object's abstractions, and not its concrete implementation. Likewise, Separation of Concerns comes to mind-
Separation of concerns results in more degrees of freedom for some aspect of the program's design, deployment, or usage. Common among these is increased freedom for simplification and maintenance of code. When concerns are well-separated, there are more opportunities for module upgrade, reuse, and independent development. Hiding the implementation details of modules behind an interface enables improving or modifying a single concern's section of code without having to know the details of other sections and without having to make corresponding changes to those other sections.
Here is how I currently run one of my machines-
while not agent.final:
agent.cycle()
So simple. The code using a machine subclass doesn't need to know anything about its internal workings. It just runs it like a clock. With the current version, that code would need to know the name of every transition, and manually call them one at a time. Am I understanding this correctly?
Here is an outline of that machine subclass (implementation details removed). Its set of states, transitions, and conditions, describe a workflow DAG with conditional branching. I am not sure I could actually get this to work under 2.0.0 just by calling agent.cycle()
anymore, because it doesn't seem that State(...)
's constructor provides for conditions, the way that Transition
does.
class UpstreamBarsAgent(StateMachine):
created = State('created', initial=True)
requested = State('requested')
received = State('received')
parsed = State('parsed')
ingested = State('ingested')
finished = State('finished', final=True)
finished_empty = State('finished_empty', final=True)
request_failed = State('request_failed', final=True)
receive_failed = State('receive_failed', final=True)
parse_failed = State('parse_failed', final=True)
ingest_failed = State('ingest_failed', final=True)
request = created.to(requested, cond="ready_to_request")
receive = requested.to(received, cond="ready_to_receive")
parse = received.to(parsed)
ingest = parsed.to(ingested, cond="ready_to_ingest")
finish_empty = parsed.to(finished_empty, cond="is_payload_empty")
finish = ingested.to(finished)
request_fail = requested.to(request_failed, cond="did_request_fail")
receive_fail = received.to(receive_failed, cond="did_receive_fail")
parse_fail = parsed.to(parse_failed, cond="did_parse_fail")
ingest_fail = ingested.to(ingest_failed, cond="did_ingest_fail")
cycle = (request | request_fail |
receive | receive_fail |
parse | finish_empty | parse_fail |
ingest | ingest_fail |
finish)
def on_create(self):
pass
def ready_to_request(self):
pass
def ready_to_receive(self):
pass
def ready_to_ingest(self):
pass
def did_request_fail(self):
pass
def did_receive_fail(self):
pass
def did_parse_fail(self):
pass
def did_ingest_fail(self):
pass
def is_payload_empty(self):
pass
def on_request(self):
pass
def on_receive(self):
pass
def on_parse(self):
pass
def on_ingest(self):
pass
def on_finish(self):
pass
def on_finished_empty(self):
pass
def on_request_fail(self):
pass
def on_receive_fail(self):
pass
def on_parse_fail(self):
pass
def on_ingest_fail(self):
pass
This Looks like an interesting package!!
I can see that you can set up states (say MONITOR, FIRE, FLOOD, PUMPING, NOTIFY)
I also see you can name transitions (say: see_smoke=monitor.to(fire)
However, the package traffic light example only shows transitions being manually activated (typed in), not reacting to various inputs (triggers?)
Programmatically, how/where do you do this checking of inputs or conditions (such as change state when time=12:45?) What triggers this transition? Say a GPIO pin was going high to show smoke detected, where is that placed in the code? (if smoke detected, in monitor state, transition to fire state)
Are the transitions automatically checked & actions taken for the particular current state? Can actions be added when entering or leaving states? Say I had the typical vending machine example & had some switches to read coins, hit the dispense button, etc...are there any more examples?
I am creating a state machine and I would like to know if it is possible to define a final state as we do in initial state
begin= State('begin', initial=True)
Any help will be appreciated
Ran into an issue where transition functions that returned a dictionary was causing an exception inside the statemachine.py transition class, in the method _get_destination_from_result. The issue is that the method inspects the return value to determine if there is any state classes returned to assign the destination class. However in the case where a dictionary is returned with 2 or more keys the len() test on the result returns true, but the object is not iterable. So when the following occurs:
if isinstance(result[-1], State):
result, destination = result[:-1], result[-1]
if len(result) == 1:
result = result[0]
The -1 index in results raises an exception.
To deal with the issue I've simple wrapped all transistion functions that return dictionary results with a method that does the following to the dictionary:
ret_val[-1] = None
return ret_val
This creates a -1 key in the dictionary to avoid the exception.
It would probably be more advantageous if the library verified the return value was iterable before attempting to index into it.
Hi!
I like your library!
I tried to take a short-cut in my state machine, validator returned a list (empty list would mean False) and the error message was not related to the actual problem.
I wonder if it would make sense to emphasize (in example, maybe) that validators need to return bool
value.
Or make it checked somehow? if a guard gets something else than a bool
, it would log a warning?
Or then, it's also possible to coerce to bool... not sure which is best here.
It would be nice to have an ability to declare a callback like so:
class Machine(StateMachine):
a = State('a', initial=True)
b = State('b')
c = State('c')
a2b = a.to(b)
b2a = b.to(a)
b2c = b.to(c)
c2a = c.to(a)
def on_b2a_test(self, *args, **kwargs):
return False # or some runtime test
def on_b2c_test(self, *args, **kwargs):
return True # or some runtime test
so that a state machine object could then do the following (pseudo-code):
def step(self, *args, **kwargs):
valid_transitions = sum([t.test() for t in m.allowed_transitions])
if valid_transitions == 0:
raise NoNextTransition
elif valid_transitions > 1:
raise AmbiguousNextTransition
else:
m.allowed_transitions[valid_transitions.index(True)]()
And a library consumer could use:
m = Machine()
runtime_state = 0
while True:
runtime_state +=1
m.step(runtime_state)
The motivation behind this is to encode the conditional transition logic into the class declaration so it doesn't get diffused across multiple locations.
Is this perhaps already available and I missed it?
Originally posted by galhyt January 30, 2023
I have a class that inherites statemachine and has no states and no transitions, all are defined in the children classes. The problem is that from version 1.0 it's mandatory to have transitions and states. Can you reverse it?
Describe what you were trying to get done.
Tell us what happened, what went wrong, and what you expected to happen.
Say we have a statemachine class and we subclass it:
class TrafficLightMachine(StateMachine):
def __init__(self):
StateMachine.__init__(self)
green = State('green', initial=True)
yellow = State('yellow')
red = State('red')
slowdown = green.to(yellow)
stop = yellow.to(red)
go = red.to(green)
class ExtTrafficLightMachine(TrafficLightMachine):
def __init__(self):
TrafficLightMachine.__init__(self)
tlm = ExtTrafficLightMachine()
The code above would result in an error:
"statemachine.exceptions.InvalidDefinition: There are no states". Now the attributes in ExtTrafficLightMachine contain the three states and transitions, but some how the subclass initialization fails to varify it.
Workaround:
class ExtTrafficLightMachine():
def __init__(self):
self.statemachine = TrafficLightMachine()
def __getattr__(self,item):
return getattr(self.statemachine,item) #I know this doesn't capture any attributes in the ExtTrafficLightMachine class and needs flow control, but this should be good enough to point out the issue.
Paste the command(s) you ran and the output.
If there was a crash, please include the traceback here.
The issue is rather a suggestion than a real problem. I am lazy. The README file seems unnecessarily verbose since it has an io documentation. It might require a refac from my perspective.
#!/usr/bin/env python3
from statemachine import StateMachine, State
class PanelFlashingMachine(StateMachine):
disconnected = State('Disconnected', initial=True)
connected = State('Connected')
running = State('Running')
connect = disconnected.to(connected)
run = connected.to(running)
stop = running.to(disconnected)
class SomFlashingMachine(StateMachine):
waiting = State('Waiting', initial=True)
flashing = State('Flashing')
finished = State('Finished')
waiting.to(flashing)
flashing.to(finished)
finished.to(waiting)
panel = PanelFlashingMachine()
unit = SomFlashingMachine()
$ python3.7 panel.py
Traceback (most recent call last):
File "panel.py", line 26, in <module>
unit = SomFlashingMachine()
File "/Users/dan/.virtualenvs/wagz/lib/python3.7/site-packages/statemachine/statemachine.py", line 272, in __init__
self.check()
File "/Users/dan/.virtualenvs/wagz/lib/python3.7/site-packages/statemachine/statemachine.py", line 285, in check
raise InvalidDefinition(_('There are no transitions.'))
statemachine.exceptions.InvalidDefinition: There are no transitions.
Hi ๐
This is my first visit to this fine repo, but it seems you have been working hard to keep all dependencies updated so far.
Once you have closed this issue, I'll create seperate pull requests for every update as soon as I find one.
That's it for now!
Happy merging! ๐ค
If you define a state machine with a transition that has an on_execute
callback, you can't return the desired state if there's only one option. The syntax return <result, desired state>
only works if multiple destination states are defined.
I'd like to track external resource states using the statemachine package. Is it possible to searialize the current state of a state machine so it can be saved to disk? And also read it back from disk and deserialize?
doc problem:
For each state, there's a dinamically created property in the form is_<state.identifier>
dinamically --> dynamically
There does not appear to be a way to implement an Internal Transition; is this something worthwhile to add?
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.