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 키는 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`에 연결하여 더 낮은 지연의 텔레메트리 스트림과 대시보드 트리거 동작을 전달하는 채널을 얻습니다. 소켓을 사용할 수 없으면 투명하게 `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 키를 사용해야 합니다. 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 패키지에는 코딩 에이전트가 Plexus API(엔드포인트, 실시간 스트리밍, 조용한 400을 유발하는 흔한 오류)를 사용하도록 가르치는 세 가지 skill이 포함되어 있습니다. `plexus skills install`은 `~/.claude/skills`에 설치하고, `--project`를 추가하면 `./.claude/skills`에 설치되어 저장소와 함께 이동합니다. 순수 Markdown이며 설치나 자격 증명이 필요 없습니다. ## 아키텍처 ``` Your code ── px.send() ── WebSocket /ws/device (또는 HTTP POST /ingest) ──> plexus-gateway ──> ClickHouse + Dashboard ``` 에이전트, 데몬, 어댑터가 없는 하나의 경량 경로입니다. 전체 HardwareOps 플랫폼(대시보드, 알림, RCA, 기단 뷰)은 app.plexus.company의 Web UI에 있습니다. ## 라이선스 Apache 2.0.