Python ZH

Python

本手册：安装与入门 · API 与配置参考 · 变量语义 · 组网与状态 · 排错与迁移

English · 首页 · 源码

使用 Python 3.10+、asyncio 和官方 NATS Python 客户端。SDK 提供内存变量和自动局域网节点，无需 Node.js 运行时。

本章目录

• 从源码开始
• 常用 API
• 连接配置
• live 通道

从源码开始

在 KinopioHub.py 中执行：

uv sync --extra dev
uv run python examples/watch.py
# In another terminal:
uv run python examples/basic.py

其他应用可在自身环境中用 python -m pip install -e /path/to/KinopioHub.py 安装当前源码。

import asyncio
from kinopio_hub import KinopioHub

async def main():
    async with KinopioHub("workshop") as hub:
        battery = hub.var("battery")
        battery.watch(lambda value, meta: print(value))
        await battery.set(80)
        await hub.flush()

asyncio.run(main())

上下文管理器负责本地初始化和退出清理，进入时不等待网络连接。

注意： 如果 Hub 持有某个值的唯一副本，需要保持它运行。

常用 API

操作	含义
hub.var(name)	稳定引用
variable.value / variable.meta	当前值副本和元数据
await variable.set(value) / delete()	更新本地内存
variable.watch(callback)	同步 (value, meta) 回调，返回取消函数
await variable.ready()	等待已知状态，也可能是不存在
await hub.connected() / flush()	等待连接 / NATS 传输
hub.status() / hub.watch(callback)	本地 SDK 状态
await hub.instances.list() / hub.instances.watch(callback)	观察到的 SDK 报告
await hub.close()	释放资源

导入 UNSET 表示本地没有可用值，None 是 JSON null。meta["exists"] 区分未知、不存在和有值。

注意： 回调必须同步且简短，可通过 on_callback_error 处理异常。

内存生命周期、版本和状态语义见工作原理。

连接配置

hub = KinopioHub(
    namespace="demo",
    servers=["tls://nats.example.com:4222"],
    mesh=False,
    discovery=False,
    tls={"handshake_first": True},
)

仅对 TLS-first 服务器启用 handshake_first。客户端直连支持 TCP/TLS/WS/WSS；鉴权使用 token 或 user/password。自定义客户端 TLS 需要 mesh=False，CA 和主机名校验保持启用。

默认自动模式与 JS、C++ 共享选举域。mesh={"group": "demo", "upstreams": ["nats://nats.example.com:7422"]} 可将托管节点连接到真正的 leaf 入口。

配置	用途
mesh["upstream_tls"]	leaf TLS：handshake_first、ca_file、cert_file、key_file
discovery=False	只关闭旧式 UDP 提示，不关闭选举

Python 不读取 HTTP 发现清单。

公开选项使用 snake_case，时间单位为秒。

参数	默认值或规则
timeout、peer_timeout	3；后者默认取前者 80%
health_interval、probe_interval	5、15
自动 connected() / flush()	默认允许 60 秒，显式设置优先
_ms 后缀字段，如 max_age_ms	毫秒
数据限制	10,000 个变量、16 MiB 记录数据、1,024 个观察实例

live 通道

live 命令与变量分开。在已连接的 Hub 中创建接收者：

channel = hub.live("control/command")
stop = await channel.subscribe(lambda value: print(value), max_age_ms=300)
# Keep this receiver running; later: await stop()

同一 namespace 内的另一个已连接 Hub 发送：await hub.live("control/command").send({"data": "step"}, timeout=3)。

整个部署中，每个通道只配置一个接收者，SDK 不进行跨进程所有权协调。相同值的每次发送都是独立命令。

注意： 接收者签发的租约拒绝过期、重复、乱序和旧连接消息；断连后租约失效，不重放。发送成功仅确认传输，不确认执行。

如果需要延后执行，订阅时传入 with_context=True，回调接收 (value, context)；实际执行前检查 context.is_valid()，expires_at 使用本地单调时钟。ROS 控制使用这条路径。Python 是唯一提供 live 通道的主机 SDK。

还提供 offline.py 和 sdk_status.py 示例，接受 KINOPIO_EXAMPLE_SERVERS、KINOPIO_TOKEN、KINOPIO_EXAMPLE_TLS_FIRST=1、KINOPIO_MESH=0、KINOPIO_LEAF_SERVERS。测试命令见开发说明。

事件与请求复用稳定引用：await ref.pub(data)、await ref.sub(handler)、await ref.req()，handle 返回值自动回复。同步 ref.get(fallback) 与 ref.watch_value(handler) 简化状态读取和观察。多响应、Headers、队列与 drain 详见消息 API。

跨 SDK 的主题、队列和生命周期约定见消息语义。