独立模式
- 无需主机计算机即可启动应用程序，并能独立连接到不同的计算机/服务器。
- 对任何不稳定性（例如网络问题，相机与主机计算机之间的连接可能中断）更具鲁棒性，因为它会自动重启应用程序。
- 启动速度更快，因为主机计算机无需传输流水线和资源（只需几秒钟）。
支持
- OAK PoE（基于RVC2的）具有板载闪存，并可通过网络协议（例如 HTTP、TCP、UDP、MQTT 等）与外部世界通信。
- OAK USB（基于RVC2的）——不支持，因为它们无法与外部世界通信。
- [已弃用] OAK IoT（基于RVC2的）具有板载内存和集成的 ESP32，可通过 WiFi/蓝牙与外部世界通信。
- OAK-D CM4 和 OAK-D CM4 PoE 集成了 Raspberry Pi Compute Module 4。
- RVC3（RAE Robot）和 RVC4（OAK4）芯片均运行 Linux 操作系统，因此用户可以直接通过 SSH 进入设备，并设置自己的应用程序在启动时运行。
OAK PoE 独立模式
DNS 解析器
独立模式缺少 DNS 解析器，因此您需要使用 IP 地址而不是域名。
转换为独立模式
示例
- 一个部分是管道定义（将刷写到设备上）
- 另一个部分是主机端代码（接收数据、可视化并显示）
oak.py）。我还会移除 XLinkOut 节点，因为它们在独立模式下会被忽略，并创建一个 Script 节点，用于通过网络发送数据。Python
1pipeline = dai.Pipeline()
2
3# Define sources and outputs
4camRgb = pipeline.create(dai.node.ColorCamera)
5detectionNetwork = pipeline.create(dai.node.YoloDetectionNetwork)
6# Properties
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# Network specific settings
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 流、神经网络输出）链接到 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('已成功清除引导加载程序刷写内容')