xovis-sdk

# Xovis 软件开发工具包 <div align="center"> | **核心软件开发工具包** | **集成** | **智能代理层** | |:---:|:---:|:---:| | [](https://pypi.org/project/xovis-sdk/1.0.0a29/) | [](https://openai.com/) | [](https://modelcontextprotocol.io/) | | [](https://www.npmjs.com/package/xovis-sdk/v/1.0.0-a29/) | [](https://www.anthropic.com/) | [](https://langchain.com/) | | [](https://github.com/xovis-open-sdk/xovis-sdk) | [](https://smithery.ai/servers/xovis-sdk/xovis-mcp) | [](https://crewai.com/) | | [](https://opensource.org/licenses/MIT) | [](https://smithery.ai/server/xovis-sdk) | [](https://cursor.sh/) | </div> Xovis 3D 传感器和 Xovis HUB 云基础设施的企业级集成软件开发工具包。 **[阅读完整文档网站 →](https://xovis-open-sdk.github.io/xovis-sdk/)** > **合规性说明：** 本项目是一个独立的开源项目。它不与 Xovis AG 官方关联、维护或背书。 --- ## ⚠️ Xovis HUB 云兼容性和速率限制 该软件开发工具包设计用于企业级车队管理。由于 `HubClient` 和 `bulk_execute` 方法的高并发性，强烈建议使用 **Xovis HUB Pro** 订阅。免费层可能导致激进的 HTTP 429 速率限制耗尽，这将破坏自动配置和遥测管道。 --- ## 概述 将原生 Xovis DataPush 协议和 REST API 集成到企业数据管道中通常需要大量的样板代码、复杂的 state 管理和严格的网络处理，以保持实时 DataPush 采集（高达 12.5Hz）。 该软件开发工具包将 Xovis 硬件的复杂性抽象为统一的、现代的和类型安全的“通用翻译器”架构。它完全解耦原始边缘遥测与下游基础设施，使工程师能够专注于空间分析、车队管理和数据仓库。 ### 系统数据流 ```mermaid graph TB %% Class Definitions for High-End Professional Design classDef hardware fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc; classDef core fill:#0f172a,stroke:#2dd4bf,stroke-width:2px,color:#f8fafc; classDef agentic fill:#1e1b4b,stroke:#818cf8,stroke-width:2px,color:#f8fafc; classDef external fill:#0c0a09,stroke:#fb923c,stroke-width:2px,color:#f8fafc; subgraph "Xovis 硬件层" direction TB A["物理传感器 / 蜘蛛"]:::hardware H["Xovis HUB 云"]:::hardware H -- "安全代理隧道（M2M）" --> A end subgraph "数据平面（高频采集）" B["Xovis 采集服务器<br/>(TCP / UDP / HTTP Webhooks)"]:::core S["XovisSink<br/>(零复制目标)"]:::core end subgraph "控制和状态平面（软件开发工具包核心）" C["设备客户端<br/>(直接本地 REST API)"]:::core F["Hub 客户端<br/>(云 REST API)"]:::core D["配置缓存 / 状态桶<br/>(三层自动完成友好)"]:::core F -- "connect_device()" --> C C <--> D end subgraph "智能代理和工具层" G["XovisAI 工具包<br/>(类型安全 AI 工具)"]:::agentic M["MCP 服务器<br/>(模型上下文协议)"]:::agentic R["Xovis 命令行和任务控制 TUI"]:::agentic end %% 数据连接 A -->|Data-Push 高达 12.5Hz| B B -->|滑动缓冲区提取| S %% 控制连接 A <-->|本地 REST API v5| C H <-->|Hub REST API| F %% 车队集成 D -. "反射状态" .-> R F -. "车队同步" .-> D %% AI 集成 D --- G F --- G G --- M M --- LLM["LLM 和自主代理<br/>(Cursor / Claude / LangChain)"]:::external ``` --- ## 架构支柱 该软件开发工具包严格分为四个不同的平面，以防止在高频操作期间阻塞异步事件循环，同时启用自主系统： 1. **数据平面（遥测采集）：** 一个零复制、无锁遥测采集引擎，支持高频 **Live-Push（高达 12.5Hz）** 坐标和分钟 **Logic-Push** 事件，支持 TCP、UDP 和 HTTP。 [阅读数据平面文档](https://xovis-open-sdk.github.io/xovis-sdk/architecture/data_plane/) 2. **控制平面（配置）：** 一个弹性、异步的 HTTP 引擎，封装 Xovis 边缘和 HUB API，具有严格的 Pydantic V2 模式验证和自动 Auth0令牌生命周期。 [阅读控制平面文档](https://xovis-open-sdk.github.io/xovis-sdk/architecture/control_plane/) 3. **拓扑和状态平面（车队管理）：** 一个内存高效的图引擎，建模复杂的 multisensor 父/子关系，具有离线优先的 **原生状态桶**。 [阅读状态和拓扑文档](https://xovis-open-sdk.github.io/xovis-sdk/architecture/state_topology/) 4. **智能代理层（AI 管理）：** 一个通用工具适配器和模型上下文协议（MCP）服务器，为现代 AI 框架和 LLM 提供自主管理能力。 [阅读智能代理层文档](https://xovis-open-sdk.github.io/xovis-sdk/architecture/agentic_layer/) --- ## 为什么使用该软件开发工具包？ | 集成挑战 | 传统方式 | `xovis-sdk` 方式 | | :--- | :--- | :--- | | **遥测采集** | 编写大量样板代码监听器，手动解析和管理高频（12.5Hz）数据包缓冲区。 | 生产就绪的零复制 `XovisTCPServer` / `XovisUDPServer`，并发流式传输帧而不阻塞事件循环。 | | **混合车队操作** | 多设备文件名冲突、版本波动和网络中具有异构固件版本的缓存覆盖竞争。 | 类别范围、版本隔离的存储在 `_local_resources/` 中，保持模式和拓扑状态完全解耦。 | | **路径遍历和路由** | 手动探测 LAN 主机、检查可用性、调整端口和解析代理隧道以访问远程传感器。 | 通过 `UnifiedDeviceClient` 自动混合网络探测，解析直接 LAN、VPN 和安全 HUB 云代理隧道。 | | **状态持久性** | 在只读环境（例如 Docker、AWS Lambda）中运行时无声失败或抛出硬错误，而没有可写的 工作空间。 | 弹性 **三层缓存系统**，自动尝试本地 CWD 工作空间、回退到系统缓存，并默认为仅 RAM。 | | **多传感器拓扑** | 复杂的递归查询以映射主 Spider 控制器、拼接的传感器上下文和子镜头层次结构。 | 动态 `child_devices` 和 `child_caches` 访问器，具有合并的批量集合聚合和并发组广播外壳。 | | **逻辑车队分组** | 通过手动 IP 循环和单独的身份验证流程管理数千个未拼接的传感器。 | 软件开发工具包原生的 `DeviceGroup` 和 `HubFleetDirectory`，用于跨独立网络的动态、IDE 自动建议的批量执行。 | | **API 代码自动完成** | 解析原始 JSON/YAML 有效负载、导航未类型化的字典和在资源名称中包含空格时崩溃。 | 基于正则表达式的动态命名空间清理（点表示法自动完成）与原始名称索引括号表示法回落。 | | **AI 代理管理** | 创建不安全的自定义脚本或将原始、未经验证的硬件 API 暴露给自主 LLM 推理循环。 | 安全设计模型上下文协议（MCP）服务器和 `XovisAIToolkit`，在严格的安全防护栏中封装操作。 | --- ## 分级编程之旅 从原始的、高速度的物理数据流式传输到完整的状态管理和多传感器车队管理，分三个渐进的层级： #### 层级 1：多环境发现和三层缓存 在本地 LAN、VPN 和安全 Xovis HUB 云代理隧道中建立与您多样化车队的连接。自动初始化弹性三层缓存系统，以持久化配置状态以供离线自动完成。 ```python import asyncio from xovis import UnifiedDeviceClient, HubClient, CachePaths async def main(): # 通过 HUB 连接以管理本地、VPN 和云代理隧道连接 async with HubClient() as hub: # 统一客户端自动探测和解析不同的连接路径 device_lan = UnifiedDeviceClient("192.168.1.10") device_vpn = UnifiedDeviceClient("10.8.0.5") device_hub = UnifiedDeviceClient("00:26:8c:12:34:56", hub_client=hub) # 并发同步和缓存配置以进行离线优先操作 async with device_lan, device_vpn, device_hub: await asyncio.gather( device_lan.cache.sync(), device_vpn.cache.sync(), device_hub.cache.sync() ) # 将聚合的、自动完成友好的状态保存到磁盘 device_lan.cache.export_to_file(CachePaths.DEVICE_STATE) print(f"成功缓存状态在：{CachePaths.DEVICE_STATE}") if __name__ == "__main__": asyncio.run(main()) ``` #### 层级 2：并发多接收器遥测采集和端口偏移 防止多设备网络上的端口冲突。通过点表示法加载离线缓存配置，动态偏移遥测端口，启动并发无锁 TCP/UDP 采集服务器，并激活实时流推送代理。 ```python import asyncio from xovis import UnifiedDeviceClient, XovisTCPServer, XovisSink class CustomTelemetrySink(XovisSink): async def write(self, frame: dict) -> None: # 零复制、高频流处理（坐标/人员跟踪） for person in frame.get("people", []): print(f"跟踪 ID {person['id']} 在 ({person['x']}, {person['y']})") async def main(): # 并发启动两个非阻塞 TCP 遥测采集服务器 server_1 = XovisTCPServer(port=9000, sink=CustomTelemetrySink()) server_2 = XovisTCPServer(port=10000, sink=CustomTelemetrySink()) await asyncio.gather(server_1.start(), server_2.start()) # 连接到设备，在连接代理上动态改变本地端口以防止冲突 async with UnifiedDeviceClient("10.8.0.5") as device: await device.cache.load_from_disk() # 通过自动完成友好的点表示法安全获取配置 conn = device.cache.connections.by_name.MainSink conn.port += 1000 # 动态偏移端口 # 应用连接有效载荷并以编程方式激活遥测流 await device.datapush.update_connection(conn.id, conn) await device.datapush.enable_agent("TelemetryAgent") print("应用连接偏移并激活代理实时流!") if __name__ == "__main__": asyncio.run(main()) ``` #### 层级 3：多传感器和车队管理 无论是解析拼接的物理多传感器拓扑还是将整个网络中的独立传感器分组到逻辑车队桶中，软件开发工具包都提供了并发广播外壳和离线优先的 IDE 自动建议。 ```python import asyncio from xovis import UnifiedDeviceClient, HubClient from xovis.api.fleet import HubFleetDirectory, DeviceGroup async def main(): # --- 示例 A：物理拼接多传感器拓扑 --- async with HubClient() as hub: # 连接到主 Spider 或 PC 传感器以遍历拼接的多传感器拓扑 async with UnifiedDeviceClient("00:26:8c:12:34:56", hub_client=hub, cache_child_devices=True) as master: await master.cache.sync() # 并发广播实时诊断操作到所有子传感器 context = master.cache.multisensors.by_name.Stitched_Context bulk_results = await context.child_devices.images.get_raw_left() print(f"捕获子诊断帧。成功：{len(bulk_results.successes)}") # --- 示例 B：逻辑车队管理和批量执行 --- # 加载车队目录（来自 Hub JSON 导出或实时 Hub API） directory = HubFleetDirectory.from_file() # 使用离线优先的 IDE 自动建议按类别对独立传感器进行分组 grp = DeviceGroup.from_directory_nodes( name="Office_Rollout", nodes=directory.by_category.Xovis_Office, password="SuperSecureFleetPassword123" ) # 对整个车队使用并发命令，并具有完整的 IDE 自动建议 # 返回一个弹性批量操作结果，映射成功和异常 fleet_result = await grp.network.get_hostname() print(f"成功访问 {len(fleet_result.successes)} 个独立设备.") if __name__ == "__main__": asyncio.run(main()) ``` [在状态和拓扑文档中了解更多关于状态管理功能的信息](https://xovis-open-sdk.github.io/xovis-sdk/architecture/state_topology/) --- ## 模型上下文协议（MCP） Xovis 软件开发工具包包括一个一流的 MCP 服务器，允许 AI 代理（如 Claude Desktop 和 Cursor）直接管理硬件。 **使用 Smithery 快速安装：** ```bash npx -y smithery install xovis-sdk ``` 查看完整的 [MCP 指南](https://xovis-open-sdk.github.io/xovis-sdk/ai/mcp/) 和 [AI 安全和防护栏](https://xovis-open-sdk.github.io/xovis-sdk/ai/safety_guardrails/) 以获取更多详细信息。 --- ## 开发者体验和命令行界面 软件开发工具包包括一个原生命令行界面工具，用于管理缓存预热，从离线缓存中提取拓扑数据，生成具有完美 IDE 自动完成的严格 Python 类型/字面值，以及交互式 REPL 访问器和完整的任务控制终端用户界面。 ### 缓存预热和同步命令行界面 直接从终端填充模式定义和本地配置： ```bash # 预热本地设备配置/模式（自动下载 OpenAPI + DataPush JSON 模式） xovis-cli warmup 192.168.1.10 --user admin --pass secret_pass # 从 Xovis HUB 预热整个云车队状态和云模式 xovis-cli warmup-hub --client-id MY_ID --client-secret MY_SECRET # 从设备（本地 IP 或通过 Hub 隧道）为完美自动完成生成静态类型 xovis-cli generate-types --device 192.168.1.10 xovis-cli generate-types --device 00:11:22:33:44:55 --via-hub # 启动 Xovis 开放软件开发工具包任务控制 TUI xovis-cli ui ``` ### 交互式 REPL 资源管理器 直接在交互式 IPython、Jupyter 笔记本或标准 Python REPL 壳中公开您的物理传感器车队。`REPLAccessor` 规范化集合绕过严格的验证锁以实现即时对象属性发现： ```python >>> from xovis import UnifiedDeviceClient >>> device = UnifiedDeviceClient("10.8.0.5") >>> await device.cache.load_from_disk() # 具有正则表达式清理名称的动态点表示法自动完成 >>> device.cache.zones.by_name.Main_Entrance <Zone id="1" name="Main Entrance"> # 无缝列表索引和原始名称包含空格或字符 >>> device.cache.zones["Main Entrance"] <Zone id="1" name="Main Entrance"> ``` --- ## 企业测试和贡献 `xovis-sdk` 遵循企业 SDET 标准的最高级别，使用具有严格幂等性和硬终止边界的 4 级测试矩阵。 要贡献，请参阅我们的 [工程指南](https://xovis-open-sdk.github.io/xovis-sdk/contributing/engineering_guidelines/)，并确保在提交 PR 之前所有检查都通过。