如何使用Python模拟本地mp4文件的RTSP流？

Question

我正在开发一个需要分析来自 IP 摄像机的 RTSP 视频流的解决方案。但是，我在通过 RTSP 访问这些摄像机时遇到问题，团队为我提供了一些示例文件（mp4 文件）。我仍然需要研究这个解决方案，为了不改变我的架构，有没有一种方法可以使用 RTSP url 和 python 来模拟来自这些 mp4 文件之一的视频流？我的解决方案架构完全基于 GCP，我也需要这个 RTSP url 才能在那里工作。

如果这是一个愚蠢的问题，我很抱歉，但这超出了我的专业范围。

自从我使用 GCP 以来，我知道使用 Vertex AI Vision 我可以通过 mp4 文件模拟流式传输，但我无法使用它（该工具的流式传输位置需要位于 us-central1 和 europe-west4，并且这些位置不允许用于我的项目）。

Answer 1

您可以启动本地客户端播放器进行流式传输，在从IP摄像机进行分析方面，需要完成一些额外的工作，如果摄像机是开源或基于库的视频源，驱动程序或控制器应该有一个选项要访问托管计算机中任何位置的实时视频流源（只需考虑是否通过服务器 URL 提供实时视频），某些摄像机可能无法仅使用自己的软件或平台轻松共享该源，我们假设视频是通过服务器和端口提供服务：

服务器端和客户端结果：

现在，因为我们需要在 GCP 中实现它，所以我的建议是首先部署一个基于 Linux 或 Linux 操作系统的虚拟机或具有兼容运行时堆栈的容器，以获取使其流式传输所需的所有资源（最快的用例是虚拟机，但随后您将需要在 GCP 中的虚拟机中授权一些防火墙规则和端口）：https://cloud.google.com/compute/docs/create-linux-vm-instance

完整的代码需要一些额外的工作，您可以对 IP 摄像机进行实时侦听，并不断保存视频的各个部分并同时流式传输它们，或者从本地设备记录视频中的文件并将其上传到您的 GCP 云存储： https://cloud.google.com/storage/docs/creating-buckets

要在文件到云到流的上下文中上传它，您可以使用：https://cloud.google.com/storage/docs/uploading-objects#storage-upload-object-python

from google.cloud import storage


def upload_blob(bucket_name, source_file_name, destination_blob_name):
    """Uploads a file to the bucket."""
    # The ID of your GCS bucket
    # bucket_name = "your-bucket-name"
    # The path to your file to upload
    # source_file_name = "local/path/to/file"
    # The ID of your GCS object
    # destination_blob_name = "storage-object-name"

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Optional: set a generation-match precondition to avoid potential race conditions
    # and data corruptions. The request to upload is aborted if the object's
    # generation number does not match your precondition. For a destination
    # object that does not yet exist, set the if_generation_match precondition to 0.
    # If the destination object already exists in your bucket, set instead a
    # generation-match precondition using its generation number.
    generation_match_precondition = 0

    blob.upload_from_filename(source_file_name, if_generation_match=generation_match_precondition)

    print(
        f"File {source_file_name} uploaded to {destination_blob_name}."
    )

完整代码（将摄像头流捕获到视频/服务 RSTP）：

#!/usr/bin/env python

import sys
#sudo apt install gir1.2-gst-rtsp-server-1.0


#the script to save the video to a file to store, from RSTP the cameras will be:
##pip install av

#script to save stream to a file
##import av
##
##source = av.open('rtsp://127.0.0.1:8554/sailormoon', metadata_encoding='utf-8') #this one comes from the gi RSTP server
##output = av.open('test-2-temp.mp4', mode='w', format='h264') #to later serve this recorded video from the cameras stream, you can reinitiate the gi server for the new video, 
##
##in_to_out = {}
##
##for i, stream in enumerate(source.streams):
##
##    if (
##            (stream.type == 'audio') or
##            (stream.type == 'video') or
##            (stream.type == 'subtitle') or
##            (stream.type == 'data')
##    ):
##        in_to_out[stream] = ostream = output.add_stream(template=stream)
##        ostream.options = {}
##
##count = 0
##for i, packet in enumerate(source.demux()):
##    try:
##        if packet.dts is None:
##            continue
##
##        packet.stream = in_to_out[packet.stream]
##        output.mux(packet)
##        count += 1
##        if count > 200:
##            break
##
##    except InterruptedError:
##        output.close()
##        break
##
##output.close()

import gi

gi.require_version('Gst', '1.0')
gi.require_version('GstRtspServer', '1.0')
from gi.repository import Gst, GstRtspServer, GObject, GLib

loop = GLib.MainLoop()
Gst.init(None)

class Jbsidis(GstRtspServer.RTSPMediaFactory):
    def __init__(self):
        GstRtspServer.RTSPMediaFactory.__init__(self)

    def do_create_element(self, url):
        #where is the mp4 video, in GCP you can specify the bucket
        src_demux = "filesrc location=/home/jbsidis/Downloads/sailormoon.mp4 ! qtdemux name=demux"
        h264_transcode = "demux.video_0"
        #this will initialize the video, but if more codecs needed add #h264_transcode = "demux.video_0 ! decodebin ! queue ! x264enc"
        pipeline = "{0} {1} ! queue ! rtph264pay name=pay0 config-interval=1 pt=96".format(src_demux, h264_transcode)
        print("Serving RSTP video: " + pipeline)
        return Gst.parse_launch(pipeline)

class ServingRSTP():
    def __init__(self):
        self.rtspServer = GstRtspServer.RTSPServer()
        factory = Jbsidis()
        factory.set_shared(True)
        mountPoints = self.rtspServer.get_mount_points()
        mountPoints.add_factory("/sailormoon", factory)
        self.rtspServer.attach(None)

if __name__ == '__main__':
    s = ServingRSTP()
    loop.run()

云中的 GCP 资源和带宽会涉及一些成本，请务必检查定价。