plexus-python

## Обзор plexus-python — легкий Python SDK для платформы Plexus. 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/с на соединение). 25 каналов на 100 Гц при отправке по одному — это 2 500 сообщений/с; переполнение отбрасывается до записи в хранилище. ```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()` возвращает run с `test_result`. ### event(name, data) Используется для записи «того, что произошло», а не «того, что непрерывно измеряется»: сбоев, переключений состояний, действий оператора, записей журнала. Платформа отображает события как маркеры на графике телеметрии, а не как линии временных рядов. Ограничения на одно событие: строковые значения 256 байт, значения dict/list 4 096 байт JSON, не более 16 тегов. ### Журналирование В пакете нет загрузки файлов журналов и нет `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`. Требуется API-ключ с правом Receive commands. 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, обучающие агентов программирования работе с API Plexus (эндпоинты, потоковая передача в реальном времени, типичные ошибки, приводящие к молчаливым 400). `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. ## Лицензия Apache 2.0.