Skip to content

Context Manager with PyHades

When you need to run hades threads integrated with web services that have a main thread, you must run the application that contains the main thread inside a PyHades Context and pass the PyHades application object to this context.

Flask Integration

Many times, you have to execute this design pattern embedded in a web application.

In this section I show you how to integrate a state machine in an application with Flask

from flask import Flask
from pyhades import PyHades, PyHadesContext, State, PyHadesStateMachine


flask_app = Flask(__name__)
hades_app = PyHades()

@hades_app.define_machine(name='TrafficLight', interval=1.0, mode="async")
class TrafficLightMachine(PyHadesStateMachine):

    # states
    green  = State('Green', initial=True)
    yellow  = State('Yellow')
    red  = State('Red')

    # transitions
    slowdown = green.to(yellow)
    stop = yellow.to(red)
    go = red.to(green)

    # parameters
    time_left = 30

    def __init__(self, name):

        super().__init__(name)

    def on_slowdown(self):

        self.time_left = 3

    def on_stop(self):

        self.time_left = 20

    def on_go(self):

        self.time_left = 30

    def while_green(self):

        if self.time_left == 0:

            self.slowdown()

        self.time_left -= 1

    def while_yellow(self):

        if self.time_left == 0:

            self.stop()

        self.time_left -= 1

    def while_red(self):

        if self.time_left == 0:

            self.go()

        self.time_left -= 1

    def __str__(self):

        return f"{self.name}: {self.get_state()} - {self.time_left} second left."

traffic_light = hades_app.get_machine('TrafficLight')

@flask_app.route("/")
def hello_world():

    return f"<p>{traffic_light} </p>"


if __name__ == "__main__":

    with PyHadesContext(hades_app):

        flask_app.run()