initial
This commit is contained in:
224
mqtt_datalogger.py
Normal file
224
mqtt_datalogger.py
Normal file
@@ -0,0 +1,224 @@
|
||||
import paho.mqtt.client as mqtt
|
||||
import psycopg2
|
||||
import json
|
||||
import time
|
||||
import os
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
# Configurazione connessione PostgreSQL
|
||||
DB_CONFIG = {
|
||||
"dbname": os.getenv("DB_NAME"),
|
||||
"dbschema": os.getenv("DB_SCHEMA"),
|
||||
"dbtable": os.getenv("DB_TABLE"),
|
||||
"user": os.getenv("DB_USER"),
|
||||
"password": os.getenv("DB_PASSWORD"),
|
||||
"host": os.getenv("DB_HOST"),
|
||||
"port": os.getenv("DB_PORT"),
|
||||
"table_dev_status": os.getenv("DB_TABLE_DEV_STATUS"),
|
||||
"table_sensor_data": os.getenv("DB_TABLE_SENSOR_DATA")
|
||||
}
|
||||
|
||||
# Configurazione MQTT
|
||||
MQTT_CONFIG = {
|
||||
"broker": os.getenv("MQTT_BROKER"),
|
||||
"port": os.getenv("MQTT_PORT"),
|
||||
"username": os.getenv("MQTT_USERNAME"),
|
||||
"password": os.getenv("MQTT_PASSWORD"),
|
||||
"topics": os.getenv("MQTT_TOPICS")
|
||||
}
|
||||
|
||||
# Creazione applicazione FastAPI
|
||||
app = FastAPI()
|
||||
|
||||
# Connessione al database
|
||||
def connect_db():
|
||||
try:
|
||||
connection = psycopg2.connect(
|
||||
dbname=DB_CONFIG["dbname"],
|
||||
user=DB_CONFIG["user"],
|
||||
password=DB_CONFIG["password"],
|
||||
host=DB_CONFIG["host"],
|
||||
port=DB_CONFIG["port"]
|
||||
)
|
||||
return connection
|
||||
except Exception as e:
|
||||
print(f"Errore di connessione al database: {e}")
|
||||
return None
|
||||
|
||||
def setup_database():
|
||||
connection = connect_db()
|
||||
if connection:
|
||||
try:
|
||||
cursor = connection.cursor()
|
||||
cursor.execute(f"""
|
||||
CREATE TABLE IF NOT EXISTS {DB_CONFIG['dbschema']}.{DB_CONFIG['table_dev_status']} (
|
||||
id SERIAL PRIMARY KEY,
|
||||
device_eui VARCHAR(50),
|
||||
event_name VARCHAR(50),
|
||||
payload JSONB,
|
||||
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
""")
|
||||
cursor.execute(f"""
|
||||
CREATE TABLE IF NOT EXISTS {DB_CONFIG['dbschema']}.{DB_CONFIG['table_sensor_data']} (
|
||||
id SERIAL PRIMARY KEY,
|
||||
device_eui VARCHAR(50),
|
||||
channel VARCHAR(10),
|
||||
measurements JSONB,
|
||||
measure_time TIMESTAMP,
|
||||
timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
);
|
||||
""")
|
||||
connection.commit()
|
||||
except Exception as e:
|
||||
print(f"Errore nella configurazione del database: {e}")
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
def insert_device_status(payload):
|
||||
connection = connect_db()
|
||||
if connection:
|
||||
try:
|
||||
cursor = connection.cursor()
|
||||
cursor.execute(
|
||||
f"INSERT INTO {DB_CONFIG['dbschema']}.{DB_CONFIG['table_dev_status']} (device_eui, event_name, payload) VALUES (%s, %s, %s)",
|
||||
(payload["deviceEui"], payload["events"][0]["name"], json.dumps(payload))
|
||||
)
|
||||
connection.commit()
|
||||
except Exception as e:
|
||||
print(f"Errore durante l'inserimento dello stato del dispositivo: {e}")
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
|
||||
def insert_sensor_data(payload):
|
||||
connection = connect_db()
|
||||
if connection:
|
||||
try:
|
||||
for event in payload["events"]:
|
||||
if event["name"] == "measure-sensor":
|
||||
for value in event["value"]:
|
||||
cursor = connection.cursor()
|
||||
cursor.execute(
|
||||
f"INSERT INTO {DB_CONFIG['dbschema']}.{DB_CONFIG['table_sensor_data']} (device_eui, channel, measurements, measure_time) VALUES (%s, %s, %s, %s)",
|
||||
(payload["deviceEui"], value["channel"], json.dumps(value["measurements"]), value["measureTime"])
|
||||
)
|
||||
connection.commit()
|
||||
cursor.close()
|
||||
except Exception as e:
|
||||
print(f"Errore durante l'inserimento dei dati dei sensori: {e}")
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
# Modelli Pydantic per l'API
|
||||
class Command(BaseModel):
|
||||
device_eui: str
|
||||
command_id: int
|
||||
value: int = None
|
||||
|
||||
# Funzioni per inviare comandi al dispositivo
|
||||
def send_downlink_command(client, device_eui, command_id, value=None):
|
||||
try:
|
||||
topic = f"$SHADOW/ipnode/{device_eui}/get/config"
|
||||
command = {
|
||||
"timestamp": int(time.time() * 1000),
|
||||
"desire": {
|
||||
str(command_id): {
|
||||
"ver": str(int(time.time() * 1000))
|
||||
}
|
||||
}
|
||||
}
|
||||
if value is not None:
|
||||
command["desire"][str(command_id)]["value"] = value
|
||||
|
||||
payload = json.dumps(command)
|
||||
client.publish(topic, payload, qos=0, retain=True)
|
||||
print(f"Comando inviato a {topic}: {payload}")
|
||||
except Exception as e:
|
||||
print(f"Errore durante l'invio del comando: {e}")
|
||||
|
||||
# Endpoint API per inviare comandi@app.post("/send_command")
|
||||
def api_send_command(command: Command):
|
||||
try:
|
||||
send_downlink_command(client, command.device_eui, command.command_id, command.value)
|
||||
return {"status": "success", "message": "Comando inviato con successo"}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Errore nell'invio del comando: {e}")
|
||||
|
||||
@app.get("/device_status")
|
||||
def get_device_status():
|
||||
connection = connect_db()
|
||||
if connection:
|
||||
try:
|
||||
cursor = connection.cursor()
|
||||
cursor.execute(f"SELECT * FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['table_dev_status']} ORDER BY timestamp DESC LIMIT 50;")
|
||||
results = cursor.fetchall()
|
||||
return {"status": "success", "data": results}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Errore durante il recupero dello stato dei dispositivi: {e}")
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
raise HTTPException(status_code=500, detail="Errore durante il recupero dei dati")
|
||||
|
||||
@app.get("/sensor_data")
|
||||
def get_sensor_data():
|
||||
connection = connect_db()
|
||||
if connection:
|
||||
try:
|
||||
cursor = connection.cursor()
|
||||
cursor.execute(f"SELECT * FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['table_sensor_data']} ORDER BY timestamp DESC LIMIT 50;")
|
||||
results = cursor.fetchall()
|
||||
return {"status": "success", "data": results}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Errore durante il recupero dei dati dei sensori: {e}")
|
||||
finally:
|
||||
cursor.close()
|
||||
connection.close()
|
||||
raise HTTPException(status_code=500, detail="Errore durante il recupero dei dati")
|
||||
|
||||
# Callback MQTT
|
||||
def on_connect(client, userdata, flags, rc):
|
||||
try:
|
||||
print("Connesso al broker MQTT con codice di stato:", rc)
|
||||
for topic in MQTT_TOPICS:
|
||||
client.subscribe(topic)
|
||||
print(f"Sottoscritto al topic: {topic}")
|
||||
except Exception as e:
|
||||
print(f"Errore durante la connessione al broker MQTT: {e}")
|
||||
|
||||
def on_message(client, userdata, msg):
|
||||
print(f"Messaggio ricevuto su {msg.topic}: {msg.payload}")
|
||||
try:
|
||||
payload = json.loads(msg.payload)
|
||||
if "events" in payload:
|
||||
if payload["events"][0]["name"] == "change-device-status":
|
||||
insert_device_status(payload)
|
||||
elif payload["events"][0]["name"] == "measure-sensor":
|
||||
insert_sensor_data(payload)
|
||||
except Exception as e:
|
||||
print(f"Errore nella gestione del messaggio: {e}")
|
||||
|
||||
# Configurazione client MQTT
|
||||
client = mqtt.Client()
|
||||
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
|
||||
client.on_connect = on_connect
|
||||
client.on_message = on_message
|
||||
|
||||
# Avvio
|
||||
def main():
|
||||
setup_database()
|
||||
try:
|
||||
client.connect(MQTT_BROKER, MQTT_PORT, 60)
|
||||
client.loop_start()
|
||||
print("Applicazione avviata. Accesso all'interfaccia web su http://127.0.0.1:8000")
|
||||
except Exception as e:
|
||||
print(f"Errore nella connessione al broker MQTT: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
main()
|
||||
uvicorn.run(app, host="127.0.0.1", port=8000)
|
||||
Reference in New Issue
Block a user