双足／双轮足 · C++ / Python 示例

3.9 双足/双轮足协议接口调用示例

3.9.1 Linux C++ 示例实现

环境准备

以 Ubuntu 20.04 系统为例，安装 websocketpp、nlohmann/json 和 boost 依赖：

sudo apt-get install libboost-all-dev libwebsocketpp-dev nlohmann-json3-dev

编译代码

g++ -std=c++11 -o websocket_client websocket_client.cpp -lssl -lcrypto -lboost_system -lpthread

运行程序

./websocket_client

3.9.2 Python 示例实现

环境准备

以 Ubuntu 20.04 系统为例，安装下面依赖：

sudo apt install python3-dev python3-pip
pip3 install websocket-client

运行脚本

python3 websocket_client.py

websocket_client.py 实现

import json
import uuid
import threading
import time
import websocket


# Replace this ACCID value with your robot's actual serial number (SN)
ACCID = None

# Atomic flag for graceful exit
should_exit = False

# WebSocket client instance
ws_client = None


# Generate dynamic GUID
def generate_guid():
    return str(uuid.uuid4())


# Send WebSocket request with title and data
def send_request(title, data=None):
    global ACCID
    global ws_client

    if data is None:
        data = {}

    # Create message structure with necessary fields
    message = {
        "accid": ACCID,
        "title": title,
        "timestamp": int(time.time() * 1000),
        "guid": generate_guid(),
        "data": data
    }

    message_str = json.dumps(message)

    # Send the message through WebSocket if client is connected
    if ws_client:
        ws_client.send(message_str)


# Handle user commands
def handle_commands():
    global should_exit

    while not should_exit:
        command = input(
            "Enter command ('stand', 'walk', 'twist', 'sit', 'stair', 'stop', 'imu') or 'exit' to quit:\n"
        )

        if command == "exit":
            should_exit = True
            break

        elif command == "stand":
            send_request("request_stand_mode")

        elif command == "walk":
            send_request("request_walk_mode")

        elif command == "twist":
            # Get twist values from user
            x = float(input("Enter x value: "))
            y = float(input("Enter y value: "))
            z = float(input("Enter z value: "))

            # Send twist command at 30 Hz for 1 second
            for _ in range(30):
                send_request("request_twist", {
                    "x": x,
                    "y": y,
                    "z": z
                })
                time.sleep(1 / 30)

        elif command == "sit":
            send_request("request_sitdown")

        elif command == "stair":
            # Get stair mode enable flag from user
            enable = input("Enable stair mode (true/false): ").strip().lower() == "true"
            send_request("request_stair_mode", {
                "enable": enable
            })

        elif command == "stop":
            send_request("request_emgy_stop")

        elif command == "imu":
            # Get IMU enable flag from user
            enable = input("Enable IMU (true/false): ").strip().lower() == "true"
            send_request("request_enable_imu", {
                "enable": enable
            })


# WebSocket on_open callback
def on_open(ws):
    print("Connected!")

    # Start handling commands in a separate thread
    threading.Thread(target=handle_commands, daemon=True).start()


# WebSocket on_message callback
def on_message(ws, message):
    global ACCID

    root = json.loads(message)
    ACCID = root.get("accid", None)

    print(f"Received message: {message}")


# WebSocket on_close callback
def on_close(ws, close_status_code, close_msg):
    print("Connection closed.")


# Close WebSocket connection
def close_connection(ws):
    ws.close()


def main():
    global ws_client

    # Create WebSocket client instance
    ws_client = websocket.WebSocketApp(
        "ws://10.192.1.2:5000",
        on_open=on_open,
        on_message=on_message,
        on_close=on_close
    )

    # Run WebSocket client loop
    print("Press Ctrl+C to exit.")
    ws_client.run_forever()


if __name__ == "__main__":
    main()