Chapter 5: MQTT Communication

Making multiple robots dance the same choreography / perform in sync relies on MQTT. This chapter explains the group-performance communication principle and time-sync mechanism, plus a copy-paste custom MQTT example for extending multi-robot play.

In this chapter:

🧭 Reading path: To understand how multi-robot sync works, see 5.1; to write your own integration, adapt the example in 5.2.

Prerequisites:

• Done Chapter 1, you can SSH into the robot;
• At least two robots (a single one works but you won’t see the sync effect);
• The robot can reach the internet (group performance uses a public MQTT broker and NTP servers).

5.1 Group Performance Principles

Why MQTT instead of point-to-point / per-robot sync: multi-robot sync is about “do the same action at the same instant”. Centrally firing commands one by one gives uncontrolled latency, and LAN broadcast can't cross network segments. MQTT goes through a public broker (cloud relay); every dog subscribes to messages under the same room id, and executes on its own clock when it receives an instruction carrying an absolute timestamp — so dogs on different networks can still sync. Who “starts” doesn't matter; everyone follows the same schedule.

Group performance (apps/group_perform/main.py) synchronizes multiple robots over a public MQTT broker:

┌─────────────┐
│ MQTT Broker  │  broker.emqx.io:1883 (MQTTv311, keepalive=60)
└──┬──┬──┬──┬──┘
   │  │  │  │
   ▼  ▼  ▼  ▼
  Dog1 Dog2 Dog3 DogN

Each device uses its own external public IP as the room id room_id (get_external_ip()). Devices in the same room communicate over three topics:

Message format (JSON):

// presence
{"type": "join", "ip": "192.168.1.100", "dog_type": "xgomini"}
{"type": "heartbeat", "ip": "192.168.1.100", "dog_type": "xgomini"}

// command
{"type": "stop"}
{"type": "exit"}

// plan
{
  "type": "start",
  "actions": [
    {"id": 13, "name": "wave", "start_at": 1710000000.500, "duration": 7},
    {"id": 16, "name": "sway", "start_at": 1710000008.000, "duration": 6}
  ],
  "music_start": 1710000000.500,
  "dog_type": "xgomini"
}

Time synchronization (the key to synchronized execution):

• Each device requests a clock offset via SNTP (UDP 123) from ntp.aliyun.com / ntp1.aliyun.com / cn.pool.ntp.org / pool.ntp.org and maintains a unified clock synced_time() = time.time() + offset (re-syncs every 60 s on success, retries every 5 s on failure)
• The initiator uses synced_time() + ACTION_PREP_DELAY(3.0s) as the base time, computes an absolute start time start_at for every action and sends it; subscribers wait for the absolute time (a deviation >30 s from the local clock is treated as unsynced and the plan is rejected)
• Without internet access it automatically falls back to UDP broadcast mode (port 5005, room id = the first 3 octets of the local IP), using the same message format

Communication details: client_id = f"xgo-{local_ip}-{time.time()%100000}"; will_set(presence, {"type":"leave","ip":local_ip}) notifies the other devices when a device drops; the client auto-reconnects on disconnect.

The timeline of one synced show (keep this in mind; it gives you the big picture for writing code):

1. At boot each dog syncs its clock from NTP every 60 s (retries every 5 s on failure) and keeps an offset;
2. The initiator picks the actions, computes the absolute start time start_at for each, and sends start on the plan topic;
3. Each subscribing dog checks its clock deviation from the initiator is <30 s, then runs locally at the time; actions are aligned by timestamp;
4. Throughout, each dog periodically sends heartbeat to report online; a drop is broadcast as leave via the will last-will message.

How to confirm sync works: have each dog print its own synced_time() - local time (i.e. offset); it should be stable within a few ms. If the deviation is >30 s it's judged “clock unsynced” and the start is rejected — first check that ntp.aliyun.com is reachable and not blocked by a firewall. You can temporarily add print(offset) in group_perform/main.py to observe.

5.2 Custom MQTT Development

Follow group_perform/main.py and join the same topic protocol with paho-mqtt:

import paho.mqtt.client as mqtt
import json

# paho-mqtt 2.x requires callback_api_version; 1.x can use mqtt.Client(client_id=..., protocol=...)
# (the source code supports both via try/except)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="xgo-dev-1", protocol=mqtt.MQTTv311)
client.connect("broker.emqx.io", 1883, 60)

room_id = "ROOM_ID"     # group performance uses the external public IP as the room id
base = f"xgo/group/{room_id}"

# 1) Publish a presence heartbeat
client.publish(f"{base}/presence", json.dumps({
    "type": "join",
    "ip": "192.168.1.100",
    "dog_type": "xgomini",
}))

# 2) Publish an action plan (start_at/music_start are Unix seconds; NTP sync first)
client.publish(f"{base}/plan", json.dumps({
    "type": "start",
    "actions": [
        {"id": 13, "name": "wave", "start_at": 1710000000.5, "duration": 7},
    ],
    "music_start": 1710000000.5,
    "dog_type": "xgomini",
}))

# 3) Send a control command
client.publish(f"{base}/command", json.dumps({"type": "stop"}))

# 4) Subscribe to the three topics
client.subscribe(f"{base}/presence")
client.subscribe(f"{base}/command")
client.subscribe(f"{base}/plan")

# 5) Handle messages
client.on_message = lambda c, u, msg: print(f"{msg.topic}: {msg.payload.decode()}")
client.loop_forever()

💡 Install it first with pip3 install --break-system-packages paho-mqtt>=2.0 (already pre-installed on the new image).

5.2.1 Run a Minimal Verification First

After writing the subscriber above, print messages only (don't drive the robot) to confirm the link works:

python3 my_listener.py   # only subscribe and print the three topics

✅ Expected: use another device (or an MQTTX client) to send a message to a topic; the subscriber prints topic: payload in real time, meaning the MQTT link is fine.

⚠️ Common issues: - Connection fails / ssl error → ping broker.emqx.io first, make sure you can reach the internet; - Message received but the robot doesn't move → start_at used local time instead of synced_time(), or the action id isn't in this model's action table; - Dogs not in sync → clock deviation >30 s (check NTP), or room_id differs (different external IP so not the same room); - Want an offline / LAN test → group performance auto-falls back to UDP broadcast mode (port 5005, room id = first 3 octets of the local IP).