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