Lesson 17 - WebSockets for Robotics

Learning Objective

• Explain how a WebSocket connection differs from a normal HTTP request.
• Open a raw WebSocket connection in Python and exchange messages by hand.
• Recognize the JSON “op” message format that rosbridge — and therefore bonicbot_bridge — speaks underneath every BonicBot call you’ve made since Lesson 1.

Introduction

Every lesson so far has started the same way:

with BonicBot(host=...) as bot:

and from that single line you’ve been able to read the lidar, watch servo angles update, and stream camera frames — all without writing a single line of networking code.

None of that is magic. It’s one persistent WebSocket connection, opened the moment BonicBot(...) runs, carrying a constant stream of small JSON messages back and forth between your laptop and the robot’s onboard rosbridge server.

This lesson opens that connection up and looks at it directly, with no SDK in the way. Lesson 18 will point it at a real robot; for now, the goal is simply to understand what a WebSocket actually is and what those JSON messages look like.

[!NOTE] When connecting to a physical robot (such as in Lesson 18 and 19), make sure to replace the host IP address with your BonicBot’s actual IP address before running the code.

Part A — WebSocket vs. HTTP

Code

import websocket
 
ws = websocket.create_connection("wss://ws.postman-echo.com/raw")
 
ws.send("Hello from Python!")
print("Server replied:", ws.recv())
 
ws.send("Still the same connection?")
print("Server replied:", ws.recv())
 
ws.close()

Expected Output

Server replied: Hello from Python!
Server replied: Still the same connection?

🔧 Under the Hood

Part B — The Message Format Robots Actually Speak

rosbridge (the server your robot runs) doesn’t send plain text over its WebSocket — it sends small JSON objects called op messages. Everything you can do — subscribe to a topic, publish a command, cancel a subscription — is one JSON dictionary with an "op" field describing which of those three it is.

Code

import json
 
subscribe_message = {
    "op": "subscribe",
    "topic": "/scan",
    "type": "sensor_msgs/LaserScan"
}
 
publish_message = {
    "op": "publish",
    "topic": "/cmd_vel",
    "msg": {
        "linear": {"x": 0.2, "y": 0, "z": 0},
        "angular": {"x": 0, "y": 0, "z": 0}
    }
}
 
unsubscribe_message = {
    "op": "unsubscribe",
    "topic": "/scan"
}
 
for label, message in [
    ("Subscribe", subscribe_message),
    ("Publish", publish_message),
    ("Unsubscribe", unsubscribe_message),
]:
    print(f"{label}:")
    print(json.dumps(message, indent=2))
    print()

Expected Output

Subscribe:
{
  "op": "subscribe",
  "topic": "/scan",
  "type": "sensor_msgs/LaserScan"
}

Publish:
{
  "op": "publish",
  "topic": "/cmd_vel",
  "msg": {
    "linear": {
      "x": 0.2,
      "y": 0,
      "z": 0
    },
    "angular": {
      "x": 0,
      "y": 0,
      "z": 0
    }
  }
}

Unsubscribe:
{
  "op": "unsubscribe",
  "topic": "/scan"
}

🔧 You’ve Been Reading These Fields Since Lesson 1

LASER_SCAN_MESSAGE_TYPE = "sensor_msgs/LaserScan"
LASER_SCAN_TOPIC = "/scan"

self.scan_sub = Topic(
    self.ros,
    LASER_SCAN_TOPIC,
    LASER_SCAN_MESSAGE_TYPE,
    throttle_rate=LASER_SCAN_THROTTLE_MS,
)
self.scan_sub.subscribe(self._scan_callback)

Part C — Why Publish/Subscribe, Not Just “Ask and Wait”

Code

class TinyBroker:
    def __init__(self):
        self.subscribers = {}
 
    def subscribe(self, topic, callback):
        self.subscribers.setdefault(topic, []).append(callback)
 
    def publish(self, topic, message):
        for callback in self.subscribers.get(topic, []):
            callback(message)
 
 
broker = TinyBroker()
 
 
def on_scan_for_navigation(msg):
    print("Navigation logic sees:", msg)
 
 
def on_scan_for_logging(msg):
    print("Logger sees:", msg)
 
 
broker.subscribe("/scan", on_scan_for_navigation)
broker.subscribe("/scan", on_scan_for_logging)
 
broker.publish("/scan", {"closest_obstacle_m": 0.8})

Expected Output

Navigation logic sees: {'closest_obstacle_m': 0.8}
Logger sees: {'closest_obstacle_m': 0.8}

🔧 This Is Exactly How CameraManager Fans Out a Frame

if self.user_callback:
    try:
        self.user_callback(image)
    except Exception as exc:
        print(f"⚠️ Error in user callback: {exc}")

Student Challenge

Challenge 1 — Python Only

Build a tiny stand-in for a rosbridge client — no network required.

• Write a class MiniRosbridgeClient with:
  • subscribe(self, topic, msg_type) — prints the properly formatted {"op": "subscribe", ...} JSON (use json.dumps(..., indent=2)).
  • publish(self, topic, msg) — prints the properly formatted {"op": "publish", ...} JSON.
  • unsubscribe(self, topic) — prints the properly formatted {"op": "unsubscribe", ...} JSON.
• Instantiate it and call all three methods once, using /scan / sensor_msgs/LaserScan for the subscribe call and /cmd_vel with a simple {"linear": {"x": 0.2}} payload for the publish call.

import json
 
 
class MiniRosbridgeClient:
    def subscribe(self, topic, msg_type):
        message = {
            "op": "subscribe",
            "topic": topic,
            "type": msg_type
        }
        print(json.dumps(message, indent=2))
 
    def publish(self, topic, msg):
        message = {
            "op": "publish",
            "topic": topic,
            "msg": msg
        }
        print(json.dumps(message, indent=2))
 
    def unsubscribe(self, topic):
        message = {
            "op": "unsubscribe",
            "topic": topic
        }
        print(json.dumps(message, indent=2))
 
 
client = MiniRosbridgeClient()
 
client.subscribe("/scan", "sensor_msgs/LaserScan")
client.publish("/cmd_vel", {"linear": {"x": 0.2}})
client.unsubscribe("/scan")

Challenge 2 — Robot

Prove to yourself that bot.ros really is a WebSocket connection object, using only what you already know from Lessons 1–16.

• Connect with with BonicBot(host='192.168.0.188') as bot:.
• Print type(bot.ros).__name__.
• Print bot.ros.is_connected.
• Print bot.host and bot.port together.

self.ros = Ros(host=self.host, port=self.port)
self.ros.run()

from bonicbot_bridge import BonicBot
 
with BonicBot(host='192.168.0.188') as bot:
    print(type(bot.ros).__name__)
    print(bot.ros.is_connected)
    print(f"{bot.host}:{bot.port}")

Challenge 3 — Robot

Watch the “persistent connection” idea from Part A play out live, using data you already know how to read.

• Connect with with BonicBot(host='192.168.0.188') as bot:.
• In a loop that runs 10 times, call bot.sensors.get_lidar_scan() and print whether it returned None or real data, with time.sleep(0.2) between calls.
• Without ever calling subscribe yourself, count how many of the 10 reads were non-None.

import time
from bonicbot_bridge import BonicBot
 
with BonicBot(host='192.168.0.188') as bot:
    received = 0
 
    for i in range(10):
        scan = bot.sensors.get_lidar_scan()
        print(f"Read {i}: {'data' if scan else 'None'}")
 
        if scan is not None:
            received += 1
 
        time.sleep(0.2)
 
    print(f"Non-empty reads: {received} / 10")

WebSocket Quick Reference

Reflection Questions