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?
Thank you for your suggestions, as they definitely helped me research the topic, but I finally figured out how to get threading working.
Turns out my issue was returning the value to a list for the main program to use.
By providing the list notes_pressed as an argument to get_next_note(), it ensures that the list is updated when return_pressed_notes() is called.
Here is the code I used to solve my issue:
from Flask import Flask
import mido
from threading import Thread
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 note
def get_next_note(notes_pressed):
# Open port to listen for note presses
with mido.open_input() as inport:
# Retreive key presses from port
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))
# Run main program
if __name__ == '__main__':
# NEW CODE
p = Thread(target=get_next_note, args=(notes_pressed,))
p.start()
app.run(debug=True, use_reloader=False)
p.join()
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?
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.