K230使用RTSP實現無線圖傳

K230 使用 RTSP 實現無線圖傳

效果圖

開發環境

開發板:LCKFB-LSPI-K230-1G-CanMV
固件:CanMV_K230_LCKFB_micropython_v1.4-0-g6cce59c_nncase_v2.9.0.img
IDE:CanMV K230
播放器:VLC 播放器

RTSP 介紹

RTSP（Real-Time Streaming Protocol，實時流協議）是一種用于控制多媒體流（如音頻和視頻）傳輸的網絡控制協議。

幾種常用的圖傳協議對比

協議用途延遲場景

RTSP 控制實時流低（200ms~2s）安防攝像頭、工業圖傳

HTTP-FLV / HLS 基于 HTTP 的流媒體中~高（HLS 通常 3s+）網頁直播、點播

WebRTC 端到端實時通信極低（<500ms）視頻通話、在線教育

協議	用途	延遲	場景
RTSP	控制實時流	低（200ms~2s）	安防攝像頭、工業圖傳
HTTP-FLV / HLS	基于 HTTP 的流媒體	中~高（HLS 通常 3s+）	網頁直播、點播
WebRTC	端到端實時通信	極低（<500ms）	視頻通話、在線教育

數據流圖

RTSP 類

class RTSPStreamer:
    """RTSP圖像流傳輸"""

    def __init__(
        self,
        port=2004,
    ):
        self.rtspserver = mm.rtsp_server()  # 實例化 RTSP 服務器
        self.video_type = mm.multi_media_type.media_h265  # 視頻編碼類型
        self.session_name = "VisionCarStream"  # 會話名稱
        self.venc_chn = VENC_CHN_ID_0  # 視頻編碼通道
        self.port = port  # RTSP 服務器端口號
        self.enable_audio = False  # 是否啟用音頻
        self.start_stream = False  # 是否啟動推流線程
        self.runthread_over = False  # 推流線程是否結束

        # 創建編碼器
        self.encoder = Encoder()
        self.encoder.SetOutBufs(self.venc_chn, 8, STREAM_WIDTH, STREAM_HEIGHT)

        # 設置編碼參數，使用H265編碼
        self.chnAttr = ChnAttrStr(
            self.encoder.PAYLOAD_TYPE_H265,
            self.encoder.H265_PROFILE_MAIN,
            STREAM_WIDTH,
            STREAM_HEIGHT,
        )
        self.streamdata = StreamData()

    def __deinit__(self):
        # 停止推流
        self.stop()

        self.encoder.Stop(self.venc_chn)
        self.encoder.Destroy(self.venc_chn)

        self.rtspserver.rtspserver_stop()
        self.rtspserver.rtspserver_deinit()

    def start(self):
        self.encoder.Create(self.venc_chn, self.chnAttr)
        self.encoder.Start(self.venc_chn)

        self.rtspserver.rtspserver_init(self.port)
        self.rtspserver.rtspserver_createsession(
            self.session_name, self.video_type, self.enable_audio
        )
        # 啟動 RTSP 服務器
        self.rtspserver.rtspserver_start()

        # 啟動推流線程
        self.start_stream = True
        _thread.start_new_thread(self._do_rtsp_stream, ())

    def stop(self):
        if not self.start_stream:
            return
        # 等待推流線程退出
        self.start_stream = False
        while not self.runthread_over:
            time.sleep(0.1)
        self.runthread_over = False

    def get_rtsp_url(self):
        return self.rtspserver.rtspserver_getrtspurl(self.session_name)

    def _do_rtsp_stream(self):
        try:
            while self.start_stream:
                os.exitpoint()
                self.encoder.GetStream(self.venc_chn, self.streamdata)
                for pack_idx in range(0, self.streamdata.pack_cnt):
                    stream_data = bytes(
                        uctypes.bytearray_at(
                            self.streamdata.data[pack_idx],
                            self.streamdata.data_size[pack_idx],
                        )
                    )
                    self.rtspserver.rtspserver_sendvideodata(
                        self.session_name,
                        stream_data,
                        self.streamdata.data_size[pack_idx],
                        time.time(),
                    )

                self.encoder.ReleaseStream(self.venc_chn, self.streamdata)

        except BaseException as e:
            print(f"Exception {e}")
        finally:
            self.runthread_over = True
            # 停止 rtsp server
            self.stop()

        self.runthread_over = True

攝像頭類

class Camera:
    def __init__(self):
        """攝像頭初始化"""
        self.sensor = Sensor()
        self.sensor.reset()

        self.sensor.set_hmirror(False)
        self.sensor.set_vflip(False)
        # 通道2：圖傳（YUV420SP）
        self.sensor.set_framesize(
            width=STREAM_WIDTH,
            height=STREAM_HEIGHT,
            chn=CAM_CHN_ID_2,
            alignment=12,
        )
        self.sensor.set_pixformat(Sensor.YUV420SP, chn=CAM_CHN_ID_2)
        """ 攝像頭通道2綁定到視頻編碼器"""
        bind_info_ID2 = self.sensor.bind_info(chn=CAM_CHN_ID_2)

        self.link = MediaManager.link(
            bind_info_ID2["src"],
            (VIDEO_ENCODE_MOD_ID, VENC_DEV_ID, VENC_CHN_ID_0),
        )

    def __deinit__(self):
        self.sensor.stop()
        del self.link

    def run(self):
        self.sensor.run()

    def capture_frame(self, chn=CAM_CHN_ID_0):
        """
        返回一幀圖像數據
        """
        return self.sensor.snapshot(chn)

    # ---------- 攝像頭初始化 ----------
    camera = Camera()

    # ---------- RTSPStreamer ----------
    print("初始化 RTSPStreamer")
    rtsp = RTSPStreamer(RTSP_PORT)

    # ---------- 媒體管理 ----------
    print("媒體初始化")
    MediaManager.init()

    # ---------- 攝像頭采集 ----------
    camera.run()

    # ---------- 啟動 RTSPStreamer ----------
    rtsp.start()
    print("rtsp server start:", rtsp.get_rtsp_url())

完善程序后運行，在輸出窗口可以看到創建的 RTSP 服務地址