zfbox/web/API/viedo.py

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
import subprocess
from collections import deque

# 配置日志
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
#     })

def stream_rtsp_to_flv(rtsp_url):
    command = [
        'ffmpeg',
        '-i', rtsp_url,  # 输入 RTSP 流
        '-vcodec', 'libx264',  # 视频编码器
        '-f', 'flv',  # 输出格式为 FLV
        'pipe:1'  # 输出到管道
    ]

    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return process

async def handle_channel_rtf(channel_id,websocket):
    rtsp_url = "rtsp://192.168.3.103/live1"
    process = stream_rtsp_to_flv(rtsp_url)

    try:
        while True:
            data = process.stdout.read(1024)  # 从管道读取 FLV 数据
            if not data:
                break
            await websocket.send(data)  # 通过 WebSocket 发送 FLV 数据
    except asyncio.CancelledError:
        process.terminate()  # 如果 WebSocket 连接关闭，则终止 ffmpeg 进程
    finally:
        process.terminate()

async def handle_channel(channel_id,websocket):
    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

    #视频传输缓冲区
    #frame_buffer = deque(maxlen=10)
    try:
        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 = mMM.verify_list.cm_get_last_frame(channel_id)  # get_last_frame 用了try
            if frame is not None:
                # frame_buffer.append(frame)  #先放入缓冲区
                icount = 0
                await websocket.send(frame)
            else:
                # print("frame is None")
                icount += 1
                if icount > error_max_count:
                    print(f"通道-{channel_id}，长时间未获取图像，休眠一段时间后再获取。")
                    #icount = 0
                    error_message = b"video_error"
                    await websocket.send(error_message)
                    await asyncio.sleep(sleep_time)  # 等待视频重连时间
    except asyncio.CancelledError:
        print(f"WebSocket connection for channel {channel_id} closed by client")
        raise
    except Exception as e:
        print(f"Unexpected error: {e}")
    finally:
        print(f"Cleaning up resources for channel {channel_id}")


@api.websocket('/ws/video_feed/<int:channel_id>')
async def ws_video_feed(channel_id):
    print(f"New connection for channel: {channel_id}")
    # 为每个通道创建独立的协程
    shendtask = asyncio.create_task(handle_channel(channel_id, websocket))
    #shendtask = asyncio.create_task(handle_channel_rtf(channel_id, websocket))
    # 等待协程完成
    await shendtask


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