如何在转换中使用超时来退出状态并停止进一步执行 on_enter_state 函数
How do I use timeout in transitions to exit a state and stop further execution of on_enter_state function
我正在尝试构建我希望是一个相当简单的状态机来控制连接到 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))
这是我期望发生的事情：
- testMachine 以 "dummy" 状态启动
- testMachine 通过调用 testMachine.to_start()
显式进入 "start" 状态
- 开始状态正在执行 "something" 15 秒，但超时在 5 秒后触发并调用 timeoutTransition
- 超时转换将 testMachine 移至等待状态
- 等待无限期地继续，等待按钮按下以触发将其移回开始状态的转换
实际情况：
(.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 和线程有关，但我希望转换和超时会为我解决这个问题。
非常欢迎任何关于为什么它没有按预期执行的想法，并就如何实际实现我正在寻找的功能提出建议（仍在使用转换，因为我希望将其用于更复杂的项目，这将非常困难track/read 有很多 if/else/while 语句。
基本上，您描述的一切都在实际发生。我猜让您感到困惑的是您的隐式步骤“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
所以，如何处理这个问题。现在，我可以想到三种不同的尝试：
1。在线程中组织繁重的处理
- 确保回调不会阻止状态机的事件处理
- 运行 在 threads/a 单个工作线程中进行大量处理。
- 建议：让你的处理线程定期检查标志，这样它们就可以在应该退出的时候优雅地退出
如果您的传感器读数永久阻塞并且您无法阻止它。您可以尝试使用 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
2。在 'internal' 转换中工作（例如在回调之前）并触发 'heartbeat' 事件
heartbeat 中发生的事情取决于当前状态。在我看来，这将带来比必须依赖线程更清晰的体验。重要提示：不要在回调中阻塞，但要为过时的读取操作超时。
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 将在超时发生时明确取消任务。
如果你有一组任务 运行ning 处于状态，AsyncMachine 将在更改状态时取消它们，即使没有异步超时。这需要 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())
我正在尝试构建我希望是一个相当简单的状态机来控制连接到 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))
这是我期望发生的事情：
- testMachine 以 "dummy" 状态启动
- testMachine 通过调用 testMachine.to_start() 显式进入 "start" 状态
- 开始状态正在执行 "something" 15 秒，但超时在 5 秒后触发并调用 timeoutTransition
- 超时转换将 testMachine 移至等待状态
- 等待无限期地继续，等待按钮按下以触发将其移回开始状态的转换
实际情况：
(.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 和线程有关，但我希望转换和超时会为我解决这个问题。
非常欢迎任何关于为什么它没有按预期执行的想法，并就如何实际实现我正在寻找的功能提出建议（仍在使用转换，因为我希望将其用于更复杂的项目，这将非常困难track/read 有很多 if/else/while 语句。
基本上，您描述的一切都在实际发生。我猜让您感到困惑的是您的隐式步骤“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
所以，如何处理这个问题。现在，我可以想到三种不同的尝试：
1。在线程中组织繁重的处理
- 确保回调不会阻止状态机的事件处理
- 运行 在 threads/a 单个工作线程中进行大量处理。
- 建议：让你的处理线程定期检查标志，这样它们就可以在应该退出的时候优雅地退出
如果您的传感器读数永久阻塞并且您无法阻止它。您可以尝试使用 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
2。在 'internal' 转换中工作（例如在回调之前）并触发 'heartbeat' 事件
heartbeat 中发生的事情取决于当前状态。在我看来，这将带来比必须依赖线程更清晰的体验。重要提示：不要在回调中阻塞，但要为过时的读取操作超时。
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 将在超时发生时明确取消任务。
如果你有一组任务 运行ning 处于状态，AsyncMachine 将在更改状态时取消它们，即使没有异步超时。这需要 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())