这篇文章主要介绍了python实现实时视频流播放代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
@action(方法=玫健?细节=True)
def视频(自我,请求,pk=None):
”“”
获取设备实时视频流
请求参数:
:param pk:
返回:
”“”
device_obj=self.get_object ()
如果device_obj #。状态==0:
#返回响应({‘错误’:“设备离线'})
如果不是device_obj.rtsp_address:
返回响应({‘错误’:“缺少rtsp地址'})
cache_id=' _video_stream_ {}“.format (device_obj.hash)
cache_status=缓存。get (cache_id,没有)
如果没有cache_status: #任务初始化,设置初始时间
缓存。集(cache_id time.time(),超时=60)
elif isinstance (cache_status浮动)和time.time ()——cache_status祝辞30:#任务已超时,返回错误信息,一段时间内不再入队
返回响应({‘错误’:“连接数目超过限制,请稍后再试'})
ret=job_queue.enqueue_video (rtsp_address=device_obj。rtsp_address device_hash=device_obj.hash)
logger.info('获取设备% s视频作业状态:% s的,pk, ret._status)
如果ret._status==b 'started”或“开始”:#视频流正常推送中,刷新播放时间,返回视频ID
缓存。集(cache_id,“继续”,超时=30)
返回响应({“视频”:" . join([设置。FFMPEG_VIDEO device_obj.hash])})
elif ret._status==b 'queued”或“排队”:#视频任务等待中
返回响应({“状态”:“等待建立视频连接'})
其他:#建立视频任务失败
返回响应({‘错误’:‘打开视频失败'})
类JobQueue:
”““实时视频播放“”“
def __init__(自我):
自我。video_queue=django_rq.get_queue(视频)#视频推流消息队列
def enqueue_video(自我、rtsp_address device_hash):
”““视频流队列”“”
job_id=' video_ {}“.format (device_hash)
工作=self.video_queue.fetch_job (job_id)
如果不是工作:
工作=self.video_queue.enqueue_call (
func=' utils.ffmpeg.ffmpeg_play ',
args=(rtsp_address device_hash),
timeout=1,
ttl=30 #最多等待30秒
result_ttl=0,
job_id=job_id
）
之前返回工作
# - * -编码:utf - 8 - *
导入子流程
进口线程
导入的时间
导入日志
从django.core。缓存进口缓存
记录器=logging.getLogger (“server.default”)
def ffmpeg_play(流、名称):
玩=True
cache_id=' _video_stream_ {}“.format(名称)
缓存。集(cache_id,“继续”,超时=30)
过程=没有
def上游():
cmd=" ffmpeg我“{}”- c: v h364 - f flv - r 25——“rtmp://127.0.0.1:1935/生活/{}’”。格式(流,名称)
过程=子流程。Popen (cmd, shell=True, stdin=子流程。管,stderr=subprocess.DEVNULL)
试一试:
logger.info(“设备:{}流线程开始:{}”。格式(名称、流))
虽然游戏:
time . sleep (1)
除了例外e:
logger.info(“设备:{}{}流线程错误”。格式(名称、e))
最后:
logger.info(“设备:{}流线程停止”.format(名字))
process.communicate (b 'q”)
用力推=threading.Thread(目标=上游)
thr.start ()
试一试:
而真正的:
玩=缓存。get (cache_id”)
如果玩!=凹绦?
logger.info(停止设备{}视频流的.format(名字))
玩=False
打破
time . sleep (1)
除了例外e:
logger.info(“设备:{}{}打流错误。格式(名称、e))
process.communicate (b 'q”)
logger.info(“等待设备{}视频线程停止.format(名字))
thr.join ()
logger.info(“设备{}视频工作停止”.format(名字))
#实时视频流播放
RQ_QUEUES={
“视频”:{
“USE_REDIS_CACHE”:“视频”,
}
}
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
