1

我Controller在主线程中进行了实例化，产生了自己的工作线程，它正在处理来自其他控制器的事件。Controller实例化，这ProductionMachine是主要的状态机，它有嵌套的机器PrepareMachine和FlashMachine.

PrepareMachine向某些设备发送连接请求并等待通过Controller工作线程收到的响应。当所有设备都连接好后，它将控制权移交给FlashMachine.

到现在为止它似乎还可以，但是当我尝试触发转换时，event_data.model.to_done()我得到TypeError了self.machine.to_connected()我应该完成的。你知道我在做什么错吗？

我在 Raspberry Pi 上使用过渡 0.8.9、python 3.7.3。

代码：

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()

输出：

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
4

1 回答 1

1

你的代码看起来不错。这显然是transitions0.8.9 及之前版本的错误。这应该在0.8.10. 不过，我确实对您的机器初始化有一些评论：当您不传递模型参数时，机器会将自己添加为模型。PrepareMachine考虑到您的示例，您不需FlashMachine要这样做。你可以初始化两者，FlashMachine(model=None, states=...)因为你只使用你ProductionMachine作为一个有状态的对象：

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()
于 2021-09-10T10:34:33.190 回答