Skip to content

Commit 7d0d116

Browse files
author
sahil parekh
committed
Initial Code without ReadMe, Will add soon
1 parent b93b320 commit 7d0d116

File tree

2 files changed

+285
-0
lines changed

2 files changed

+285
-0
lines changed

‎main_prg.py‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
2+
'''\
3+
This Simple program Demonstrates how to use G-Streamer and capture RTSP Frames in Opencv using Python
4+
- Sahil Parekh
5+
'''
6+
7+
import multiprocessing as mp
8+
import time
9+
import vid_streamv3 as vs
10+
import cv2
11+
import sys
12+
13+
'''
14+
Main class
15+
'''
16+
class mainStreamClass:
17+
def __init__(self):
18+
19+
#Current Cam
20+
self.camProcess = None
21+
self.cam_queue = None
22+
self.stopbit = None
23+
self.camlink = '' #Add your RTSP cam link
24+
25+
def startMain(self):
26+
27+
#set queue size
28+
self.cam_queue = mp.Queue(maxsize=100)
29+
30+
#get all cams
31+
time.sleep(3)
32+
33+
self.stopbit = mp.Event()
34+
self.camProcess = vs.StreamCapture(self.camlink,
35+
self.stopbit,
36+
self.cam_queue)
37+
self.camProcess.start()
38+
39+
# calculate FPS
40+
lastFTime = time.time()
41+
42+
try:
43+
while True:
44+
45+
if not self.cam_queue.empty():
46+
# print('Got frame')
47+
cmd, val = self.cam_queue.get()
48+
49+
'''
50+
#calculate FPS
51+
diffTime = time.time() - lastFTime`
52+
fps = 1 / diffTime
53+
# print(fps)
54+
55+
'''
56+
lastFTime = time.time()
57+
58+
# if cmd == vs.StreamCommands.RESOLUTION:
59+
# pass #print(val)
60+
61+
if cmd == vs.StreamCommands.FRAME:
62+
if val is not None:
63+
cv2.imshow('Cam: ' + self.camlink, val)
64+
cv2.waitKey(1)
65+
66+
except KeyboardInterrupt:
67+
print('Caught Keyboard interrupt')
68+
69+
except:
70+
e = sys.exc_info()
71+
print('Caught Main Exception')
72+
print(e)
73+
74+
self.stopCamStream()
75+
cv2.destroyAllWindows()
76+
77+
78+
def stopCamStream(self):
79+
print('in stopCamStream')
80+
81+
if self.stopbit is not None:
82+
self.stopbit.set()
83+
while not self.cam_queue.empty():
84+
try:
85+
_ = self.cam_queue.get()
86+
except:
87+
break
88+
self.cam_queue.close()
89+
90+
self.camProcess.join()
91+
92+
93+
if __name__ == "__main__":
94+
mc = mainStreamClass()
95+
mc.startMain()

‎vid_streamv3.py‎

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
1+
#cython: language_level=3, boundscheck=False
2+
import multiprocessing as mp
3+
from enum import Enum
4+
import numpy as np
5+
import gi
6+
gi.require_version('Gst', '1.0')
7+
from gi.repository import Gst
8+
Gst.init(None)
9+
10+
'''Konwn issues
11+
12+
* if format changes at run time system hangs
13+
'''
14+
15+
class StreamMode(Enum):
16+
INIT_STREAM = 1
17+
SETUP_STREAM = 1
18+
READ_STREAM = 2
19+
20+
21+
class StreamCommands(Enum):
22+
FRAME = 1
23+
ERROR = 2
24+
HEARTBEAT = 3
25+
RESOLUTION = 4
26+
STOP = 5
27+
28+
29+
class StreamCapture(mp.Process):
30+
31+
def __init__(self, link, stop, outQueue):
32+
"""
33+
Initialize the stream capturing process
34+
link - rstp link of stream
35+
stop - to send commands to this process
36+
outPipe - this process can send commands outside
37+
"""
38+
39+
super().__init__()
40+
self.streamLink = link
41+
self.stop = stop
42+
self.outQueue = outQueue
43+
self.currentState = StreamMode.INIT_STREAM
44+
self.pipeline = None
45+
self.source = None
46+
self.decode = None
47+
self.convert = None
48+
self.sink = None
49+
self.image_arr = None
50+
self.newImage = False
51+
self.frame1 = None
52+
self.frame2 = None
53+
54+
55+
56+
def gst_to_opencv(self, sample):
57+
buf = sample.get_buffer()
58+
caps = sample.get_caps()
59+
60+
# Print Height, Width and Format
61+
# print(caps.get_structure(0).get_value('format'))
62+
# print(caps.get_structure(0).get_value('height'))
63+
# print(caps.get_structure(0).get_value('width'))
64+
65+
arr = np.ndarray(
66+
(caps.get_structure(0).get_value('height'),
67+
caps.get_structure(0).get_value('width'),
68+
3),
69+
buffer=buf.extract_dup(0, buf.get_size()),
70+
dtype=np.uint8)
71+
return arr
72+
73+
def new_buffer(self, sink, _):
74+
sample = sink.emit("pull-sample")
75+
arr = self.gst_to_opencv(sample)
76+
self.image_arr = arr
77+
self.newImage = True
78+
return Gst.FlowReturn.OK
79+
80+
def run(self):
81+
# Create the empty pipeline
82+
self.pipeline = Gst.parse_launch(
83+
'rtspsrc name=m_rtspsrc ! rtph264depay name=m_rtph264depay ! avdec_h264 name=m_avdech264 ! videoconvert name=m_videoconvert ! appsink name=m_appsink')
84+
85+
# source params
86+
self.source = self.pipeline.get_by_name('m_rtspsrc')
87+
self.source.set_property('latency', 0)
88+
self.source.set_property('location', self.streamLink)
89+
self.source.set_property('protocols', 'tcp')
90+
self.source.set_property('retry', 50)
91+
self.source.set_property('timeout', 50)
92+
self.source.set_property('tcp-timeout', 5000000)
93+
self.source.set_property('drop-on-latency', 'true')
94+
95+
# decode params
96+
self.decode = self.pipeline.get_by_name('m_avdech264')
97+
self.decode.set_property('max-threads', 2)
98+
self.decode.set_property('output-corrupt', 'false')
99+
100+
# convert params
101+
self.convert = self.pipeline.get_by_name('m_videoconvert')
102+
103+
# sink params
104+
self.sink = self.pipeline.get_by_name('m_appsink')
105+
106+
# Maximum number of nanoseconds that a buffer can be late before it is dropped (-1 unlimited)
107+
# flags: readable, writable
108+
# Integer64. Range: -1 - 9223372036854775807 Default: -1
109+
self.sink.set_property('max-lateness', 500000000)
110+
111+
# The maximum number of buffers to queue internally (0 = unlimited)
112+
# flags: readable, writable
113+
# Unsigned Integer. Range: 0 - 4294967295 Default: 0
114+
self.sink.set_property('max-buffers', 5)
115+
116+
# Drop old buffers when the buffer queue is filled
117+
# flags: readable, writable
118+
# Boolean. Default: false
119+
self.sink.set_property('drop', 'true')
120+
121+
# Emit new-preroll and new-sample signals
122+
# flags: readable, writable
123+
# Boolean. Default: false
124+
self.sink.set_property('emit-signals', True)
125+
126+
# # sink.set_property('drop', True)
127+
# # sink.set_property('sync', False)
128+
129+
# The allowed caps for the sink pad
130+
# flags: readable, writable
131+
# Caps (NULL)
132+
caps = Gst.caps_from_string(
133+
'video/x-raw, format=(string){BGR, GRAY8}; video/x-bayer,format=(string){rggb,bggr,grbg,gbrg}')
134+
self.sink.set_property('caps', caps)
135+
136+
if not self.source or not self.sink or not self.pipeline or not self.decode or not self.convert:
137+
print("Not all elements could be created.")
138+
self.stop.set()
139+
140+
self.sink.connect("new-sample", self.new_buffer, self.sink)
141+
142+
# Start playing
143+
ret = self.pipeline.set_state(Gst.State.PLAYING)
144+
if ret == Gst.StateChangeReturn.FAILURE:
145+
print("Unable to set the pipeline to the playing state.")
146+
self.stop.set()
147+
148+
# Wait until error or EOS
149+
bus = self.pipeline.get_bus()
150+
151+
while True:
152+
153+
if self.stop.is_set():
154+
print('Stopping CAM Stream by main process')
155+
break
156+
157+
message = bus.timed_pop_filtered(10000, Gst.MessageType.ANY)
158+
# print "image_arr: ", image_arr
159+
if self.image_arr is not None and self.newImage is True:
160+
161+
if not self.outQueue.full():
162+
163+
# print("\r adding to queue of size{}".format(self.outQueue.qsize()), end='\r')
164+
self.outQueue.put((StreamCommands.FRAME, self.image_arr), block=False)
165+
166+
self.image_arr = None
167+
168+
169+
if message:
170+
if message.type == Gst.MessageType.ERROR:
171+
err, debug = message.parse_error()
172+
print("Error received from element %s: %s" % (
173+
message.src.get_name(), err))
174+
print("Debugging information: %s" % debug)
175+
break
176+
elif message.type == Gst.MessageType.EOS:
177+
print("End-Of-Stream reached.")
178+
break
179+
elif message.type == Gst.MessageType.STATE_CHANGED:
180+
if isinstance(message.src, Gst.Pipeline):
181+
old_state, new_state, pending_state = message.parse_state_changed()
182+
print("Pipeline state changed from %s to %s." %
183+
(old_state.value_nick, new_state.value_nick))
184+
else:
185+
print("Unexpected message received.")
186+
187+
188+
print('terminating cam pipe')
189+
self.stop.set()
190+
self.pipeline.set_state(Gst.State.NULL)

0 commit comments

Comments
 (0)