EH/KIM/fr-a2-mqtt-event-handling by kimmynoto · Pull Request #961 · DataBytes-Organisation/Project-Echo

kimmynoto · 2026-08-05T12:56:45Z

Summary

This implements FR-A2 (Live MQTT Event Handling). There wasn't actually any MQTT connection in the codebase before this just a placeholder variable waiting for a handler that never got built. So this PR builds that whole piece from scratch such as connecting to the broker, tracking connection state, cleaning up incoming data, and letting users know when the connection drops or comes back.

What's in this PR

Built a new MQTT client (mqtt_client.py) that connects to the broker and subscribes to live event topics
Hooked it into the backend's startup, with error handling so a failed connection just retries quietly in the background instead of crashing the whole API
Added automatic reconnect with backoff
Added a /mqtt/connection-state endpoint so the connection status is actually checkable
Added normalization for incoming vocalization events so they come out in one consistent shape instead of raw/inconsistent MQTT payloads
Added a frontend piece that polls the connection state and shows a banner/toast (using the existing showPageBanner/showToast helpers) when the connection drops or recovers

How I tested it

Confirmed the MQTT client connects successfully via the container logs
Manually published a test message with mosquitto_pub and confirmed it comes out normalized correctly
Stopped and restarted the MQTT broker container to confirm the frontend banner/toast actually show up and clear correctly

- Build MQTT client service that connects to broker and subscribes to live event topics - Add graceful startup handling so a failed connection retries in background instead of crashing the API - Implement reconnect-with-backoff - Add /mqtt/connection-state endpoint to expose live connection status - Normalize incoming vocalization event payloads into a consistent shape - Add frontend polling + page banner/toast notifications showing connection state (connected/reconnecting/disconnected) and alerting users on connection loss/recovery

AnDo081105

P1 findings from the FR-A2 review.

- Subscribe to the correct live-event topic (MQTT_PUBLISH_URL / projectecho/engine/2) instead of the wrong hardcoded topic, and make broker/topic config env-var driven in docker-compose - Store normalized events (keyed by sensorId) and expose them via a new /mqtt/latest-events endpoint, so the frontend/dashboard can actually consume live data - Fix frontend polling: a failed connection-state request now shows an unavailable banner instead of silently logging to console

AnDo081105

Additional FR-A2 review findings.

- /mqtt/latest-events now sources data from the Backend/DB (Events collection) instead of the raw MQTT topic, matching the architecture: Engine prediction -> Backend/DB -> /mqtt/latest-events -> HMI -> map - Added show_latest_events() in hmi.py, reusing the same aggregate + species pattern already used by /events_time, so commonName/type/status/diet are real enriched values instead of placeholders - Restored the /mqtt/connection-state endpoint, which was accidentally dropped while cleaning up a duplicate route earlier - Verified end-to-end against real inserted test data: species metadata correctly enriched, events reaching and rendering on the map with no errors

AnDo081105 · 2026-08-21T04:24:39Z

+@app.get("/mqtt/latest-events", tags=["mqtt"])
+def mqtt_latest_events():
+    from app.routers.hmi import show_latest_events
+    return {"events": show_latest_events(limit=20)}


Suggested change

return {"events": show_latest_events(limit=20)}

@app.get("/mqtt/latest-events", tags=["mqtt"])

def mqtt_latest_events():

from app.routers.hmi import show_latest_events

events = [

{"eventType": "vocalization", **event}

for event in show_latest_events(limit=20)

]

events += [

event for event in get_latest_events()

if event.get("eventType") != "vocalization"

]

return {"events": events}

AnDo081105 · 2026-08-21T04:28:25Z

+
+MQTT_BROKER_URL = os.environ.get("MQTT_BROKER_URL", "ts-mqtt-server-cont")
+MQTT_BROKER_PORT = int(os.environ.get("MQTT_BROKER_PORT", 1883))
+MQTT_TOPIC = os.environ.get("MQTT_PUBLISH_URL", "projectecho/engine/2")


Suggested change

MQTT_TOPIC = os.environ.get("MQTT_PUBLISH_URL", "projectecho/engine/2")

MQTT_TOPICS = os.environ.get(

"MQTT_TOPICS",

"projectecho/engine/2,projectecho/movement,iot/data/test",

).split(",")

def on_connect(client, userdata, flags, rc):

global connection_state

connection_state = "connected"

for topic in MQTT_TOPICS:

client.subscribe(topic.strip())

AnDo081105 · 2026-08-21T04:32:11Z

Code suggestion: publish movement events to the MQTT feed.

GitHub cannot attach this inline because src/production/simulator/src/comms_manager.py is not changed by this PR. Insert after movement_event is constructed (currently line 165):

self.mqtt_client.publish(
    os.environ.get("MQTT_MOVEMENT_TOPIC", "projectecho/movement"),
    json.dumps(movement_event),
)

This keeps the existing HTTP persistence and adds the MQTT-derived movement event required by FR-A2.

AnDo081105

I have made some suggestions based on my findings. Have a look and re-check to see whether I missed anything. Thank you so much:D

- Merge database-sourced (enriched vocalization) events and in-memory MQTT-derived events (movement, sensor-health, IoT-node) into a single /mqtt/latest-events response - Subscribe to multiple MQTT topics (vocalization, movement, IoT) via a comma-separated MQTT_TOPICS env var, instead of a single hardcoded topic - Simulator now also publishes movement events to MQTT (projectecho/movement), alongside its existing HTTP persistence, so movement data is actually available on the live feed - Frontend now dispatches a mqtt:<eventType> custom event for sensor_health/iot_node instead of silently dropping them, so future rendering code can listen for them - Fixed a bug this introduced: added placeholder type/status/diet fields to normalized movement events, since the existing convertJSONtoAnimalMovementEvent function required them and was throwing on undefined values

kimmynoto added 2 commits August 5, 2026 22:47

Update project-echo-openapi.json

14cf37c

kimmynoto requested a review from Vamshi-Gollapelly as a code owner August 5, 2026 12:56

AnDo081105 reviewed Aug 8, 2026

View reviewed changes

Comment thread src/production/backend/app/services/mqtt_client.py Outdated

Comment thread src/production/backend/app/services/mqtt_client.py

Comment thread src/production/hmi/ui/public/js/HMI.js

AnDo081105 reviewed Aug 9, 2026

View reviewed changes

Comment thread src/production/backend/app/main.py

Comment thread src/production/backend/app/services/mqtt_client.py

kimmynoto and others added 2 commits August 13, 2026 01:04

Merge branch 'main' into hmi/kim/fr-a2-mqtt-event-handling

215ca17

AnDo081105 reviewed Aug 21, 2026

View reviewed changes

AnDo081105 changed the title ~~Hmi/kim/fr-a2-mqtt-event-handling~~ EH/KIM/fr-a2-mqtt-event-handling Aug 21, 2026

AnDo081105 approved these changes Sep 2, 2026

View reviewed changes

Navigation Menu

Uh oh!

EH/KIM/fr-a2-mqtt-event-handling - #961

EH/KIM/fr-a2-mqtt-event-handling#961
kimmynoto wants to merge 6 commits into
mainfrom
hmi/kim/fr-a2-mqtt-event-handling

kimmynoto commented Aug 5, 2026

Uh oh!

AnDo081105 left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

AnDo081105 left a comment

Uh oh!

Uh oh!

Uh oh!

AnDo081105 Aug 21, 2026

Uh oh!

AnDo081105 Aug 21, 2026

Uh oh!

AnDo081105 commented Aug 21, 2026 •

edited

Loading

Uh oh!

AnDo081105 left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

-    return {"events": show_latest_events(limit=20)}
+@app.get("/mqtt/latest-events", tags=["mqtt"])
+def mqtt_latest_events():
+    from app.routers.hmi import show_latest_events
+    events = [
+        {"eventType": "vocalization", **event}
+        for event in show_latest_events(limit=20)
+    ]
+    events += [
+        event for event in get_latest_events()
+        if event.get("eventType") != "vocalization"
+    ]
+    return {"events": events}

-MQTT_TOPIC = os.environ.get("MQTT_PUBLISH_URL", "projectecho/engine/2")
+MQTT_TOPICS = os.environ.get(
+    "MQTT_TOPICS",
+    "projectecho/engine/2,projectecho/movement,iot/data/test",
+).split(",")
+def on_connect(client, userdata, flags, rc):
+    global connection_state
+    connection_state = "connected"
+    for topic in MQTT_TOPICS:
+        client.subscribe(topic.strip())

Uh oh!

Conversation

kimmynoto commented Aug 5, 2026

Summary

What's in this PR

How I tested it

Uh oh!

AnDo081105 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

AnDo081105 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

AnDo081105 Aug 21, 2026

Choose a reason for hiding this comment

Uh oh!

AnDo081105 Aug 21, 2026

Choose a reason for hiding this comment

Uh oh!

AnDo081105 commented Aug 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AnDo081105 left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

AnDo081105 commented Aug 21, 2026 •

edited

Loading