websocket_server 是一个与业务内容无关的 App Lab Custom Brick。它只负责建立 WebSocket 连接、接收原始消息和向客户端发送原始消息，不解析 JSON，也不知道客户端 是不是 ROS 2 节点。

这样可以把系统拆成两个可复用组件：

websocket_server Brick
  负责端口、路径、连接、文本帧、二进制帧和客户端管理
        ↓ client_id + str/bytes
上层协议 Brick（后续单独实现）
  负责 JSON、版本、消息类型、序号、时间戳和 ROS 2/App 语义

本文只实现并验证第一层。

组件支持 UNO Q 和 VENTUNO Q，只使用 Linux 网络，不依赖 MCU Sketch。本轮生命周期修复在 UNO Q 上实测；原 VENTUNO Q / App CLI 0.12.1 环境仅核对对应官方源码，没有重新实机验证。

实现结果

测试 App 提供以下端点：

ws://<开发板IP>:8765/ws

客户端发来文本帧时，App 返回内容相同的文本帧；发来二进制帧时，App 返回内容相同的 二进制帧。回环是测试 App 的行为，不是 Brick 强制规定的协议。

Brick 本身提供：

  • 精确路径检查；
  • 文本帧和二进制帧透传；
  • 多客户端连接和独立 client_id；
  • 单播、广播和主动断开；
  • 最大消息大小和最大客户端数限制；
  • WebSocket 协议层 ping/pong；
  • 跟随 App Lab 启停并释放端口。

它不提供 JSON 编解码、应用层握手、业务时间戳、ROS 2 接口、电机控制或安全停车。

配套源码

文章左侧“配套源码”可以逐个打开本文使用的完整文件：

app/
├── app.yaml
├── README.md
├── python/
│   └── main.py
├── tests/
│   ├── run.sh
│   ├── test_lifecycle.py
│   ├── test_app_lifecycle.py
│   └── websocket_test.py
└── bricks/
    └── websocket_server/
        ├── __init__.py
        ├── server.py
        ├── brick_config.yaml
        ├── requirements.txt
        └── README.md

其中 bricks/websocket_server/ 是可复制到其他 App 的组件；python/main.py、 tests/ 用于验证通信、启停、超时及 App 的失败退出。

Brick 配置

brick_config.yaml 声明 Brick ID、端口和可配置变量：

id: websocket_server
name: WebSocket Server
description: A reusable protocol-neutral WebSocket server for Arduino App Lab applications.
category: communication
supported_boards:
  - ventunoq
  - unoq
ports:
  - 8765
variables:
  - name: WEBSOCKET_SERVER_HOST
    default_value: "0.0.0.0"
    hidden: true
  - name: WEBSOCKET_SERVER_PORT
    default_value: "8765"
    hidden: true
  - name: WEBSOCKET_SERVER_PATH
    default_value: "/ws"
  - name: WEBSOCKET_SERVER_MAX_MESSAGE_BYTES
    default_value: "16384"
  - name: WEBSOCKET_SERVER_MAX_CLIENTS
    default_value: "4"
  - name: WEBSOCKET_SERVER_PING_INTERVAL_S
    default_value: "10"
  - name: WEBSOCKET_SERVER_PING_TIMEOUT_S
    default_value: "10"

requirements.txt 固定已经验证的依赖版本：

websockets==17.1

在 App 中加载 Brick

将完整 websocket_server 文件夹放入 App 的 bricks/，然后在 app.yaml 中引用：

name: WebSocket Brick Loopback
ports: []
bricks:
  - websocket_server:
      variables:
        WEBSOCKET_SERVER_HOST: "0.0.0.0"
        WEBSOCKET_SERVER_PORT: "8765"
        WEBSOCKET_SERVER_PATH: "/ws"
        WEBSOCKET_SERVER_MAX_MESSAGE_BYTES: "16384"
        WEBSOCKET_SERVER_MAX_CLIENTS: "4"
        WEBSOCKET_SERVER_PING_INTERVAL_S: "10"
        WEBSOCKET_SERVER_PING_TIMEOUT_S: "10"
icon: 🔌

端口已经由 Brick 声明，所以 App 顶层保持 ports: []。目录名、brick_config.yaml 中的 id 和 app.yaml 中的引用名必须都是 websocket_server。

最小回环 App

下面的 App 注册消息回调，并把 payload 原样发回来源客户端：

import time

from arduino.app_utils import App
from websocket_server import WebSocketServer


server = WebSocketServer()


def handle_message(client_id, payload):
    """
    @description         : 将收到的文本帧或二进制帧原样发回来源客户端
    @param client_id     : Brick 为当前连接分配的客户端标识
    @param payload       : 未解析的 str 或 bytes 消息
    @return              : 无返回值
    """
    server.send(client_id, payload)


def loop():
    """
    @description         : 让出 CPU 并保持 App Lab 主循环运行
    @param               : 无参数
    @return              : 无返回值
    """
    server.require_running()
    time.sleep(0.05)


server.on_message(handle_message)
try:
    App.run(user_loop=loop)
finally:
    server.request_stop()

@brick 会把 WebSocketServer 实例纳入 App Lab 生命周期，由 App.run() 自动启停。 但不能省略 require_running()：已核对的 app_utils 0.11.0 / 0.12.0 会捕获并记录 start() 的异常，然后继续运行 App。主循环检查让启动失败再次传播，使示例以退出码 1 结束，并在日志中显示端口占用等原因。finally 则为未进入 App 运行队列的失败 Brick 请求清理。

Python API

API 返回值 用途
on_connect(callback) 当前实例 注册 callback(client_info)
on_message(callback) 当前实例 注册 callback(client_id, payload)
on_disconnect(callback) 当前实例 注册 callback(client_info, code, reason)
send(client_id, payload) bool 向指定客户端发送消息
broadcast(payload) int 向全部当前客户端发送并返回成功数量
disconnect(client_id, code, reason) bool 主动关闭指定客户端
get_clients() list[dict] 获取不含底层 socket 的客户端快照
get_status() dict 获取监听状态、连接数、限制和服务错误
start() True 监听就绪才返回；启动失败或取消抛出 RuntimeError，超时抛出 TimeoutError
stop() bool 请求停止并在总预算内等待；全部清理完成才返回 True
request_stop() bool 只请求停止，不等待；尚未清理完成返回 False
require_running() None 服务正常时返回，否则抛出带原因的 RuntimeError

payload 只有两种类型：文本帧是 str，二进制帧是 bytes。Brick 不自动把 bytes 转换为 UTF-8，也不自动调用 json.loads()。

client_info 包含：

{
    "client_id": "由 Brick 分配的连接标识",
    "remote_address": "远端地址",
    "path": "/ws",
    "connected_monotonic_s": 12345.67,
}

connected_monotonic_s 用于计算本次系统启动期间的连接时长，不是 Unix 时间戳，不能 写入跨设备通信协议。

连接和消息回调运行在 WebSocket 工作线程中，应快速返回。耗时计算、磁盘操作或硬件任务 应由 App 放入自己的有界队列，再交给专用线程处理。

启停状态和等待上限

App Lab 会检查方法签名，start(self) 和 stop(self) 不能增加其他参数。 等待时间应在创建对象时指定，下面也是默认值：

server = WebSocketServer(start_timeout_s=5.0, stop_timeout_s=3.0)

start() 最多等待 5 秒。端口占用等错误会保留原始原因并抛出异常；超时会请求取消本次启动， 迟到的启动结果只能清理，不能重新变成正常运行。

stop() 的 3 秒是一次调用的总等待预算，包括等待关闭监听、客户端和连接回调， 不是每个客户端各等 3 秒。底层 shutdown() 在清理线程中执行，不会先无限阻塞 stop()。

state 含义 是否允许新一轮启动
starting 正在启动 重复 start() 等待同一次启动
running 已就绪、没有停止请求 重复 start() 返回成功，不新增线程
stopping 已请求停止，清理尚未结束 不允许
stopped 未启动过，或正常停止 确认旧线程结束后允许
failed 启动、监听或清理失败 只有旧线程结束且没有清理异常时才允许

用 get_status() 观察状态时，还要注意：

  • listening 才表示服务已就绪且未请求停止。running 只表示后台工作线程还存在。
  • cleanup_pending=True 表示清理未完成，此时端口或连接可能尚未释放。
  • server_error 保存启动或监听失败原因；cleanup_error 保存清理失败原因。
  • stop() 返回 False 不是“停止成功”。等待清理完成后才能重新启动；出现 cleanup_error 应重启 App 进程。

在回调中调用 stop() 会立即返回 False，仅请求停止，避免回调等待自己退出。 Python 无法强制终止卡住的回调或系统调用，所以这里只保证调用者的等待预算。 迟到的绑定操作可能短暂占用端口，但会立即清理，不发布就绪、不调用业务回调。 send()、broadcast()、disconnect() 仍是同步操作，不属于上述停机预算。

WebSocket 约定

项目 默认值
监听地址 0.0.0.0
端口 8765
路径 /ws
最大消息 16384 字节
最大同时连接 4
协议 ping 间隔 10 秒
pong 超时 10 秒
压缩 关闭

请求路径必须精确匹配 /ws，查询参数不参与路径比较。错误路径使用关闭码 1008；超过 最大客户端数时，新连接使用关闭码 1013。

WebSocket ping/pong 只能判断传输连接是否仍然响应。机器人协议中的控制权、命令超时和 安全停车仍然需要上层协议 Brick 单独实现。

创建并启动测试 App

在开发板终端创建一个不包含 MCU Sketch 的独立 App：

arduino-app-cli app new websocket-brick-loopback \
  --no-sketch \
  --description "Protocol-neutral reusable WebSocket Brick loopback test"

将配套源码中的 app/ 内容复制到：

/home/arduino/ArduinoApps/websocket-brick-loopback/

启动 App：

cd /home/arduino/ArduinoApps/websocket-brick-loopback
arduino-app-cli app start .

查看运行日志：

arduino-app-cli app logs . --tail 100

日志中应出现：

[loopback] state=running listening=True clients=0 error=None

CLI 提示 App 启动成功表示容器启动流程完成，不保证 WebSocket 已就绪。 请以这里的日志和下一步的回环测试为准；首次安装依赖可能需要等待。

运行端到端测试

测试入口自动使用当前 App 的虚拟环境，不需要知道 Python 次版本，也不需要设置 PYTHONPATH：

cd /home/arduino/ArduinoApps/websocket-brick-loopback
sh tests/run.sh

run.sh 根据自己的位置找到 App 根目录和 App CLI 生成的 .cache/app-compose.yaml， 再在正在运行的 main 容器中调用 App 虚拟环境。它不依赖固定的 Python 次版本、容器 名称或 site-packages 路径。因此同一份 App 复制到另一台受支持的开发板后，只要先由 App CLI 成功启动，就可以使用同一个测试命令。测试脚本还会自动读取 Brick 当前配置的 端口、路径和最大客户端数。

需要测试另一台设备的 WebSocket 地址，或者需要覆盖自动检测值时执行：

sh tests/run.sh \
  --url ws://192.168.1.50:8765/ws \
  --max-clients 4

测试范围：

  1. 文本消息保持类型和内容原样回环；
  2. 二进制消息保持类型和内容原样回环；
  3. 两个客户端同时连接且消息互不串线；
  4. 错误路径以 1008 拒绝；
  5. 连接达到配置上限后，下一个连接以 1013 拒绝。

实测输出：

PASS text and binary echo
PASS two independent clients
PASS wrong path rejected
PASS maximum client limit
ALL TESTS PASSED

再运行生命周期回归测试：

sh tests/run.sh --unit -v

它使用独立的本机端口，检查并发启停、启动取消、停止总预算、阻塞回调、握手未完成和清理失败。 其中两个测试运行真实的 App.run() 子进程，验证端口占用时退出码为 1，以及 Linux SIGTERM 后五秒内退出、释放端口并可重新启动。故意制造故障时会打印错误日志，以最终测试汇总为准。

完整 App 的停止、重启测试方法：

arduino-app-cli app stop .
arduino-app-cli app start .
sh tests/run.sh

开发电脑可安装 websockets==17.1，在配套 app/ 目录执行 python -B -m unittest discover -s tests -p 'test_*.py'。 缺少真实 app_utils 时两个 App 集成测试会跳过，不能将本机测试等同于 App Lab 实测。

本轮 UNO Q 验证结果：18 项生命周期测试全部通过；App CLI 完整停止约 1.1 秒， 容器退出码为 143，监听端口释放；重新启动后，容器内和开发电脑直连的回环测试均通过。 开发电脑测试通过 16 项，另外两项因没有真实 app_utils 而跳过。

如果开发电脑启用了网络代理，访问开发板局域网 IP 时需要在代理工具中设为直连， 或给测试进程设置 NO_PROXY=<开发板IP>。否则可能出现开发板内回环正常、电脑访问超时。

在其他 App 中复用

复用时只需要复制：

bricks/websocket_server/

回环用的 python/main.py 和 tests/ 不是 Brick 的 组成部分。新的 App 可以注册自己的回调，或者让后续的协议 Brick 使用 on_message() 接收原始 payload，再完成 JSON、ROS 2/App 消息和机器人安全规则的处理。

复制后保留主循环中的运行检查和退出清理。修改监听端口时，也要同步修改 brick_config.yaml 的 ports 与 App 中的端口配置；仅修改 Python 对象的端口不能自动改变容器端口映射。

实现依据

这页先记到这里。问题拆开一点，答案就会近一点。