alecs-examples/send_stream.py at dev · alliedvision/alecs-examples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
"""
  Copyright (C) 2025 Allied Vision Technologies. All Rights Reserved.
  Subject to the BSD 3-Clause License.
"""

import sys
from vmbpy import *
from vmbpy.camera import CamerasTuple
import gi
gi.require_version('Gst', '1.0')
from gi.repository import GLib, Gst
import os


def print_preamble():
    print('////////////////////////////////////////')
    print('/// Smart Camera Send Stream Example ///')
    print('////////////////////////////////////////\n')


def print_usage():
    print('Usage:')
    print('    python send_stream.py <target ip>')
    print('    python send_stream.py -h')
    print()


def abort(reason: str, return_code: int = 1, usage: bool = False):
    print(reason + '\n')

    if usage:
        print_usage()

    sys.exit(return_code)


def parse_args() -> str:
    args = sys.argv[1:]
    argc = len(args)

    for arg in args:
        if arg == '-h':
            print_usage()
            sys.exit(0)

    if argc > 1:
        abort(reason="Invalid number of arguments. Abort.", return_code=2, usage=True)

    return args[0]

if __name__ == '__main__':
    print_preamble()
    target_ip = parse_args()

    with VmbSystem.get_instance() as vmb_system:
        cameras: CamerasTuple = vmb_system.get_all_cameras()
        if len(cameras) == 0:
            abort('No camera found.')

        with cameras[0] as camera:
            device_id = camera.get_id()
            width = camera.Width.get()
            height = camera.Height.get()
            pixel_formats = camera.get_pixel_formats()
            format = "UYVY" if PixelFormat.YCbCr422_8_CbYCrY in pixel_formats else "GRAY8" if PixelFormat.Mono8 in pixel_formats else ""
            if len(format) == 0:
                abort('No supported pixel formats.')

    # The following is essentially equivalent to running the following GStreamer command:
    # gst-launch-1.0 vmbsrc camera=<device_id> ! 'video/x-raw,format=<format>, width=<width>, height=<height>' ! nvvidconv ! queue max-size-buffers=1 leaky=downstream ! nvv4l2h265enc bitrate=2000000 iframeinterval=1 insert-sps-pps=true profile=Main preset-level=UltraFastPreset ! queue ! rtph265pay aggregate-mode=zero-latency ! udpsink port=51372 host=<target ip>

    Gst.init(None)

    mainloop = GLib.MainLoop()

    pipeline =  Gst.Pipeline.new("pipe")

    # create elements and set their properties

    vmbsrc = Gst.ElementFactory.make("vmbsrc")
    Gst.util_set_object_arg(vmbsrc, "camera", device_id)
    vmbsrc.set_property('exposuretime', 25000)
    vmbsrc.set_property('balancewhiteauto', 1)

    caps0 = Gst.Caps.from_string(f"video/x-raw, format={format}, width={width}, height={height}")
    filter0 = Gst.ElementFactory.make("capsfilter", "filter")
    filter0.set_property("caps", caps0)

    nvvidconv = Gst.ElementFactory.make("nvvidconv")

    queue = Gst.ElementFactory.make("queue")
    queue.set_property("max-size-buffers", 1)
    queue.set_property("leaky", "downstream")

    h265enc = Gst.ElementFactory.make("nvv4l2h265enc")
    h265enc.set_property("bitrate", 2000000)
    h265enc.set_property("insert-sps-pps", True)
    h265enc.set_property("iframeinterval", 1)
    h265enc.set_property("profile", "Main")
    h265enc.set_property("preset-level", "UltraFastPreset")


    rtph265pay = Gst.ElementFactory.make("rtph265pay")
    rtph265pay.set_property('aggregate-mode','zero-latency')


    udpsink = Gst.ElementFactory.make("udpsink")
    udpsink.set_property("host", target_ip)
    udpsink.set_property("port", 51372)

    # add elements to pipeline
    pipeline.add(vmbsrc)
    pipeline.add(filter0)
    pipeline.add(queue)
    pipeline.add(nvvidconv)

    pipeline.add(h265enc)
    pipeline.add(rtph265pay)
    pipeline.add(udpsink)

    # link elements

    vmbsrc.link(filter0)
    filter0.link(nvvidconv)
    nvvidconv.link(queue)
    queue.link(h265enc)
    h265enc.link(rtph265pay)
    rtph265pay.link(udpsink)

    # stream until Ctrl-C is pressed

    pipeline.set_state(Gst.State.PLAYING)

    try:
        mainloop.run()
    except KeyboardInterrupt:
        pipeline.set_state(Gst.State.NULL)
        mainloop.quit()