I have Controller instantiated in the main thread, spawning its own worker thread, which is processing events from other controllers. The Controller instantiates ProductionMachine, which is the main state machine and it has nested machines PrepareMachine and FlashMachine.
PrepareMachine sends connect requests to some devices and waiting for the response received through the Controller's worker thread. When all devices are connected, it hands over the control to FlashMachine.
Till now it seems OK, but when I try to trigger transition event_data.model.to_done() I get TypeError from self.machine.to_connected() which I supposed to be finished. Do you have an idea what am I doing wrong?
I'm using transitions 0.8.9, python 3.7.3 on Raspberry Pi.
Code:
from transitions.extensions import LockedHierarchicalMachine
from threading import Thread
from time import sleep
import logging as log
class Controller:
def __init__(self):
self.machine = ProductionMachine()
self.worker_thread = Thread(target=self.worker, name="controller")
self.worker_thread.start()
def worker(self):
for i in range(3):
sleep(0.2)
self.machine.to_connected()
class ProductionMachine(LockedHierarchicalMachine):
def __init__(self):
prep = PrepareMachine()
flash = FlashMachine()
states = [
{"name": "prepare", "children": prep, "remap": {"done": "flash"}},
{"name": "flash", "children": flash},
]
super().__init__(states=states, queued=True, send_event=True)
class PrepareMachine(LockedHierarchicalMachine):
def __init__(self):
self.counter = 3
states = [
{"name": "connected", "on_enter": self.entry_connected},
{"name": "done"},
]
super().__init__(states=states, queued=True, send_event=True)
def entry_connected(self, event_data):
self.counter -= 1
if self.counter == 0:
event_data.model.to_done()
class FlashMachine(LockedHierarchicalMachine):
def __init__(self):
states = [
{"name": "initial", "on_enter": self.entry_initial},
{"name": "flashing"},
]
super().__init__(states=states, queued=True, send_event=True)
def entry_initial(self, event_data):
event_data.model.to_flashing()
log.basicConfig(level=log.INFO)
controller = Controller()
controller.machine.to_prepare()
controller.worker_thread.join()
Output:
Traceback (most recent call last):
File "/usr/lib/python3.8/threading.py", line 932, in _bootstrap_inner
self.run()
File "/usr/lib/python3.8/threading.py", line 870, in run
self._target(*self._args, **self._kwargs)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/examples/hfsm_locked.py", line 16, in worker
self.machine.to_connected()
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/locking.py", line 196, in _locked_method
return func(*args, **kwargs)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/nesting.py", line 854, in trigger_event
res = self._trigger_event(_model, _trigger, None, *args, **kwargs)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/nesting.py", line 1050, in _trigger_event
tmp = self._trigger_event(_model, _trigger, value, *args, **kwargs)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/nesting.py", line 1054, in _trigger_event
tmp = self.events[_trigger].trigger(_model, self, *args, **kwargs)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/nesting.py", line 118, in trigger
return _machine._process(func)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/core.py", line 1200, in _process
self._transition_queue[0]()
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/nesting.py", line 136, in _trigger
return self._trigger_scoped(_model, _machine, *args, **kwargs)
File "/home/jankrejci/Drive/projekty/210430_pdx_calib/.venv/lib/python3.8/site-packages/transitions/extensions/nesting.py", line 153, in _trigger_scoped
state_tree = reduce(dict.get, _machine.get_global_name(join=False), state_tree)
TypeError: descriptor 'get' for 'dict' objects doesn't apply to a 'NoneType' object
Your code looks okay. This is clearly a bug with transitions 0.8.9 and before. This should be fixed in 0.8.10. I do have some remarks about your machine initializations though: When you don't pass a model parameter, the machine will add itself as a model. Considering your example, you do not need PrepareMachine and FlashMachine to do this. You could initialized both with FlashMachine(model=None, states=...) since you only use your ProductionMachine as a stateful object:
from transitions.extensions import LockedHierarchicalMachine
from threading import Thread
from time import sleep
import logging as log
class Controller:
def __init__(self):
self.machine = ProductionMachine()
self.worker_thread = Thread(target=self.worker, name="controller")
self.worker_thread.start()
def worker(self):
for i in range(3):
sleep(0.2)
self.machine.to_connected()
class ProductionMachine(LockedHierarchicalMachine):
def __init__(self):
prep = PrepareMachine()
flash = FlashMachine()
states = [
{"name": "prepare", "children": prep, "remap": {"done": "flash"}},
{"name": "flash", "children": flash},
]
super().__init__(states=states, queued=True, send_event=True)
class PrepareMachine(LockedHierarchicalMachine):
def __init__(self):
self.counter = 3
states = [
{"name": "connected", "on_enter": self.entry_connected},
{"name": "done"},
]
super().__init__(model=None, states=states, queued=True, send_event=True)
def entry_connected(self, event_data):
self.counter -= 1
if self.counter == 0:
event_data.model.to_done()
class FlashMachine(LockedHierarchicalMachine):
def __init__(self):
states = [
{"name": "initial", "on_enter": self.entry_initial},
{"name": "flashing"},
]
super().__init__(model=None, states=states, queued=True, send_event=True)
def entry_initial(self, event_data):
event_data.model.to_flashing()
log.basicConfig(level=log.INFO)
controller = Controller()
controller.machine.to_prepare()
controller.worker_thread.join()
assert controller.machine.is_flash_flashing()