1+ __author__ = 'ddragosd'
2+ # coding=utf-8
3+
4+ """
5+ Controller for the FFmpeg process
6+ """
7+
8+ import subprocess
9+ import json
10+ from urllib2 import Request , urlopen , URLError
11+ import logging
12+ import re
13+ import time
14+ import datetime
15+ from configobj import ConfigObj
16+ import os
17+ import argparse
18+
19+
20+ class LiveTranscoder :
21+ def __init__ (self ):
22+ # initialize config
23+ # Initialize Blank Configs
24+ self .config = ConfigObj ()
25+
26+ # Initialize Logger
27+ self .log = logging .getLogger ("LiveTranscoder" )
28+ self .log .setLevel (logging .DEBUG )
29+ log_format = logging .Formatter ("%(asctime)s %(name)s [%(levelname)s]: %(message)s" )
30+ log_filename = "/var/log/streamkit/live_transcoder_%s.log" % \
31+ (datetime .datetime .fromtimestamp (time .time ()).strftime ('%Y-%m-%d_%H_%M_%S' ))
32+ try :
33+ file_handler = logging .FileHandler (log_filename )
34+ file_handler .setFormatter (log_format )
35+ self .log .addHandler (file_handler )
36+ except :
37+ pass
38+
39+ # when executing the collector separately, log directly on the output stream
40+ stream_handler = logging .StreamHandler ()
41+ stream_handler .setFormatter (log_format )
42+ self .log .addHandler (stream_handler )
43+
44+ logging .getLogger ('LiveTranscoder' ).addHandler (stream_handler )
45+
46+ def _get_default_config (self , file_name = '/etc/live-transcoder/default_config.json' ):
47+ config_file = open (file_name )
48+ cfg = json .load (config_file )
49+ config_file .close ()
50+
51+ return cfg
52+
53+ def _get_user_config (self , user_config_json ):
54+ try :
55+ self .log .info ("Loading user-data: %s" , user_config_json )
56+ config = json .loads (user_config_json )
57+ self .log .info ("User-Data: %s" , json .dumps (config ))
58+ return config
59+ except Exception , e :
60+ self .log .exception (e )
61+ return None
62+
63+
64+ def _updateStreamMetadataInConfig (self , config ):
65+ """
66+ Reads config.source metadata (width, height, bitrate, HD) and adds it back into the config object
67+ """
68+ self .log .info ("Reading metadata for %s" , config .get ("source" ))
69+ cfg = config
70+
71+ proc = subprocess .Popen (["ffmpeg" , "-i" , config .get ("source" )], stdout = subprocess .PIPE , stderr = subprocess .PIPE )
72+
73+ stdoutdata , stderrdata = proc .communicate ()
74+ ffmpeg_response = stderrdata
75+
76+ self .log .debug ("FFMPEG result %s" , ffmpeg_response )
77+ regex = re .compile ("bitrate:\s(\d+)\s" , re .IGNORECASE | re .MULTILINE | re .DOTALL )
78+ bitrate = regex .search (ffmpeg_response )
79+ if bitrate is not None :
80+ bitrate = bitrate .group (1 )
81+ cfg ["bitrate" ] = bitrate
82+ self .log .info ("Source bitrate: %s" , bitrate )
83+
84+ regex = re .compile ("\s(\d+)x(\d+)\s" , re .IGNORECASE | re .MULTILINE | re .DOTALL )
85+ size = regex .search (ffmpeg_response )
86+ width = 1
87+ height = 1
88+ ratio = 0
89+ isHD = False
90+ if size is not None :
91+ width = int (size .group (1 ))
92+ height = int (size .group (2 ))
93+ ratio = round (width / float (height ), 2 )
94+ isHD = ratio == 1.77
95+ self .log .info ("Source size: width=%d, height=%d, ratio=%.4f, HD=%s" , width , height , ratio , isHD )
96+
97+ cfg ["width" ] = width
98+ cfg ["height" ] = height
99+ cfg ["HD" ] = isHD
100+
101+ return cfg
102+
103+
104+ def _getTranscodingCmd (self , config ):
105+ bitrates = None
106+ if config ["HD" ] == True :
107+ bitrates = config ["hd_streams" ]
108+ else :
109+ bitrates = config ["sd_streams" ]
110+
111+ cmd = 'ffmpeg -i %s ' % (config ["source" ])
112+ sub_commands = []
113+ for quality in bitrates :
114+ if int (quality ["bitrate" ]) <= int (config ["bitrate" ]):
115+ sub_cmd_template = """-f flv -c:a copy -c:v libx264 -s %dx%d -x264opts bitrate=%d -rtmp_playpath %s -rtmp_app %s %s """
116+ sub_cmd_template_audio = """-f flv -c:a copy -b:a %dk -c:v libx264 -s %dx%d -x264opts bitrate=%d -rtmp_playpath %s -rtmp_app %s %s """
117+ target_stream = config ["target_stream" ]
118+ target_stream = target_stream .replace ("$width" , str (quality ["width" ]))
119+ target_stream = target_stream .replace ("$height" , str (quality ["height" ]))
120+ target_stream = target_stream .replace ("$bitrate" , str (quality ["bitrate" ]))
121+ sub_cmd = ''
122+ if "audio_bitrate" in quality :
123+ sub_cmd = sub_cmd_template_audio % (
124+ quality ["audio_bitrate" ], quality ["width" ], quality ["height" ], quality ["bitrate" ],
125+ target_stream ,
126+ config ["target_app" ], config ["target_host" ] )
127+ else :
128+ sub_cmd = sub_cmd_template % (
129+ quality ["width" ], quality ["height" ], quality ["bitrate" ], target_stream ,
130+ config ["target_app" ], config ["target_host" ] )
131+ cmd = cmd + sub_cmd
132+ return cmd
133+
134+ def _runTranscodingCommand (self , command_with_args ):
135+ try :
136+ s = subprocess .Popen (command_with_args , stdout = subprocess .PIPE , stderr = subprocess .STDOUT )
137+ while True :
138+ line = s .stdout .readline ()
139+ if not line :
140+ break
141+ self .log .info (line )
142+ self .log .info ("FFmpeg process stopped normally." )
143+ return 1
144+ except Exception , e :
145+ # Note that FFMpeg will always exit with error code for live transcoding
146+ self .log .info ("FFmpeg process stopped. Reason:" )
147+ self .log .exception (e )
148+ return - 1
149+
150+ def startLiveTranscoding (self , user_config_json ):
151+ # Load default
152+ self .config .merge (self ._get_default_config ())
153+ # Merge with user data
154+ user_config = self ._get_user_config (user_config_json )
155+ if user_config is not None :
156+ self .config .merge (user_config )
157+ self .log .info ("Running live-transcoder with configuration: %s" , self .config )
158+
159+ max_retries = int (self .config ["max_retries" ])
160+ max_retries_delay = int (self .config ["max_retries_delay_sec" ])
161+ for i in range (1 , max_retries + 1 ):
162+ self .config = self ._updateStreamMetadataInConfig (self .config )
163+
164+ cmd = self ._getTranscodingCmd (self .config )
165+ self .log .info ("Executing FFmpeg command:\n %s\n " , cmd )
166+
167+ # start live transcoding
168+ cmd_args = cmd .split ()
169+ self .log .info ("Running command. (run=%d/%d)" , i , max_retries )
170+ r = self ._runTranscodingCommand (cmd_args )
171+ self .log .info ("Transcoding command stopped. (run=%d/%d). Code=%d" , i , max_retries , r )
172+ time .sleep (max_retries_delay )
173+
174+ self .log .info ("Live-Transcoder has completed ! You can now shutdown the instance." )
175+
176+
177+ transcoder = LiveTranscoder ()
178+ user_config_json = None
179+ if __name__ == '__main__' :
180+ parser = argparse .ArgumentParser ()
181+ parser .add_argument ('-u' , '--user-config-json' , dest = 'user_config_json' )
182+ args = parser .parse_args ()
183+ user_config_json = args .user_config_json
184+
185+ transcoder .startLiveTranscoding (user_config_json )
0 commit comments