zfbox/web/API/viedo.py

import cv2
import asyncio
import time
from . import api
from quart import jsonify, request,websocket
from aiortc import RTCPeerConnection, RTCSessionDescription, VideoStreamTrack,RTCConfiguration
from core.ModelManager import mMM
from core.DBManager import mDBM
from myutils.ConfigManager import myCongif
from fractions import Fraction
import threading
import logging
from quart.helpers import Response

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
#-------------------基于WEBRTC实现拉流
#pcs = set()     #创建一个空的集合，去重复且无序
pcs = {}

'''
--------------基础信息--后续封装在函数里---------------
'''
# 打开摄像头
# def get_camera():
#     cap = cv2.VideoCapture(0)
#     if not cap.isOpened():
#         raise IOError("Cannot open webcam")
#     return cap
#
# cap = get_camera()
# last_image = None
# read_lock = threading.Lock()
# def camera_thread():
#     global cap, last_image
#     while True:
#         try:
#             ret, frame = cap.read()
#             if not ret:
#                 logger.warning("Camera disconnected. Reconnecting...")
#                 cap.release()
#                 cap = get_camera()
#                 continue
#             frame = cv2.resize(frame, (640, 480))  # 降低视频分辨率
#             with read_lock:
#                 last_image = frame
#             time.sleep(1.0/20)
#         except Exception as e:
#             logger.error(f"Error in camera thread: {e}")
#             continue

#threading.Thread(target=camera_thread, daemon=True).start()
'''
---------------------传输--------------------------
'''
# class VideoTransformTrack(VideoStreamTrack):
#     kind = "video"
#     def __init__(self,cid,itype=1):  #0-usb,1-RTSP,2-海康SDK --itype已没有作用
#         super().__init__()
#         self.channel_id = cid
#
#         self.frame_rate = myCongif.get_data("frame_rate")
#         self.time_base = Fraction(1, self.frame_rate)
#         self.frame_count = 0
#         self.start_time = time.time()
#
#     async def recv(self):
#         new_frame = None
#         time.sleep(1.0 / self.frame_rate)
#         new_frame = mMM.verify_list[self.channel_id][4]
#         # while True:
#         #     new_frame = mMM.verify_list[self.channel_id][4]
#         #     #new_frame = av.VideoFrame.from_ndarray(img, format="rgb24")
#         #     if new_frame is not None:
#         #         break
#         #     else:
#         #         time.sleep(1.0 / self.frame_rate)
#         # 设置时间戳和时间基数 -- 根据耗时实时计算帧率
#         # elapsed_time = time.time() - self.start_time
#         # self.frame_count += 1
#         # actual_frame_rate = self.frame_count / elapsed_time
#         # self.time_base = Fraction(1, int(actual_frame_rate))
#         #设置时间戳和帧率
#         self.frame_count += 1
#         if self.frame_count > myCongif.get_data("RESET_INTERVAL"):
#             self.frame_count = 0
#         new_frame.pts = self.frame_count
#         new_frame.time_base = self.time_base
#         #print(f"{self.channel_id} -- Frame pts: {new_frame.pts}, time_base: {new_frame.time_base}")
#
#         # 将检测结果图像转换为帧
#         # new_frame.pts = int(self.cap.get(cv2.CAP_PROP_POS_FRAMES))
#         # new_frame.time_base = self.time_base
#         # end_time = time.time()  # 结束时间
#         # print(f"Processing time: {end_time - start_time} seconds")
#         #print("webRTC recv at:",time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
#         return new_frame

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
#     })

async def generate_frames(stream_id):
    #video_capture = streams[stream_id]
    start_time = time.time()
    icount = 0
    frame_rate_time = 1.0 / myCongif.get_data("frame_rate")
    while True:
        try:
            await asyncio.sleep(frame_rate_time)
            frame = mMM.verify_list[int(stream_id)][4]
            if frame is not None:
                img = frame.to_ndarray(format="bgr24")
                ret, buffer = cv2.imencode('.jpg', img)
                if not ret:
                    continue
                frame = buffer.tobytes()
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n')
            end_time = time.time()  # 结束时间
            icount += 1
            print(f"icount:{icount},Processing time: {end_time - start_time} seconds")
        except asyncio.CancelledError:
            print("Task was cancelled")
            break  # 处理任务取消
        except Exception as e:
            print(f"Error: {e}")
            break  # 在出现异常时退出循环
    print("退出循环")

@api.route('/video_feed/<stream_id>')
async def video_feed(stream_id):
    if int(stream_id) != 2:
        return "None"
    print("执行一次：video_feed")
    return Response(generate_frames(stream_id),mimetype='multipart/x-mixed-replace; boundary=frame')

@api.websocket('/ws/video_feed/<int:channel_id>')
async def ws_video_feed(channel_id):
    while True:
        frame = mMM.verify_list[int(channel_id)][4]
        if frame is not None:
            img = frame.to_ndarray(format="bgr24")
            ret, buffer = cv2.imencode('.jpg', img)
            if not ret:
                continue
            frame = buffer.tobytes()
            await websocket.send(frame)
        await asyncio.sleep(1.0 / myCongif.get_data("frame_rate"))  # Adjust based on frame rate

@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('/close_stream', methods=['POST'])
async def close_stream():
    channel_id = (await request.form)['channel_id']
    reStatus = 0
    reMsg =""
    if channel_id in pcs:
        await pcs[channel_id].close()
        pcs.pop(channel_id, None)
        print(f'Stream {channel_id} closed.')
        #数据库中该通道的关联关系要清除
        strsql = f"update channel set element_id = NULL where ID={channel_id};"
        ret = mDBM.do_sql(strsql)
        if ret == True:
            reStatus = 1
            reMsg = '关闭画面成功！'
        else:
            reStatus = 0
            reMsg = '删除通道与组件关联关系失败，请联系技术支持！'
    else:
        reMsg = "通道编号不在系统内，请检查！"
    return jsonify({'status': reStatus, 'msg': reMsg})