54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
import logging
|
|
import paho.mqtt.client as mqtt
|
|
from app.interfaces.agent_gateway import AgentGateway
|
|
from app.entities.agent_data import AgentData, GpsData
|
|
from app.usecases.data_processing import process_agent_data
|
|
from app.interfaces.hub_gateway import HubGateway
|
|
|
|
|
|
class AgentMQTTAdapter(AgentGateway):
|
|
def __init__(
|
|
self,
|
|
broker_host,
|
|
broker_port,
|
|
topic,
|
|
hub_gateway: HubGateway,
|
|
):
|
|
# MQTT
|
|
self.broker_host = broker_host
|
|
self.broker_port = broker_port
|
|
self.topic = topic
|
|
self.client = mqtt.Client()
|
|
# Hub
|
|
self.hub_gateway = hub_gateway
|
|
|
|
def on_connect(self, client, userdata, flags, rc):
|
|
if rc == 0:
|
|
logging.info("Connected to MQTT broker")
|
|
self.client.subscribe(self.topic)
|
|
else:
|
|
logging.info(f"Failed to connect to MQTT broker with code: {rc}")
|
|
|
|
def on_message(self, client, userdata, msg):
|
|
"""Processing agent data and sent it to hub gateway"""
|
|
try:
|
|
payload: str = msg.payload.decode("utf-8")
|
|
|
|
agent_data = AgentData.model_validate_json(payload, strict=True)
|
|
processed_data = process_agent_data(agent_data)
|
|
|
|
if self.hub_gateway.save_data(processed_data):
|
|
logging.info("Processed data successfully forwarded to the Hub.")
|
|
else:
|
|
logging.error("Failed to send data: Hub gateway is unavailable.")
|
|
except Exception as e:
|
|
logging.error(f"Error processing MQTT message: {e}")
|
|
|
|
def connect(self):
|
|
self.client.on_connect = self.on_connect
|
|
self.client.on_message = self.on_message
|
|
self.client.connect(self.broker_host, self.broker_port, 60)
|
|
|
|
def loop_forever(self):
|
|
self.client.loop_forever()
|