52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
from paho.mqtt import client as mqtt_client
|
|
from schema.aggregated_data_schema import AggregatedDataSchema
|
|
from file_datasource import FileDatasource
|
|
import logging
|
|
import config
|
|
|
|
|
|
def connect_mqtt(broker, port):
|
|
"""Create MQTT client"""
|
|
print(f"CONNECT TO {broker}:{port}")
|
|
|
|
def on_connect(client, userdata, flags, rc):
|
|
if rc == 0:
|
|
print(f"Connected to MQTT Broker ({broker}:{port})!")
|
|
else:
|
|
print("Failed to connect {broker}:{port}, return code %d\n", rc)
|
|
exit(rc) # Stop execution
|
|
|
|
logging.info(f"Acting as USER_ID = {config.USER_ID}")
|
|
|
|
client = mqtt_client.Client()
|
|
client.on_connect = on_connect
|
|
client.connect(broker, port)
|
|
client.loop_start()
|
|
return client
|
|
|
|
|
|
def publish(client, topic, datasource):
|
|
datasource.startReading()
|
|
while True:
|
|
data = datasource.read()
|
|
msg = AggregatedDataSchema().dumps(data)
|
|
result = client.publish(topic, msg)
|
|
logging.debug(f"Published to {topic}: {msg[:50]}...")
|
|
status = result[0]
|
|
if status != 0:
|
|
logging.error(f"Failed to send message to topic {topic}")
|
|
|
|
|
|
def run():
|
|
logging.basicConfig(level = logging.INFO)
|
|
# Prepare mqtt client
|
|
client = connect_mqtt(config.MQTT_BROKER_HOST, config.MQTT_BROKER_PORT)
|
|
# Prepare datasource
|
|
datasource = FileDatasource(16384.0, "data/accelerometer.csv", config.GPS_SOURCE, "data/parking.csv")
|
|
# Infinity publish data
|
|
publish(client, config.MQTT_TOPIC, datasource)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|