chongming-mqtt (0.1.0)

Published 2026-08-25 14:30:39 +08:00 by chakcy

To install the package using pip, run the following command:

pip install --index-url http://124.71.68.6:3000/api/packages/chongming/pypi/simple/ --extra-index-url https://pypi.org/simple chongming-mqtt

For more information on the PyPI registry, see the documentation.

chongming MQTT 工具包：外部转发、外部订阅、内部请求-响应

chongming-mqtt

MQTT 工具包，封装基于 amqtt 的三类客户端，覆盖常见 MQTT 通信场景。

安装

uv pip install -e utils/mqtt

依赖：amqtt>=0.11.3，Python >=3.12。

客户端一览

客户端	方向	用途
`MQTTForwardClient`	单向发布	主动把数据上报/转发到外部 Broker
`MQTTSubscribeClient`	订阅 + 回复	订阅外部主题，处理请求并按 `reply_topic` 回复（请求-响应）
`InternalMQTTClient`	请求-响应	对任意应用进程发起 MQTT 请求-响应，显式指定请求/响应主题

三者均支持：

断线自动重连（无限重试，最大重连间隔 30s）
URI percent-encoded 凭据解码（密码中的 @ 写为 %40）
mqtt://user:pass@host:port 连接串

1. MQTTForwardClient — 外部转发（单向发布）

主动把数据推送到外部 Broker 的指定主题。

from chongming_mqtt import MQTTForwardClient

client = MQTTForwardClient(uri="mqtt://admin:Mixc%402021@host:48081", topic="/board_ping")
await client.start()
await client.publish({"BoardId": "B001", "Temp": 45.6})   # 发布到默认主题
await client.publish({"x": 1}, topic="/other_topic")       # 指定主题
await client.stop()

方法	说明
`start()`	建立长连接（自动重连）
`publish(payload, topic=None)`	发布 JSON；`topic` 缺省用构造时的主题
`stop()`	断开连接

2. MQTTSubscribeClient — 外部订阅（请求-响应）

订阅外部主题，收到消息后调用绑定的处理函数；配置 reply_topic 后，把处理函数返回值发布到回复主题。

from chongming_mqtt import MQTTSubscribeClient

async def handler(payload: dict, topic: str):
    # 处理请求，返回回复内容（返回 None 表示不回复）
    return {"code": 0, "msg": "ok", "data": payload}

client = MQTTSubscribeClient(
    uri="mqtt://admin:Mixc%402021@host:48081",
    routes=[
        {
            "topic": "/edge_app_controller",        # 订阅主题（支持 + / # 通配符）
            "handler": handler,                      # async (payload, topic) -> dict | None
            "reply_topic": "/edge_app_controller_reply",  # 回复主题（支持 {1}{2} 占位符）
        }
    ],
)
await client.start()
# ... 运行中 ...
await client.stop()

不回复约定：处理函数返回 None 时不发布回复（静默命令）
reply_topic 占位符：{1}、{2}… 填充订阅主题通配符捕获值，如 reply_topic="/dev/{1}/resp"

3. InternalMQTTClient — 内部请求-响应

对任意应用进程发起 MQTT 请求-响应：显式指定"请求主题"与"响应主题"，等待并返回响应报文。不依赖特定协议（如 chongming/*），适用于与其它应用/进程通信。

from chongming_mqtt import InternalMQTTClient

client = InternalMQTTClient(uri="mqtt://127.0.0.1:1883")   # 默认连内部 broker
await client.start()

resp = await client.request(
    request_topic="/other_app/request",      # 目标应用监听的请求主题
    response_topic="/other_app/reply/#",     # 目标应用回复的主题（支持 +/# 或 {rid}）
    payload={"BoardId": "B001"},
    timeout=30,
)
# resp: dict | None（None 表示超时/异常）
await client.stop()

响应关联策略

请求会自动注入 request_id 字段。收到响应时按以下优先级匹配：

对方回显 request_id → 精确匹配（推荐）
response_topic 含 {rid} 模板 → 订阅唯一主题匹配（对方把 request_id 拼进响应主题）
兜底 FIFO（固定主题、不回显，串行场景）

支持并发请求与响应乱序，互不串扰。

与 aibox_sophon_adapter 插件的集成

插件在 [workers.config] 下分别配置这三个客户端，启动逻辑封装在插件的 mqtt_clients.py 能力组件中（start_mqtt_forward / start_mqtt_subscribe / start_internal_mqtt，启动失败自动降级为 None 并告警），由 AiboxSophonAdapterPlugin.on_init 调用：

[workers.config.mqtt_forward]       # -> MQTTForwardClient
enabled = true
uri = "mqtt://admin:Mixc%402021@dg.datagoo.cn:48081"
topic = "/board_ping"

[workers.config.mqtt_subscribe]     # -> MQTTSubscribeClient
enabled = true
uri = "mqtt://admin:Mixc%402021@dg.datagoo.cn:48081"
  [[workers.config.mqtt_subscribe.topics]]
  topic = "/edge_app_controller"
  handler = "edge_app_controller"
  reply_topic = "/edge_app_controller_reply"

[workers.config.internal_mqtt]      # -> InternalMQTTClient（handler 内调用）
enabled = true
uri = "mqtt://127.0.0.1:1883"

handler 内使用 InternalMQTTClient：

# edge_app_controller.py
from chongming_mqtt import InternalMQTTClient

async def _cmd_some_op(payload: dict) -> dict:
    from aibox_sophon_adapter.global_var import internal_mqtt
    if internal_mqtt is None:
        return {"error": "内部 MQTT 未初始化"}
    resp = await internal_mqtt.request(
        request_topic="/other_app/request",
        response_topic="/other_app/reply/#",
        payload={"BoardId": payload.get("BoardId")},
        timeout=30,
    )
    return resp or {"error": "内部请求超时"}

测试

单元测试位于 utils/mqtt/test/，通过注入 FakeMQTT 模拟底层连接，不依赖真实 Broker，可离线快速运行。

uv run pytest utils/mqtt/test -v

覆盖范围：

测试文件	覆盖内容
`test_forward.py`	凭据解码、默认/指定主题发布、未连接与异常时的行为、stop 断开
`test_subscribe.py`	主题通配符（`+`/`#`）编译、无效路由跳过、消息分发、`reply_topic` 回复、`None` 不回复、`{1}` 占位符
`test_internal.py`	凭据解码、回显 `request_id` 关联、`{rid}` 模板、超时、FIFO 兜底、并发乱序、订阅去重

若需对真实 Broker 做端到端验证，可参考项目根目录的 test_mqtt_e2e.py 模式，自行启动本地 broker 后联测。

构建集成

若通过 PyInstaller/Nuitka 打包，需在 config.toml 的 [pyinstaller_build_import] base 中加入：

"chongming_mqtt",

并确保 chongming_mqtt 已安装到目标 Python 环境（uv pip install -e utils/mqtt）。

Requires Python: >=3.12

Details

PyPI

2026-08-25 14:30:39 +08:00

83

25 KiB

Assets (2)

chongming_mqtt-0.1.0-py3-none-any.whl 13 KiB

chongming_mqtt-0.1.0.tar.gz 12 KiB

Versions (1) View all

0.1.0

2026-08-25