我正试图构建一个我希望是相当简单的状态机来控制一个连接到LCD显示器和按钮的程序。我有一个叫做buttonPressedCallback的回调来在不同的状态之间转换，而且我希望在给定的时间后，如果按钮没有被按下，就会有一个超时来停止当前的LCD显示信息。我以为我已经想好了，但它似乎并没有像我期望的那样响应。按钮回调我还没有真正玩过，但就在实验超时功能的时候，我注意到我的代码退出的时间比预期的要早很多。我想把它作为一个更复杂的状态机用于其他项目，所以我需要把基本的东西做好。
这是我的代码。
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
from gpiozero import Button
import time
BUTTON_PIN = 18
@add_state_features(Timeout)
class CustomStateMachine(Machine):
pass
class simpleMachine(object):
states = [{'name': 'dummy', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
{'name': 'start', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
'waiting']
def __init__(self, button):
self.button = button
self.machine = CustomStateMachine(model=self, states=simpleMachine.states, initial='dummy')
self.machine.add_transition(trigger='buttonPressCallback', source='start', dest='waiting')
self.machine.add_transition(trigger='buttonPressCallback', source='waiting', dest='start')
self.machine.add_transition('timeoutTransition', '*', 'waiting')
self.button.when_pressed = self.buttonPressCallback
def on_enter_start(self):
self.printState()
print("doing 'things' for 15 secs, timeout should happen first")
time.sleep(15)
print("Start state time.sleep() ended")
print("Spent %s seconds in start state" % (time.time() - start_time))
def on_enter_dummy(self):
self.printState()
def on_enter_waiting(self):
self.printState()
print("Nothing happens here, just waiting")
while True:
time.sleep(1)
print("Waiting state time.sleep() ended")
def printState(self):
print("Entered state {}".format(self.state))
if __name__ == "__main__":
start_time = time.time()
btn = Button(pin=BUTTON_PIN, bounce_time=0.1)
testMachine = simpleMachine(btn)
print("State Machine started")
testMachine.to_start()
print("Program ran for %s seconds" % (time.time() - start_time))
以下是我希望发生的事情：
实际上发生了什么。
(.env) dietpi@DietPi:~/rgb_clock$ sudo -E .env/bin/python test.py State Machine started Entered state start doing 'things' for 15 secs, timeout should happen first Entered state waiting Nothing happens here, just waiting Start state time.sleep() ended Spent 15.149317979812622 seconds in start state Program ran for 15.153512001037598 seconds (.env) dietpi@DietPi:~/rgb_clock$ ```
我希望这与asyncio和线程有关，但我希望过渡和超时能帮我解决这个问题。
任何关于为什么它没有按照预期执行的想法都非常欢迎，并建议如何实际实现我正在寻找的功能（仍然使用transition，因为我希望将此用于一个更复杂的项目，这将是非常困难的跟踪阅读与大量的ifelsewhile语句。
基本上，你所描述的一切都在实际发生。我想让你困惑的是你的隐式步骤 "3a"(回调)。on_enter_start 被取消，主线程停止睡眠）没有发生。此外，超时线程是守护进程线程的事实导致了第二个问题，即当 on_enter_start 就完成了。
我修改了一下你的例子，用 DEBUG 日志来获取这里实际发生的所有步骤。transitions 日志的使用相当广泛。所以，最好是打开了 logging 如果事情没有按预期进行。为了高效执行 INFO 通常情况下是足够的。
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
import time
import logging
@add_state_features(Timeout)
class CustomStateMachine(Machine):
pass
class SimpleMachine(object):
states = [{'name': 'dummy', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
{'name': 'start', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
'waiting']
def __init__(self):
self.machine = CustomStateMachine(model=self, states=SimpleMachine.states, initial='dummy')
self.machine.add_transition('timeoutTransition', '*', 'waiting')
def on_enter_start(self):
print("doing 'things' for 15 secs, timeout should happen first")
time.sleep(15)
print("Start state time.sleep() ended")
print("Spent %s seconds in start state" % (time.time() - start_time))
def on_enter_waiting(self):
print("Nothing happens here, just waiting")
while True:
time.sleep(1)
print("Waiting state time.sleep() ended")
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
start_time = time.time()
test_machine = SimpleMachine()
print("State Machine started")
test_machine.to_start()
print("Program ran for %s seconds" % (time.time() - start_time))
assert test_machine.state == 'waiting'
日志输出。
State Machine started
doing 'things' for 15 secs, timeout should happen first
DEBUG:transitions.core:Executed machine preparation callbacks before conditions.
DEBUG:transitions.core:Initiating transition from state dummy to state start...
DEBUG:transitions.core:Executed callbacks before conditions.
DEBUG:transitions.core:Executed callback before transition.
DEBUG:transitions.core:Exiting state dummy. Processing callbacks...
INFO:transitions.core:Exited state dummy
DEBUG:transitions.core:Entering state start. Processing callbacks...
# This is where on_enter_start is called and will block due to time.sleep
DEBUG:transitions.extensions.states:Timeout state start. Processing callbacks...
# The next event is the timeout be triggered (in a Thread!) and timeout callbacks
# will be processed (timeoutTransition)
DEBUG:transitions.core:Executed machine preparation callbacks before conditions.
DEBUG:transitions.core:Initiating transition from state start to state waiting...
DEBUG:transitions.core:Executed callbacks before conditions.
DEBUG:transitions.core:Executed callback before transition.
DEBUG:transitions.core:Exiting state start. Processing callbacks...
# state start is left!
INFO:transitions.core:Exited state start
DEBUG:transitions.core:Entering state waiting. Processing callbacks...
# state waiting is entered. Your callback on_enter_waiting will be executed in
# the Timeout thread and block there
Nothing happens here, just waiting
Start state time.sleep() ended
Spent 15.001700162887573 seconds in start state
# in your main thread your on_enter_start callback is now done
Program ran for 15.001909732818604 seconds
INFO:transitions.core:Executed callback 'on_enter_start'
INFO:transitions.core:Entered state start
DEBUG:transitions.core:Executed callback after transition.
DEBUG:transitions.core:Executed machine finalize callbacks
# The program will exit since timeout threads are daemon threads.
# the reason is that waiting timeouts do not block a program's exit
Process finished with exit code 0
那么，如何处理这个问题呢？现在，我可以想到三种不同的尝试。
如果你的传感器读数长期堵塞，而你又无法阻止它。您可以尝试使用 multiprocessing 来杀死回调，而不需要一个标志来检查...
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
import time
import logging
import threading
@add_state_features(Timeout)
class CustomStateMachine(Machine):
pass
class SimpleMachine(object):
states = [{'name': 'dummy', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
{'name': 'start', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
'waiting']
def __init__(self):
self.running = False # our flag which will tell threads whether they should exit
self.current_job = None # where we save the running thread for joining
self.machine = CustomStateMachine(model=self, states=SimpleMachine.states, initial='dummy')
self.machine.add_transition('timeoutTransition', '*', 'waiting')
def change_jobs(self, func):
if self.current_job:
self.running = False
self.current_job.join() # wait until job and thread exits
self.running = True
self.current_job = threading.Thread(target=func)
self.current_job.daemon = False # depends on your use case
self.current_job.start()
def on_enter_start(self):
self.change_jobs(self.do_start_things)
def do_start_things(self):
print("doing 'things' for 15 secs, timeout should happen first")
counter = 0
start_time = time.time()
while self.running and counter < 15:
print("work work")
time.sleep(1)
counter += 1
print("Spent %s seconds in start state" % (time.time() - start_time))
def waiting(self):
while self.running:
print("wait for input")
time.sleep(1)
def on_enter_waiting(self):
self.change_jobs(self.waiting)
if __name__ == "__main__":
#logging.basicConfig(level=logging.DEBUG)
test_machine = SimpleMachine()
print("State Machine started")
test_machine.to_start()
while True:
time.sleep(1) # make sure your main thread isnt exiting
在心跳中发生的事情取决于当前的状态。在我看来，这将导致一个更干净的体验，而不是一个必须依靠线程。重要的是。不要在回调中进行BLOCK，而是要超时，比如说，在读取操作时，要超时。
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
import time
import logging
@add_state_features(Timeout)
class CustomStateMachine(Machine):
pass
class SimpleMachine(object):
states = [{'name': 'dummy', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
{'name': 'start', 'timeout': 5, 'on_timeout': 'timeoutTransition'},
'waiting']
def __init__(self):
self.running = False
self.current_job = None
self.machine = CustomStateMachine(model=self, states=SimpleMachine.states, initial='dummy')
self.machine.add_transition('timeoutTransition', '*', 'waiting')
self.machine.add_transition(trigger='tick', source='start', dest=None, before='start_tick')
self.machine.add_transition(trigger='tick', source='waiting', dest=None, before='waiting_tick')
def start_tick(self):
print("work work")
def waiting_tick(self):
print("wait for input")
if __name__ == "__main__":
#logging.basicConfig(level=logging.DEBUG)
test_machine = SimpleMachine()
print("State Machine started")
test_machine.to_start()
while True:
time.sleep(1)
test_machine.tick()
AsyncMachine 从状态转换时取消任务asyncio.wait_for 将会在超时时明确取消一个任务.如果你有一个任务集合在状态下运行。AsyncMachine 将在改变状态时取消它们，即使没有asyncio超时。这需要 transitions > 0.8 和 Python > 3.7. 注意到 AsyncMachine 是一个相当新的补充。transitions.
from transitions.extensions.asyncio import AsyncMachine
import asyncio
import logging
class SimpleMachine(object):
states = ['dummy', 'start', 'waiting']
def __init__(self):
self.machine = AsyncMachine(model=self, states=SimpleMachine.states, initial='dummy')
self.machine.add_transition('run', 'dummy', 'start')
self.machine.add_transition('timeoutTransition', '*', 'waiting')
async def doing_things(self):
while True:
print("work work")
await asyncio.sleep(1)
async def on_enter_start(self):
try:
await asyncio.wait_for(self.doing_things(), 5)
except asyncio.TimeoutError:
print("Timeout!")
await self.timeoutTransition()
async def on_enter_waiting(self):
while True:
print("wait for input")
await asyncio.sleep(1)
if __name__ == "__main__":
# logging.basicConfig(level=logging.DEBUG)
test_machine = SimpleMachine()
print("State Machine started")
asyncio.get_event_loop().run_until_complete(test_machine.run())