cis/cisd.py

   1 #!/usr/bin/env python
   2
   3 import sys
   4 import os
   5 import shutil
   6 import time
   7 import threading
   8 from Queue import Queue
   9
  10 import config
  11 import bt
  12
  13
  14 class CIWorker(threading.Thread):
  15     """
  16     Content Ingestion Worker. A class which executes content ingestion jobs
  17     on a separate thread.
  18
  19     CIWorker shares a Queue with its master where jobs are submitted.
  20     """
  21
  22     raw_videos_dir = 'tmp/raw'
  23     transcoded_videos_dir = 'tmp/media'
  24     thumbs_dir = 'tmp/thumbs'
  25     torrents_dir = 'tmp/torrents'
  26
  27     def __init__(self, queue, bit_torrent):
  28         """
  29         Initialize Content Ingestion Worker.
  30
  31         @param queue a list of dictionaries with the following keys:
  32         <ul>
  33             <li>raw_video</li>
  34             <li>name: a video name which must be a valid file name</li>
  35             <li>transcode_configs: a list of transcode configuration
  36             dictionaries having the keys as the parameters of
  37             api.BaseTranscoder.transcode(...)</li>
  38             <li>thumbs: string 'random' for extracting a thumbnail
  39             image from a random video position or a positive integer which
  40             represents the number of summary thumbnails to be extracted</li>
  41         </ul>
  42         """
  43
  44         threading.Thread.__init__(self, name='CIWorker')
  45
  46         self.queue = queue
  47         self.bit_torrent = bit_torrent
  48
  49     def transfer_in(self, raw_video):
  50         """
  51         Transfers a raw video file from the Web Server.
  52
  53         @param raw_video raw video file name
  54         """
  55
  56         file_transfer = config.FILE_TRANSFERER_CLASS( \
  57                 self.raw_videos_dir, config.INPUT_PATH)
  58         file_transfer.get([raw_video])
  59         file_transfer.close()
  60
  61         print '** Transfering in finished.'
  62
  63     def transcode(self, input_video, video_name, transcode_configs):
  64         """
  65         Transcodes a video in each requested formats.
  66
  67         @param input_video input video file name
  68         @param video_name a video name which must be a valid file name
  69         @param transcode_configs a list of dictionaries with format settings
  70         """
  71
  72         transcoder = config.TRANSCODER_CLASS( \
  73                 input_file = os.path.join(self.raw_videos_dir, input_video), \
  74                 name = video_name, prog_bin = config.TRANSCODER_BIN)
  75         transcoder.dest_path = self.transcoded_videos_dir
  76
  77         # Transcode the raw video in each requested format.
  78         # TODO report partial errors
  79         for transcode_config in transcode_configs:
  80             transcode_config['output_file'] = \
  81                     transcoder.transcode(**transcode_config)
  82
  83         print '** Transcoding finished.'
  84
  85     def extract_thumbs(self, input_video, video_name, thumbs):
  86         """
  87         Extracts thumbnail images from a video.
  88
  89         @param input_video input video file name
  90         @param video_name a video name which must be a valid file name
  91         @param thumbs use 'random' to extract a thumbnail image from a random
  92         point of the video or use a positive integer n to extract n summary
  93         thumbnail
  94         """
  95
  96         # TODO report partial errors
  97         thumb_extractor = config.THUMB_EXTRACTOR_CLASS( \
  98                 input_file = os.path.join(self.raw_videos_dir, input_video), \
  99                 name = video_name, \
 100                 prog_bin = config.THUMB_EXTRACTOR_BIN)
 101         thumb_extractor.dest_path = self.thumbs_dir
 102         if thumbs == 'random':
 103             thumb_extractor.extract_random_thumb()
 104         elif type(thumbs) is int and thumbs > 0:
 105             thumb_extractor.extract_summary_thumbs(thumbs)
 106
 107         print '** Extracting thumbs finished.'
 108
 109     def seed(self, transcode_configs):
 110         """
 111         Creates torrents from the videos passed and then stats seeding them.
 112
 113         @param transcode_configs a list of dictionaries with format settings
 114         """
 115
 116         for transcode_config in transcode_cofigs:
 117             # * CREATE TORRENTS FOR EACH TRANSCODED VIDEO
 118             # Create torrent file.
 119             bt.create_torrent(transcode_config['output_file'])
 120
 121             # The torrent file is created in the same directory with the
 122             # source file. Move it to the torrents directory.
 123             shutil.move(transcode_config['output_file'] + '.tstream', \
 124                     self.torrents_dir)
 125
 126             # * SEED TORRENTS
 127             bit_torrent.start_download( \
 128                     transcode_config['output_file'] + '.tstream',
 129                     self_transcoded_videos_dir)
 130
 131         print '** Creating torrents and seeding finished.'
 132
 133     def transfer_out(self, local_files, local_path, remote_path):
 134         """
 135         Transfers some local files to a remote path of the Web Server.
 136
 137         @param local_files list local files to transfer
 138         @param remote_path destination path on the Web Server
 139         """
 140
 141         file_transfer = config.FILE_TRANSFERER_CLASS( \
 142                 local_path, remote_path)
 143         file_transfer.put(local_files)
 144         file_transfer.close()
 145
 146         print '** Creating torrents and seeding finished.'
 147
 148     def remove_file(self, files, path):
 149         """
 150         Deletes files from a specified path.
 151         """
 152
 153         for f in files:
 154             os.unlink(os.path.join(path, f))
 155
 156         print '** Cleaning up finished.'
 157
 158     def run(self):
 159         while True:
 160             job = self.queue.get()
 161
 162             # * TRANSFER RAW VIDEO IN
 163             self.transfer_in(job['raw_video'])
 164
 165             # * TRANSCODE RAW VIDEO
 166             self.transcode(job['raw_video'], job['name'], \
 167                     job['transcode_configs'])
 168
 169             # * EXTRACT THUMBNAIL IMAGES
 170             if job['thumbs'] != 0:
 171                 self.extract_thumbs(job['raw_video'], job['name'], \
 172                         job['thumbs'])
 173
 174 #            # * CREATE TORRENTS AND START SEEDING OF TRANSCODED VIDEOS
 175 #            self.seed(job['transcode_configs'])
 176 #
 177 #            # Torrent files.
 178 #            files = [f for f in os.listdir(self.torrents_dir) \
 179 #                    if os.path.isfile(os.path.join( \
 180 #                            self.torrents_dir, f))]
 181 #            torrent_files = fnmatch.filter(files, name + "_*")
 182 #
 183 #            # Thumbnail images files.
 184 #            files = [f for f in os.listdir(self.thumbs_dir) \
 185 #                    if os.path.isfile(os.path.join( \
 186 #                            self.thumbs_dir, f))]
 187 #            thumb_files = fnmatch.filter(files, name + "_*")
 188 #
 189 #            # Raw video files.
 190 #            raw_files = [f for f in os.listdir(self.raw_videos_dir) \
 191 #                    if os.path.isfile(os.path.join( \
 192 #                            self.raw_videos_dir, f))]
 193 #
 194 #            # * TRANSFER TORRENTS AND THUMBNAIL IMAGES OUT
 195 #            self.transfer_out(torrent_files, self.torrents_dir, \
 196 #                    config.OUTPUT_TORRENTS_PATH)
 197 #            self.transfer_out(thumb_files, self.thumbs_dir, \
 198 #                    config.OUTPUT_THUMBS_PATH)
 199 #
 200 #            # * CLEANUP RAW VIDEOS AND THUMBNAIL IMAGES
 201 #            self.remove_files(raw_files, self.raw_videos_dir)
 202 #            self.remove_files(thumb_files, self.thumbs_dir)
 203
 204             # * JOB FINISHED
 205             queue.task_done()
 206
 207
 208 if __name__ == '__main__':
 209     # Jobs queue.
 210     queue = Queue()
 211
 212     # The BitTorrent object implements a NextShare (Tribler) BitTorrent client
 213     # for seeding, downloading etc.
 214     bit_torrent = bt.BitTorrent()
 215
 216     # Worker thread.
 217     ci_worker = CIWorker(queue, bit_torrent)
 218     ci_worker.daemon = True
 219     ci_worker.start()
 220
 221     while True:
 222         raw_video = sys.stdin.readline().strip()
 223         if raw_video == 'x':
 224             break
 225
 226         container = 'webm'
 227         a_codec = 'vorbis'
 228         a_bitrate = '128k'
 229         v_codec = 'vp8'
 230         v_bitrate = '480k'
 231         v_resolution = '640x480'
 232
 233         name = raw_video[:raw_video.rindex('.')]
 234         transcode_config = {
 235             'container': container,
 236             'a_codec': a_codec,
 237             'a_bitrate': a_bitrate,
 238             'v_codec': v_codec,
 239             'v_bitrate': v_bitrate,
 240             'v_resolution': v_resolution
 241         }
 242         thumbs = 4
 243
 244         job = {
 245             'raw_video': raw_video,
 246             'name': name,
 247             'transcode_configs': [transcode_config],
 248             'thumbs': thumbs
 249         }
 250
 251         queue.put(job)
 252
 253     queue.join()