Gráfica en tiempo real con Python y matplotlib

Actualización 2: He hecho un vídeo de Youtube con todo el proceso por si no se entiende algo! Disfrutadlo

Holas!

En la entrada anterior vimos como conectar la IMU y el ESP32 para leer los datos de aceleración y giroscopio, ahora vamos a ver como podemos hacer para mostrarlos por pantalla en una bonita gráfica en tiempo real. Para ello, nos vamos a ayudar de Python y la librería matplotlib. Además de esto, vamos a necesitar usar las librerías para ejecución en hilos (threading) y la librería para leer del serial (pySerial).

Preparación

Antes de empezar a pegarnos con el código necesitamos instalar las librerías que hemos comentado anteriormente. Por suerte la librería threading ya está incluida en las librerías incluidas por defecto. Sin embargo, pySerial y matplotlib no, y las tenemos que instalar. La manera más sencilla es hacer uso del gestor ‘pip’ para ello nos basta con ejecutar: pip install matplotlib pyserial. Con esto, la parte referente al Python está terminada pero aún hay que preparar los datos que se envían por el serial para ser tratados.

En nuestro caso, aunque se puede hacer de maneras más fáciles (y que explicaré al final del post), hemos decidido sacar el valor del ‘Roll’ leído de la IMU con el formato: printf("Roll: %.2f\n", roll);. De esta manera sabemos que si aparece la palabra Roll, luego va a venir un valor en coma flotante que tenemos que extraer. Lo que tendríamos en el serial sería algo tal que:

Roll: 0.25
Roll: 0.32
Roll: 0.33
Roll: 0.48
Roll: 1.50
Roll: 2.20
...

A picar Python!

Lo primero que tenemos que hacer es importar las librerías que usaremos en nuestro proyecto. En nuestro caso sabemos que necesitamos:

• matplotlib: Para generar y mostrar las gráficas. Por comodidad vamos a incluir solo ‘pyplot’ y ‘animation’ como ‘plt’ y ‘animation’.
• serial: Es la librería del ‘pySerial’, que usaremos para acceder a los datos del puerto serial.
• re: Vamos a usar expresiones regulares para extraer los datos de la cadena recibida por el puerto serial
• threading: Necesitamos generar hilos que vayan en paralelo a nuestro hilo principal.

¿Por qué necesitamos hilos?: En matplotlib el hilo principal tiene que ser el que lleve la gráfica mostrada, si intentásemos hacer la gráfica y el recolector de datos en el mismo hilo, tendríamos problemas ya que cuando se está ejecutando el recolector, la gráfica estaría completamente bloqueada y no podríamos mover la ventana o movernos por los valores. Es por eso que decidimos sacar el recolector en un hilo a parte que vaya rellenando los datos que luego la gráfica coge cada poco.

Esta parte del código quedaría:

import serial
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import re
import threading

Preparamos la gráfica

Una vez con las librerías cargadas tenemos que preparar el array de datos que vamos a compartir entre el hilo recolector y la gráfica, y la gráfica en sí. Para los datos vamos a generar una array bidimensional. Luego ponemos los límites de la gráfica a 200 muestras en el eje X (Vamos a hacer que cuando se superen las 200 muestras, la gráfica empiece a moverse borrando los datos anteriores) y el eje Y entre 90 y -90.

gData = []
gData.append([0])
gData.append([0])

#Configuramos la gráfica
fig = plt.figure()
ax = fig.add_subplot(111)
hl, = plt.plot(gData[0], gData[1])
plt.ylim(-90, 90)
plt.xlim(0,200)

Recolector serial

Para recoger los datos vamos a preparar la función que se ejecutará en otro hilo. La forma de hacerlo es: tenemos el array de datos globales, al cual pueden acceder tanto el hilo recolector como el hilo principal, así que vamos a recibir el array como argumento, y vamos a ir actualizándolo con los nuevos datos que tengamos del serial. Para ello:

• • Definimos la función ‘getData’: def GetData(out_data)
  • Abrimos el puerto serie, en nuestro caso el ‘dev/ttyUSB1’: with serial.Serial('/dev/ttyUSB1',115200, timeout=1) as ser:
  • Nos quedamos en un bucle infinito, leyendo todos los datos del serial. Si encontramos la palabra ‘Roll’ parseamos el texto.

while True:
    line = ser.readline().decode('utf-8')
    if "Roll" in line:
    #...

• • Usando una expresión regular podemos extraer el valor (Las expresiones regulares las trataremos en otro momento, de momento tened fe). Si ya hay 200 valores en la lista, añadimos el nuestro y quitamos el primero. Recordad que ‘out_data‘ es el array global de datos, que al entrar como parámetro ahora se llama out_data

res = re.search("Roll: ([-+]?(\d+(\.\d*)?|\.\d+)([eE][-+]?\d+)?)", line)
out_data[1].append( float(res.group(1)) )
if len(out_data[1]) > 200:
    out_data[1].pop(0)

Con esto tendríamos nuestra función que abre el serial, lee cada vez que hay datos, y si hay un ‘Roll’ en la línea actual, obtiene el valor flotante y lo añade a la lista para ser pintado por el hilo principal. Aun así, la función no se va a ejecutar si nadie le llama, nos quedaría configurar el hilo para que se comience a ejecutar en paralelo. Para ello hacemos uso de la librería threading que nos ofrece una API muy sencilla para poder lanzar funciones como si fuesen hilos separados. Con dos simples lineas de código ejecutamos la función y le pasamos el array global a la función para que se modifique el mismo que va a leer más adelante nuestra gráfica.

dataCollector = threading.Thread(target = GetData, args=(gData,))
dataCollector.start()

Y por fin pintamos algo!

Ya tenemos nuestro hilo recolector cogiendo datos del serial y rellenando el array global con ellos. Ya solo nos queda mostrar la gráfica y decirle que se actualice cada poco tiempo con los nuevos datos que hay en el array. Para ello usamos la función ‘FuncAnimation‘ del paquete ‘animation. En nuestro caso nos interesan los siguientes argumentos:

• fig: figura que tiene que ser actualizada
• func: función que se va a ejecutar cada cierto tiempo y que tiene que saber como actualizar la gráfica.
• fargs: lista con los argumentos que se tiene que pasar a func en cada ciclo.
• interval: para configurar el periodo de ejecución de la función.
• blit: parámetro para optimizar el repintado de la gráfica, lo ponemos a True.

Antes de poder usar el paquete ‘animation‘ necesitamos definir la función que se ejecutará y posteriormente la enlazamos. Según la documentación del paquete la definición de la función que se le puede pasar a ‘FuncAnimation‘ tiene que ser: def func(frame, *fargs) -> iterable_of_artists. Es decir, el primer argumento será el frame que se está ejecutando y luego una lista variable de argumentos. En nuestro caso vamos a pasarle el frame, el controlador de la línea que se está pintando y el array de datos que tiene que poner en la línea. La función se explica bastante bien por sí sola así que el resultado directamente quedaría:

def update_line(num, hl, data):
    #python 2.7
    hl.set_data(range(len(data[1])), data[1])
    #python 3.0
    hl.set_data(list(range(1,len(data)+1)), data)
    return hl,

line_ani = animation.FuncAnimation(fig, update_line, fargs=(hl, gData),
    interval=50, blit=False)

Pero si no se pinta nada todavía!

Ya está todo listo, pero queda lo más importante, mostrar la gráfica  plt.show() y unir los hilos de ejecución dataCollector.join() para que cuando muera uno, muera el otro. Así que os dejo ya todo el código junto para que podáis verlo.

import serial
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import re
import threading

gData = []
gData.append([0])
gData.append([0])

#Configuramos la gráfica
fig = plt.figure()
ax = fig.add_subplot(111)
hl, = plt.plot(gData[0], gData[1])
plt.ylim(-90, 90)
plt.xlim(0,200)

# Función que se va a ejecutar en otro thread
# y que guardará los datos del serial en 'out_data'
def GetData(out_data):
    with serial.Serial('/dev/ttyUSB1',115200, timeout=1) as ser:
        print(ser.isOpen())
        while True:
        line = ser.readline().decode('utf-8')
        # Si la línea tiene 'Roll' la parseamos y extraemos el valor
        if "Roll" in line:
        res = re.search("Roll: ([-+]?(\d+(\.\d*)?|\.\d+)([eE][-+]?\d+)?)", line)

        # Añadimos el nuevo valor, si hay más de 200 muestras quitamos la primera
        # para que no se nos acumulen demasiados datos en la gráfica
        out_data[1].append( float(res.group(1)) )
        if len(out_data[1]) > 200:
            out_data[1].pop(0)

# Función que actualizará los datos de la gráfica
# Se llama periódicamente desde el 'FuncAnimation'
def update_line(num, hl, data):
    hl.set_data(range(len(data[1])), data[1])
    return hl,

# Configuramos la función que "animará" nuestra gráfica
line_ani = animation.FuncAnimation(fig, update_line, fargs=(hl, gData),
    interval=50, blit=False)

# Configuramos y lanzamos el hilo encargado de leer datos del serial
dataCollector = threading.Thread(target = GetData, args=(gData,))
dataCollector.start()

plt.show()

dataCollector.join()

Mejoras

Las principales mejoras de este sistemas van relacionadas con como se envían los datos y como se extraen del serial. Si, no hace falta que los datos sean vistos por un humano según salen del serial (es decir, que no puedas ver el número de forma que lo entiendas), la mejor forma es mandando los bytes del float directamente y usando la librería struct.

Aún cuando los datos tienen que tener un formato humano, existe un método mejor y es sacando los datos como si fueran un CSV. Por ejemplo:

11.6855913537; 3.74792472633; 2.39604982471;
19.7334382385; 6.01010710602; 7.09775689259;
17.6626564028; 15.4836368254; 18.7422603611;
11.2272216627; 19.2792673438; 7.00984442616;
3.41966723958; 18.6907153699; 11.9929061892;
8.17878894973; 5.62236474639; 16.264799115;
13.6283387514; 19.8690485849; 3.55410108408;
10.2682397876; 8.97191210715; 7.98331495641;
7.55256049223; 7.20755374132; 17.7867241302;
16.3053496637; 6.83186213206; 17.1387359398;

De esta forma, si sabemos que va en cada columna podemos tratarlos con python de manera mucho más eficiente (la librería ‘re‘ es bastante pesada) con un simple ‘split‘

line = ser.readline().decode('utf-8') # Recibimos por ejemplo: '11.6855913537; 3.74792472633; 2.39604982471;'
column = 0

for i in line.split(";"): # Recorremos los campos que hemos recibido
        try: # Hacemos un try, porque el último campo estará vacío, en ese caso nos saltará excepción, así que simplemente pasamos
        out_data(column).append(float(i)) # Asignamos a cada columna de los datos para pintar, la columna correspondiente del input
        column = column + 1
    except:
    pass

Con este método podéis mostrar varias columnas, recibiendo los datos en paralelo del puerto serie, y si sigue el mismo formato, se puede adaptar el getData a cualquier protocolo.

Espero que os haya servido de ayuda 😉
Dejad vuestras dudas en los comentarios