238 lines
8.9 KiB
Python
238 lines
8.9 KiB
Python
import paho.mqtt.client as mqtt
|
|
import psycopg2
|
|
import json
|
|
import time
|
|
import logging
|
|
import os
|
|
import uvicorn
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
# Configurazione Logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 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:
|
|
logger.error(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()
|
|
logger.info("Database configurato correttamente.")
|
|
except Exception as e:
|
|
logger.error(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()
|
|
logger.info(f"Stato del dispositivo inserito: {payload['deviceEui']}")
|
|
except Exception as e:
|
|
logger.error(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()
|
|
logger.info(f"Dati del sensore inseriti per il canale {value['channel']} del dispositivo {payload['deviceEui']}")
|
|
cursor.close()
|
|
except Exception as e:
|
|
logger.error(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)
|
|
logger.info(f"Comando inviato a {topic}: {payload}")
|
|
except Exception as e:
|
|
logger.error(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:
|
|
logger.error(f"Errore nell'invio del comando: {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()
|
|
logger.info("Stato dei dispositivi recuperato.")
|
|
return {"status": "success", "data": results}
|
|
except Exception as e:
|
|
logger.error(f"Errore durante il recupero dello stato dei dispositivi: {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()
|
|
logger.info("Dati dei sensori recuperati.")
|
|
return {"status": "success", "data": results}
|
|
except Exception as e:
|
|
logger.error(f"Errore durante il recupero dei dati dei sensori: {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, ppp):
|
|
try:
|
|
logger.info("Connesso al broker MQTT con codice di stato: %s", rc)
|
|
for topic in [MQTT_CONFIG['topics']]:
|
|
client.subscribe(topic)
|
|
logger.info(f"Sottoscritto al topic: {topic}")
|
|
except Exception as e:
|
|
logger.error(f"Errore durante la connessione al broker MQTT: {e}")
|
|
|
|
def on_message(client, userdata, msg):
|
|
logger.info(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:
|
|
logger.error(f"Errore nella gestione del messaggio: {e}")
|
|
|
|
# Configurazione client MQTT
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
|
|
client.username_pw_set(MQTT_CONFIG['username'], MQTT_CONFIG['password'])
|
|
client.on_connect = on_connect
|
|
client.on_message = on_message
|
|
|
|
# Avvio
|
|
def main():
|
|
setup_database()
|
|
try:
|
|
client.connect(MQTT_CONFIG['broker'], int(MQTT_CONFIG['port']), 60)
|
|
client.loop_start()
|
|
logger.info("Applicazione avviata. Accesso all'interfaccia web su http://127.0.0.1:8000")
|
|
except Exception as e:
|
|
logger.error(f"Errore nella connessione al broker MQTT: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
main()
|
|
uvicorn.run(app, host="127.0.0.1", port=8000)
|