device_message_listaner mqtt python mysql

Author: dexif

README.md

@@ -43,7 +43,7 @@ A simple example of using mqtt to receive and display telemetry data from device
 
 * [HTML+JS+CSS](mqtt-telemetry-dashboard) with [mqtt.js](https://github.com/mqttjs/MQTT.js)
 
-## MQTT channel message handler
+## MQTT device message handler
 
 * [Python + MySQL](mqtt-message-handler/python)
 

mqtt-message-handler/python/README.md

@@ -1,8 +1,8 @@
-# flespi.io MQTT channel message listener
+# flespi.io MQTT device message listener
 
 MySQL database, [gmqtt](https://github.com/wialon/gmqtt) library, python 3.5+
 
-**Example:** [channel_message_listener.py](channel_message_listener.py)
+**Example:** [device_message_listener.py](device_message_listener.py)
 
 First of all you should **create a new table** in your database to handle messages.
 
@@ -27,7 +27,7 @@ First of all you should **create a new table** in your database to handle messag
 
 **Also you should configure the listener by changing these configuration lines:**
 
-    channel_id = '12345'
+    device_id = '12345'
     flespi_mqtt_host = 'mqtt.flespi.io'
     flespi_token = 'YOUR FLESPI TOKEN'
     mysql_host = 'localhost'
@@ -41,4 +41,4 @@ First of all you should **create a new table** in your database to handle messag
 
 **Then start the example:**
 
-    python3 channel_message_listener.py
+    python3 device_message_listener.py

mqtt-message-handler/python/device_message_listener.py

@@ -0,0 +1,121 @@
+device_id = '12345' # you can use + to subscrube to all devices or list device IDs separated by commas (123,124,135)
+flespi_mqtt_host = 'mqtt.flespi.io'
+flespi_token = 'YOUR FLESPI TOKEN'
+mysql_host = 'localhost'
+mysql_user = 'root'
+mysql_passwd = 'password'
+mysql_db = 'message_receiver'
+
+# First of all you should create a new table in your database to handle messages.
+# You can change columns as you wish. But then you should change the respective insert query in "on_message" function.
+#
+# CREATE TABLE message_receiver.python_message_listener (
+# 	ident varchar(100) NOT NULL,
+# 	`server.timestamp` DOUBLE NOT NULL,
+# 	`position.longitude` DOUBLE NULL,
+# 	`position.latitude` DOUBLE NULL,
+# 	`timestamp` DOUBLE NULL,
+# 	`position.altitude` DOUBLE NULL,
+# 	`position.direction` DOUBLE NULL,
+# 	`position.speed` DOUBLE NULL,
+# 	`position.satellites` INT NULL,
+# 	CONSTRAINT python_message_listener_PK PRIMARY KEY (ident,`server.timestamp`,`timestamp`)
+# )
+# ENGINE=MyISAM
+# DEFAULT CHARSET=utf8
+# COLLATE=utf8_general_ci;
+
+
+import asyncio
+import os
+import signal
+import time
+import json
+import pymysql
+import uvloop
+from gmqtt import Client as MQTTClient
+
+
+
+asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+STOP = asyncio.Event()
+con = None
+counter = 0
+
+def on_connect(client, flags, rc, properties):
+    print('Connected')
+    client.subscribe('flespi/message/gw/devcies/' + device_id, qos=0)
+
+
+def on_message(client, topic, payload, qos, properties):
+    global counter, con
+    data = json.loads(payload)
+    if con is not None:
+        cursor = con.cursor()
+        cursor.execute(
+            """
+                INSERT INTO python_message_listener (
+                    `ident`,
+                    `server.timestamp`,
+                    `position.longitude`,
+                    `position.latitude`,
+                    `timestamp`,
+                    `position.altitude`,
+                    `position.direction`,
+                    `position.speed`,
+                    `position.satellites`
+                ) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
+            """,
+            (
+                data["ident"],
+                data["server.timestamp"],
+                data["position.longitude"],
+                data["position.latitude"],
+                data["timestamp"],
+                data["position.altitude"],
+                data["position.direction"],
+                data["position.speed"],
+                data["position.satellites"]
+            )
+            )
+        counter+=1
+        con.commit()
+    else:
+        print('Not inserted', json.dumps(data, indent=4))
+
+
+def on_disconnect(client, packet, exc=None):
+    print('Disconnected')
+    if con is not None:
+        con.close()
+
+def on_subscribe(client, mid, qos, properties):
+    print('SUBSCRIBED')
+
+def ask_exit(*args):
+    print('Inserted:', counter)
+    STOP.set()
+
+async def main(broker_host, token):
+    global con
+    con = pymysql.connect(host = mysql_host, user = mysql_user, passwd = mysql_passwd, db = mysql_db, autocommit=True)
+    client = MQTTClient('message_listener')
+
+    client.on_connect = on_connect
+    client.on_message = on_message
+    client.on_disconnect = on_disconnect
+    client.on_subscribe = on_subscribe
+
+    client.set_auth_credentials(token, None)
+    await client.connect(broker_host)
+
+    await STOP.wait()
+    await client.disconnect()
+
+
+if __name__ == '__main__':
+    loop = asyncio.get_event_loop()
+    loop.add_signal_handler(signal.SIGINT, ask_exit)
+    loop.add_signal_handler(signal.SIGTERM, ask_exit)
+
+    loop.run_until_complete(main(flespi_mqtt_host, flespi_token))