4.14 Robot Subscription & Publish Interface

Overview

The SubPub module handles the robot controller’s WebSocket-based publish/subscribe channels: it establishes the connection, subscribes to topics such as robot state, registers, and I/O, and delivers real-time data via callback or blocking calls.

Core Features

• Support for connecting to and disconnecting from WebSocket servers
• Support for subscribing to robot status data
• Support for subscribing to register data
• Support for subscribing to IO signal data
• Support for receiving messages via callback functions
• Support for blocking receive of the next message
• Support for setting subscription frequency
• Support for handling receive message errors

Use Cases

• Real-time monitoring of robot operation status from host PC
• Implementation of robot data visualization
• Data linkage with external systems
• Real-time monitoring of register and IO status
• Building robot monitoring and control systems
• Implementation of real-time alarm and notification for robot status

4.14.1 Connect to WebSocket Server

Method Name	sub_pub.connect() -> StatusCodeEnum
Description	Connect to the robot controller WebSocket server
Request Parameters	None
Return Value	StatusCodeEnum: Connection status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.2 Disconnect WebSocket Connection

Method Name	sub_pub.disconnect() -> StatusCodeEnum
Description	Disconnect from the robot controller WebSocket server
Request Parameters	None
Return Value	StatusCodeEnum: Disconnection status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.3 Add Robot Status Subscription

Method Name	sub_pub.subscribe_status( topics : List[RobotTopicType], frequency : int = 200) -> StatusCodeEnum
Description	Add robot status data subscriptions
Request Parameters	topics : List[RobotTopicType] Robot topic type list frequency : int Subscription frequency (Hz, default 200)
Return Value	StatusCodeEnum: Subscription status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.4 Add Register Subscription

Method Name	sub_pub.subscribe_register( reg_type : RegTopicType, reg_ids : List[int], frequency : int = 200) -> StatusCodeEnum
Description	Add register data subscriptions
Request Parameters	reg_type : RegTopicType Register type reg_ids : List[int] Register ID list frequency : int Subscription frequency (Hz, default 200)
Return Value	StatusCodeEnum: Subscription status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.5 Add IO Subscription

Method Name	sub_pub.subscribe_io( io_list : List[tuple[IOTopicType, int]], frequency : int = 200) -> StatusCodeEnum
Description	Subscribe to IO signal data, including digital inputs/outputs
Request Parameters	io_list : List[tuple[IOTopicType, int]] IO list (each element is (IO type, IO ID) ) frequency : int Subscription frequency (Hz, default 200)
Return Value	StatusCodeEnum: Subscription status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.6 Start Receiving Messages

Method Name	sub_pub.start_receiving( on_message_received : Callable[[Dict[str, Any]], None]) -> StatusCodeEnum
Description	Start receiving subscription messages and process received data via the callback function
Request Parameters	on_message_received : Callable[[Dict[str, Any]], None] Message receive callback function
Return Value	StatusCodeEnum: Receiving status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.7 Handle Receive Message Error

Method Name	sub_pub.handle_receive_error() -> StatusCodeEnum
Description	Handle errors when receiving subscription messages; exits the loop if the receive callback raises an exception.
Request Parameters	None
Return Value	StatusCodeEnum: Receiving status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

4.14.8 Receive Next Message

Method Name	sub_pub.receive() -> tuple[Dict[str, Any], StatusCodeEnum]
Description	Receive the next message and return it
Request Parameters	None
Return Value	tuple[Dict[str, Any], StatusCodeEnum]: Received message dictionary and status code
Compatible robot software version	Collaborative (Copper): v7.7.0.0+ Industrial (Bronze): v7.7.0.0+

Example Code

sub_pub.py

#!python
"""
Copyright © 2016 Agilebot Robotics Ltd. All rights reserved.
Instruction: SubPub使用示例 / SubPub usage example
"""

import asyncio
import logging

from Agilebot import Arm, IOTopicType, RegTopicType, RobotTopicType, StatusCodeEnum

# 配置日志 / Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


async def message_handler(message):
    """
    消息处理函数 / Message handler function

    :param message: 接收到的消息字典 / Received message dictionary
    """
    logger.info(f"收到消息 / Received message: {message}")


async def main():
    """
    主函数，演示SubPub的使用 / Main function demonstrating SubPub usage
    """
    # 创建Arm实例 / Create Arm instance
    arm = Arm()

    # 连接到控制器 / Connect to controller
    ret = arm.connect("10.27.1.254")
    if ret != StatusCodeEnum.OK:
        logger.error(f"连接失败 / Connection failed: {ret}")
        return

    logger.info("Arm连接成功 / Arm connected successfully")

    try:
        # 连接WebSocket / Connect WebSocket
        ret = await arm.sub_pub.connect()
        if ret != StatusCodeEnum.OK:
            logger.error(f"WebSocket连接失败 / WebSocket connection failed: {ret}")
            return

        logger.info("WebSocket连接成功 / WebSocket connected successfully")

        # 订阅机器人状态 / Subscribe to robot status
        topic_types = [
            RobotTopicType.JOINT_POSITION,
            RobotTopicType.CARTESIAN_POSITION,
            RobotTopicType.ROBOT_STATUS,
            RobotTopicType.CTRL_STATUS,
            RobotTopicType.SERVO_STATUS,
            RobotTopicType.INTERPRETER_STATUS,
            RobotTopicType.TRAJECTORY_RECORD,
            RobotTopicType.USER_OP_MODE,
            RobotTopicType.USER_FRAME,
            RobotTopicType.TOOL_FRAME,
            RobotTopicType.VELOCITY_RATIO,
            RobotTopicType.TRAJECTORY_RECORD,
        ]

        ret = await arm.sub_pub.subscribe_status(topic_types, frequency=200)
        if ret != StatusCodeEnum.OK:
            logger.error(f"订阅机器人状态失败 / Failed to subscribe to robot status: {ret}")
            return

        logger.info("机器人状态订阅成功 / Robot status subscription successful")

        # 订阅寄存器 / Subscribe to registers
        ret = await arm.sub_pub.subscribe_register(RegTopicType.R, [1, 2, 3], frequency=200)
        if ret != StatusCodeEnum.OK:
            logger.error(f"订阅寄存器失败 / Failed to subscribe to registers: {ret}")
            return

        logger.info("寄存器订阅成功 / Register subscription successful")

        # 订阅IO信号 / Subscribe to IO signals
        io_list = [(IOTopicType.DI, 1), (IOTopicType.DO, 1)]

        ret = await arm.sub_pub.subscribe_io(io_list, frequency=200)
        if ret != StatusCodeEnum.OK:
            logger.error(f"订阅IO失败 / Failed to subscribe to IO: {ret}")
            return

        logger.info("IO订阅成功 / IO subscription successful")

        # 单次接收消息示例 / Single message receive example
        logger.info("尝试单次接收消息 / Attempting single message receive")
        try:
            # 设置超时 / Set timeout
            message, ret = await asyncio.wait_for(arm.sub_pub.receive(), timeout=5.0)
            if ret == StatusCodeEnum.OK:
                logger.info(f"单次接收消息成功 / Single message receive successful: {message}")
            else:
                logger.error(f"单次接收消息失败 / Single message receive failed: {ret}")
        except asyncio.TimeoutError:
            logger.warning("接收消息超时 / Message receive timeout")

        # 启动消息接收 / Start message receiving
        ret = await arm.sub_pub.start_receiving(message_handler)
        if ret != StatusCodeEnum.OK:
            logger.error(f"启动消息接收失败 / Failed to start message receiving: {ret}")
            return

        logger.info("开始接收消息 / Started receiving messages")

        # 运行一段时间 / Run for a while
        logger.info("运行10秒钟... / Running for 10 seconds...")
        await asyncio.sleep(10)

    except Exception as e:
        logger.error(f"运行过程中发生错误 / Error occurred during execution: {str(e)}")

    finally:
        # 断开连接 / Disconnect
        await arm.sub_pub.disconnect()
        arm.disconnect()
        logger.info("连接已断开 / Connection disconnected")


if __name__ == "__main__":
    # 运行异步主函数 / Run async main function
    asyncio.run(main())