Estoy intentando ejecutar una función en el fondo de una aplicación de matraz. Actualmente, la aplicación intenta leer la entrada de un teclado midi y enviar la información a un sitio web. He investigado la creación de subprocesos, pero mis hallazgos/intentos hasta ahora no han tenido éxito.
Esencialmente, JavaScript/JQuery solicita datos de la ruta de la aplicación '/get_notes' cada segundo. Teóricamente, si get_notes() se ejecuta entre estos segundos, puede enviar una lista con formato json que contiene todas las teclas presionadas en ese período de tiempo específico.
He parafraseado e incluido el código relevante de app.py a continuación. ¡Cualquier sugerencia sera apreciada!
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?Gracias por sus sugerencias, ya que definitivamente me ayudaron a investigar el tema, pero finalmente descubrí cómo hacer que los hilos funcionen.
Resulta que mi problema fue devolver el valor a una lista para que lo use el programa principal.
Al proporcionar la lista notes_pressed como argumento para get_next_note() , garantiza que la lista se actualice cuando se llame a return_pressed_notes() .
Aquí está el código que usé para resolver mi problema:
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()Puedes usar un APScheduler
ejemplo:
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()¿Qué opinas de este método?
Me he encontrado con un problema similar al crear algunos juegos de casino en línea usando matraz.
La solución que encontré fue ejecutar una tarea de apio con un bucle infinito y guardar el estado actual de la tarea para que el matraz pueda acceder a ella en cualquier momento.
Celery es una solución de grado de producción para ejecutar tareas en segundo plano que funciona bien con el matraz.
No voy a profundizar en el apio aquí ya que es un tema complejo, pero esta guía es excelente para implementar el apio con frasco: https://blog.miguelgrinberg.com/post/using-celery-with-flask
En su caso, no estoy realmente seguro de cómo funciona la biblioteca mido, pero según su pregunta parece que "with mido.open_input() as inport:" ya abre un canal de E/S y lo mantiene perpetuamente. En este caso, ni siquiera necesitará volver a formatear la función, solo agregue la lista de salida como el estado de la tarea.
@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})Luego puede crear dos puntos finales diferentes en su aplicación de matraz. Uno que inicializa la tarea de apio y otro que lee su estado:
@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)Avíseme si eso funcionaría para usted y si hay algún problema al usar mido con apio.