add merge filters ci test

.codebase/pipelines/ci.yaml

@@ -66,6 +66,7 @@ jobs:
           - (cd output/test/push_data_into_graph    && python3 test_push_data.py)
           - (cd output/test/complex_edit_case       && python3 test_complex_case.py)
           - (cd output/test/complex_edit_case       && python3 test_compare_with_edit.py)
+          - (cd output/test/test_merge_filters      && python3 test.py)
           # test hmp
           - (cd bmf/hml/tests/data                  && ./gen.sh $(pwd)/../../../../output/files)
           - (cd bmf/hml/tests                       && pytest)

bmf/test/test_merge_filters/switch_module.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from bmf import (
+    Module,
+    Log,
+    LogLevel,
+    ProcessResult,
+    Packet,
+    Timestamp,
+    VideoFrame,
+)
+import json
+import time
+
+
+class switch_module(Module):
+    main_stream = 0
+    switch_stream = 1
+    def __init__(self, node, option=None):
+        self.input_packets = []
+        # None means wait
+        # 1 means passthrough
+        # 0 means not passthrough
+        self.switch = None
+        self.process_done = False
+        self.main_eof_skip = False
+        self.version = self.get_version()
+
+    def process_pkt(self, task, pkt):
+        if self.switch is None:
+            self.input_packets.append(pkt)
+            return
+        output_queue = task.get_outputs().get(0, None)
+        if self.switch == 0:
+            self.process_done = True
+            if output_queue is not None:
+                output_queue.put(Packet.generate_eof_packet())
+            task.timestamp = Timestamp.DONE
+
+        if self.switch == 1:
+            for packet in self.input_packets:
+                if output_queue is not None:
+                    output_queue.put(packet)
+            self.input_packets = []
+            if output_queue is not None:
+                output_queue.put(pkt)
+
+    def process_main_eof(self, task):
+        Log.log(LogLevel.INFO, "process main eof, switch: ", self.switch)
+        if self.switch is None:
+            return True
+        output_queue = task.get_outputs().get(0, None)
+
+        if self.switch == 1:
+            for packet in self.input_packets:
+                if output_queue is not None:
+                    output_queue.put(packet)
+            self.input_packets = []
+
+        self.process_done = True
+        if output_queue is not None:
+            output_queue.put(Packet.generate_eof_packet())
+        task.timestamp = Timestamp.DONE
+        return False
+
+
+    def process(self, task):
+        output_queue = task.get_outputs().get(0, None)
+        main_queue = task.get_inputs().get(0, None)
+        switch_queue = task.get_inputs().get(1, None)
+
+        if self.process_done:
+            return ProcessResult.OK
+
+        #save main_queue
+        while not main_queue.empty():
+            pkt = main_queue.get()
+            if pkt.timestamp == Timestamp.EOF:
+                self.main_eof_skip = self.process_main_eof(task)
+            else:
+                self.process_pkt(task,pkt)
+
+        while not switch_queue.empty() and self.switch is None:
+            pkt = switch_queue.get()
+            if pkt.timestamp == Timestamp.EOF:
+                Log.log(LogLevel.INFO, "get eof, set switch 0")
+                self.switch = 0
+            else:
+                Log.log(LogLevel.INFO, "get pkt, set switch 1")
+                self.switch = 1
+
+        if self.main_eof_skip and self.switch is not None:
+            Log.log(LogLevel.INFO, "process main eof last time")
+            self.process_main_eof(task)
+
+        return ProcessResult.OK
+
+    def get_version(self):
+        return "v1"

bmf/test/test_merge_filters/test.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import bmf
+
+
+def test():
+    graph = bmf.graph({'dump_graph':1})
+    video = graph.decode({"input_path": "../../files/big_bunny_10s_30fps.mp4"})[
+        "video"
+    ]
+    vo = video.fps(24).pass_through()
+
+    v = vo.scale(848,480).setsar(1).pass_through()
+
+    s = bmf.module(
+        [vo, v],
+        "switch_module",
+        input_manager="immediate",
+    )
+
+    v1 = s.scale(480,320)
+    v2 = v.scale(640, 360)
+    v3 = v.scale(360, 240)
+
+    bmf.encode(
+        v1,
+        None,
+        {
+            "output_path": "v1.mp4",
+            "video_params": {
+                "vsync": "vfr",
+                "codec": "h264",
+                "preset": "medium",
+            },
+        },
+    )
+
+    bmf.encode(
+        v2,
+        None,
+        {
+            "output_path": "v2.mp4",
+            "video_params": {
+                "vsync": "vfr",
+                "codec": "h264",
+                "preset": "medium",
+            },
+        },
+    )
+
+    bmf.encode(
+        v3,
+        None,
+        {
+            "output_path": "v3.mp4",
+            "video_params": {
+                "vsync": "vfr",
+                "codec": "h264",
+                "preset": "medium",
+            },
+        },
+    )
+    graph.run()
+
+
+if __name__ == "__main__":
+    test()