# 线程主机节点

本指南介绍了使用三个自定义节点实现自定义 DepthAI 管道的实现：TestSource、TestPassthrough 和 TestSink。这些节点演示了 DepthAI 管道中的基本数据流，其中缓冲区由源节点生成，通过透传节点传递，最后由汇节点接收。

## 概述

管道由三个自定义节点组成：

 * TestSource：按固定间隔生成并发送缓冲区。
 * TestPassthrough：从源节点接收缓冲区并转发给汇节点。
 * TestSink：从透传节点接收缓冲区并进行处理。

这些节点演示了 DepthAI 中线程主机节点的基本结构，可用于在管道内实现自定义逻辑。

## 演示输出

```bash
Hello, this is passthrough
Hello, this is passthroughSubnodes_subnode1
Hello, this is passthroughSubnodes_subnode2
Hello, this is sink1
Hello, this is sink2
source node is sending a buffer!
The passthrough node received a buffer!
The passthrough node received a buffer!
The passthrough node received a buffer!
sink2 node received a buffer!
sink1 node received a buffer!
Pipeline is running...
source node is sending a buffer!
The passthrough node received a buffer!
The passthrough node received a buffer!
sink1 node received a buffer!
The passthrough node received a buffer!
sink2 node received a buffer!
...
```

## 源代码

#### Python

```python
import depthai as dai
import time

class TestPassthrough(dai.node.ThreadedHostNode):
    def __init__(self, name: str):
        super().__init__()
        self.name = name
        self.input = self.createInput()
        self.output = self.createOutput()

        # Possible API 1:
        self.input.setPossibleDatatypes([dai.Node.DatatypeHierarchy(dai.DatatypeEnum.ImgFrame, True)])
        self.output.setPossibleDatatypes([dai.Node.DatatypeHierarchy(dai.DatatypeEnum.ImgFrame, True)])

        # Possible API 2:
        self.input.setPossibleDatatypes([
            (dai.DatatypeEnum.ImgFrame, True),
            (dai.DatatypeEnum.Buffer, True)
        ])
        self.output.setPossibleDatatypes([
            (dai.DatatypeEnum.ImgFrame, True),
            (dai.DatatypeEnum.Buffer, True)
        ])

    def onStart(self):
        print("Hello, this is", self.name)

    def onStop(self):
        print("Goodbye from", self.name)

    def run(self):
        while self.mainLoop():
            buffer = self.input.get()
            print("The passthrough node received a buffer!")
            self.output.send(buffer)

class TestSink(dai.node.ThreadedHostNode):
    def __init__(self, name: str):
        super().__init__()
        self.input = self.createInput()

        self.name = name

    def onStart(self):
        print("Hello, this is", self.name)

    def run(self):
        while self.mainLoop():
            buffer = self.input.get()
            del buffer
            print(f"{self.name} node received a buffer!")

class TestSource(dai.node.ThreadedHostNode):
    def __init__(self, name: str):
        super().__init__()
        self.name = name
        self.output = self.createOutput()

    def run(self):
        while self.mainLoop():
            buffer = dai.Buffer()
            print(f"{self.name} node is sending a buffer!")
            self.output.send(buffer)
            time.sleep(1)

class TestPassthroughSubnodes(dai.node.ThreadedHostNode):
    def __init__(self, name: str):
        super().__init__()
        self.passthrough1 = self.createSubnode(TestPassthrough, name + "_subnode1")
        self.passthrough2 = self.createSubnode(TestPassthrough, name + "_subnode2")
        self.input = self.passthrough1.input
        self.output = self.passthrough2.output

        # Link the two subnodes together
        self.passthrough1.output.link(self.passthrough2.input)

    def run(self):
        while self.mainLoop():
            buffer = self.input.get()
            self.output.send(buffer)

with dai.Pipeline(False) as p:
    """
    Create the following pipeline:
    source -> passthrough -> sink1
           |
           -> passthroughSubnodes -> sink2
    """

    # Create nodes
    source = p.create(TestSource, "source")
    passthrough = p.create(TestPassthrough, "passthrough")
    passthroughSubnodes = p.create(TestPassthroughSubnodes, "passthroughSubnodes")
    sink1 = p.create(TestSink, "sink1")
    sink2 = p.create(TestSink, "sink2")

    # Link nodes
    source.output.link(passthrough.input)
    source.output.link(passthroughSubnodes.input)
    passthrough.output.link(sink1.input)
    passthroughSubnodes.output.link(sink2.input)

    p.start()
    while p.isRunning():
        time.sleep(1)
        print("Pipeline is running...")
```

#### C++

```cpp
#include <chrono>
#include <iostream>
#include <thread>

#include "depthai/depthai.hpp"

class TestPassthrough : public dai::node::CustomThreadedNode<TestPassthrough> {
   public:
    Input input = dai::Node::Input{*this, {}};
    Output output = dai::Node::Output{*this, {}};

    void run() override {
        while(mainLoop()) {
            auto buffer = input.get<dai::Buffer>();
            if(buffer) {
                std::cout << "The passthrough node received a buffer!" << std::endl;
                output.send(buffer);
            }
        }
    }
};

class TestSink : public dai::node::CustomThreadedNode<TestSink> {
   public:
    Input input = dai::Node::Input{*this, {}};

    void run() override {
        while(mainLoop()) {
            auto buffer = input.get<dai::Buffer>();
            if(buffer) {
                std::cout << "The sink node received a buffer!" << std::endl;
            }
        }
    }
};

class TestSource : public dai::node::CustomThreadedNode<TestSource> {
   public:
    Output output = dai::Node::Output{*this, {}};

    void run() override {
        while(mainLoop()) {
            auto buffer = std::make_shared<dai::Buffer>();
            std::cout << "The source node is sending a buffer!" << std::endl;
            output.send(buffer);
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }
};

int main() {
    dai::Pipeline pipeline(false);

    auto source = pipeline.create<TestSource>();
    auto passthrough = pipeline.create<TestPassthrough>();
    auto sink = pipeline.create<TestSink>();

    source->output.link(passthrough->input);
    passthrough->output.link(sink->input);

    pipeline.start();

    while(true) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        std::cout << "Pipeline is running..." << std::endl;
    }

    return 0;
}
```

### 需要帮助？

请前往 [OAKChina 官网](https://www.oakchina.cn/) 获取技术支持或解答您的任何疑问。