独立模式
- 无需主机启动应用程序，可独立连接到不同的计算机/服务器
- 对各种不稳定性（例如网络问题导致相机与主机连接断开）更具鲁棒性，因为它会自动重启应用程序
- 启动速度更快，因为主机无需发送流水线+资源（只需几秒钟）
支持
- OAK PoE（基于RVC2）具有板载闪存，可通过网络协议（例如HTTP、TCP、UDP、MQTT...）与外界通信。
- OAK USB（基于RVC2）-不支持，因为它们无法与外界通信。
- [已弃用] OAK IoT（基于RVC2）具有板载内存和集成ESP32，能够通过WiFi/蓝牙与外界通信。
- OAK-D CM4 和 OAK-D CM4 PoE 集成了树莓派计算模块4
- RVC3（RAE Robot）和 RVC4（OAK4）芯片均运行Linux操作系统，因此用户可以直接SSH进入设备，并设置自己的应用程序在启动时运行。
OAK PoE 独立模式
DNS resolver
独立模式缺少DNS解析器，因此您需要使用IP地址而不是域名。
转换为独立模式
示例
- 一部分将是流水线定义（将刷写到设备上）
- 另一部分将是主机代码（接收数据、可视化并显示）
oak.py）。同时，我会移除 XLinkOut 节点，因为在独立模式下它们会被忽略，并创建一个 Script 节点，用于通过网络发送数据。Python
1pipeline = dai.Pipeline()
2
3# 定义源和输出
4camRgb = pipeline.create(dai.node.ColorCamera)
5detectionNetwork = pipeline.create(dai.node.YoloDetectionNetwork)
6# 属性
7camRgb.setPreviewSize(640, 352)
8camRgb.setResolution(dai.ColorCameraProperties.SensorResolution.THE_1080_P)
9camRgb.setInterleaved(False)
10camRgb.setColorOrder(dai.ColorCameraProperties.ColorOrder.BGR)
11camRgb.setFps(40)
12
13# 网络特定设置
14detectionNetwork.setConfidenceThreshold(0.5)
15detectionNetwork.setNumClasses(80)
16detectionNetwork.setCoordinateSize(4)
17detectionNetwork.setIouThreshold(0.5)
18detectionNetwork.setBlobPath(nnPath)
19detectionNetwork.setNumInferenceThreads(2)
20detectionNetwork.input.setBlocking(False)
21
22# 创建处理 TCP 通信的 Script 节点
23script = pipeline.create(dai.node.Script)
24script.setProcessor(dai.ProcessorType.LEON_CSS)
25script.setScript("""
26while True:
27 detections = node.io["detection_in"].get().detections
28 img = node.io["frame_in"].get()
29""")
30# 将输出（RGB 流、NN 输出）链接到 Script 节点
31detectionNetwork.passthrough.link(script.inputs['frame_in'])
32detectionNetwork.out.link(script.inputs['detection_in'])Python
1import socket
2import time
3import threading
4node.warn("Server up")
5server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
6server.bind(("0.0.0.0", 5000)) # 在端口 5000 上创建 TCP 服务器
7server.listen()
8
9while True:
10 conn, client = server.accept()
11 node.warn(f"Connected to client IP: {client}")
12 try:
13 while True:
14 detections = node.io["detection_in"].get().detections # 读取 ImgDetections 消息，仅获取检测结果
15 img = node.io["frame_in"].get() # 读取 ImgFrame 消息
16 node.warn('Received frame + dets')
17 img_data = img.getData()
18 ts = img.getTimestamp()
19
20 det_arr = []
21 for det in detections:
22 det_arr.append(f"{det.label};{(det.confidence*100):.1f};{det.xmin:.4f};{det.ymin:.4f};{det.xmax:.4f};{det.ymax:.4f}")
23 det_str = "|".join(det_arr) # 将检测结果序列化为字符串，用于发送给客户端
24
25 header = f"IMG {ts.total_seconds()} {len(img_data)} {len(det_str)}".ljust(32)
26 node.warn(f'>{header}<')
27 conn.send(bytes(header, encoding='ascii')) # 发送头部
28 if 0 < len(det_arr): # 如果有检测结果，则发送序列化后的检测数据
29 conn.send(bytes(det_str, encoding='ascii'))
30 conn.send(img_data) # 发送实际图像帧
31 except Exception as e:
32 node.warn("Client disconnected")host.py 脚本来连接相机的 TCP 服务器，接收视频流 + 元数据，并可视化显示数据。我们可以使用 host.py 脚本 作为基础，并修改它以接收和可视化检测结果：Python
1import socket
2import re
3import cv2
4import numpy as np
5
6# 输入你自己的 IP！在运行 oak.py 脚本后，终端会打印出 IP 地址
7OAK_IP = "10.12.101.188"
8
9labels = [ "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" ]
10
11def get_frame(socket, size):
12 bytes = socket.recv(4096)
13 while True:
14 read = 4096
15 if size-len(bytes) < read:
16 read = size-len(bytes)
17 bytes += socket.recv(read)
18 if size == len(bytes):
19 return bytes
20
21sock = socket.socket()
22sock.connect((OAK_IP, 5000))
23
24try:
25 COLOR = (127,255,0)
26 while True:
27 header = str(sock.recv(32), encoding="ascii")
28 chunks = re.split(' +', header)
29 if chunks[0] == "IMG":
30 print(f">{header}<")
31 ts = float(chunks[1])
32 imgSize = int(chunks[2])
33 det_len = int(chunks[3])
34
35 if 0 < det_len: # 如果有检测结果，则读取它们
36 det_str = str(sock.recv(det_len), encoding="ascii")
37
38 img = get_frame(sock, imgSize) # 获取图像帧
39 img_planar = np.frombuffer(img, dtype=np.uint8).reshape(3, 352, 640) # 重塑形状（平面格式）
40 img_interleaved = img_planar.transpose(1, 2, 0).copy() # 转换为交错格式（cv2 需要）
41 # 可视化检测结果：
42 if 0 < det_len:
43 dets = det_str.split("|") # 反序列化检测结果
44 for det in dets:
45 det_section = det.split(";")
46 class_id = int(det_section[0])
47 confidence = float(det_section[1])
48 bbox = [ # 从相对坐标转换为绝对坐标
49 int(float(det_section[2]) * img_interleaved.shape[1]),
50 int(float(det_section[3]) * img_interleaved.shape[0]),
51 int(float(det_section[4]) * img_interleaved.shape[1]),
52 int(float(det_section[5]) * img_interleaved.shape[0])
53 ]
54 cv2.putText(img_interleaved, labels[class_id], (bbox[0] + 10, bbox[1] + 20), cv2.FONT_HERSHEY_TRIPLEX, 0.5, COLOR)
55 cv2.putText(img_interleaved, f"{int(confidence)}%", (bbox[0] + 10, bbox[1] + 40), cv2.FONT_HERSHEY_TRIPLEX, 0.5, COLOR)
56 cv2.rectangle(img_interleaved, (bbox[0], bbox[1]), (bbox[2], bbox[3]), COLOR, 2)
57
58 # 显示带有可视化检测结果的帧
59 cv2.imshow("Img", img_interleaved)
60
61 if cv2.waitKey(1) == ord('q'):
62 break
63except Exception as e:
64 print("Error:", e)
65
66sock.close()oak.py 和 host.py）可在此处找到。请注意，对于独立模式，您可能需要烧录静态 IP，这样就不必每次更改代码中的 IP 地址。引导加载程序
烧录管线
Python
1import depthai as dai
2
3pipeline = dai.Pipeline()
4
5# 定义独立管线；添加节点并连接它们
6# cam = pipeline.create(dai.node.ColorCamera)
7# script = pipeline.create(dai.node.Script)
8# ...
9
10# 烧录管线
11(f, bl) = dai.DeviceBootloader.getFirstAvailableDevice()
12bootloader = dai.DeviceBootloader(bl)
13progress = lambda p : print(f'烧录进度: {p*100:.1f}%')
14bootloader.flash(progress, pipeline)DepthAI 应用程序包 (.dap)
Python
1import depthai as dai
2
3pipeline = dai.Pipeline()
4
5# 定义独立管线；添加节点并连接它们
6# cam = pipeline.create(dai.node.ColorCamera)
7# script = pipeline.create(dai.node.Script)
8# ...
9
10# 创建 DepthAI 应用程序包 (.dap)
11(f, bl) = dai.DeviceBootloader.getFirstAvailableDevice()
12bootloader = dai.DeviceBootloader(bl)
13bootloader.saveDepthaiApplicationPackage(pipeline=pipeline, path=<新_dap文件的路径>)清除闪存
Python
1import depthai as dai
2(f, bl) = dai.DeviceBootloader.getFirstAvailableDevice()
3if not f:
4 print('未发现设备，正在退出...')
5 exit(-1)
6
7with dai.DeviceBootloader(bl) as bootloader:
8 bootloader.flashClear()
9 print('成功清除引导加载程序的闪存')