import asyncio from . import api from quart import jsonify, request,websocket from core.ModelManager import mMM from core.DBManager import mDBM from myutils.ConfigManager import myCongif import logging import time # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) #-------------------基于WEBRTC实现拉流 #pcs = set() #创建一个空的集合，去重复且无序 pcs = {} ''' ---------------------传输-------------------------- ''' async def get_stats(peer_connection): stats = await peer_connection.getStats() for report in stats.values(): if report.type == 'outbound-rtp': print(f"RTT: {report.roundTripTime} seconds") print(f"Packets Sent: {report.packetsSent}") print(f"Bytes Sent: {report.bytesSent}") # @api.route('/offer', methods=['POST']) # async def offer(): # #接收客户端的连接请求 # params = await request.json # channel_id = params["cid"] #需要传参过来 # itype = params["mytype"] # element_id = params["element_id"] # reStatus = 0 # reMsg = "" # ret = False # if itype == 1: # #添加通道和页面控件ID的关系 # strsql = f"select * from channel where element_id = {element_id}" # data = mDBM.do_select(strsql,1) # if data: # reStatus = 0 # reMsg = '该控件已经关联某一视频通道，请确认数据是否正确！' # return jsonify({'status': reStatus, 'msg': reMsg}) # strsql = f"update channel set element_id = {element_id} where ID = {channel_id};" # ret = mDBM.do_sql(strsql) # if ret == True: # reStatus = 1 # reMsg = '播放画面成功，请稍等！' # else: # reStatus = 0 # reMsg = '播放画面失败，请联系技术支持！' # #提取客户端发来的SDP，生成服务器端的SDP # offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"]) # # # 配置 STUN 服务器 # #ice_servers = [{"urls": []}]# 禁用 STUN 服务器 # # pc = RTCPeerConnection(configuration={"iceServers": ice_servers}) # config = RTCConfiguration(iceServers=[]) # pc = RTCPeerConnection(configuration=config) # # t = threading.Thread(target=get_stats,args=(pc,)) # # t.start() # # #pc = RTCPeerConnection() # 实例化一个rtcp对象 # pcs[channel_id] = pc # 集合中添加一个对象，若已存在则不添 # # @pc.on("datachannel") # def on_datachannel(channel): # @channel.on("message") # def on_message(message): # print(f'Message received: {message}') # # 处理接收到的数据 # if message == 'pause': # # 执行暂停操作 # pass # elif message == 'resume': # # 执行继续操作 # pass # # #监听RTC连接状态 # @pc.on("iconnectionstatechange") #当ice连接状态发生变化时 # async def iconnectionstatechange(): # if pc.iceConnectionState == "failed": # await pc.close() # pcs.pop(channel_id, None) #移除对象 # # # 添加视频轨道 # video_track = VideoTransformTrack(channel_id) # pc.addTrack(video_track) # # # @pc.on('track') --- stream.getTracks().forEach(track => pc.addTrack(track, stream)); 猜测是这里触发的添加了摄像头通道 # # def on_track(track): # # if track.kind == 'video': # # local_video = VideoTransformTrack(track) # # pc.addTrack(local_video) # # # 记录客户端 SDP # await pc.setRemoteDescription(offer) # # 生成本地 SDP # answer = await pc.createAnswer() # # 记录本地 SDP # await pc.setLocalDescription(answer) # # print("返回sdp") # return jsonify({ # "sdp": pc.localDescription.sdp, # "type": pc.localDescription.type, # "status":reStatus, #itype =1 的时候才有意义 # "msg":reMsg # }) @api.websocket('/ws/video_feed/') async def ws_video_feed(channel_id): print(f"New connection for channel: {channel_id}") channel_data = mMM.verify_list.get_channel(channel_id) try: if channel_data: verify_rate = int(myCongif.get_data("verify_rate")) error_max_count = verify_rate * int(myCongif.get_data("video_error_count")) #视频帧捕获失败触发提示的上限 sleep_time = int(myCongif.get_data("cap_sleep_time")) frame_interval = 1.0 / verify_rate #用于帧率控制 last_frame_time = time.time() # 初始化个读帧时间 icount = 0 while True: #这里的多线程并发，还需要验证检查 # 帧率控制帧率 ---输出暂时不控制帧率，有就传输 current_time = time.time() elapsed_time = current_time - last_frame_time if elapsed_time < frame_interval: await asyncio.sleep(frame_interval - elapsed_time) # 若小于间隔时间则休眠 last_frame_time = time.time() #执行视频传输 frame = channel_data.get_last_frame() if frame is not None: #img = frame.to_ndarray(format="bgr24") # ret, buffer = cv2.imencode('.jpg', frame) # if not ret: # continue # frame = buffer.tobytes() icount = 0 await websocket.send(frame) else: #print("frame is None") icount += 1 if icount > error_max_count: print("视频源无图像") icount = 0 error_message = b"video_error" await websocket.send(error_message) await asyncio.sleep(sleep_time) #等待视频重连时间 #await asyncio.sleep(0.02) else: print("没有通道数据！") error_message = b"client_error" await websocket.send(error_message) await asyncio.sleep(0.1) #等0.1秒前端处理时间 except Exception as e: print(f"WebSocket connection for channel {channel_id} closed: {e}") finally: print(f"Cleaning up resources for channel {channel_id}") @api.route('/shutdown', methods=['POST']) async def shutdown():#这是全关 --需要修改 coros = [pc.close() for pc in pcs] await asyncio.gather(*coros) pcs.clear() return 'Server shutting down...' @api.route('/viewlist', methods=['GET']) async def viewlist():#视频列表 count = request.args.get('count') channel_list = [] element_list = [] name_list = [] if count: strsql = (f"select t1.ID,t1.element_id,t1.channel_name,t2.area_name from channel t1 left join " f"area t2 on t1.area_id =t2.id where element_id between 0 and {count};") datas = mDBM.do_select(strsql) for data in datas: channel_list.append(data[0]) element_list.append(data[1]) name_list.append(f"{data[3]}--{data[2]}") return jsonify({'clist': channel_list, 'elist': element_list,'nlist':name_list}) @api.route('/start_stream', methods=['POST']) async def start_stream(): #开启视频通道，把视频通道编号和元素编号进行关联 json_data = await request.get_json() channel_id = json_data.get('channel_id') element_id = json_data.get('element_id') reStatus = 0 reMsg = "" strsql = f"select element_id from channel where ID = {channel_id};" data = mDBM.do_select(strsql,1) if data: if not data[0] or data[0] == '': strsql = f"update channel set element_id = '{element_id}' where ID={channel_id};" ret = mDBM.do_sql(strsql) if ret == True: reStatus = 1 reMsg = '播放视频配置成功！' else: reMsg = '播放视频配置失败，请联系技术支持！' else: index = int(data[0]) +1 reMsg = f"该视频通道已在:Video Stream {index}播放，请先关闭" return jsonify({'status': reStatus, 'msg': reMsg}) @api.route('/close_stream', methods=['POST']) async def close_stream(): #关闭视频通道 json_data = await request.get_json() element_id = json_data.get('element_id') reStatus = 0 reMsg ="" #数据库中该通道的关联关系要清除 strsql = f"update channel set element_id = '' where element_id={element_id};" ret = mDBM.do_sql(strsql) if ret == True: reStatus = 1 reMsg = '关闭画面成功！' else: reMsg = '删除通道与组件关联关系失败，请联系技术支持！' # ?关闭视频播放--业务逻辑-待确认后端是否需要执行 return jsonify({'status': reStatus, 'msg': reMsg})