【python小脚本】摄像头rtsp流转hls m3u8 格式web端播放

写在前面

• 工作需要，简单整理
• 实际上这种方式延迟太高了，后来前端直接接的海康的本地解码插件，走的 websockt
• 博文内容为 摄像头 rtsp 实时流转 hls m3u8 的一个 Python 脚本
• 理解不足小伙伴帮忙指正 :),生活加油

99%的焦虑都来自于虚度时间和没有好好做事，所以唯一的解决办法就是行动起来，认真做完事情，战胜焦虑，战胜那些心里空荡荡的时刻，而不是选择逃避。不要站在原地想象困难，行动永远是改变现状的最佳方式

摄像头 rtsp 实时流转 hls m3u8 格式 web 端播放

方案介绍：

• 在服务器上安装并配置 FFmpeg,从 RTSP 摄像头获取实时视频流
• 使用 FFmpeg并将其转码为 HLS 格式,生成 m3u8 播放列表和 TS 分段文件。
• 将生成的 HLS 文件托管到 Nginx 服务器的 Web 根目录下,并在 Nginx 配置文件中添加相应的配置,以正确处理 HLS 文件的 MIME 类型和跨域访问等。
• 在 Web 页面中使用 HTML5 的<video>标签或 HLS.js库来播放 Nginx 托管的 HLS 视频流。

这里使用的 Nginx 是有 rtmp 模块的 nginx https://github.com/dreammaker97/nginx-rtmp-win32-dev

rtsp 常见的两个转码方式：

rtsp 转 rtmp ffmpeg rtsp 2 rtmp

ffmpeg.exe -i rtsp://admin:hik12345@10.112.205.103:554/Streaming/Channels/101?transportmode=multicast -acodec aac -strict experimental -ar 44100 -ac 2 -b:a 96k -r 25 -b:v 500k   -c:v libx264 -c:a copy -f flv rtmp://127.0.0.1:1935/live/demo

ffmpeg rtsp 2 hls rtsp 转 hls

ffmpeg -f rtsp -rtsp_transport tcp -i rtsp://admin:hik12345@10.112.205.103:554/Streaming/Channels/101?transportmode=multicast -acodec aac -strict experimental -ar 44100 -ac 2 -b:a 96k -r 25 -b:v 500k -s 640*480  -c:v libx264 -c:a copy -cpu-used 0  -threads 1  -f hls -hls_time 2.0 -hls_list_size 3 -hls_wrap 50 X:\nginx-rtmp-win32-dev\nginx-rtmp-win32-dev\html\hls\test777.m3u8

名词解释:

RTSP 协议: RTSP (Real-Time Streaming Protocol) 是一种用于实时音视频流传输的网络协议,通常用于监控摄像头等设备的实时视频流传输。

HLS 格式: HLS (HTTP Live Streaming) 是苹果公司开发的自适应比特率流式传输协议,可以将视频流转码为 HTTP 可访问的 TS 分段文件和 m3u8 播放列表。HLS 具有良好的跨平台和兼容性。

FFmpeg : FFmpeg 是一个强大的多媒体框架,可以用于音视频的编码、解码、转码等操作。它可以将 RTSP 流转码为 HLS 格式。

Nginx: Nginx 是一款高性能的 Web 服务器,也可作为反向代理服务器使用。它可以托管 HLS 格式的 m3u8 播放列表和 TS 分段文件,为 Web 端提供 HLS 流的访问。

HLS.js: HLS.js 是一款 JavaScript 库,可以在不支持 HLS 原生播放的浏览器上实现 HLS 流的播放。

编码

通过 fastapi 启了一个Web服务，前端获取某个摄像头的流的时候，会启动一个 ffmpeg 子进程来处理流，同时会给前端返回一个 Nginx 推流的 地址

逻辑比较简单,涉及到进程处理，项目启动会自动启动 nginx，当取流时会自动启动 ffmpeg，nginx 和 ffmpge 都为 当前 Python 服务的子进程，当web 服务死掉，对应子进程全部死掉。

项目地址： https://github.com/LIRUILONGS/rtsp2hls-M3U8.git

requirements.txt

APScheduler==3.10.4
fastapi==0.111.1
ping3==4.0.8
pyinstaller==6.9.0
pytest==8.3.1
traitlets==5.14.3
uvicorn==0.30.3  

配置文件

# windows 环境配置文件，目录需要修改为 `/` 分割符
ngxin:
  # 启动的推流服务IP，取流的时候使用的IP地址
  nginx_ip : 127.0.0.1 
  # 启动 ng 端口，取流时使用的端口
  nginx_port: 8080
  # 启动的推流服务前缀 
  nginx_fix : /hls/
  # nginx 程序路径，这里不加 `nginx.exe` 实际执行需要跳转到这个目录
  nginx_path: "X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/"
  # nginx 配置文件位置
  nginx_config_path: "X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/conf/nginx.conf"

fastapi:
  # 服务端口
  port: 8991
  # 流存放nginx目录
  hls_dir: "X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/html/hls/"
  # ffmpeg 执行路径
  ffmpeg_dir:  'W:/ffmpeg-20200831-4a11a6f-win64-static/bin/ffmpeg.exe'
  # 最大取流时间
  max_stream_threads : 60
  # 扫描时间
  max_scan_time : 3*60
  # 最大转码数
  max_code_ff_size : 6
  # ffmpeg 转化执行的路径 
  comm: "{ffmpeg_dir} -f rtsp -rtsp_transport tcp -i rtsp://admin:hik12345@{ip}:554/Streaming/Channels/101?transportmode=multicast -acodec aac -strict experimental -ar 44100 -ac 2 -b:a 96k -r 25 -b:v 300k -s {width}*{height} -live 1  -c:v libx264 -c:a copy -cpu-used 0  -threads 1  -f hls -hls_time 0.5 -hls_list_size 1 -hls_wrap 100 {hls_dir}{ip}-{uuid_v}.m3u8"
  

核心代码

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@File    :   main.py
@Time    :   2024/07/24 17:20:21
@Author  :   Li Ruilong
@Version :   1.0
@Contact :   liruilonger@gmail.com
@Desc    :   rtmp 转码 到 hls 
"""

........................................


@app.get("/sc_view/get_video_stream")
async def get_video_stream(
    ip: str = Query("192.168.2.25", description="IP地址"),  # 设置默认值为 1
    width: int = Query(320, description=" 流宽度"),  # 设置默认值为 10
    height: int = Query(170, description=" 流高度"),  # 设置默认值为 'name'
):
    """
    @Time    :   2024/07/23 11:04:31
    @Author  :   liruilonger@gmail.com
    @Version :   1.0
    @Desc    :    ffmag 解码推流
    """

    if width is None or ip is None or height is None:
        raise HTTPException(status_code=400, detail="参数不能为空")
    import time
    # 获取前端传递的参数
    uuid_v = str(uuid.uuid4())
    if validate_ip_address(ip) is False:
        return {"message": "no validate_ip_address", "code": 600}

    if ping_test(ip) is False:
        return {"message": "ping no pong", "code": 600}
    with lock:
        if ip in chanle:
            return chanle[ip]
        if len(chanle) >= max_code_ff_size:
            return {"status": 400, "message": f"超过最大取流数：{max_code_ff_size}"}
        hls_dir = fastapi['hls_dir']
        ffmpeg_dir = fastapi["ffmpeg_dir"]
        print(vars())
        command = comm.format_map(vars())
        try:
            print(command.strip())
            process = subprocess.Popen(
                command,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL
            )
            if process.pid:
                t_d = {
                    "pid": process.pid,
                    "v_url": f'http://{locad_id}:{locad_port}{locad_fix}{ip}-{uuid_v}.m3u8',
                    "ip": ip
                }
                print(t_d)
                print("==============================摄像头数据更新完成...，重新确认子进程是否运行")
                pss = get_process_by_name("ffmpeg.exe", process.pid)
                print("创建的进程为:", pss)
                if len(pss) > 0:
                    chanle[ip] = t_d
                    print(f"返回取流路径为：{t_d}")
                    return t_d
                else:
                    return {"status": 400, "message": "IP 取流失败!,请重新尝试"}
        except subprocess.CalledProcessError as e:
            return {"error": f"Error running ffmpeg: {e}"}


@app.get("/sc_view/stop_video_stream")
async def stop_video_stream(pid: int = Query(2000, description="进程ID")):
    """
    @Time    :   2024/07/24 14:10:43
    @Author  :   liruilonger@gmail.com
    @Version :   1.0
    @Desc    :   结束推流
    """

    if pid is None:
        raise HTTPException(status_code=400, detail="参数不能为空")

    pss = get_process_by_name("ffmpeg.exe", pid)
    print(pss)
    if len(pss) == 0:
        print("未获取到进程信息", pid)
        return {
            "status": 200,
            "message": "未获取到进程信息"
        }
    print("获取到进程信息：", pss)
    try:
        # 发送 SIGTERM 信号以关闭进程
        os.kill(int(pid), signal.SIGTERM)
        chanle.pop(pid)
        print(f"Process {pid} has been terminated.{str(pss)}")
        return {"status": 200, "message": "关闭成功!"}
    except OSError as e:
        # 调用 kill 命令杀掉
        pss[0].kill()
        print(f"Error terminating process {pid}: {e}")
        return {"status": 200, "message": "关闭成功!"}


@app.get("/sc_view/all_stop_video_stream")
async def all_stop_video_stream():
    """
    @Time    :   2024/07/24 14:10:43
    @Author  :   liruilonger@gmail.com
    @Version :   1.0
    @Desc    :   批量结束推流
    """
    pss = get_process_by_name("ffmpeg.exe")
    print(pss)
    if len(pss) == 0:
        return {
            "status": 200,
            "message": "转码全部结束"
        }
    print("获取到进程信息：", pss)
    process_list = []
    for p in pss:
        process_list.append({
            "pid": p.info['pid'],
            "name":  p.info['name'],
            "status": p.status(),
            "started": datetime.datetime.fromtimestamp(p.info['create_time']),
            "memory_info": str(p.memory_info().rss / (1024 * 1024)) + " MB",
            "cpu_percent": str(p.cpu_percent()) + " %",
            "cmdline": p.cmdline()
        })
        try:
            # 发送 SIGTERM 信号以关闭进程
            os.kill(int(p.info['pid']), signal.SIGTERM)
            #chanle.pop(p.info['pid'])
            ips =  [ k for k,v in chanle.items() if v.pid == p.info['pid']  ]
            if len(ips) >0:
               chanle.pop(ips[0]) 
            print(f"Process {p.info['pid']} has been terminated.{str(pss)}")
        except OSError as e:
            # 调用 kill 命令杀掉
            pss[0].kill()
            print(f"Error terminating process {p.info['pid']}: {e}")
    return {"status": 200, "message": "关闭成功!", "close_list": process_list}


@app.get("/sc_view/get_video_stream_process_list")
async def get_video_stream_process_list():
    """
    @Time    :   2024/07/24 15:46:38
    @Author  :   liruilonger@gmail.com
    @Version :   1.0
    @Desc    :   返回当前在采集的流处理进程信息
    """

    pss = get_process_by_name("ffmpeg.exe")
    process_list = []
    for p in pss:
        ip_file = str(p.info['cmdline'][-1]).split("/")[-1]
        process_list.append({
            "pid": p.info['pid'],
            "name":  p.info['name'],
            "status": p.status(),
            "started": datetime.datetime.fromtimestamp(p.info['create_time']),
            "memory_info": str(p.memory_info().rss / (1024 * 1024)) + " MB",
            "cpu_percent": str(p.cpu_percent()) + " %",
            "cmdline": p.cmdline(),
            "v_url":  f'http://{locad_id}:{locad_port}{locad_fix}{ip_file}',
        })
    return {"message": "当前在采集的流信息", "process_list": process_list}

nginx 启动相关

# 启动 Nginx
def start_nginx():
    """
    @Time    :   2024/07/24 21:13:25
    @Author  :   liruilonger@gmail.com
    @Version :   1.0
    @Desc    :   启动 nginx
    """
    try:
        os.chdir(nginx_path)
        print("当前执行路径：" + str(nginx_path + "nginx.exe" + " -c " + nginx_config_path))
        subprocess.Popen([nginx_path + "nginx.exe", "-c", nginx_config_path], stdout=subprocess.DEVNULL,
                         stderr=subprocess.DEVNULL)
        print("\n===================  Nginx has been started successfully.\n")
    except subprocess.CalledProcessError as e:
        print(f"Failed to start Nginx: {e}")
    finally:
        os.chdir(os.path.dirname(__file__))  # 切换回用户主目录

# 停止 Nginx


def stop_nginx():
    """
    @Time    :   2024/07/24 21:13:41
    @Author  :   liruilonger@gmail.com
    @Version :   1.0
    @Desc    :   关闭 nginx
    """
    try:
        os.chdir(nginx_path)
        print("当前执行路径：" + str(nginx_path + "nginx.exe" + " -s " + "stop"))
        subprocess.Popen([nginx_path + "nginx.exe", "-s", "stop"], stdout=subprocess.DEVNULL,
                         stderr=subprocess.DEVNULL)
        print("\n============  Nginx has been stopped successfully.\n")
    except subprocess.CalledProcessError as e:
        print(f"Failed to stop Nginx: {e}")
    finally:
        os.chdir(os.path.dirname(__file__))  # 切换回用户主目录

进程相关方法

def get_process_by_name(process_name, pid=None):
    """
    @Time    :   2024/07/24 14:21:31
    @Author  :   liruilonger@gmail.com
    @Version :   1.1
    @Desc    :   获取指定进程名和进程 ID 的进程列表

    Args:
        process_name (str): 进程名称
        pid (int, optional): 进程 ID，默认为 None 表示不筛选 ID

    Returns:
        list: 包含指定进程名和进程 ID 的进程对象的列表
    """

    processes = []
    attrs = ['pid', 'memory_percent', 'name', 'cmdline', 'cpu_times',
             'create_time', 'memory_info', 'status', 'nice', 'username']
    for proc in psutil.process_iter(attrs):
        # print(proc.info['name'])
        try:
            if proc.info['name'] == process_name:
                if pid is None or proc.info['pid'] == pid:
                    processes.append(proc)
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            pass
    print("Process==================end")
    return processes


def get_process_by_IP(process_name, ip=None):
    """
    @Time    :   2024/07/24 14:21:31
    @Author  :   liruilonger@gmail.com
    @Version :   1.1
    @Desc    :   获取指定进程名和 IP 的进程列表

    Args:
        process_name (str): 进程名称
        pid (int, optional): IP，默认为 None 表示不筛选 IP

    Returns:
        list: 包含指定进程名和进程 IP 的进程对象的列表
    """
    attrs = ['pid', 'memory_percent', 'name', 'cmdline', 'cpu_times',
             'create_time', 'memory_info', 'status', 'nice', 'username']
    press = []
    for proc in psutil.process_iter(attrs):
        try:
            if proc.info['name'] == process_name:

                if ip is None or any(ip in s for s in proc.info['cmdline']):
                    ip_file = str(proc.info['cmdline'][-1]).split("/")[-1]
                    press.append({
                        "pid": proc.info['pid'],
                        "v_url":  f'http://{locad_id}:{locad_port}{locad_fix}{ip_file}',
                        "ip": ip
                    })
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            pass
    return press

打包

pyinstaller --add-data "config.yaml;."  --add-data "templates/*;templates"   main.py   

exe 路径

rtsp2hls2M3U8\dist\main

配置文件路径

rtsp2hls2M3U8\dist\main\_internal

部署测试

2024-08-13 15:57:03,404 - win32.py[line:58] - DEBUG: Looking up time zone info from registry
2024-08-13 15:57:03,410 - yaml_util.py[line:62] - INFO: 加载配置数据：{'ngxin': {'nginx_ip': '127.0.0.1', 'nginx_port': 8080, 'nginx_fix': '/hls/', 'nginx_path': 'X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/', 'nginx_config_path': 'X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/conf/nginx.conf'}, 'fastapi': {'port': 8991, 'hls_dir': 'X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/html/hls/', 'ffmpeg_dir': 'W:/ffmpeg-20200831-4a11a6f-win64-static//bin/ffmpeg.exe', 'max_stream_threads': 60, 'max_scan_time': '3*60', 'max_code_ff_size': 6, 'comm': '{ffmpeg_dir} -f rtsp -rtsp_transport tcp -i rtsp://admin:hik12345@{ip}:554/Streaming/Channels/101?transportmode=multicast -acodec aac -strict experimental -ar 44100 -ac 2 -b:a 96k -r 25 -b:v 300k -s {width}*{height} -live 1  -c:v libx264 -c:a copy -cpu-used 0  -threads 1  -f hls -hls_time 0.5 -hls_list_size 1 -hls_wrap 100 {hls_dir}{ip}-{uuid_v}.m3u8'}}
2024-08-13 15:57:03,413 - base.py[line:454] - INFO: Adding job tentatively -- it will be properly scheduled when the scheduler starts
2024-08-13 15:57:03,414 - proactor_events.py[line:630] - DEBUG: Using proactor: IocpProactor
INFO:     Started server process [30404]
INFO:     Waiting for application startup.
2024-08-13 15:57:03,441 - base.py[line:895] - INFO: Added job "scan_video_stream_list" to job store "default"
2024-08-13 15:57:03,441 - base.py[line:181] - INFO: Scheduler started
Process==================end
当前执行路径：X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/nginx.exe -c X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/conf/nginx.conf

===================  Nginx has been started successfully.

2024-08-13 15:57:09,256 - base.py[line:954] - DEBUG: Looking for jobs to run
2024-08-13 15:57:09,256 - base.py[line:1034] - DEBUG: Next wakeup is due at 2024-08-13 16:57:03.413311+08:00 (in 3594.156780 seconds)
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

API 文档:http://127.0.0.1:8000/docs#

测试页面

{'ip': '192.168.2.25', 'width': 320, 'height': 170, 'time': <module 'time' (built-in)>, 'uuid_v': 'dbeda9ce-01ec-41cd-8315-8145954d1ea0', 'hls_dir': 'X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/html/hls/', 'ffmpeg_dir': 'W:/ffmpeg-20200831-4a11a6f-win64-static//bin/ffmpeg.exe'}
W:/ffmpeg-20200831-4a11a6f-win64-static//bin/ffmpeg.exe -f rtsp -rtsp_transport tcp -i rtsp://admin:hik12345@192.168.2.25:554/Streaming/Channels/101?transportmode=multicast -acodec aac -strict experimental -ar 44100 -ac 2 -b:a 96k -r 25 -b:v 300k -s 320*170 -live 1  -c:v libx264 -c:a copy -cpu-used 0  -threads 1  -f hls -hls_time 0.5 -hls_list_size 1 -hls_wrap 100 X:/nginx-rtmp-win32-dev/nginx-rtmp-win32-dev/html/hls/192.168.2.25-dbeda9ce-01ec-41cd-8315-8145954d1ea0.m3u8
{'pid': 32416, 'v_url': 'http://127.0.0.1:8080/hls/192.168.2.25-dbeda9ce-01ec-41cd-8315-8145954d1ea0.m3u8', 'ip': '192.168.2.25'}
==============================摄像头数据更新完成...，重新确认子进程是否运行
Process==================end
创建的进程为: [psutil.Process(pid=32416, name='ffmpeg.exe', status='running', started='15:59:38')]
返回取流路径为：{'pid': 32416, 'v_url': 'http://127.0.0.1:8080/hls/192.168.2.25-dbeda9ce-01ec-41cd-8315-8145954d1ea0.m3u8', 'ip': '192.168.2.25'}
INFO:     127.0.0.1:64650 - "GET /sc_view/get_video_stream?ip=192.168.2.25&width=320&height=170 HTTP/1.1" 200 OK

博文部分内容参考

© 文中涉及参考链接内容版权归原作者所有，如有侵权请告知 :)

© 2018-2024 liruilonger@gmail.com, 保持署名-非商用-相同方式共享(CC BY-NC-SA 4.0)