I have one rtsp server with multiple videos hosted using multiprocessing queues as channels using sensor factory objects using gstreamer. The reason for using multi processing queue is because of the limitation of nvh264enc encoder in the pipeline(The teardown causes the ports to be nonusable in subsequent client connections, no problem in x264enc). The problem happens when multiple clients are connected to the same link, even though the factory is shared between clients with same media element. This problem occurs. The results can be seen in clients such as vlc player.
"""
This module will provide encoder functionality
"""
from base64 import decode
from gc import get_stats
from os import stat
from queue import Empty
import sys
import itertools
import numpy as np
import logging
import threading
import multiprocessing as mp
import gi
from pprint import pprint
gi.require_version('Gst', '1.0')
gi.require_version('GstRtspServer', '1.0')
from gi.repository import Gst, GstRtspServer, GObject, GLib
import cv2
import subprocess
import time
GObject.threads_init()
Gst.init(None)
# def on_debug(category, level, dfile, dfctn, dline, source, message, user_data):
# if source:
# print('Debug {} {}: {}'.format(
# Gst.DebugLevel.get_name(level), source.name, message.get()))
# else:
# print('Debug {}: {}'.format(
# Gst.DebugLevel.get_name(level), message.get()))
# if not Gst.debug_is_active():
# Gst.debug_set_active(True)
# level = Gst.debug_get_default_threshold()
# Gst.debug_set_default_threshold(Gst.DebugLevel.INFO)
# if level < Gst.DebugLevel.ERROR:
# Gst.debug_set_default_threshold(Gst.DebugLevel.WARNING)
# Gst.debug_add_log_function(on_debug, None)
# Gst.debug_remove_log_function(Gst.debug_log_default)
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s:%(levelname)s:%(name)s:%(message)s',
handlers=[logging.FileHandler('GPU_enabled_frame_encode.log'), logging.StreamHandler(sys.stdout)])
logging.debug('Debug message')
logging.info('Info message')
logging.warning('Warning message')
logging.error('Error message')
logging.critical('Critical message')
class SensorFactory(GstRtspServer.RTSPMediaFactory):
def __init__(self, fps, img_shape, cols, verbosity=1, cap=None, speed_preset='medium', properties={}):
super(SensorFactory, self).__init__(**properties)
logging.info("sensory factory")
self.rtsp_media = None
self.height = int(img_shape[0])
self.width = int(img_shape[1] * cols)
self.number_frames = 0
self.stream_timestamp = 0.0
self.timestamp = time.time()
self.dt = 0.0
self.streamed_frames = 0
self.verbosity = verbosity
fps = int(fps)
self.cap = cap
self.appsrc = None
# duration of a frame in nanoseconds nvh264enc x264enc
self.duration = 1.0 / fps * Gst.SECOND
key_int_max = ' key-int-max={} '.format(fps)
caps_str = 'caps=video/x-raw,format=BGR,width={},height={},framerate={}/1 '.format(self.width,
self.height,
fps)
self.launch_string = 'appsrc name=source is-live=true block=true do-timestamp=true \
format=GST_FORMAT_TIME ' + caps_str + \
' ! queue' \
' ! videoconvert' \
' ! video/x-raw,format=I420' \
' ! nvh264enc' \
' ! rtph264pay config-interval=1 pt=96 name=pay0' \
''
def set_cap(self, cap):
self.cap = cap
def on_need_data(self, src, length):
# this method executes when client requests data
# logging.info("this method executes when client requests data")
# if self.cap.isOpened():
# _, frame = self.cap.get_frame()
# ret = True
if self.cap.isOpened():
frame_queue = self.cap.get_queue()
frame = frame_queue.get()
ret = True
if ret:
if frame.shape[:2] != (self.height, self.width):
frame = cv2.resize(frame, (self.width, self.height))
data = frame.tostring()
buf = Gst.Buffer.new_allocate(None, len(data), None)
buf.fill(0, data)
buf.duration = self.duration
timestamp = self.number_frames * self.duration
buf.pts = buf.dts = int(timestamp)
buf.offset = timestamp
self.number_frames += 1
retval = self.appsrc.emit('push-buffer', buf)
# print('pushed buffer, frame {}, duration {} ns, durations {} s'.format(self.number_frames,
# self.duration,
# self.duration / Gst.SECOND))
if retval != Gst.FlowReturn.OK:
logging.info("[INFO]: retval not OK: {}".format(retval))
if retval == Gst.FlowReturn.FLUSHING:
logging.info('Offline')
elif self.verbosity > 0:
logging.info("[INFO]: Unable to read frame from cap.")
# time.sleep(0.05)
def do_create_element(self, url):
if self.verbosity > 0:
request_uri = url.get_request_uri()
logging.info('[INFO]: stream request on {}'.format(request_uri))
return Gst.parse_launch(self.launch_string)
def do_configure(self, rtsp_media):
self.rtsp_media = rtsp_media
rtsp_media.set_reusable(True)
self.number_frames = 0
self.appsrc = rtsp_media.get_element().get_child_by_name('source')
# executes when client requests data
self.appsrc.connect('need-data', self.on_need_data)
def get_rtsp_media(self):
if self.rtsp_media:
return self.rtsp_media
def __del__(self):
print('Destructor called, factory deleted.')
class RTSP_utility_server(GstRtspServer.RTSPServer):
def __init__(self, fps, suffix='test', rtp_port=8554,
ip='12.0.0.0', caps=(None,), Sizes=[[1080, 1920]],
speed_preset='medium', verbosity=1, Indexes=[]):
GObject.threads_init()
Gst.init(None)
super(RTSP_utility_server, self).__init__(**{})
self.verbosity = verbosity
self.rtp_port = "{}".format(rtp_port)
if int(self.rtp_port) < 1024 and self.verbosity > 0:
logging.info(
'[INFO]: Note, admin privileges are required because port number < 1024.')
self.set_service(self.rtp_port)
self.speed_preset = speed_preset
self.caps = caps
self.factory = [None] * len(self.caps)
self.suffix = suffix
self.fps = fps
self.Sizes = Sizes
self.Indexes = Indexes
self.attach(None)
self.ip = self.get_ip()
self.media_path_list = [None] * len(self.caps)
self.clients_list = []
if len(self.suffix):
self.full_suffix = '/' + self.suffix.lstrip('/')
else:
self.full_suffix = ''
self.connect("client-connected", self.client_connected)
logging.info(
'[INFO]: streaming on:\n\trtsp://{}:{}/{}#'.format(self.ip, self.rtp_port, self.suffix))
self.status_thread = threading.Thread(target=self.status_thread_loop)
self.status_thread.daemon = True
self.status_thread.start()
self.context = GLib.MainContext()
print(self.attach(self.context))
def set_caps(self, caps):
if not isinstance(caps, (list, tuple)):
caps = [caps]
self.caps = caps
def create_media_factories(self):
mount_points = self.get_mount_points()
media_path_list = []
for i, cap in enumerate(self.caps):
img_shape = self.Sizes[i]
if len(self.Indexes) == 0:
N_Index = str(i + 1)
else:
N_Index = str(self.Indexes[i])
factory = SensorFactory(fps=self.fps, img_shape=img_shape, speed_preset=self.speed_preset,
cols=1, verbosity=self.verbosity, cap=cap)
factory.set_shared(True)
factory.set_stop_on_disconnect(True)
logging.info('inside media_factories Stream on ' +
self.full_suffix + N_Index)
logging.info('inside media_factories Stream on ' +
self.full_suffix + N_Index)
mount_points.add_factory(self.full_suffix + N_Index, factory)
self.factory[i] = factory
media_path_list.append(self.full_suffix + N_Index)
self.media_path_list = media_path_list
self.get_status()
def destroy_media_factories(self):
session_pool = self.get_session_pool()
logging.info("Number of sessions are :" +
str(session_pool.get_n_sessions()))
sessions_list = session_pool.filter()
for session in sessions_list:
for path in self.get_paths():
media_matched, _ = session.get_media(path)
if media_matched:
rtsp_media = media_matched.get_media()
rtsp_media.set_eos_shutdown(True)
rtsp_media.unprepare()
logging.debug("media removed for path "+path)
number_of_disconnects = session_pool.cleanup()
if number_of_disconnects > 0:
logging.info("number of disconnects:"+str(number_of_disconnects))
def destroy_media_factories_by_path(self,path_to_remove="/video1"):
session_pool = self.get_session_pool()
logging.info("Number of sessions are :" +
str(session_pool.get_n_sessions()))
sessions_list = session_pool.filter()
for session in sessions_list:
for path in self.get_paths():
media_matched, _ = session.get_media(path)
if media_matched and path==path_to_remove:
rtsp_media = media_matched.get_media()
rtsp_media.set_eos_shutdown(True)
rtsp_media.unprepare()
logging.debug("media removed for path "+path)
number_of_disconnects = session_pool.cleanup()
if number_of_disconnects > 0:
logging.info("number of disconnects:"+str(number_of_disconnects))
def client_connected(self, gst_server_obj, rtsp_client_obj):
logging.info('[INFO]: Client has connected')
self.create_media_factories()
self.clients_list.append(rtsp_client_obj)
if self.verbosity > 0:
logging.info('[INFO]: Client has connected')
def stop_all(self):
self.destroy_media_factories()
def stop_by_index(self,path):
self.destroy_media_factories_by_path(path)
def get_paths(self):
return self.media_path_list
def get_status(self):
mount_points = self.get_mount_points()
session_pool = self.get_session_pool()
# logging.info("Number of sessions are :" +
# str(session_pool.get_n_sessions()))
number_of_disconnects = session_pool.cleanup()
if number_of_disconnects > 0:
logging.info("number of disconnects:"+str(number_of_disconnects))
for path in self.get_paths():
sessions_list = session_pool.filter()
for session in sessions_list:
session.set_timeout(1)
media_matched, _ = session.get_media(path)
if media_matched:
rtsp_media = media_matched.get_media()
status = rtsp_media.get_status() #<enum GST_RTSP_MEDIA_STATUS_PREPARED of type GstRtspServer.RTSPMediaStatus>
if "GST_RTSP_MEDIA_STATUS_UNPREPARING" in str(status):
# print(self.context)
# GObject.Object.unref(self)
# print(status)
# transport = media_matched.get_transport(0)
# transport.set_active(False)
# print(enum_list)
# print(rtsp_media.unprepare())
# rtsp_media.set_eos_shutdown(True)
# session.set_timeout(1)
# session.release_media(media_matched)
# session.allow_expire()
# session.release_media(media_matched)
# media_matched.set_state(Gst.State.PAUSED)
# media_matched.set_state(Gst.State.READY)
# media_matched.set_state(Gst.State.NULL)
# rtsp_media.suspend()
# print(rtsp_media.unprepare())
# rtsp_media.set_pipeline_state(Gst.State.READY)
# rtsp_media.set_pipeline_state(Gst.State.NULL)
# rtsp_media.unprepare()
# self.client_filter()
print("removing")
def status_thread_loop(self):
while True:
time.sleep(2)
try:
self.get_status()
except Exception as e:
print(e)
def __del__(self):
print("Destroy called")
def get_current_encoders_details(self):
enc_details = dict()
for i,cap in enumerate(self.caps):
enc_details[i]={"path":self.media_path_list[i],"decoder":cap,"factory":self.factory[i]}
return enc_details
@staticmethod
def get_ip():
return subprocess.check_output("hostname -I", shell=True).decode('utf-8').split(' ')[0]
class encoders():
"""
This is the containter for multiple encoders
"""
def __init__(self,decoders=[],head=[],resolutions=[],suffix="video") -> None:
self.encoder_list = []
self.caps = decoders
self.fps = 60
Sizes = resolutions
self.enc_obj = RTSP_utility_server(self.fps, Sizes=Sizes, speed_preset="medium", caps=decoders, suffix=suffix,
verbosity=1, rtp_port=8554, ip='10.5.1.130')
def get_encoders(self):
return self.enc_obj.get_current_encoders_details()
def set_frame(self,enc_id=0,frame=np.zeros((360, 640, 3))):
encoder_objects = self.enc_obj.get_current_encoders_details()
decoder_obj = None
for id in self.enc_obj.get_current_encoders_details():
if id == enc_id:
decoder_obj = encoder_objects[id]["decoder"]
break
if decoder_obj:
decoder_obj.set_canvas(frame)
else:
print("Decoder not available")
def stop_all(self):
self.enc_obj.stop_all()
def stop_by_index(self,path="/video1"):
self.enc_obj.stop_by_index(path)
def get_status(self):
self.enc_obj.get_status()
the mp queue logic is as following
while True:
# print(f'Thread {self.media_id} Looking')
if self.stop_signal.is_set():
self.stop_signal.clear()
self.active = False
break
ret, frame = self.capture.read()
if ret and type(frame) != type(None):
self.frame = frame.copy()
self.latest_frame = frame.copy()
self.recieving_signal.set()
self.latest_frame_time = time.ctime()
try:
self.frame_queue.put(self.frame,block=False)
except Full:
pass
Here capture is cv2.VideoCapture from an offline file. Is there any way to improve the quality of the video without stuttering with new connections.