主机节点
HostNode 和 ThreadedHostNode 类实现，这些类提供了在流水线内创建自定义主机端逻辑的框架。DepthAI 带有几个预定义的主机节点，例如 BasaltVIO、RTABMapSLAM 以及 Record / Replay。这些节点提供了强大的开箱即用功能。然而，主机节点最令人兴奋的一个方面是开发者能够创建自己的自定义节点。这些自定义节点可以设计用于捕获帧、操作数据、解码神经网络输出等。此外，开发者可以将自定义主机节点打包为 Python 包，并通过 pip 分发。这使得在不同的项目中以及社区内共享和复用自定义节点变得更加容易。例如，像 geaxgx 的手部追踪器 这样的自定义节点可以打包并发布到 PyPI，从而允许其他人通过简单的 pip install 轻松安装并集成到他们的 DepthAI 流水线中。在宿主机上运行
- 在 RVC2 设备上（OAK-D、OAK-1） - 宿主机是通过 USB 连接到 DepthAI 设备的设备。这意味着主机节点将在连接到 DepthAI 设备的设备（通常是 PC）上运行。
- 在 RVC4 设备上（OAK4-S、OAK4-D） - 主机侧取决于设备运行的模式。在外设模式下，宿主机是通过 USB 连接到 DepthAI 设备的设备 - 与 RVC2 设备相同。在独立模式下，OAK4 设备本身是宿主机（运行 Linux），主机节点在设备的 CPU 上运行（见下图）。
主机节点的类型
- 线程化主机节点（Threaded Host Node） - 标准线程化主机节点
- 主机节点（Host Node） - 具有额外机制用于在宿主机上处理数据及同步的线程化主机节点
线程化主机节点（Threaded Host Node）
ThreadedHostNode 是需要多线程的主机节点的基类。该类允许节点在其自己的专用线程中异步运行逻辑，非常适合不应阻塞流水线主执行流程的操作。模板代码
ThreadedHostNode 派生一个新类，并实现以下方法：__init__：构造方法，用于初始化节点。在这里可以创建输入和输出队列。run()方法是节点的主执行循环，在其中实现自定义逻辑。onStart()和onStop()方法分别在节点启动和停止时被调用。这些方法是可选的，可用于执行任何初始化或清理任务。
代码示例
来自 主机摄像头 示例的简单线程化主机节点示例。一旦定义了节点，就可以使用
Python
1class HostCamera(dai.node.ThreadedHostNode):
2 def __init__(self):
3 super().__init__() # 调用基类构造方法
4 self.output = self.createOutput() # 创建一个输出队列 - 这将发送 ImgFrame 消息
5
6 def run(self): # 节点的主执行循环
7 cap = cv2.VideoCapture(0) # 为主机摄像头创建一个 VideoCapture 对象
8 if not cap.isOpened():
9 p.stop()
10 raise RuntimeError("错误：无法打开主机摄像头")
11 while self.isRunning(): # 循环直到节点停止
12 ret, frame = cap.read() # 从摄像头读取帧
13 if not ret:
14 break
15 # 创建一个 ImgFrame 消息并设置其数据、宽度、高度和类型
16 imgFrame = dai.ImgFrame()
17 imgFrame.setData(frame)
18 imgFrame.setWidth(frame.shape[1])
19 imgFrame.setHeight(frame.shape[0])
20 imgFrame.setType(dai.ImgFrame.Type.BGR888i)
21 # 将消息发送到输出队列
22 self.output.send(imgFrame)
23 # 等待下一帧
24 time.sleep(0.1)pipeline.create() 将其添加到流水线中。Python
1with dai.Pipeline() as p:
2 hostCamera = p.create(HostCamera) # 在流水线内创建自定义主机摄像头节点的实例
3 camQueue = hostCamera.output.createOutputQueue() # 创建一个输出队列以接收帧
4
5 p.start() # 启动流水线，这也会启动主机摄像头，隐式运行 `onStart()` 以及 `run()` 方法
6 while p.isRunning():
7 image : dai.ImgFrame = camQueue.get() # 从输出队列获取帧
8 cv2.imshow("HostCamera", image.getCvFrame())
9 key = cv2.waitKey(1)
10 if key == ord('q'):
11 p.stop() # 停止流水线和主机摄像头，隐式运行 `onStop()` 方法
12 break将多个主机节点连接在一起
代码示例
此处显示的代码是 线程化主机节点 示例的简化版本。定义自定义节点后，可以将它们添加到管道中并进行连接：
Python
1class TestSink(dai.node.ThreadedHostNode): # 用于接收数据的自定义线程化主机节点
2 def __init__(self):
3 super().__init__()
4 self.input = self.createInput() # 只需要一个输入队列，因为此节点接收数据
5
6 def onStart(self):
7 print("Hello, this is", __class__.__name__) # 节点启动时打印一条消息
8
9 def run(self):
10 while self.isRunning():
11 buffer = self.input.get() # 从输入队列获取一个缓冲区
12 print("The sink node received a buffer!")
13
14class TestSource(dai.node.ThreadedHostNode): # 用于发送数据的自定义线程化主机节点
15 def __init__(self):
16 super().__init__()
17 self.output = self.createOutput() # 只需要一个输出队列，因为此节点发送数据
18
19 def run(self):
20 while self.isRunning():
21 buffer = dai.Buffer() # 创建一个缓冲区
22 print("The source node is sending a buffer!")
23 self.output.send(buffer) # 将缓冲区发送到输出队列
24 time.sleep(1)Python
1with dai.Pipeline() as p:
2 source = TestSource() # 创建源节点的实例
3 sink = TestSink() # 创建汇节点的实例
4 source.output.link(sink.input) # 将源节点的输出连接到汇节点的输入
5 p.start()
6 while p.isRunning():
7 time.sleep(1)
8 print("Pipeline is running...")主机节点
HostNode 是一个基于 ThreadedHostNode 构建的、更专门的类。它提供了额外的机制用于处理主机上的数据和同步。该类旨在处理更复杂的场景，其中主机端的数据处理需要与 DepthAI 设备管道紧密集成，例如同步多个数据流或显示处理后的数据。模板代码
HostNode 派生一个新类，并实现以下方法：process方法在输入队列收到消息时调用。这是实现自定义逻辑的地方。onStart和onStop方法均可用（因为HostNode继承自ThreadedHostNode），可用于执行任何初始化或清理任务。
代码示例
来自 主机显示 的简单主机节点示例。定义节点后，可以使用
Python
1class HostDisplay(dai.node.HostNode):
2 def build(self, frameOutput: dai.Node.Output):
3 self.link_args(frameOutput) # 必须与 `process` 方法的输入匹配
4
5 # 这将所有处理任务发送到管道，由 `pipeline.runTasks()` 或隐式地由 `pipeline.run()` 方法执行。
6 # 这是必要的，因为 GUI 窗口需要在主线程中更新，而 `process` 方法默认在单独的线程中调用。
7 self.sendProcessingToPipeline(True)
8 return self
9
10 def onStart(self) -> None: # 可选方法
11 print("HostDisplay started")
12
13 def process(self, message: dai.ImgFrame):
14 cv2.imshow("HostDisplay", message.getCvFrame())
15 key = cv2.waitKey(1)
16 if key == ord('q'):
17 print("Detected 'q' - stopping the pipeline...")
18 self.stopPipeline()pipeline.create() 将其添加到管道中。Python
1p = dai.Pipeline()
2with p:
3 camera = p.create(dai.node.Camera).build()
4 hostDisplay = p.create(HostDisplay).build(camera.requestOutput((300, 300)))
5
6 p.run() # 将阻塞直到管道被其他方式停止（此处为显示节点）同步
HostNode 提供了对通过输入发送的消息的隐式同步。这意味着如果有多个消息发送到输入队列，它们将根据时间戳进行同步。我们修改了上面的示例以演示此功能：代码示例
Python
1class HostDisplay(dai.node.HostNode):
2
3 def build(self, *args):
4 rgb_frame, mono_frame = args # 从管道解包输入帧。此处我们期望两个帧：RGB 和单色
5 self.link_args(rgb_frame, mono_frame) # 必须与 `process` 方法的输入匹配
6
7 # 这将所有处理任务发送到管道，由 `pipeline.runTasks()` 或隐式地由 `pipeline.run()` 方法执行。
8 # 这是必要的，因为 GUI 窗口需要在主线程中更新，而 `process` 方法默认在单独的线程中调用。
9 self.sendProcessingToPipeline(True)
10 return self
11
12 def process(self, rgb_frame, mono_frame):
13 # 显示来自输入节点的每个帧
14 cv2.imshow('rgb', rgb_frame.getCvFrame())
15 cv2.imshow('mono', mono_frame.getCvFrame())
16
17 key = cv2.waitKey(1)
18 if key == ord('q'):
19 print("Detected 'q' - stopping the pipeline...")
20 self.stopPipeline()
21
22p = dai.Pipeline()
23with p:
24 rgb_camera = p.create(dai.node.Camera).build(dai.CameraBoardSocket.CAM_A) # 为 RGB 相机创建相机节点
25 mono_camera = p.create(dai.node.Camera).build(dai.CameraBoardSocket.CAM_B) # 为左单色相机创建相机节点
26
27 # 创建自定义主机显示节点的实例，并将其连接到两个相机节点
28 display = p.create(HostDisplay).build(rgb_camera.requestOutput((500, 500)), mono_camera.requestOutput((500, 500)))
29
30 p.run() # 将阻塞直到管道被其他方式停止（此处为显示节点）附加方法
runSyncingOnDevice()- 该方法可用于在设备而非主机上运行process方法。默认情况下，process方法在主机上执行。sendProcessingToPipeline()- 该方法可用于将处理任务发送到管道。默认情况下，process方法在单独的线程中执行。当处理涉及 GUI（如使用 OpenCV 显示帧）时，此为必选项，因为 GUI 操作必须在主线程中完成。
需要帮助？
请前往 OAKChina 官网 获取技术支持或解答您的任何疑问。