Pass non-image data between modules using Packet

[xiaoweiw-nv]

When building the controlnet demo, I am trying to build a pipeline that looks like this:

image decoder ---> controlnet inference module ---> image encoder ^ | prompt reader ---------+

The prompt is read from files and passed to the controlnet inference module through bmf.Packet. The type of the prompt is Python dict, and bmf/python/py_module_sdk.cpp shows that bmf.Packet support all python types. But when I run the pipeline, I get the following error:

[2023-10-11 02:09:40.995] [error] node id:2 catch exception: BMF(0.0.8) /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/engine/c_engine/src/node.cpp:352: error: (-5:Bad argument) [Node_2_c_ffmpeg_filter] Process result != 0.
 in function 'process_node'

[2023-10-11 02:09:40.996] [error] node id:2 Process node failed, will exit.
[2023-10-11 02:09:40.996] [info] node 2 got exception, close directly
[2023-10-11 02:09:40.996] [info] node id:2 process eof, add node to scheduler
[2023-10-11 02:09:40.996] [info] schedule queue 0 start to join thread
[ipp1-2035:1225 :0:1315] Caught signal 11 (Segmentation fault: address not mapped to object at address 0x1e0)
==== backtrace (tid:   1315) ====
 0 0x0000000000042520 __sigaction()  ???:0
 1 0x000000000015856d CFFFilter::init_filtergraph()  /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/c_modules/src/ffmpeg_filter.cpp:242
 2 0x0000000000159b9c CFFFilter::process_filter_graph()  /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/c_modules/src/ffmpeg_filter.cpp:393
 3 0x000000000015ae0e CFFFilter::process()  /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/c_modules/src/ffmpeg_filter.cpp:578
 4 0x0000000000372a9e bmf_engine::Node::process_node()  /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/engine/c_engine/src/node.cpp:348
 5 0x00000000003a5a7a bmf_engine::SchedulerQueue::exec()  /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/engine/c_engine/src/scheduler_queue.cpp:153
 6 0x00000000003a5678 bmf_engine::SchedulerQueue::exec_loop()  /home/scratch.xiaoweiw_sw/bytedance/babitmf/bmf/engine/c_engine/src/scheduler_queue.cpp:111
 7 0x00000000003a8b1e std::__invoke_impl<int, int (bmf_engine::SchedulerQueue::*)(), bmf_engine::SchedulerQueue*>()  /usr/include/c++/11/bits/invoke.h:74
 8 0x00000000003a8a72 std::__invoke<int (bmf_engine::SchedulerQueue::*)(), bmf_engine::SchedulerQueue*>()  /usr/include/c++/11/bits/invoke.h:96
 9 0x00000000003a89d3 std::thread::_Invoker<std::tuple<int (bmf_engine::SchedulerQueue::*)(), bmf_engine::SchedulerQueue*> >::_M_invoke<0ul, 1ul>()  /usr/include/c++/11/bits/std_thread.h:259
10 0x00000000003a898a std::thread::_Invoker<std::tuple<int (bmf_engine::SchedulerQueue::*)(), bmf_engine::SchedulerQueue*> >::operator()()  /usr/include/c++/11/bits/std_thread.h:266
11 0x00000000003a896a std::thread::_State_impl<std::thread::_Invoker<std::tuple<int (bmf_engine::SchedulerQueue::*)(), bmf_engine::SchedulerQueue*> > >::_M_run()  /usr/include/c++/11/bits/std_thread.h:211
12 0x00000000000dc253 std::error_code::default_error_condition()  ???:0
13 0x0000000000094b43 pthread_condattr_setpshared()  ???:0
14 0x0000000000126a00 __xmknodat()  ???:0
=================================
Segmentation fault (core dumped)

I haven't used ffmpeg filter module in the graph, tt seems that bmf will insert ffmpeg filter modules in the graph. Test code as follows.

test_controlnet.py:

import sys

sys.path.append("../../")
import bmf

sys.path.pop()

def test():
    input_video_path = "./ControlNet/test_imgs/bird.png"
    input_prompt_path = "./prompt.txt"
    output_path = "./output.jpg"

    graph = bmf.graph()

    video = graph.decode({'input_path': input_video_path})
    prompt = graph.module('text_module', {'path': input_prompt_path})
    concat = bmf.concat(video['video'], prompt)
    concat.module('controlnet_module', {}).run()

if __name__ == '__main__':
    test()

text_module.py:

import sys
import random
from typing import List, Optional
import pdb

from bmf import *
import bmf.hml.hmp as mp

class text_module(Module):
    def __init__(self, node, option=None):
        self.node_ = node
        self.eof_received_ = False
        self.prompt_path = './prompt.txt'
        if 'path' in option.keys():
            self.prompt_path = option['path']

    def process(self, task):
        pdb.set_trace()
        output_queue = task.get_outputs()[0]

        if self.eof_received_:
            output_queue.put(Packet.generate_eof_packet())
            Log.log_node(LogLevel.DEBUG, self.node_, 'output text stream', 'done')
            task.set_timestamp(Timestamp.DONE)
            return ProcessResult.OK

        prompt_dict = dict()
        with open(self.prompt_path) as f:
            for line in f:
                pk, pt = line.partition(":")[::2]
                prompt_dict[pk] = pt

        out_pkt = Packet(prompt_dict)
        out_pkt.timestamp = 0
        output_queue.put(out_pkt)
        self.eof_received_ = True

        return ProcessResult.OK

def register_inpaint_module_info(info):
    info.module_description = "Text file IO module"