alex/sensecap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
alex
2024-12-27 15:54:09 +01:00
parent eb1da74943
commit 5272a044e6
1 changed files with 31 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
mqtt_datalogger.py
View File

@@ -2,11 +2,16 @@ import paho.mqtt.client as mqtt
import psycopg2 import psycopg2
import json import json
import time import time
import logging
import os import os
import uvicorn import uvicorn
from fastapi import FastAPI, HTTPException from fastapi import FastAPI, HTTPException
from pydantic import BaseModel 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 # Configurazione connessione PostgreSQL
DB_CONFIG = { DB_CONFIG = {
"dbname": os.getenv("DB_NAME"), "dbname": os.getenv("DB_NAME"),
@@ -44,7 +49,7 @@ def connect_db():
) )
return connection return connection
except Exception as e: except Exception as e:
print(f"Errore di connessione al database: {e}") logger.error(f"Errore di connessione al database: {e}")
return None return None
def setup_database(): def setup_database():
@@ -72,8 +77,9 @@ def setup_database():
); );
""") """)
connection.commit() connection.commit()
logger.info("Database configurato correttamente.")
except Exception as e: except Exception as e:
print(f"Errore nella configurazione del database: {e}") logger.error(f"Errore nella configurazione del database: {e}")
finally: finally:
cursor.close() cursor.close()
connection.close() connection.close()
@@ -88,8 +94,9 @@ def insert_device_status(payload):
(payload["deviceEui"], payload["events"][0]["name"], json.dumps(payload)) (payload["deviceEui"], payload["events"][0]["name"], json.dumps(payload))
) )
connection.commit() connection.commit()
logger.info(f"Stato del dispositivo inserito: {payload['deviceEui']}")
except Exception as e: except Exception as e:
print(f"Errore durante l'inserimento dello stato del dispositivo: {e}") logger.error(f"Errore durante l'inserimento dello stato del dispositivo: {e}")
finally: finally:
cursor.close() cursor.close()
connection.close() connection.close()
@@ -107,9 +114,10 @@ def insert_sensor_data(payload):
(payload["deviceEui"], value["channel"], json.dumps(value["measurements"]), value["measureTime"]) (payload["deviceEui"], value["channel"], json.dumps(value["measurements"]), value["measureTime"])
) )
connection.commit() connection.commit()
logger.info(f"Dati del sensore inseriti per il canale {value['channel']} del dispositivo {payload['deviceEui']}")
cursor.close() cursor.close()
except Exception as e: except Exception as e:
print(f"Errore durante l'inserimento dei dati dei sensori: {e}") logger.error(f"Errore durante l'inserimento dei dati dei sensori: {e}")
finally: finally:
connection.close() connection.close()
@@ -136,9 +144,9 @@ def send_downlink_command(client, device_eui, command_id, value=None):
payload = json.dumps(command) payload = json.dumps(command)
client.publish(topic, payload, qos=0, retain=True) client.publish(topic, payload, qos=0, retain=True)
print(f"Comando inviato a {topic}: {payload}") logger.info(f"Comando inviato a {topic}: {payload}")
except Exception as e: except Exception as e:
print(f"Errore durante l'invio del comando: {e}") logger.error(f"Errore durante l'invio del comando: {e}")
# Endpoint API per inviare comandi@app.post("/send_command") # Endpoint API per inviare comandi@app.post("/send_command")
def api_send_command(command: Command): def api_send_command(command: Command):
@@ -146,6 +154,7 @@ def api_send_command(command: Command):
send_downlink_command(client, command.device_eui, command.command_id, command.value) send_downlink_command(client, command.device_eui, command.command_id, command.value)
return {"status": "success", "message": "Comando inviato con successo"} return {"status": "success", "message": "Comando inviato con successo"}
except Exception as e: 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}") raise HTTPException(status_code=500, detail=f"Errore nell'invio del comando: {e}")
@app.get("/device_status") @app.get("/device_status")
@@ -156,8 +165,10 @@ def get_device_status():
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(f"SELECT * FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['table_dev_status']} ORDER BY timestamp DESC LIMIT 50;") cursor.execute(f"SELECT * FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['table_dev_status']} ORDER BY timestamp DESC LIMIT 50;")
results = cursor.fetchall() results = cursor.fetchall()
logger.info("Stato dei dispositivi recuperato.")
return {"status": "success", "data": results} return {"status": "success", "data": results}
except Exception as e: 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}") raise HTTPException(status_code=500, detail=f"Errore durante il recupero dello stato dei dispositivi: {e}")
finally: finally:
cursor.close() cursor.close()
@@ -172,8 +183,10 @@ def get_sensor_data():
cursor = connection.cursor() cursor = connection.cursor()
cursor.execute(f"SELECT * FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['table_sensor_data']} ORDER BY timestamp DESC LIMIT 50;") cursor.execute(f"SELECT * FROM {DB_CONFIG['dbschema']}.{DB_CONFIG['table_sensor_data']} ORDER BY timestamp DESC LIMIT 50;")
results = cursor.fetchall() results = cursor.fetchall()
logger.info("Dati dei sensori recuperati.")
return {"status": "success", "data": results} return {"status": "success", "data": results}
except Exception as e: 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}") raise HTTPException(status_code=500, detail=f"Errore durante il recupero dei dati dei sensori: {e}")
finally: finally:
cursor.close() cursor.close()
@@ -181,17 +194,17 @@ def get_sensor_data():
raise HTTPException(status_code=500, detail="Errore durante il recupero dei dati") raise HTTPException(status_code=500, detail="Errore durante il recupero dei dati")
# Callback MQTT # Callback MQTT
def on_connect(client, userdata, flags, rc): def on_connect(client, userdata, flags, rc, ppp):
try: try:
print("Connesso al broker MQTT con codice di stato:", rc) logger.info("Connesso al broker MQTT con codice di stato: %s", rc)
for topic in MQTT_TOPICS: for topic in [MQTT_CONFIG['topics']]:
client.subscribe(topic) client.subscribe(topic)
print(f"Sottoscritto al topic: {topic}") logger.info(f"Sottoscritto al topic: {topic}")
except Exception as e: except Exception as e:
print(f"Errore durante la connessione al broker MQTT: {e}") logger.error(f"Errore durante la connessione al broker MQTT: {e}")
def on_message(client, userdata, msg): def on_message(client, userdata, msg):
print(f"Messaggio ricevuto su {msg.topic}: {msg.payload}") logger.info(f"Messaggio ricevuto su {msg.topic}: {msg.payload}")
try: try:
payload = json.loads(msg.payload) payload = json.loads(msg.payload)
if "events" in payload: if "events" in payload:
@@ -200,11 +213,11 @@ def on_message(client, userdata, msg):
elif payload["events"][0]["name"] == "measure-sensor": elif payload["events"][0]["name"] == "measure-sensor":
insert_sensor_data(payload) insert_sensor_data(payload)
except Exception as e: except Exception as e:
print(f"Errore nella gestione del messaggio: {e}") logger.error(f"Errore nella gestione del messaggio: {e}")
# Configurazione client MQTT # Configurazione client MQTT
client = mqtt.Client() client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD) client.username_pw_set(MQTT_CONFIG['username'], MQTT_CONFIG['password'])
client.on_connect = on_connect client.on_connect = on_connect
client.on_message = on_message client.on_message = on_message
@@ -212,11 +225,11 @@ client.on_message = on_message
def main(): def main():
setup_database() setup_database()
try: try:
client.connect(MQTT_BROKER, MQTT_PORT, 60) client.connect(MQTT_CONFIG['broker'], int(MQTT_CONFIG['port']), 60)
client.loop_start() client.loop_start()
print("Applicazione avviata. Accesso all'interfaccia web su http://127.0.0.1:8000") logger.info("Applicazione avviata. Accesso all'interfaccia web su http://127.0.0.1:8000")
except Exception as e: except Exception as e:
print(f"Errore nella connessione al broker MQTT: {e}") logger.error(f"Errore nella connessione al broker MQTT: {e}")
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn import uvicorn