I'm attempting to run a function in the background of a flask app. The app currently attempts to read input from a midi keyboard and send the information to a website. I've looked into threading, but my findings/attempts so far have been unsuccessful.

Essentially, JavaScript/JQuery requests for data from the '/get_notes' app route every second. Theoretically, if get_notes() is running between these seconds it can send a json-formatted list containing all keys pressed in that specific time period.

I've paraphrased & included the relevant code from app.py below. Any suggestions would be appreciated!

from Flask import Flask
import mido

app = Flask(__name__)

# List to store pressed keys
notes_pressed = []

@app.route('/get_notes', methods=['GET'])
def return_pressed_notes():
    return json.dumps(notes_pressed)

# Function to translate midi key numbers to note letters
def translate_key(key_num):
    ...

# Function that returns recently played notes within time period
def get_notes():
    # Open port to listen for key presses
    with mido.open_input() as inport:
        for msg in inport:
            # If key press is valid
            # - note is pressed
            # - velocity!=0 (prevents ghost notes)
            if (msg.type=='note_on' and msg.velocity!=0 and msg.channel==0):
                # Add new note to list
                notes_pressed.append(translate_key(msg.note - lowest_key))

if __name__ == '__main__':
    app.run()
    # Somehow run get_notes() in background?

CodePudding user response：

You can use an APScheduler

example:

from Flask import Flask
from flask_apscheduler import APScheduler

app = Flask(__name__)
scheduler = APScheduler()
scheduler.init_app(app)

@scheduler.task('cron', id='get_note', hour=12) # every day at noon
def get_notes:
    return 'blablabla'

if __name__ == '__main__':
    scheduler.start()
    app.run()

What do you think about this method?

CodePudding user response：

I have run into a similar problem building some online cassino games using flask.

The solution I found was to run a celery task with an infinite loop and save the current state of the task so that flask can access it anytime.

Celery is a production grade solution for running background tasks that works well with flask.

I'm not going to dig deep into celery here as it is a complex topic, but these guide is excellent for implementing celery with flask: https://blog.miguelgrinberg.com/post/using-celery-with-flask

In your case, I am not really sure how the mido library works, but per your question it looks like "with mido.open_input() as inport:" already opens a I/O channel and maitains it perpetually. In this case you won't even need to reformat the function, just add the output list as the task's state.

@celery.task(bind=True)
def get_notes(self):
    notes_pressed = []
    # Open port to listen for key presses
    with mido.open_input() as inport:
        for msg in inport:
            # If key press is valid
            # - note is pressed
            # - velocity!=0 (prevents ghost notes)
            if (msg.type=='note_on' and msg.velocity!=0 and msg.channel==0):
                # Add new note to list
                notes_pressed.append(translate_key(msg.note - lowest_key))
                self.update_state(state="PROGRESS",
                                   meta={data: notes_pressed})

Then you can create two different endpoints in your flask app. One that initializes the celery task and another that reads its status:

@app.route('/longtask', methods=['POST'])
def longtask():
    task = get_notes.apply_async()
    return jsonify({}), 202, {'Location': url_for('taskstatus',
                                                  task_id=task.id)}

@app.route('/status/<task_id>')
def taskstatus(task_id):
    task = get_notes.AsyncResult(task_id)
    notes_pressed = task.info.get("data")
    #do whatever you want with it
    return json.dumps(notes_pressed)

Please let me know if that would work for you and if there are any issues using mido with celery.