Proyecto aTIC (IV) - Bridge MQTT/InfluxDB

-

 Bridge MQTT/InfluxDB. 

En el post anterior (Proyecto aTIC (III) - Broker MQTT) ya se ha comentado que la RPI2 funciona como bridge entre los datos publicados en el topic MQTT y la base de datos de InfluxDB. Esta función de brigde la realiza ejecutando un script en python que se define a continuación. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
from influxdb import InfluxDBClient
from typing import NamedTuple
import random
import time
import paho.mqtt.client as mqtt
import json
from ast import literal_eval

MQTT_CLIENT = 'RPI2'
MQTT_ADDRESS = 'aticrpi2.duckdns.org'
MQTT_TOPIC = 'Datos OBD'


def on_connect(client, userdata, flags, rc):    # Conexión y suscripción al topic
    print('Connected with result code ' + str(rc))
    client.subscribe(MQTT_TOPIC)


def on_message(client, userdata, msg):      #Callback cuando se recibe mensaje en topic
    json_message = [json.loads(msg.payload)]
    print (str(json_message))
    _send_sensor_data_to_influxdb(json_message)

INFLUXDB_ADDRESS = 'aticrpi1.duckdns.org'
INFLUXDB_USER = 'root'
INFLUXDB_PASSWORD = 'root'
INFLUXDB_DATABASE = 'OBD_DB'

influxdb_client = InfluxDBClient(INFLUXDB_ADDRESS, 8086, INFLUXDB_USER, INFLUXDB_PASSWORD, None)


def _send_sensor_data_to_influxdb(sensor_data):

    return influxdb_client.write_points(sensor_data)


def _init_influxdb_database():
    databases = influxdb_client.get_list_database()
    if len(list(filter(lambda x: x['name'] == INFLUXDB_DATABASE, databases))) == 0:
        influxdb_client.create_database(INFLUXDB_DATABASE)
    influxdb_client.switch_database(INFLUXDB_DATABASE)

def main():
    _init_influxdb_database()

    mqtt_client = mqtt.Client(MQTT_CLIENT)
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = on_message

    mqtt_client.connect(MQTT_ADDRESS)
    mqtt_client.loop_forever()


if __name__ == "__main__":
    print('Bridge running...')
    main()

Los pasos principales que ha de realizar la aplicación para funcionar como bridge correctamente se pueden resumir en los siguientes:

1.- Inicializar una tabla en InfluxDB para escribir los datos. Se realiza mediante la función _init_influxdb_database() definida también en este script con los datos de la base de datos de la RPI1.

2.- Acceder al broker de MQTT de la RPI2 y suscribirse al topic "Datos OBD". En el script se comienza definiendo los datos necesarios para establecer la conexión MQTT con el broker de la RPI2. Se define el cliente y con la función on_connect se suscribe al topic "Datos OBD".

3.- Mandar los datos recogidos del topic "Datos OBD" y enviarlos por HTTP a InfluxDB. Una vez suscrito al topic "Datos OBD", mediante la función on_message, la aplicación queda a la espera de que se publiquen datos en el topic, y cuando se recibe un dato, se guarda con formato json y se envía a la base de datos mediante HTTP con la función wite_points.


Para continuar al siguiente post del proyecto, accede a Proyecto aTIC (V) - Grafana.

Comentarios

Publicar un comentario

Entradas populares de este blog

Proyecto aTIC (I) - Introducción

-
Imagen
INTRODUCCIÓN. El objetivo principal del proyecto es realizar una comunicación HTTP con un servidor, aplicando los conocimimentos aprendidos en clase. El proyecto que se ha desarrollado no sólo realiza una comunicación HTTP con un servidor, sino que también se hace uso de MQTT y contenedores para transmitir datos desde un sensor a una base de datos. El diagrama principal del proyecto desarrollado se muestra en la siguiente figura, donde se ve cómo intervienen cuatro dispositivos: un sensor, un ordenador y dos RaspberryPi.   El sensor es un dispositivo conectado a la ECU de un vehículo que permite realizar lecturas de distintas variables del vehículo y se comunica por bluetooth, en este caso con un ordenador. Esta comunicación se ha configurado y realizado mediante un script de python que también realiza la tarea de publicar las lecturas realizadas en el broker MQTT alojado en la RPI2.  Esta RaspberryPi (RPI2), además, ejecuta una aplicación que se suscribe al topic MQTT donde e...
Leer más

Proyecto aTIC (V) - Grafana

-
Imagen
 Grafana.  Como se ha podido ver en Proyecto aTIC (IV) - Bridge MQTT/InfluxDB , la RPI2 también dispone de la herramienta Grafana. Ésta está asignada a su puerto 3000 y es una herramienta de software libre que permite crear cuadros de mando y gráficos ( dashboards ). Se ha escogido esta herramienta para la visualización ya que se puede implementar sencillamente con Docker y porque tiene un API HTTP completo, además de que es compatible con InfluxDB. En este proyecto esta herramienta ha sido empleada para crear el dashboard que se muestra a continuación. En él se visualizan las lecturas de velocidad, nivel de batería y revoluciones por minuto del vehículo al que está conectado el sensor. Se emplean distiontos tipo s de gráficos, así como distintos colores para diferenciar fácilmente las lecturas. Además, se han añadido las visualizaciones de los valores medios, mínimos y máximos de la velocidad y el nivel de batería.    Estos datos se obtienen de la base de datos de ...
Leer más