Json transfer

.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/json_param_transfer     && python3 json_param_transfer.py)
           # test hmp
           - (cd bmf/hml/tests/data                  && ./gen.sh $(pwd)/../../../../output/files)
           - (cd bmf/hml/tests                       && pytest)
@@ -148,6 +149,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/json_param_transfer     && python3 json_param_transfer.py)
           # test hmp
           - (cd bmf/hml/tests/data                  && ./gen.sh $(pwd)/../../../../output/files)
           - (cd bmf/hml/tests                       && pytest)

bmf/test/json_param_transfer/CMakeLists.txt

@@ -0,0 +1,30 @@
+cmake_minimum_required(VERSION 3.12)
+project(cpp_module)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_FLAGS "-Wno-deprecated-declarations")
+set(CMAKE_INSTALL_PREFIX ${PROJECT_SOURCE_DIR})
+
+SET(MODULE_SRCS ${PROJECT_SOURCE_DIR}/cpp_module.cpp)
+
+add_library(cpp_module SHARED ${MODULE_SRCS})
+
+target_include_directories(${PROJECT_NAME} PRIVATE ${PROJECT_SOURCE_DIR}/../../../output/bmf/include
+                           ${CUDA_INCLUDE_DIR}
+                           /usr/local/include)
+target_link_directories(${PROJECT_NAME} PRIVATE ${PROJECT_SOURCE_DIR}/../../../output/bmf/lib)
+target_link_libraries(cpp_module
+    PRIVATE
+    bmf_module_sdk
+)
+
+if(WIN32)
+    set_target_properties(copy_module PROPERTIES
+        RUNTIME_OUTPUT_DIRECTORY ${BMF_ASSEMBLE_ROOT}/bmf/lib
+        RUNTIME_OUTPUT_DIRECTORY_DEBUG ${BMF_ASSEMBLE_ROOT}/bmf/lib
+        RUNTIME_OUTPUT_DIRECTORY_RELEASE ${BMF_ASSEMBLE_ROOT}/bmf/lib
+    )
+endif()
+
+#set(CMAKE_INSTALL_PREFIX ${PROJECT_SOURCE_DIR})
+#install(TARGETS cpp_module)

bmf/test/json_param_transfer/cpp_module.cpp

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2023 Babit Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "cpp_module.h"
+
+int cpp_module::process(Task &task) {
+    PacketQueueMap &input_queue_map = task.get_inputs();
+    PacketQueueMap &output_queue_map = task.get_outputs();
+    PacketQueueMap::iterator it;
+
+    for (it = input_queue_map.begin(); it != input_queue_map.end(); it++) {
+        // input stream label
+        int label = it->first;
+        Packet pkt;
+        // process all packets in one input queue
+        while (task.pop_packet_from_input_queue(label, pkt)) {
+            // if packet is eof, set module done
+            if (pkt.timestamp() == BMF_EOF) {
+                task.set_timestamp(DONE);
+                auto out_it = output_queue_map.begin();
+                if (out_it != output_queue_map.end()) {
+                    task.fill_output_packet(out_it->first, Packet::generate_eof_packet());
+                }
+                return 0;
+            }
+
+            // parse the input json param if there's JsonParam
+            if (pkt.is<JsonParam>()) {
+                auto jp = pkt.get<JsonParam>();
+                std::string jstring = jp.dump();
+                BMFLOG(BMF_INFO) << "Cpp module parsed json: " << jstring;     
+            }
+
+            JsonParam transfer_json;
+            transfer_json.parse(fmt::format("{{\"input_id\": {}, \"frame_number\": {}}}",
+                                           label, frame_number++));
+            auto output_pkt = Packet(transfer_json);
+            auto out_it = output_queue_map.begin();
+            if (out_it != output_queue_map.end()) {
+                task.fill_output_packet(out_it->first, output_pkt);
+            }
+        }
+    }
+    return 0;
+}
+REGISTER_MODULE_CLASS(cpp_module)

bmf/test/json_param_transfer/cpp_module.h

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2023 Babit Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef BMF_CPP_MODULE_H
+#define BMF_CPP_MODULE_H
+
+#include <bmf/sdk/bmf.h>
+#include <bmf/sdk/packet.h>
+
+USE_BMF_SDK_NS
+
+class cpp_module : public Module {
+  public:
+    cpp_module(int node_id, JsonParam option) : Module(node_id, option) {}
+
+    ~cpp_module() {}
+
+    int frame_number = 0;
+
+    virtual int process(Task &task);
+};
+
+#endif

bmf/test/json_param_transfer/json_param_transfer.py

@@ -0,0 +1,32 @@
+import sys
+sys.path.append("../../")
+import bmf
+import time
+
+def json_param_transfer():
+    graph = bmf.graph({'dump_graph': 1})
+    video = graph.decode({
+        "input_path": "../../files/big_bunny_10s_30fps.mp4",
+        "video_codec": "copy"
+    })["video"]
+    fps = 1
+    module_path = './libcpp_module.so'
+    module_entry = 'cpp_module:cpp_module'
+    # the graph is triggered by a decoder and then connect cpp module -> python module -> cpp module
+    # to show the json can be transferred and parsed between cpp and python module easily
+    cpp_json = video.c_module("cpp_module",
+                                      module_path,
+                                      module_entry,
+                                      )
+    python_paser = cpp_json.module('python_module')
+    cpp_paser = python_paser.c_module("cpp_module",
+                                      module_path,
+                                      module_entry,
+                                      ).run()
+
+
+if __name__ == "__main__":
+    start_time = time.time()
+    json_param_transfer()
+    end_time = time.time()
+    print("process time:", end_time - start_time)

bmf/test/json_param_transfer/python_module.py

@@ -0,0 +1,33 @@
+from bmf import Module, Log, LogLevel, InputType, ProcessResult, Packet, Timestamp, scale_av_pts, av_time_base, \
+    BmfCallBackType, VideoFrame, AudioFrame
+
+
+class python_module(Module):
+
+    def __init__(self, node, option=None):
+        self.node_ = node
+        self.option_ = option
+        pass
+
+    def process(self, task):
+        for (input_id, input_packets) in task.get_inputs().items():
+            while not input_packets.empty():
+                pkt = input_packets.get()
+                # process EOS
+                if pkt.timestamp == Timestamp.EOF:
+                    Log.log_node(LogLevel.DEBUG, task.get_node(),
+                                 "Receive EOF")
+                    for output_id in task.get_outputs().keys():
+                        output_packets = task.get_outputs()[output_id]
+                        output_packets.put(Packet.generate_eof_packet())
+                    task.timestamp = Timestamp.DONE
+                    return ProcessResult.OK
+
+                if pkt.is_(dict):
+                    json = pkt.get(dict)
+                    print("Python module parsed json:", json)
+                for output_id in task.get_outputs().keys():
+                    output_packets = task.get_outputs()[output_id]
+                    output_packets.put(pkt)
+
+        return ProcessResult.OK