stream_2016/gstconf.py

   1 #!/usr/bin/env python3.4
   2 # -*- coding: utf-8 -*-
   3
   4 # This file is part of ABYSS.
   5 # ABYSS Broadcast Your Streaming Successfully
   6 #
   7 # ABYSS is free software: you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation, either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # ABYSS is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with ABYSS.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20 # Copyright (c) 2016 David Testé
  21
  22 from os import rename
  23 from os import listdir
  24 from os import path
  25 from os import mkdir
  26 from time import localtime, strftime
  27 import configparser
  28
  29 import gi
  30 from gi.repository import Gst
  31 from gi.repository import GstVideo
  32
  33 DIR_NAME = 'FILES_RECORDED'
  34 AUDIO_DEFAULT = './' + DIR_NAME + '/' + 'AUDIO_DEFAULT'
  35 RAWVIDEO_DEFAULT = './' + DIR_NAME + '/' + 'RAWVIDEO_DEFAULT'
  36 STREAM_DEFAULT = './' + DIR_NAME + '/' + 'STREAM_DEFAULT'
  37 BACKUP_SUFFIX = '_BACKUP'
  38 FAILED_SUFFIX = '_FAILED_'
  39 fail_counter = 1
  40 AUDIO_BACKUP = AUDIO_DEFAULT + BACKUP_SUFFIX
  41 RAWVIDEO_BACKUP = RAWVIDEO_DEFAULT + BACKUP_SUFFIX
  42 STREAM_BACKUP = STREAM_DEFAULT + BACKUP_SUFFIX
  43 ERROR = '[ERROR] '
  44 INFO = '[INFO] '
  45 WARN = '[WARN] '
  46 CONFIG = '.abyss'
  47
  48 sources = {'RTSP_IP' : None,
  49            'AUDIO_INPUT' : None,}
  50 sinks = {'AUDIO_OUTPUT' : None,
  51          'DIR': None,
  52          'STREAM_SERVER_IP' : None,
  53          'SERVER_PORT' : None,
  54          'PASSWORD' : None,
  55          'AUDIO_MOUNT' : None,
  56          'VIDEO_MOUNT' : None,}
  57
  58 ##AUDIO_INPUT = 'alsa_input.usb-Burr-Brown_from_TI_USB_Audio_CODEC-00-CODEC.analog-stereo'
  59 ##AUDIO_OUTPUT = 'alsa_output.pci-0000_00_1b.0.analog-stereo'
  60
  61 config = configparser.RawConfigParser()
  62 if path.exists(CONFIG):
  63     config.read(CONFIG)
  64     try:
  65         sources = {key : config.get('sources', key) for key in sources}
  66         sinks = {key : config.get('sinks', key) for key in sinks}
  67     except:
  68         print(ERROR, gettime(), 'Failed to parse config file.')
  69 else:
  70     print(ERROR, gettime(), '".abyss" config file doesn\'t exist.')
  71
  72 if not path.isdir(DIR_NAME):
  73     mkdir(DIR_NAME)
  74
  75 class New_user_pipeline():
  76
  77
  78     def __init__(self, feed='main'):
  79         self.rtsp_address = 'rtsp://' + sources['RTSP_IP']
  80         self.feed = feed
  81         self.user_pipeline = self.create_gstreamer_pipeline()
  82
  83     def create_video_sources(self):
  84         """Create video inputs from various sources."""
  85         self.videosrc = Gst.ElementFactory.make('rtspsrc', 'videosrc')
  86         self.videosrc.set_property('location', self.rtsp_address)
  87         self.videosrc.set_property('latency', 0)
  88 ##        self.videosrc.set_property('debug', True)
  89         if self.feed == 'backup':
  90             self.videosrc_backup = Gst.ElementFactory.make('v4l2src',
  91                                                            'videosrc_backup')
  92             device_location = self.find_webcam_device()
  93             self.videosrc_backup.set_property('device', device_location)
  94
  95     def find_webcam_device(self):
  96         """Look out for the USB webcam device."""
  97         devices = [dev for dev in listdir('/dev/') if 'video' in dev]
  98         for item in devices:
  99             # In case of computer having a built-in webcam
 100             if item != 'video0' and len(devices) > 1:
 101                 return '/dev/' + item
 102             # Without built-in webcam
 103             elif len(devices) == 1:
 104                 return '/dev/video0'
 105         print(ERROR, gettime(), 'No webcam device found.')
 106
 107     def find_mixingdesk_device(self):
 108         """Look out for the USB mixing desk device.
 109         Product used here: Behringer XENYX Q1002USB.
 110         """
 111         # shell cmd : 'pactl list | grep alsa_input'
 112         # AUDIO_INPUT --> const used currently
 113         pass
 114
 115     def create_pipeline_callbacks(self):
 116         """Callbacks to connect dynamically created pads."""
 117         self.videosrc.connect('pad-added', self.on_pad_added_to_rtspsrc)
 118
 119     def on_pad_added_to_rtspsrc(self, rtspsrc, pad):
 120         """Connect the dynamic 'src'pad of an RTSP source."""
 121         sinkpad = self.queuev_1.get_static_pad('sink')
 122         pad.link(sinkpad)
 123
 124     def create_audio_sources(self):
 125         """Create audio inputs from various sources."""
 126         self.audiosrc = Gst.ElementFactory.make('pulsesrc', 'audiosrc')
 127         self.audiosrc.set_property('device', sources['AUDIO_INPUT'])
 128
 129     def create_audiolevel_plugin(self):
 130         """Create audio level plugin to feed a vu-meter."""
 131         self.audiolevel = Gst.ElementFactory.make('level', 'audiolevel')
 132         self.audiolevel.set_property('interval', 200000000)
 133
 134     def create_filesink(self):
 135         """Create storable output elements."""
 136         self.disksink_rawvideo = Gst.ElementFactory.make('filesink')
 137         self.disksink_rawvideo.set_property('location', RAWVIDEO_DEFAULT)
 138         self.disksink_audio = Gst.ElementFactory.make('filesink')
 139         self.disksink_audio.set_property('location', AUDIO_DEFAULT)
 140         self.disksink_stream = Gst.ElementFactory.make('filesink')
 141         self.disksink_stream.set_property('location', STREAM_DEFAULT)
 142         if self.feed == 'backup':
 143             self.disksink_rawvideo.set_property('location', RAWVIDEO_BACKUP)
 144             self.disksink_audio.set_property('location', AUDIO_BACKUP)
 145             self.disksink_stream.set_property('location', STREAM_BACKUP)
 146
 147     def create_streamsink(self):
 148         """Create streamable output elements."""
 149         # To local screen:
 150         self.screensink = Gst.ElementFactory.make('xvimagesink', 'screensink')
 151         self.screensink.set_property('sync', False)
 152         # To local audio output (headphones):
 153         self.audiosink = Gst.ElementFactory.make('pulsesink', 'audiosink')
 154         self.audiosink.set_property('device', sinks['AUDIO_OUTPUT'])
 155         self.audiosink.set_property('sync', False)
 156         # To icecast server:
 157         self.icecastsink_audio = Gst.ElementFactory.make('shout2send', 'icecastsink_audio')
 158         self.icecastsink_audio.set_property('sync', False)
 159         self.icecastsink_audio.set_property('ip', sinks['STREAM_SERVER_IP'])
 160         self.icecastsink_audio.set_property('port', int(sinks['SERVER_PORT']))
 161         self.icecastsink_audio.set_property('mount', sinks['AUDIO_MOUNT'])
 162         self.icecastsink_audio.set_property('password', sinks['PASSWORD'])
 163         self.icecastsink_stream = Gst.ElementFactory.make('shout2send', 'icecastsink_stream')
 164         self.icecastsink_stream.set_property('sync', False)
 165         self.icecastsink_stream.set_property('ip', sinks['STREAM_SERVER_IP'])
 166         self.icecastsink_stream.set_property('port', int(sinks['SERVER_PORT']))
 167         self.icecastsink_stream.set_property('mount', sinks['VIDEO_MOUNT'])
 168         self.icecastsink_stream.set_property('password', sinks['PASSWORD'])
 169
 170     def create_payloader_elements(self):
 171         pass
 172
 173     def create_depayloader_elements(self):
 174         self.rtpjpegdepay = Gst.ElementFactory.make('rtpjpegdepay', 'rtpjpegdepay')
 175
 176     def create_encoder_elements(self):
 177         # Audio encoders:
 178         self.vorbisenc = Gst.ElementFactory.make('vorbisenc', 'vorbisenc')
 179         # Video encoders:
 180         self.vp8enc = Gst.ElementFactory.make('vp8enc', 'vp8enc')
 181         self.vp8enc.set_property('min_quantizer', 1)
 182         self.vp8enc.set_property('max_quantizer', 13)
 183         self.vp8enc.set_property('cpu-used', 5)
 184         self.vp8enc.set_property('deadline', 1)
 185 ##        self.vp8enc.set_property('target-bitrate', 2000000)
 186         self.vp8enc.set_property('threads', 2)
 187         self.vp8enc.set_property('sharpness', 7)
 188
 189     def create_decoder_elements(self):
 190         self.jpegdec = Gst.ElementFactory.make('jpegdec', 'jpegdec')
 191         self.jpegdec.set_property('max-errors', -1)
 192
 193     def create_muxer_elements(self):
 194         self.oggmux = Gst.ElementFactory.make('oggmux', 'oggmux')
 195         self.mkvmux = Gst.ElementFactory.make('matroskamux', 'mkvmux')
 196         self.webmmux = Gst.ElementFactory.make('webmmux', 'webmmux')
 197         self.webmmux.set_property('streamable', True)
 198
 199     def create_demuxer_elements(self):
 200         pass
 201
 202     def create_filtering_elements(self):
 203         self.scaling = Gst.ElementFactory.make('videoscale', 'scaling')
 204         caps = Gst.caps_from_string(
 205             'video/x-raw, width=(int)640, height=(int)360, framerate=(float)25/1')
 206         self.capsfilter = Gst.ElementFactory.make('capsfilter', 'capsfilter')
 207         self.capsfilter.set_property('caps', caps)
 208
 209         caps_backup = Gst.caps_from_string('video/x-raw, width=(int)640, height=(int)360')
 210         self.capsfilter_backup = Gst.ElementFactory.make('capsfilter', 'capsfilter_backup')
 211         self.capsfilter_backup.set_property('caps', caps_backup)
 212
 213     def create_tee_elements(self):
 214         """Create tee elements to divide feeds."""
 215         self.tee_rawvideo = Gst.ElementFactory.make('tee', 'tee_rawvideo')
 216         self.tee_videodecoded = Gst.ElementFactory.make('tee', 'tee_videodecoded')
 217         self.tee_streamfull = Gst.ElementFactory.make('tee', 'tee_streamfull')
 218         self.tee_rawaudio = Gst.ElementFactory.make('tee', 'tee_rawaudio')
 219         self.tee_streamaudio = Gst.ElementFactory.make('tee', 'tee_streamaudio')
 220
 221     def connect_tee(self,
 222                     tee_element,
 223                     input_element,
 224                     output_element_1,
 225                     output_element_2,
 226                     output_element_3=None,):
 227         """Links input and outputs of a given tee element."""
 228         # Find a way to check if the element given are in the pipeline
 229         # then pass the result to the 'if' statement.
 230         ##        argcheck = [True for arg in locals() if arg in 'the_list_of_elements_added']
 231         ##        print('[DEBUG] ArgList check: ', argcheck)
 232         ##        if False not in argcheck
 233         if True:
 234             input_element.link(tee_element)
 235             tee_element.link(output_element_1)
 236             tee_element.link(output_element_2)
 237             if output_element_3:
 238                 tee_element.link(output_element_3)
 239         else:
 240             print(ERROR,
 241                   gettime(),
 242                   'Couldn\'t link the tee. Element(s) probably not in the pipeline ')
 243
 244     def create_queues(self):
 245         # For video feed:
 246         self.queuev_1 = Gst.ElementFactory.make('queue', 'queuev_1')
 247         self.queuev_2 = Gst.ElementFactory.make('queue', 'queuev_2')
 248         self.queuev_3 = Gst.ElementFactory.make('queue', 'queuev_3')
 249         self.queuev_4 = Gst.ElementFactory.make('queue', 'queuev_4')
 250         self.queuev_5 = Gst.ElementFactory.make('queue', 'queuev_5')
 251         # For audio feed:
 252         self.queuea_1 = Gst.ElementFactory.make('queue', 'queuea_1')
 253         self.queuea_2 = Gst.ElementFactory.make('queue', 'queuea_2')
 254         self.queuea_3 = Gst.ElementFactory.make('queue', 'queuea_3')
 255         self.queuea_4 = Gst.ElementFactory.make('queue', 'queuea_4')
 256         self.queuea_4.set_property('leaky', 2)
 257         self.queuea_5 = Gst.ElementFactory.make('queue', 'queuea_5')
 258         # For audio+video muxer:
 259         self.queuem_1 = Gst.ElementFactory.make('queue', 'queuem_1')
 260         self.queuem_2 = Gst.ElementFactory.make('queue', 'queuem_2')
 261         self.queuem_2.set_property('leaky', 2)
 262
 263     def create_pipeline_elements(self):
 264         print(INFO, gettime(), 'Pipeline creation state: creating elements... ', end='')
 265         # Inputs elements:
 266         self.create_video_sources()
 267         self.create_audio_sources()
 268         # Middle elements:
 269         self.create_audiolevel_plugin()
 270         self.create_payloader_elements()
 271         self.create_depayloader_elements()
 272         self.create_encoder_elements()
 273         self.create_decoder_elements()
 274         self.create_muxer_elements()
 275         self.create_filtering_elements()
 276         self.create_tee_elements()
 277         self.create_queues()
 278         # Output elements:
 279         self.create_filesink()
 280         self.create_streamsink()
 281         if self.feed == 'test':
 282             print('TEST OK...', end='')
 283         print('created')
 284         if self.feed == 'backup':
 285             print (INFO,
 286                    gettime(),
 287                    'Webcam device location: ',
 288                    self.videosrc_backup.get_property('device'))
 289
 290
 291     def add_elements_to_pipeline(self):
 292         print(INFO, gettime(), 'Pipeline creation state: adding elements... ', end='')
 293         cond = self.feed != 'test'
 294
 295         # Inputs elements:
 296         self.streampipe.add(self.audiosrc)
 297         # Middle elements:
 298         self.streampipe.add(self.audiolevel)
 299         self.streampipe.add(self.queuea_1)
 300         self.streampipe.add(self.queuev_3)
 301         if cond:
 302             self.streampipe.add(self.vorbisenc)
 303             self.streampipe.add(self.oggmux)
 304             self.streampipe.add(self.queuea_2)
 305             self.streampipe.add(self.queuea_3)
 306             self.streampipe.add(self.vp8enc)
 307             self.streampipe.add(self.mkvmux)
 308             self.streampipe.add(self.webmmux)
 309             self.streampipe.add(self.tee_rawaudio)
 310             self.streampipe.add(self.tee_rawvideo)
 311             self.streampipe.add(self.tee_streamaudio)
 312             self.streampipe.add(self.tee_streamfull)
 313             self.streampipe.add(self.queuev_2)
 314             self.streampipe.add(self.queuev_4)
 315             self.streampipe.add(self.queuev_5)
 316             self.streampipe.add(self.queuea_4)
 317             self.streampipe.add(self.queuea_5)
 318             self.streampipe.add(self.queuem_1)
 319             self.streampipe.add(self.queuem_2)
 320         # Outputs elements:
 321         self.streampipe.add(self.screensink)
 322         if cond:
 323             self.streampipe.add(self.disksink_rawvideo)
 324             self.streampipe.add(self.disksink_audio)
 325             self.streampipe.add(self.disksink_stream)
 326             self.streampipe.add(self.icecastsink_audio)
 327             self.streampipe.add(self.icecastsink_stream)
 328         else:
 329             self.streampipe.add(self.audiosink)
 330
 331         if self.feed == 'main' or self.feed == 'test':
 332             # Inputs elements:
 333             self.streampipe.add(self.videosrc)
 334             # Middle elements:
 335             self.streampipe.add(self.rtpjpegdepay)
 336             self.streampipe.add(self.jpegdec)
 337             self.streampipe.add(self.scaling)
 338             self.streampipe.add(self.capsfilter)
 339             self.streampipe.add(self.tee_videodecoded)
 340             self.streampipe.add(self.queuev_1)
 341             if self.feed == 'test':
 342                 print ('TEST OK...', end='')
 343         elif self.feed == 'backup':
 344             # Inputs elements:
 345             self.streampipe.add(self.videosrc_backup)
 346             # Middle elements:
 347             self.streampipe.add(self.capsfilter_backup)
 348             print ('BACKUP OK...', end='')
 349         print('added')
 350
 351     def link_pipeline_elements(self):
 352         """Link all elements with static pads."""
 353         print(INFO, gettime(), 'Pipeline creation state: linking elements... ', end='')
 354         cond = self.feed != 'test'
 355
 356         # Audio feed:
 357         self.audiosrc.link(self.audiolevel)
 358         self.audiolevel.link(self.queuea_1)
 359         if cond:
 360             self.queuea_1.link(self.vorbisenc)
 361             self.connect_tee(self.tee_rawaudio,
 362                              self.vorbisenc,
 363                              self.queuea_2,
 364                              self.queuea_5,)
 365             self.queuea_2.link(self.oggmux)
 366             self.connect_tee(self.tee_streamaudio,
 367                              self.oggmux,
 368                              self.queuea_3,
 369                              self.queuea_4,)
 370             self.queuea_3.link(self.disksink_audio)
 371             self.queuea_4.link(self.icecastsink_audio)
 372             self.queuea_5.link(self.webmmux)
 373         else:
 374             self.queuea_1.link(self.audiosink)
 375
 376         # Video feed:
 377         if cond:
 378             self.queuev_2.link(self.mkvmux)
 379             self.mkvmux.link(self.queuev_4)
 380             self.queuev_4.link(self.disksink_rawvideo)
 381         else:
 382             self.queuev_1.link(self.rtpjpegdepay)
 383             self.rtpjpegdepay.link(self.jpegdec)
 384             self.jpegdec.link(self.queuev_3)
 385         self.queuev_3.link(self.screensink)
 386
 387         # Stream (audio+video) feed:
 388         if cond:
 389             self.vp8enc.link(self.queuev_5)
 390             self.queuev_5.link(self.webmmux)
 391             self.connect_tee(self.tee_streamfull,
 392                              self.webmmux,
 393                              self.queuem_1,
 394                              self.queuem_2,)
 395             self.queuem_1.link(self.disksink_stream)
 396             self.queuem_2.link(self.icecastsink_stream)
 397             if self.feed == 'main':
 398                 # linking here RTSP feed
 399                 self.queuev_1.link(self.rtpjpegdepay)
 400                 self.connect_tee(self.tee_rawvideo,
 401                                  self.rtpjpegdepay,
 402                                  self.queuev_2,
 403                                  self.jpegdec,)
 404                 self.connect_tee(self.tee_videodecoded,
 405                                  self.jpegdec,
 406                                  self.queuev_3,
 407                                  self.scaling,)
 408                 # Stream (video) feed:
 409                 self.scaling.link(self.capsfilter)
 410                 self.capsfilter.link(self.vp8enc)
 411             elif self.feed == 'backup':
 412                 # linking here backup feed (WEBCAM)
 413                 self.videosrc_backup.link(self.capsfilter_backup)
 414                 self.connect_tee(self.tee_rawvideo,
 415                                  self.capsfilter_backup,
 416                                  self.queuev_2,
 417                                  self.queuev_3,
 418                                  output_element_3=self.vp8enc)
 419                 print('BACKUP OK...', end='')
 420         if not cond:
 421             print('TEST OK...', end='')
 422         print('linked')
 423
 424     def create_gstreamer_pipeline(self):
 425         # New empty pipeline:
 426         self.streampipe = Gst.Pipeline()
 427         self.create_pipeline_elements()
 428         # Setting-up:
 429         self.add_elements_to_pipeline()
 430         self.link_pipeline_elements()
 431         if self.feed == 'main' or self.feed == 'test':
 432             self.create_pipeline_callbacks()
 433
 434         global bus
 435         bus = self.streampipe.get_bus()
 436         bus.add_signal_watch()
 437         bus.enable_sync_message_emission()
 438         # Used to get messages that GStreamer emits.
 439         bus.connect("message", self.on_message)
 440
 441         print(INFO, gettime(), 'Pipeline creation state: successfully done.')
 442         return self.streampipe
 443
 444     def on_message(self, bus, message):
 445         t = message.type
 446         if t == Gst.MessageType.EOS:
 447             self.streampipe.set_state(Gst.State.NULL)
 448         elif t == Gst.MessageType.ERROR:
 449             err, debug = message.parse_error()
 450             print (ERROR, '%s' % err, debug)
 451
 452     def stream_play(self):
 453         self.streampipe.set_state(Gst.State.PLAYING)
 454         if self.feed == 'backup':
 455             print(WARN, gettime(), 'Backup pipeline started.')
 456         print(INFO, gettime(), 'PLAYING State resquested.')
 457
 458     def stream_stop(self):
 459         self.streampipe.set_state(Gst.State.NULL)
 460         print(INFO, gettime(), 'STOPPED State resquested.')
 461
 462     def set_filenames(self, string, streamfailed=False):
 463         """Sets filename and location for each sink."""
 464         global fail_counter
 465         filename = string
 466         audio = './' + DIR_NAME + '/' + filename + '_AUDIO'
 467         rawvideo = './' + DIR_NAME + '/' + filename + '_RAWVIDEO'
 468         stream = './' + DIR_NAME + '/' + filename + '_STREAM'
 469         if self.feed == 'main':
 470             if streamfailed and filename:
 471                 audio = audio + FAILED_SUFFIX + str(fail_counter)
 472                 rawvideo = rawvideo + FAILED_SUFFIX + str(fail_counter)
 473                 stream = stream + FAILED_SUFFIX + str(fail_counter)
 474                 self.rename_files(audio, rawvideo, stream)
 475                 fail_counter += 1
 476             elif streamfailed:
 477                 audio = AUDIO_DEFAULT + FAILED_SUFFIX + str(fail_counter)
 478                 rawvideo = RAWVIDEO_DEFAULT + FAILED_SUFFIX + str(fail_counter)
 479                 stream = STREAM_DEFAULT + FAILED_SUFFIX + str(fail_counter)
 480                 self.rename_files(audio, rawvideo, stream)
 481                 fail_counter += 1
 482             else:
 483                 self.rename_files(audio, rawvideo, stream)
 484         elif self.feed == 'backup':
 485 ##            print('INSIDE BACKUP RENAMING')
 486             rename(AUDIO_BACKUP, audio)
 487             rename(RAWVIDEO_BACKUP, rawvideo)
 488             rename(STREAM_BACKUP, stream)
 489
 490         print(INFO, gettime(), 'Audio file written on disk.')
 491         print(INFO, gettime(), 'Raw video file written on disk.')
 492         print(INFO, gettime(), 'Streamed file written on disk.')
 493
 494     def rename_files(self, audio_name, rawvideo_name, stream_name):
 495         rename(AUDIO_DEFAULT, audio_name)
 496         rename(RAWVIDEO_DEFAULT, rawvideo_name)
 497         rename(STREAM_DEFAULT, stream_name)
 498
 499 def get_gstreamer_bus():
 500     return bus
 501
 502 def gettime():
 503     return strftime('%y-%m-%d_%H:%M:%S ', localtime())