Python下的毫秒级延迟RTSP|RTMP播放器技术探究和AI视觉算法对接

# SmartPlayerPythonDemo.py
# Created by daniusdk.com
# WeChat: xinsheng120
def start_playback(self):
	if not self.player_handle or not self.player_handle.value:
		self.update_status("play handle is None")
		return

	print(f"start_playback")

	self.init_common_sdk_param()

	hwnd = ctypes.c_void_p(self.canvas.winfo_id())
	print(f"Canvas hwnd: 0x{hwnd.value:x}")
	if self.smart_player_sdk_api.SetRenderWindow(self.player_handle, hwnd) != NTBaseCodeDefine.NT_ERC_OK:
		self.update_status("设置渲染窗口失败")
		return

	# 设置硬解码
	if self.hardware_decode.get():
		self.smart_player_sdk_api.SetH264HardwareDecoder(self.player_handle, 1 if self.is_support_h264_hardware_decoder else 0, 0)
		self.smart_player_sdk_api.SetH265HardwareDecoder(self.player_handle, 1 if self.is_support_h265_hardware_decoder else 0, 0)

	self.smart_player_sdk_api.SetAudioVolume(self.player_handle, int(self.volume_scale.get()))

	if self.smart_player_sdk_api.StartPlay(self.player_handle) != NTBaseCodeDefine.NT_ERC_OK:
		self.update_status("开始播放失败")
		return

	if self.is_enable_frame_callback:
		# 启动帧处理线程
		self.stop_event.clear()
		self.frame_thread = threading.Thread(target=self.process_frames, daemon=True)
		self.frame_thread.start()

	self.is_playing = True
	self.play_btn.config(text="停止")
	self.update_status("正在播放...")

def stop_playback(self):
	if not self.player_handle or not self.player_handle.value:
		return

	self.smart_player_sdk_api.StopPlay(self.player_handle)
	self.is_playing = False
	self.play_btn.config(text="播放")
	self.update_status("已停止")

# 设置硬解码
if self.hardware_decode.get():
	self.smart_player_sdk_api.SetH264HardwareDecoder(self.player_handle, 1 if self.is_support_h264_hardware_decoder else 0, 0)
	self.smart_player_sdk_api.SetH265HardwareDecoder(self.player_handle, 1 if self.is_support_h265_hardware_decoder else 0, 0)

def toggle_record(self):
	if not self.is_recording:
		self.start_recording()
	else:
		self.stop_recording()

def start_recording(self):
	if not self.player_handle or not self.player_handle.value:
		self.update_status("play handle is None")
		return

	# 确保记录配置已初始化
	if self.record_config is None:
		messagebox.showinfo("提示", "请先配置录像参数")
		self.show_record_config()  # 打开配置对话框
		return

	print(f"start_recording")

	self.init_common_sdk_param()

	# 设置录像参数
	ruler = NT_SP_RecorderFileNameRuler()
	ruler.type_ = 0
	if self.record_config is not None:
		ruler.file_name_prefix_ = self.record_config["file_prefix"].encode()
		ruler.append_date_ = 1 if self.record_config["is_append_date"] else 0
		ruler.append_time_ = 1 if self.record_config["is_append_time"] else 0

	self.smart_player_sdk_api.SetRecorderDirectoryW(
		self.player_handle,
		ctypes.c_wchar_p(self.record_config["dir_path"])
	)
	self.smart_player_sdk_api.SetRecorderFileNameRuler(self.player_handle, byref(ruler))
	self.smart_player_sdk_api.SetRecorderVideo(self.player_handle,
		c_int(1 if self.record_config["is_record_video"] else 0))
	self.smart_player_sdk_api.SetRecorderAudio(self.player_handle,
		c_int(1 if self.record_config["is_record_audio"] else 0))

	if self.smart_player_sdk_api.StartRecorder(self.player_handle) == NTBaseCodeDefine.NT_ERC_OK:
		self.is_recording = True
		self.record_btn.config(text="停止录像")
		self.update_status("录像中...")
	else:
		self.update_status("启动录像失败")

def stop_recording(self):
	if not self.player_handle or not self.player_handle.value:
		return

	print(f"stop_recording Player handle: 0x{self.player_handle.value:x}")

	self.smart_player_sdk_api.StopRecorder(self.player_handle)
	self.is_recording = False
	self.record_btn.config(text="录像")
	self.update_status("录像已停止")

def capture_image(self):
	if not self.player_handle or not self.player_handle.value:
		return

	filename = f"capture_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"

	# 定义回调函数
	def capture_callback(handle, user_data, result, file_name):
		status = "成功" if result == NTBaseCodeDefine.NT_ERC_OK else "失败"
		self.capture_queue.put((result, file_name.decode('utf-8') if file_name else filename))

	# 创建回调对象并保持引用
	self.capture_image_cb = CAPTURE_IMAGE_CALLBACK(capture_callback)

	# 调用SDK接口
	ret = self.smart_player_sdk_api.CaptureImage(
		self.player_handle,
		filename.encode('utf-8'),
		None,
		self.capture_image_cb)

	if ret != NTBaseCodeDefine.NT_ERC_OK:
		self.update_status("截图请求发送失败")

def video_frame_callback(self, handle, user_data, status, frame):
    if not frame:
        return
 
    frame_data = frame.contents
    if frame_data.format_ == NT_SP_E_VIDEO_FRAME_FORMAT.NT_SP_E_VIDEO_FRAME_FORMAT_YUV420P.value:
        # 处理YUV420P数据
        yuv_data = bytes(ctypes.cast(frame_data.plane0_, ctypes.POINTER(ctypes.c_ubyte * frame_data.size_)).contents)
        self.process_yuv_frame(yuv_data, frame_data.width_, frame_data.height_)
    elif frame_data.format_ == NT_SP_E_VIDEO_FRAME_FORMAT.NT_SP_E_VIDEO_FRAME_FORMAT_RGB24.value:
        # 处理RGB24数据
        rgb_data = bytes(ctypes.cast(frame_data.plane0_, ctypes.POINTER(ctypes.c_ubyte * frame_data.size_)).contents)
        self.process_rgb_frame(rgb_data, frame_data.width_, frame_data.height_)

self.frame_cb = VIDEO_FRAME_CALLBACK(self.video_frame_callback)
self.smart_player_sdk_api.SetVideoFrameCallBack(self.player_handle,
                           NT_SP_E_VIDEO_FRAME_FORMAT.NT_SP_E_VIDEO_FRAME_FORMAT_YUV420P.value,
                           None, self.frame_cb)

def process_yuv_frame(self, yuv_data, width, height):
    # 将YUV数据转换为numpy数组
    yuv_array = np.frombuffer(yuv_data, dtype=np.uint8).reshape((height * 3 // 2, width))
 
    # 调用视觉算法
    result = self.visual_algorithm.process_yuv(yuv_array)
 
    # 在主线程更新UI
    self.root.after(0, self.update_visual_result, result) 

def process_rgb_frame(self, rgb_data, width, height):
    # 将RGB数据转换为numpy数组
    rgb_array = np.frombuffer(rgb_data, dtype=np.uint8).reshape((height, width, 3))
 
    # 调用视觉算法
    result = self.visual_algorithm.process_rgb(rgb_array)
 
    # 在主线程更新UI
    self.root.after(0, self.update_visual_result, result)

class VisualAlgorithm:
    def process_yuv(self, yuv_array):
        # 在这里实现视觉算法对YUV数据的处理
        pass
 
    def process_rgb(self, rgb_array):
        # 在这里实现视觉算法对RGB数据的处理
        pass

def __init__(self, root):
    # 其他初始化代码
    self.visual_algorithm = VisualAlgorithm()
 

def update_visual_result(self, result):
    # 在这里更新UI以显示视觉算法的结果
    pass