边缘计算×视频传输：人形机器人低延迟控制的SDK层解决方案

def init_sdk(self):  
    # 初始化SDK核心模块  
    self.smart_player_sdk_api = get_smart_player_sdk_api_instance()  
    if self.smart_player_sdk_api.Init(0, None) != NTBaseCodeDefine.NT_ERC_OK:  
        raise RuntimeError("SDK初始化失败")  
      
    # 检测硬件解码能力  
    self.is_support_h264_hardware_decoder = (  
        self.smart_player_sdk_api.IsSupportH264HardwareDecoder() == NTBaseCodeDefine.NT_ERC_OK  
    )  
    self.is_support_h265_hardware_decoder = (  
        self.smart_player_sdk_api.IsSupportH265HardwareDecoder() == NTBaseCodeDefine.NT_ERC_OK  
    )  
      
    # 创建播放器实例  
    self.player_handle = NT_HANDLE()  
    if self.smart_player_sdk_api.Open(byref(self.player_handle), None, 0, None) != NTBaseCodeDefine.NT_ERC_OK:  
        raise RuntimeError("播放器创建失败")  
      
    # 设置三重回调机制  
    self.setup_callbacks()  

    # region 回调处理
    def event_callback(self, handle, user_data, event_id, param1, param2, param3, param4, param5, param6):
        """事件回调处理（通过主线程更新UI）"""
        event = NT_SP_E_EVENT_ID(event_id)
        msg = None

        if event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_CONNECTING:
            msg = "链接状态: 链接中"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_CONNECTION_FAILED:
            msg = "链接状态: 链接失败"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_CONNECTED:
            msg = "链接状态: 链接成功"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_DISCONNECTED:
            msg = "链接状态: 链接断开"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_START_BUFFERING:
            msg = "开始缓冲"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_BUFFERING:
            msg = f"缓冲中: {param1}%"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_STOP_BUFFERING:
            msg = "结束缓冲"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_DOWNLOAD_SPEED:
            speed_kbps = str(int(param1 * 8 / 1000)) + "kbps " + str(int(param1 / 1024)) + "KB/s"
            msg = f"下载速度: {speed_kbps}"
        elif event == NT_SP_E_EVENT_ID.NT_SP_E_EVENT_ID_PLAYBACK_REACH_EOS:
            msg = "播放结束"

        if msg:
            # 将消息放入队列, 通过主线程更新状态栏
            self.event_queue.put(msg)

    # 新增一个方法用于更新状态栏
    def update_status_from_queue(self):
        """从队列中取出事件信息并更新到状态栏"""
        while not self.event_queue.empty():
            try:
                message = self.event_queue.get_nowait()
                self.update_status(message)
            except queue.Empty:
                break
                
        # 处理录像队列
        while not self.recorder_queue.empty():
            try:
                status, filename = self.recorder_queue.get_nowait()
                if status == 1:
                    self.update_status(f"新录像文件: {filename}")
                elif status == 2:
                    self.update_status(f"录像文件完成: {filename}")
            except queue.Empty:
                break

        # 处理截图队列
        while not self.capture_queue.empty():
            try:
                result, filename = self.capture_queue.get_nowait()
                if result == NTBaseCodeDefine.NT_ERC_OK:
                    self.update_status(f"截图已保存: {filename}")
                else:
                    self.update_status(f"截图失败: {filename}")
            except queue.Empty:
                break
        # 每隔一段时间检查队列
        self.root.after(100, self.update_status_from_queue)

    def video_size_callback(self, handle, user_data, width, height):
        """视频尺寸变化回调"""
        print(f"视频尺寸变化: {width}*{height}")
        self.root.after(0, self.update_status, f"视频分辨率: {width}*{height}")

   def video_frame_callback(self, handle, user_data, status, frame):
        """视频帧回调（RGB32格式）"""
        if not frame:
            return

        frame_data = frame.contents
        if frame_data.format_ != NT_SP_E_VIDEO_FRAME_FORMAT.NT_SP_E_VIDEO_FRAME_FORMAT_RGB32.value:
            return

        buffer_size = frame_data.stride0_ * frame_data.height_
        byte_array = bytes(ctypes.cast(frame_data.plane0_, ctypes.POINTER(ctypes.c_ubyte * buffer_size)).contents)
        try:
            self.frame_queue.put_nowait((byte_array, frame_data.width_, frame_data.height_, frame_data.stride0_))
        except queue.Full:
            pass

# 处理录像队列
while not self.recorder_queue.empty():
	try:
		status, filename = self.recorder_queue.get_nowait()
		if status == 1:
			self.update_status(f"新录像文件: {filename}")
		elif status == 2:
			self.update_status(f"录像文件完成: {filename}")
	except queue.Empty:
		break

def start_playback(self):  
    # 设置硬件解码（关键性能优化）  
    if self.hardware_decode.get():  
        self.smart_player_sdk_api.SetH264HardwareDecoder(  
            self.player_handle,  
            int(self.is_support_h264_hardware_decoder), 0  
        )  
        self.smart_player_sdk_api.SetH265HardwareDecoder(  
            self.player_handle,  
            int(self.is_support_h265_hardware_decoder), 0  
        )  
      
    # 绑定渲染窗口  
    hwnd = ctypes.c_void_p(self.canvas.winfo_id())  
    self.smart_player_sdk_api.SetRenderWindow(self.player_handle, hwnd)  
      
    # 启动播放引擎  
    if self.smart_player_sdk_api.StartPlay(self.player_handle) != NTBaseCodeDefine.NT_ERC_OK:  
        raise RuntimeError("播放启动失败")  

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("录像已停止")