RKVision开发日志(十一)

一、Tee测试

又又又简单封装了一下…
每次新建一条支路都要重写tee，queue, valve 太麻烦了，封装成了一个_add_branch

def _add_branch(self, downstream_bin: Gst.Element, drop: bool = True):
    """
    添加分支
    @param name: 分支名称
    @param downstream_bin: 下游处理单元
    @param drop: 阀门初始状态, True: 丢弃; False: 放行

    @ return: 分支的阀门
    """
    name = downstream_bin.get_name()
    q = Gst.ElementFactory.make("queue", f"q_{name}")
    v = Gst.ElementFactory.make("valve", f"v_{name}")
    if not all([q, v, downstream_bin]):
        raise RuntimeError(f"Failed to create branch {name}")

    v.set_property("drop", drop)

    self.pipeline.add(q)
    self.pipeline.add(v)
    self.pipeline.add(downstream_bin)

    tee_pad = self.tee.get_request_pad("src_%u")
    if not tee_pad:
        raise RuntimeError(f"Failed to request tee pad for {name}")
    if tee_pad.link(q.get_static_pad("sink")) != Gst.PadLinkReturn.OK:
        raise RuntimeError(f"tee -> q_{name} link failed")

    if not q.link(v):
        raise RuntimeError(f"q_{name} -> v_{name} link failed")
    if not v.link(downstream_bin):
        raise RuntimeError(f"v_{name} -> {name}_bin link failed")

    self.branches[name] = {"valve": v, "bin": downstream_bin}

主管道中就非常简单的两句话了：

        self.display_bin = DisplayBin(name = "displaybin", config=self.config, gst_config=self.gst_config)
        self._add_branch(self.display_bin, drop=False)

主管道中依旧定义好了源和分流器，新的管道从这里直接接入就可以了，非常方便

二、control测试

为了实现外部进程动态的开关，我写了一个控制接口：

def control_branch(self, name: str, enabled: bool):
    """
    控制分支的阀门状态
    @param name: 分支名称
    @param enabled: True: 放行; False: 丢弃
    """
    def _do():
        # 关
        if not enabled:
            self.branches[name]["valve"].set_property("drop", True)
            self.branches[name]["bin"].set_state(Gst.State.NULL)
            return False

        # 开
        self.branches[name]["bin"].set_state(Gst.State.PLAYING)
        self.branches[name]["valve"].set_property("drop", False)
        return False
        
    GLib.idle_add(_do)
    logger.info(f"Branch '{name}' set to {'ENABLED' if enabled else 'DISABLED'}")

我在主管道中，维护了一个字典：

self.branches = {}  # name -> {"valve": v, "bin": downstream_bin}

当外部控制阀门开关的时候，只需知道需要开关支路的名字，如：

pipeline.control_branch("displaybin", False)

就可以实现开关了。需要注意的是，这里给的参数是enabled，gst中的valve组件这个参数是drop，True的时候代表扔掉（不放行），False的时候代表不扔（放行），逻辑上过于反人类，封装control的时候取了一下反。
吐槽一下，gst的接口名称设计的过于反人类，前面的src_pad，sink_pad也很令人费解

三、嵌入主程序

gst这里自然是要嵌入到相机进程中的，最后绑定到web接口上控制。在前几天我就一直在想这个问题，control进程作为主进程，为了不空跑，最好的方案是用异步函数去监听前端发来的请求，但是这不就和fastapi重合了吗？fastapi后端也是异步的等待触发，那我为什么要单独写一个control进程呢？之前可能考虑qt的交互问题，问题是qt也可以去调用request命令去做http请求呀

既然如此，那就尝试重构一下主进程吧，为了做依赖注入，还是要写服务层…

写到这里感觉很多东西都是工程化的必然选择，之前vibe coding的时候都是看ai写，然后自己理解，感觉很多地方没什么用，现在自己写起来，自己曾经删掉的东西最后又要慢慢加回来，经常写一会要推到重来一次，反反复复的重构代码的感觉很好又很不好

为了不和之前的control进程冲突，我打算还是先写一个backend，里面启动一个fastapi服务器，其中backend.py负责一些start，当然这部分可以放到main中实现，这个后面再慢慢学习慢慢写，这样也不用再去监听退出事件了