# 独立模式 独立模式允许基于[RVC2](https://docs.luxonis.com/hardware/platform/rvc/rvc2.md)的相机在通电时自动启动（[刷写的应用程序](#Flash%20the%20pipeline)），而无需连接到任何特定的主机。这在多相机架构中尤其有用，每个相机可以独立运行自己的应用程序，不依赖主机。每个相机既可以是服务器也可以是客户端，并可通过网络协议（例如HTTP、TCP、UDP、MQTT...）与其他相机或服务器通信。 相比之下，外设模式意味着相机直接连接到特定的主机。主机连接到一个空闲设备，上传流水线+资源（如神经网络模型），然后与设备通信（例如获取视频流）。 与外设模式相比，独立模式： * 无需主机启动应用程序，可独立连接到不同的计算机/服务器 * 对各种不稳定性（例如网络问题导致相机与主机连接断开）更具鲁棒性，因为它会自动重启应用程序 * 启动速度更快，因为主机无需发送流水线+资源（只需几秒钟） ## 支持 对于基于RVC2的相机，独立模式的支持取决于闪存和与外界通信的能力： * [OAK PoE](https://docs.luxonis.com/hardware/platform/deploy/poe-deployment-guide.md)（基于[RVC2](https://docs.luxonis.com/hardware/platform/rvc/rvc2.md)）具有板载闪存，可通过网络协议（例如HTTP、TCP、UDP、MQTT...）与外界通信。 * [OAK USB](https://docs.luxonis.com/hardware/platform/deploy/usb-deployment-guide.md)（基于[RVC2](https://docs.luxonis.com/hardware/platform/rvc/rvc2.md)）-不支持，因为它们无法与外界通信。 * [已弃用] OAK IoT（基于[RVC2](https://docs.luxonis.com/hardware/platform/rvc/rvc2.md)）具有板载内存和集成ESP32，能够通过WiFi/蓝牙与外界通信。 集成了Linux计算机的设备可以设置自己的应用程序在启动时运行： * [OAK-D CM4](https://shop71313603.taobao.com/?spm=pc_detail.30350276.shop_block.dshopinfo.27a17dd635FNDA) 和 [OAK-D CM4 PoE](https://shop71313603.taobao.com/?spm=pc_detail.30350276.shop_block.dshopinfo.27a17dd635FNDA) 集成了树莓派计算模块4 * [RVC3](https://docs.luxonis.com/hardware/platform/rvc/rvc3.md)（[RAE Robot](https://docs.luxonis.com/hardware/rae/get-started.md)）和 [RVC4](https://docs.luxonis.com/hardware/platform/rvc/rvc4.md)（[OAK4](https://docs.luxonis.com/hardware/platform/rvc/rvc4.md#Robotics%20Vision%20Core%204%20(RVC4)-OAK4%20cameras)）芯片均运行Linux操作系统，因此用户可以直接SSH进入设备，并设置自己的应用程序在启动时运行。 ## OAK PoE 独立模式 为了与外界（计算机）“通信”，OAK POE相机可以使用[Script节点](https://docs.luxonis.com/software-v3/depthai/depthai-components/nodes/script.md)来发送/接收网络数据包（HTTP/TCP/UDP...）。以下是一些示例： * [TCP流](https://github.com/luxonis/oak-examples/tree/master/gen2-poe-tcp-streaming)（相机作为服务器或客户端） * [HTTP服务器](https://docs.luxonis.com/software/depthai/examples/script_mjpeg_server.md#script-mjpeg-server) * [HTTP客户端](https://docs.luxonis.com/software/depthai/examples/script_http_client.md) * [MQTT客户端](https://github.com/luxonis/oak-examples/tree/master/gen2-poe-mqtt) > **DNS resolver** > 独立模式缺少DNS解析器，因此您需要使用IP地址而不是域名。 ### 转换为独立模式 由于主机和设备之间将没有直接通信，您首先需要移除所有XLinkIn节点，这些节点在外设模式下用于主机与设备之间的通信。 我们需要向流水线中添加Script节点，并将所有我们想要的数据（例如编码视频、神经网络元数据、IMU结果等）首先流式传输到Script节点，然后通过网络发送。 #### 示例 让我们将[YOLOv8](https://docs.luxonis.com/software-v3/depthai/examples/yolov8_nano.md)示例更新为独立模式。该示例使用YOLOv8模型检测视频流中的物体，并将视频流+元数据发送到主机，在那里进行可视化（边界框）并显示给用户。 首先，我们需要将代码分为两部分： * 一部分将是流水线定义（将刷写到设备上） * 另一部分将是主机代码（接收数据、可视化并显示） 在示例的源代码（Python）中，我们可以看到管道定义代码在第 41 到 75 行。我们可以将此定义复制到一个单独的文件中（例如 oak.py）。同时，我会移除 XLinkOut 节点，因为在独立模式下它们会被忽略，并创建一个 Script 节点，用于通过网络发送数据。 ```python pipeline = dai.Pipeline() # 定义源和输出 camRgb = pipeline.create(dai.node.ColorCamera) detectionNetwork = pipeline.create(dai.node.YoloDetectionNetwork) # 属性 camRgb.setPreviewSize(640, 352) camRgb.setResolution(dai.ColorCameraProperties.SensorResolution.THE_1080_P) camRgb.setInterleaved(False) camRgb.setColorOrder(dai.ColorCameraProperties.ColorOrder.BGR) camRgb.setFps(40) # 网络特定设置 detectionNetwork.setConfidenceThreshold(0.5) detectionNetwork.setNumClasses(80) detectionNetwork.setCoordinateSize(4) detectionNetwork.setIouThreshold(0.5) detectionNetwork.setBlobPath(nnPath) detectionNetwork.setNumInferenceThreads(2) detectionNetwork.input.setBlocking(False) # 创建处理 TCP 通信的 Script 节点 script = pipeline.create(dai.node.Script) script.setProcessor(dai.ProcessorType.LEON_CSS) script.setScript(""" while True: detections = node.io["detection_in"].get().detections img = node.io["frame_in"].get() """) # 将输出（RGB 流、NN 输出）链接到 Script 节点 detectionNetwork.passthrough.link(script.inputs['frame_in']) detectionNetwork.out.link(script.inputs['detection_in']) ``` 现在，我们只需要关注 Script 节点部分，即启动 TCP 服务器并将帧和检测结果发送到连接的客户端。我们已经有了 [TCP 流式传输示例代码](https://github.com/luxonis/oak-examples/blob/master/gen2-poe-tcp-streaming/oak.py#L21-L45)，可以将其作为基础。我们还需要将检测结果（元数据）流式传输到客户端，因此需要添加相应代码。 最终的 Script 节点代码如下： ```python import socket import time import threading node.warn("Server up") server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.bind(("0.0.0.0", 5000)) # 在端口 5000 上创建 TCP 服务器 server.listen() while True: conn, client = server.accept() node.warn(f"Connected to client IP: {client}") try: while True: detections = node.io["detection_in"].get().detections # 读取 ImgDetections 消息，仅获取检测结果 img = node.io["frame_in"].get() # 读取 ImgFrame 消息 node.warn('Received frame + dets') img_data = img.getData() ts = img.getTimestamp() det_arr = [] for det in detections: det_arr.append(f"{det.label};{(det.confidence*100):.1f};{det.xmin:.4f};{det.ymin:.4f};{det.xmax:.4f};{det.ymax:.4f}") det_str = "|".join(det_arr) # 将检测结果序列化为字符串，用于发送给客户端 header = f"IMG {ts.total_seconds()} {len(img_data)} {len(det_str)}".ljust(32) node.warn(f'>{header}<') conn.send(bytes(header, encoding='ascii')) # 发送头部 if 0 < len(det_arr): # 如果有检测结果，则发送序列化后的检测数据 conn.send(bytes(det_str, encoding='ascii')) conn.send(img_data) # 发送实际图像帧 except Exception as e: node.warn("Client disconnected") ``` 现在，我们已经准备好带有 Script 节点的管道部分，需要创建 host.py 脚本来连接相机的 TCP 服务器，接收视频流 + 元数据，并可视化显示数据。我们可以使用 [host.py 脚本](https://github.com/luxonis/oak-examples/blob/master/gen2-poe-tcp-streaming/host.py) 作为基础，并修改它以接收和可视化检测结果： ```python import socket import re import cv2 import numpy as np # 输入你自己的 IP！在运行 oak.py 脚本后，终端会打印出 IP 地址 OAK_IP = "10.12.101.188" labels = [ "person", "bicycle", "car", "motorbike", "aeroplane", "bus", "train", "truck", "boat", "traffic light", "fire hydrant", "stop sign", "parking meter", "bench", "bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe", "backpack", "umbrella", "handbag", "tie", "suitcase", "frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket", "bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "chair", "sofa", "pottedplant", "bed", "diningtable", "toilet", "tvmonitor", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "sink", "refrigerator", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush" ] def get_frame(socket, size): bytes = socket.recv(4096) while True: read = 4096 if size-len(bytes) < read: read = size-len(bytes) bytes += socket.recv(read) if size == len(bytes): return bytes sock = socket.socket() sock.connect((OAK_IP, 5000)) try: COLOR = (127,255,0) while True: header = str(sock.recv(32), encoding="ascii") chunks = re.split(' +', header) if chunks[0] == "IMG": print(f">{header}<") ts = float(chunks[1]) imgSize = int(chunks[2]) det_len = int(chunks[3]) if 0 < det_len: # 如果有检测结果，则读取它们 det_str = str(sock.recv(det_len), encoding="ascii") img = get_frame(sock, imgSize) # 获取图像帧 img_planar = np.frombuffer(img, dtype=np.uint8).reshape(3, 352, 640) # 重塑形状（平面格式） img_interleaved = img_planar.transpose(1, 2, 0).copy() # 转换为交错格式（cv2 需要） # 可视化检测结果： if 0 < det_len: dets = det_str.split("|") # 反序列化检测结果 for det in dets: det_section = det.split(";") class_id = int(det_section[0]) confidence = float(det_section[1]) bbox = [ # 从相对坐标转换为绝对坐标 int(float(det_section[2]) * img_interleaved.shape[1]), int(float(det_section[3]) * img_interleaved.shape[0]), int(float(det_section[4]) * img_interleaved.shape[1]), int(float(det_section[5]) * img_interleaved.shape[0]) ] cv2.putText(img_interleaved, labels[class_id], (bbox[0] + 10, bbox[1] + 20), cv2.FONT_HERSHEY_TRIPLEX, 0.5, COLOR) cv2.putText(img_interleaved, f"{int(confidence)}%", (bbox[0] + 10, bbox[1] + 40), cv2.FONT_HERSHEY_TRIPLEX, 0.5, COLOR) cv2.rectangle(img_interleaved, (bbox[0], bbox[1]), (bbox[2], bbox[3]), COLOR, 2) # 显示带有可视化检测结果的帧 cv2.imshow("Img", img_interleaved) if cv2.waitKey(1) == ord('q'): break except Exception as e: print("Error:", e) sock.close() ``` 完整代码（oak.py 和 host.py）可在[此处找到](https://github.com/luxonis/oak-examples/tree/master/gen2-poe-tcp-streaming/poe-host-yolo)。请注意，对于独立模式，您可能需要烧录静态 IP，这样就不必每次更改代码中的 IP 地址。 ## 引导加载程序 [引导加载程序](https://docs.luxonis.com/software-v3/depthai/depthai-components/bootloader.md) 处理已烧录应用程序（独立模式）的启动，我们建议使用最新的引导加载程序，该程序可以使用 [设备管理器](https://docs.luxonis.com/software-v3/depthai/depthai-components/bootloader.md)（GUI 工具）进行烧录。若要查看其背后的 API 代码，请参阅 [烧录引导加载程序](https://docs.luxonis.com/software-v3/depthai/examples/flash_bootloader.md) 示例代码（仅 API 脚本）。 ## 烧录管线 在您定义好（独立模式的）[管线](https://docs.luxonis.com/software-v3/depthai/depthai-components/pipeline.md) 并将最新的 [引导加载程序](https://docs.luxonis.com/software-v3/depthai/depthai-components/bootloader.md) 烧录到设备上之后，您可以将管线及其资源（例如神经网络模型）烧录到设备。您可以使用以下代码片段烧录管线： ```python import depthai as dai pipeline = dai.Pipeline() # 定义独立管线；添加节点并连接它们 # cam = pipeline.create(dai.node.ColorCamera) # script = pipeline.create(dai.node.Script) # ... # 烧录管线 (f, bl) = dai.DeviceBootloader.getFirstAvailableDevice() bootloader = dai.DeviceBootloader(bl) progress = lambda p : print(f'烧录进度: {p*100:.1f}%') bootloader.flash(progress, pipeline) ``` 成功烧录管线后，每次给设备上电时，该管线会自动启动。如果您想更改已烧录的管线，只需再次重新烧录即可。 ### DepthAI 应用程序包 (.dap) 或者，您也可以使用 [设备管理器](https://docs.luxonis.com/software-v3/depthai/depthai-components/bootloader.md) 烧录管线。对于这种方法，您需要一个 DepthAI 应用程序包 (.dap)，可以使用以下脚本创建： ```python import depthai as dai pipeline = dai.Pipeline() # 定义独立管线；添加节点并连接它们 # cam = pipeline.create(dai.node.ColorCamera) # script = pipeline.create(dai.node.Script) # ... # 创建 DepthAI 应用程序包 (.dap) (f, bl) = dai.DeviceBootloader.getFirstAvailableDevice() bootloader = dai.DeviceBootloader(bl) bootloader.saveDepthaiApplicationPackage(pipeline=pipeline, path=<新_dap文件的路径>) ``` ## 清除闪存 由于管线会在设备上电时启动，这可能导致不必要的发热。如果您想清除已烧录的管线，请使用下面的代码片段。 ```python import depthai as dai (f, bl) = dai.DeviceBootloader.getFirstAvailableDevice() if not f: print('未发现设备，正在退出...') exit(-1) with dai.DeviceBootloader(bl) as bootloader: bootloader.flashClear() print('成功清除引导加载程序的闪存') ``` ## 恢复出厂设置 如果您不小心让设备变砖，或者只是想清除所有内容（已烧录的管线/资源和引导加载程序配置），我们建议使用 [设备管理器](https://docs.luxonis.com/software-v3/depthai/depthai-components/bootloader.md)。恢复出厂设置还会烧录最新的引导加载程序。