我想设置一个appsrc来流式传输自定义数据，例如使用python中的gstreamer管道传输任意图像，如numpy矩阵。为此，我使用了推送模式的GstAppSrc。这对于使用autovideosink显示数据很好，但似乎不适用于任何编码器，例如将jpeg写入为.avi或将h264流式传输为rtsp流。
由于完整的应用程序太复杂了，我试着做了一个小例子，在numpy中创建了一个100 x100的RGB矩阵，并使用自定义的GstAppSrc推送它。
我尝试遵循this example（它是用C编写的）并将其解析为Python代码。
只要我不需要对数据进行编码并使用autovideosink直接显示它，管道就可以工作。使用任何类型的编码器都会导致管道阻塞或抛出错误，如：
（python：178667）：GStreamer-Video-CRITICAL**：15：50：56.078：gst_pad_set_caps：Assert'caps！= NULL && gst_caps_is_fixed（caps）'失败
这是没有意义的，因为我设置和固定上限的缓冲区和我的GstAppSrc。
对于设置rtsp流，我遵循以下答案：https://stackoverflow.com/a/73186851/20580911
下面是我的演示程序，用于测试管道：

import time
from dataclasses import dataclass
import numpy as np
import gi
gi.require_version('Gst', '1.0')
gi.require_version('GstVideo', '1.0')
gi.require_version('GstApp', '1.0')
gi.require_version('GLib', '2.0')
from gi.repository import Gst, GstVideo, GstApp, GLib
Gst.init()
@dataclass
class AppSrcSharedData:
    pipeline: Gst.Pipeline
    app_src: GstApp.AppSrc
    source_id: int
    main_loop: GLib.MainLoop
    frame_ctr: int
    color_val: int
def create_color_tile(hue_val):
    red = 1
    green = 1
    blue = 1
    if hue_val <= 120:
        red = 1 - (hue_val / 120)
        green = hue_val / 120
        blue = 0
    elif hue_val <= 240:
        red = 0
        green = 1 - ((hue_val-120) / 120)
        blue = ((hue_val-120) / 120)
    elif hue_val <= 360:
        red = ((hue_val-240) / 120)
        green = 0
        blue = 1 - ((hue_val-240) / 120)
    return np.full((100, 100, 3), (255 * np.asarray([red, green, blue])), dtype=np.uint8)
def _push_data(data: AppSrcSharedData):
    buffer = Gst.Buffer.new_wrapped(create_color_tile(data.color_val).tobytes())
    buffer.add_reference_timestamp_meta(data.app_src.get_caps(), data.app_src.get_current_running_time(), Gst.CLOCK_TIME_NONE)
    buffer.pts = data.app_src.get_current_running_time()
    buffer.dts = Gst.CLOCK_TIME_NONE
    buffer.duration = Gst.CLOCK_TIME_NONE
    buffer.offset = data.frame_ctr
    data.frame_ctr += 1
    ret = data.app_src.push_buffer(buffer)
    data.color_val += 1
    time.sleep(0.03)
    if data.color_val > 360:
        data.app_src.end_of_stream()
        data.pipeline.send_event(Gst.Event.new_eos())
        data.pipeline.set_state(Gst.State.NULL)
        data.main_loop.quit()
        return False
    if ret != Gst.FlowReturn.OK:
        return True
    else:
        return False
def _start_feed(appsrc, length, udata):
    udata.source_id = GLib.idle_add(_push_data, udata)
def _stop_feed(appsrc, udata):
    GLib.source_remove(udata.appsrc_data.source_id)
    udata.appsrc_data.source_id = 0
if __name__ == '__main__':
    pipe_str = "appsrc name=source ! videoconvert ! autovideosink"
    # pipe_str = ("appsrc name=source format=time is_live=True ! videoconvert ! openh264enc ! rtspclientsink location=rtsp://localhost:8554/test")
    # pipe_str = "appsrc name=source format=time is_live=True ! videoconvert ! jpegenc ! avimux ! filesink location=output.avi"
    pipeline = Gst.parse_launch(pipe_str)
    src = None
    for element in pipeline.children:
        if element.name == "source":
            src = element
    # configure appsrc
    src.set_stream_type(GstApp.AppStreamType.STREAM)
    src.set_live(True)
    src.set_do_timestamp(False)
    src.set_format(Gst.Format.TIME)
    # set caps
    video_info = GstVideo.VideoInfo()
    video_info.set_format(GstVideo.VideoFormat.RGB, 100, 100)
    video_caps = video_info.to_caps()
    video_caps.fixate()
    src.set_caps(video_caps)
    main_loop = GLib.MainLoop()
    appsrc_data = AppSrcSharedData(pipeline, src, 0, main_loop, 0, 0)
    # connect callbacks
    src.connect("need-data", _start_feed, appsrc_data)
    src.connect("enough-data", _stop_feed, appsrc_data)
    ret = pipeline.set_state(Gst.State.PLAYING)
    main_loop.run()

字符串
有人能帮我弄清楚我是否缺少了一个参数设置或配置，以使编码正常工作吗？