plexus-python

## 概述 plexus-python 是 Plexus 平台的轻量 Python SDK。Plexus 面向硬件团队提供时序数据存储与仪表盘：把无人机、机器人、IoT 设备的数据流式写入 Plexus Time Series，或接入已有数据库，获得实时仪表盘与告警。本包只负责把数据送到 Plexus 网关，存储、仪表盘、告警和机队管理都在平台侧。 ## 快速开始 ```bash pip install plexus-python ``` ```python from plexus import Plexus px = Plexus(api_key="plx_xxx", source_id="device-001") px.send("temperature", 72.5) ``` API key 可在 app.plexus.company/api 获取，也可运行 `plexus init` 在浏览器中授权本机。 ## 设备标识 每台设备需要唯一的 `source_id`。推荐用引导脚本设置，它要求先给出设备名： ```bash curl -sL https://app.plexus.company/setup | bash -s -- \ --key plx_xxx --name drone-01 ``` 名称会转成设备的 `source_id`，需匹配 `^[a-z0-9][a-z0-9._-]*$`（最长 256 字符）。若既没有 `--name` 也没有代码里的 `source_id=...`，SDK 首次运行会生成随机 id（如 `source-1a2b3c4d`）并保存到 `~/.plexus/config.json`。不要用主机名：克隆的 SD 卡镜像都会以 `raspberrypi` 启动，遥测会合并到同一个 source。名称不会自动去重，网关原样回显声明的 `source_id`，两台设备声明同名就会写入同一 source。 ## 核心方法 ### send(metric, value) 最常用的方法，每次有新读数时调用。`metric` 是点分命名空间字符串（如 `"motor.rpm"`），`value` 接受任意可 JSON 序列化类型：float/int（传感器读数、计数器）、str（状态机、错误码）、bool（二值标志）、dict（向量、结构化读数）、list（波形、关节角）。可选参数 `tags={"motor_id": "A1"}` 用于仪表盘过滤，`timestamp=t` 指定 Unix 秒级时间戳。 ### send_batch(points) 一次发送多个读数，共享时间戳并合并为一次网络调用。`points` 是 `(metric, value)` 元组列表，或需要逐点时间戳时用 `(metric, value, timestamp)` 三元组。 ### batch() 每秒超过几个读数时使用。每次 `send()` 是一条 WebSocket 消息，网关限制的是消息数而非点数（每连接 2,000/s）。25 个通道 100 Hz 逐条发送就是 2,500 消息/s，溢出会在存储前被丢弃。 ```python with px.batch(interval_ms=50) as b: while running: b.send("att.pos_x", att.x) ``` 后台线程每 `interval_ms` 刷新队列，离开代码块时刷新剩余数据，读数保留采集时的时间戳。若网关丢弃帧会报告 `RATE_LIMITED`，SDK 在 `px.rate_limited_frames` 计数并在下次发送时抛出 `RateLimitedError`。 ### run(name) run 是 source 上的命名时间窗口，可在 `/runs` 回看、按 T+0 对齐比较，并在关闭时按通过标准检查。离开代码块将 run 标记为 `completed`，异常则标记为 `aborted` 并重新抛出。也可用 `px.start_run()` / `px.end_run()` 分开调用，`end_run()` 返回带 `test_result` 的 run。 ### event(name, data) 用于记录“发生的事”而非“持续测量的量”：故障、状态切换、操作员动作、日志条目。平台把事件显示为遥测图上的标记而非时序线。单个事件限制：字符串值 256 字节，dict/list 值 4,096 字节 JSON，最多 16 个 tag。 ### 日志 本包没有日志文件上传，也没有 `logging.Handler`。要把重要日志行送进 Plexus，用事件发送：`px.event("log", {"level": "error", "msg": "..."})`。只转发希望出现在时间线上的行（错误、警告、状态变化），不要每条调试日志都发。 ## 视频流 视频需要付费计划：帧走 WebSocket，免费计划网关会拒绝。帧实时转发给观看者，仅在应用内按下 Record 时存储，单次录制最长 4 小时。 - `send_video_frame(frame, camera_id)`：自己掌控采集循环时使用（picamera2 回调、OpenCV VideoCapture 循环、自管 FFmpeg 管道）。接受 numpy ndarray（需 opencv-python）、JPEG 字节（原样透传）、其他图像字节（经 Pillow 解码并重编码为 JPEG，需 `pip install plexus-python[video]`）。 - `stream_camera(url, camera_id)`：有 RTSP 流或视频文件、不想自己管理采集循环时使用，SDK 内部运行 FFmpeg（需 FFmpeg 在 `$PATH`）。返回 `threading.Event`，调用 `.set()` 停止，在后台线程运行。 ## 自带协议 本包不含适配器、自动检测或守护进程，只有客户端。用你本来就会用的库把值送进 `px.send()` 即可，README 给出 MAVLink（pymavlink）、CAN（python-can）、MQTT（paho-mqtt）、I2C 传感器（Adafruit CircuitPython）的示例，`examples/` 中有可运行版本。 ## 可靠性 每次发送先在本地缓冲再走网络，带指数退避重试，跨断网保数据。缓冲默认在磁盘（SQLite），可跨重启和断电；`persistent_buffer=False` 则仅内存。可用 `px.buffer_size()` 和 `px.flush_buffer()` 查看点数与刷新。 ## 时间戳与时钟校正 默认情况下 SDK 自己选时间。走 WebSocket 时每次连接都会与网关时钟同步，即使设备系统时钟不准（首次启动无 NTP、RTC 过期、全新系统镜像），数据也能落在时间线正确位置。有可靠外部时间源（GPS、可信 RTC、主机 NTP）或回放已知时间戳的历史数据时，应显式传 `timestamp`。已知限制：时钟同步在 WebSocket 重连时刷新，长时间连接且 RTC 漂移的设备会在重连间隔内累积未校正漂移；HTTP 回退路径不接收时钟同步；`send_batch()` 默认共享一个时间戳。 ## 传输 默认通过 WebSocket 连接网关的 `/ws/device`，获得更低延迟的遥测流，以及承载仪表盘触发动作的通道。socket 不可用时透明回退到 `POST /ingest`，不丢数据。没有传输选择器，SDK 总是优先 WebSocket。免费计划下网关拒绝设备 WebSocket（`streaming_requires_plan`），SDK 自行回退 HTTP，`send()`、`send_batch()`、`batch()`、`event()` 仍可用；实时流和视频需付费计划，免费计划限 3 台设备和 7 天历史。 ## 命令 可声明代码能接受哪些命令，需在首次 `send()` 前声明（声明随认证帧发送）。用 `@px.command(...)` 装饰处理器，参数支持 string（maxLength、enum）、integer/number（minimum、maximum、unit）、boolean，外加 title、description、default、required，最多 16 个且扁平无嵌套。`danger` 分 normal/dangerous/critical，`idempotent` 为真时掉线后才会重投，`expires_in` 为 5–3600 秒，`concurrency` 可选 accept/reject。处理器以 `handler(run, **params)` 调用，参数已校验并转换，返回值成为 run 结果，异常使 run 变为 `failed`。需使用带 Receive commands 权限的 API key。SDK 会确认每个 run、不重复执行同一 run id、用单调时钟判断过期、重连后重放未确认状态。`px.on_command()` 已弃用但仍以旧签名工作。 ## 环境变量 `PLEXUS_API_KEY`（必填）、`PLEXUS_GATEWAY_URL`（默认 `https://gateway.plexus.company`）、`PLEXUS_GATEWAY_WS_URL`（默认 `wss://gateway.plexus.company`）。 ## Agent skills 包内附带三个 skill，教编码代理使用 Plexus API（端点、实时流、导致静默 400 的常见错误）。`plexus skills install` 安装到 `~/.claude/skills`，加 `--project` 则安装到 `./.claude/skills` 随仓库走。纯 Markdown，无需安装或凭据。 ## 架构 ``` Your code ── px.send() ── WebSocket /ws/device (或 HTTP POST /ingest) ──> plexus-gateway ──> ClickHouse + Dashboard ``` 一条轻量路径，无 agent、无守护进程、无适配器。完整 HardwareOps 平台（仪表盘、告警、RCA、机队视图）在 app.plexus.company 的 Web UI。 ## 许可证 Apache 2.0。