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` をそのまま返すため、2 台のデバイスが同じ名前を宣言すると同じ 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) 複数の読み取り値を一度に送信し、タイムスタンプを共有して 1 回のネットワーク呼び出しにまとめます。`points` は `(metric, value)` タプルのリスト、またはポイントごとのタイムスタンプが必要な場合は `(metric, value, timestamp)` の 3 要素タプルです。 ### batch() 毎秒数個を超える読み取り値がある場合に使用します。各 `send()` は 1 つの WebSocket メッセージであり、ゲートウェイが制限するのはポイント数ではなくメッセージ数です（接続あたり 2,000/s）。25 チャンネルを 100 Hz で 1 件ずつ送信すると 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、tag は最大 16 個です。 ### ログ 本パッケージにはログファイルのアップロードも `logging.Handler` もありません。重要なログ行を Plexus へ送るにはイベントとして送信します：`px.event("log", {"level": "error", "msg": "..."})`。タイムラインに表示したい行（エラー、警告、状態変化）のみを転送し、すべてのデバッグログを送らないでください。 ## 動画ストリーム 動画には有料プランが必要です。フレームは WebSocket を通り、無料プランではゲートウェイが拒否します。フレームは視聴者へリアルタイムで転送され、アプリ内で Record を押したときのみ保存され、1 回の録画は最長 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()` はデフォルトで 1 つのタイムスタンプを共有します。 ## トランスポート デフォルトでは 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 パッケージには 3 つの 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 ``` 軽量な経路が 1 つだけで、agent もデーモンもアダプターもありません。完全な HardwareOps プラットフォーム（ダッシュボード、アラート、RCA、フリートビュー）は app.plexus.company の Web UI にあります。 ## ライセンス Apache 2.0。