当前位置：首页 > news >正文

K230使用RTSP实现无线图传

news 2026/5/12 13:15:35

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 = Falsewhile not self.runthread_over:time.sleep(0.1)self.runthread_over = Falsedef 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 serverself.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.linkdef 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 服务地址