trafficlight

[hstarmans]

Hey everyone,

I really like the transitions package. Before. I always used the Qt StateMachine class and PyQt. If only a Statemachine is required, transitions is leaner and not such a large dependency. I have two questions. I have created a traffic light, which can be turned on and run in the background. To do so the machine is dispatched to a separate thread which keeps calling next. The interpreter is constantly updated on the state of the traffic light via print outs and a user can stop or pause it. Update: Threads were replaced with coroutines, the interpreter was replaced with a web front end, see example at the end. The machine serves as a simple illustrations for more complicated machines, handled in the background, e.g. 3D printers. The two questions are as follows:

1. Why is False or True returned if the traffic light is paused!? I also get this if I transition a machine from the documentation's code samples. Update: This is to notify transition is successful, documentation has been updated.
2. Is this is the right way to do it, could you give a more elaborate example in the chapter Threadsafe(-ish) State Machine, any other comments!? Update: No threads are expensive, use asyncio. The code works in Python 3, not 2.7.

'''
Created on Jan 8, 2016
'''

from transitions import LockedHierarchicalMachine as Machine
from time import sleep, time
from threading import Thread, Event

class ThreadClass(Thread):
        def __init__(self,trafficlight):
            super().__init__()
            self.trafficlight=trafficlight
            self.stopevent=Event()  
            self.pauseevent=Event()
        def stop(self):  
            self.stopevent.set()
            self.pauseevent.set()    
        def pause(self):
            self.pauseevent.set()            
        def run(self):
            while(True):
                sleep(self.trafficlight.printstatetimeout()) 
                self.trafficlight.lock.acquire() 
                if self.stopevent.isSet():
                    print("Stop pressed, traffic light stops")
                    self.trafficlight.lock.release()
                    break
                else:
                    if self.pauseevent.isSet() and not self.stopevent.isSet():
                        self.trafficlight.lock.release()
                        print("Pause pressed")
                        self.pauseevent.clear()
                        self.pauseevent.wait()
                        print("Resume or stop pressed")
                        self.pauseevent.clear()
                        self.trafficlight.lock.acquire()
                    else:
                        self.trafficlight.next()
                self.trafficlight.lock.release()          

class Trafficlight(Machine):
    def __init__(self):
        self.timeoutdic = {'run_green': 3, 'run_red': 5, 'run_yellow': 2}
        states=['paused',{'name':'run', 'children':['green','yellow','red']}]
        transitions = [
                       { 'trigger': 'next', 'source': 'run_green', 'dest': 'run_yellow'},
                       { 'trigger': 'next', 'source': 'run_yellow', 'dest': 'run_red'},
                       { 'trigger': 'next', 'source': 'run_red', 'dest': 'run_green'},
                       {'trigger':'pause','source': 'run', 'dest':'paused', 'before':'storestate'},
                       {'trigger':'pause','source':'paused','dest':'run', 'after':'restorestate'},
                       {'trigger':'reset', 'source':'*','dest':'run_green'}
                       ]
        Machine.__init__(self, states=states,transitions=transitions, initial='run_green')
        self.reinstall()
    def reinstall(self):
        self.thread=ThreadClass(self) 
        self.oldtime=None
        self.elapsedtime=None
        self.oldstate=None
    def printstatetimeout(self):
        self.oldtime=time()
        self.oldstate=self.state
        if self.elapsedtime:
            naptime=self.timeoutdic[self.state]-self.elapsedtime
            self.elapsedtime=None
        else:
            naptime=self.timeoutdic[self.state]
        print('Light is '+self.state+' and on for '+str(naptime)+'s')
        return naptime
    def start(self):
        self.thread.start()
    def stop(self):
        print('Traffic light is turned off')
        print('Waiting '+str(self.timeoutdic[self.state])+'s for all processes to finish')
        self.thread.stop()
        sleep(self.timeoutdic[self.state])
        self.reset()
        self.reinstall()
    def storestate(self):
        print('Light is turned off temporarily')
        self.elapsedtime=time()-self.oldtime
        self.thread.pause()
    def restorestate(self):
        self.set_state(self.oldstate)
        self.thread.pause()  

trafficlight = Trafficlight()
trafficlight.start()  
sleep(10)
trafficlight.stop()   
trafficlight.start()  
sleep(11)
trafficlight.pause() 
sleep(5)   # TODO: False returned why!?
# Resumes traffic light, interruption is accounted for see; storestate, restorestate
trafficlight.pause()  

[wtgee]

Hi @hstarmans,

I don't see the need for the Thread and it is probably giving you problems the way it is set up. Are you just trying to get some kind of loop that actually runs the machine or do you have a need for threads?

[hstarmans]

Hi @wtgee, My idea is indeed to use a separate loop to run the actual machine, do computations and listen to responses.

[tyarkoni]

Glad you like the package. I'm also not sure that threading is needed here, but maybe @aleneum can comment, since they contributed the threading functionality. Also, take a look at #18 and #36 in case they address your issue.

That said, please submit questions about package usage on Stack Overflow; GitHub issues are for bugs, feature requests, etc. If you think that the False/True values being returned reflect a bug (off the top of my head, I'm not sure what would be causing this, but I don't think that's been flagged as an issue before), you can open a separate issue for that, but please add a minimal code sample that reproduces the problem. Thanks!

[aleneum]

Hello @hstarmans,

triggered Events return whether the transition was successful or not. pause should return True because you try to transit from a substate of run into another state and this transition should be successful. I tested your code and that I could observe:

Light is run_green and on for 3s
Light is run_yellow and on for 2s
Light is run_red and on for 5s
Traffic light is turned off
Waiting 5s for all processes to finish
Stop pressed, traffic light stops
Light is run_green and on for 3s
Light is run_yellow and on for 2s
Light is run_red and on for 5s
Light is run_green and on for 3s
Light is turned off temporarily
True # <-- I changed the code to 'print trafficlight.pause()'
Pause pressed

Looks fine to me.

Your code does not work for python 2.7 because super requires an argument like super(ThreadClass, self).__init__() before python 3.

The LockedMachine RLock is not meant for public access. I decided to make in public rather than protected (_lock) since LockedTransition requires to acquire it too. However, maybe this leads to confusion and a protected _lock would have been the better choice. I will consider changing it as well as extending the documention of the HSM.

Best regards, Alex

[hstarmans]

Dear @aleneum, thank you for your reply. The fault is in my system, not transitions, see #64. If i run the code with ipython from the shell, it works. If I run the code using the interactive in Eclipse; Eclipse IDE Version: Mars.1 Release (4.5.1) Build id: 20150924-1200, PyDev for Eclipse 4.5.1.201601132212, Ubuntu 14.04 LTS, Python 3.4 Oct 14 2015, ipython 4.0.1, latest git clone from Transitions. I , still, get False if I transitions between states. I have not been able to resolve it. I look forward to the extended documentation. I am especially interested in something similar to the above, that is a machine which is handled by a background process and user can still control in the foreground.

[wtgee]

Hi @hstarmans. I've created a machine that runs as you describe using asyncio (in python3.5). I'm not sure we ever want to pull this into the machine itself as it is pretty light-weight, but I can put together a document describing some of the process. I'll try to get to that this weekend. Thanks

[hstarmans]

Hey @wtgee , I am definitely interested in other solutions. So please share! My objective is to use a flask webserver to control a statemachine.

[hstarmans]

Hey @wtgee, @aleneum and @tyarkoni, thank you for your suggestions. I updated the trafficlight example and moved from threads to coroutines as @wtgee suggested. Control is done via Flask and Tornado with a very basic web frontend. It is based on Python 3.4, so the syntax is a bit dated (no async and await). It might need some further polishing but should run.

from flask import Flask, render_template, jsonify 
from tornado.wsgi import WSGIContainer
from tornado.web import FallbackHandler, Application
from tornado.ioloop import IOLoop
from tornado import gen
from tornado.locks import Event
from transitions import LockedHierarchicalMachine as Machine
from time import time

class Trafficlight(Machine):
    def __init__(self):
        self.timeoutdic = {'run_green': 3, 'run_red': 5, 'run_yellow': 2}
        states=['paused',{'name':'run', 'children':['green','yellow','red']}]
        transitions = [
                       { 'trigger': 'next', 'source': 'run_green', 'dest': 'run_yellow'},
                       { 'trigger': 'next', 'source': 'run_yellow', 'dest': 'run_red'},
                       { 'trigger': 'next', 'source': 'run_red', 'dest': 'run_green'},
                       {'trigger':'pause','source': 'run', 'dest':'paused', 'before':'storestate', 'conditions':'is_running'},
                       {'trigger':'pause','source':'paused','dest':'run', 'after':'restorestate','unless':'is_running'},
                       {'trigger':'stop', 'source':'run','dest':'run_green','after':'stopafter','conditions':'is_running'},
                       {'trigger':'stop', 'source':'paused','dest':'run_green','after':['restorestate','stopafter'],'conditions':'is_paused'}
                       ]
        Machine.__init__(self, states=states,transitions=transitions, initial='run_green')
        self.running=False
        self.stopped=Event()
        self.reinstall()
    def reinstall(self):
        self.paused=Event()
        self.oldtime=None
        self.elapsedtime=None
        self.oldstate=None
    def is_running(self):
        try:
            running=self.running.running()
        except:
            running=self.running
        return running
    def is_paused(self):
        return self.running.is_set() and self.paused.is_set()
    def printstatetimeout(self):
        self.oldtime=time()
        self.oldstate=self.state
        if self.elapsedtime:
            naptime=self.timeoutdic[self.state]-self.elapsedtime
            self.elapsedtime=None
        else:
            naptime=self.timeoutdic[self.state]
        print('Light is '+self.state+' and on for '+str(naptime)+'s')
        return naptime
    @gen.coroutine
    def loop(self):
        while(True):
                nxt = gen.sleep(self.printstatetimeout())
                yield nxt    
                if self.stopped.is_set():
                    print("Stop pressed, traffic light stops")
                    self.running=False
                    self.stopped=Event()
                    break
                elif self.paused.is_set():
                    print("Pause pressed")
                    break
                else:
                    self.next()
    def start(self):
        if self.is_running():
            print("Traffic light already running")   
        else:
            self.running=self.loop()
    def stopafter(self):
        print('Traffic light is turned off')
        print('Waiting '+str(self.timeoutdic[self.state])+'s for all processes to finish')
        self.stopped.set()
        self.reinstall()
    def storestate(self):
        print('Light is turned off temporarily')
        self.elapsedtime=time()-self.oldtime
        self.paused.set()
    def restorestate(self):
        self.paused.clear()
        self.set_state(self.oldstate)
        self.running=self.loop() 

trafficlight=Trafficlight()

# Flask
#
# Flask application
app = Flask(__name__)

@app.route("/")
def index():
    return render_template('main.html')
@app.route('/_pause') 
def pause():
    print("Command pause received")
    trafficlight.pause()
    return jsonify()
@app.route('/_stop') 
def stop():
    print("Command stop recevied")
    trafficlight.stop()
    return jsonify()
@app.route('/_start') 
def start():
    print("Command start received")
    trafficlight.start()
    return jsonify()

tr = WSGIContainer(app)
application = Application([
(r".*", FallbackHandler, dict(fallback=tr)),
])

if __name__ == "__main__":  
    print("Server started")
    application.listen(5000)
    IOLoop.instance().start()

The main.html file is as follows:

<!doctype html>
<html lang="en">
    <head>
    <!-- User access to intenet is required -->
    <script src="//ajax.googleapis.com/ajax/libs/jquery/1.9.1/jquery.min.js"></script>
    <script type=text/javascript>
        // Buttons
        function buttonpause(){
        $('#buttonpause').bind('click', function() {
            $.getJSON('/_pause', {}, function() {});
            })
        }

        function buttonstop(){
            $('#buttonstop').bind('click', function() {
                $.getJSON('/_stop', {}, function() {});
                })
            }

        function buttonstart(){
            $('#buttonstart').bind('click', function() {
                $.getJSON('/_start', {}, function() {});
                })
            }

        // Calls documentloaded after page and binds buttons
        function documentloaded(){
                buttonstart();
                buttonstop();
                buttonpause();
            }
        $(function() {documentloaded()});

     </script>
  </head>
  <body>
    <button type="button" id="buttonstart">Start</button>
    <button type="button" id="buttonstop">Stop</button>
    <button type="button" id="buttonpause">Pause</button>
  </body>
</html>