plugins/dvdvideo/vobstream.py

   1 import os
   2 import subprocess
   3 from threading import Thread
   4 from dvdtitlestream import DVDTitleStream
   5
   6 import logging
   7 import math
   8 import os
   9 import re
  10 import shutil
  11 import subprocess
  12 import sys
  13 import tempfile
  14 import threading
  15 import time
  16
  17 import lrucache
  18
  19 import config
  20 import metadata
  21 import virtualdvd
  22
  23 logger = logging.getLogger('pyTivo.dvdvideo.vobstream')
  24
  25 info_cache = lrucache.LRUCache(1000)
  26 mpgcat_procs = {}
  27 reapers = {}
  28
  29 BLOCKSIZE = 512 * 1024
  30 MAXBLOCKS = 2
  31 TIMEOUT = 600
  32 USE_FFMPEG = True
  33
  34 # XXX BIG HACK
  35 # subprocess is broken for me on windows so super hack
  36 def patchSubprocess():
  37     o = subprocess.Popen._make_inheritable
  38
  39     def _make_inheritable(self, handle):
  40         if not handle: return subprocess.GetCurrentProcess()
  41         return o(self, handle)
  42
  43     subprocess.Popen._make_inheritable = _make_inheritable
  44
  45 mswindows = (sys.platform == "win32")
  46 if mswindows:
  47     patchSubprocess()
  48
  49 def debug(msg):
  50     if type(msg) == str:
  51         try:
  52             msg = msg.decode('utf8')
  53         except:
  54             if sys.platform == 'darwin':
  55                 msg = msg.decode('macroman')
  56             else:
  57                 msg = msg.decode('iso8859-1')
  58     logger.debug(msg)
  59
  60 def WriteSectorStreamToSubprocess( fhin, sub, event, blocksize ):
  61
  62     # Write all the data till either end is closed or done
  63     while not event.isSet():
  64
  65         # Read in the block and escape if we got nothing
  66         data = fhin.read( blocksize )
  67         if len(data) == 0:
  68             break
  69
  70         if sub.poll() != None and sub.stdin != None:
  71             break
  72
  73         # Write the data and flush it
  74         try:
  75             sub.stdin.write( data )
  76             sub.stdin.flush()
  77         except IOError:
  78             break
  79
  80         # We got less data so we must be at the end
  81         if len(data) < blocksize:
  82             break
  83
  84     # Close the input if it's not already closed
  85     if not fhin.closed:
  86         fhin.close()
  87
  88     # Close the output if it's not already closed
  89     if sub.stdin != None and not sub.stdin.closed:
  90         sub.stdin.close()
  91
  92 def vobstream(isQuery, inFile, outFile, tsn=''):
  93     settings = {'TBD': 'TBD'}
  94
  95     if isQuery:
  96         return settings
  97
  98     ffmpeg_path = config.get_bin('ffmpeg')
  99
 100     dvd = virtualdvd.VirtualDVD( inFile )
 101     title = dvd.FileTitle()
 102     ts = DVDTitleStream( title.Stream() )
 103
 104     vinfo = video_info( inFile )
 105     vmap = vinfo['mapVideo'].replace( '.', ':' )
 106     amap = vinfo['mapAudio'].replace( '.', ':' )
 107
 108     if USE_FFMPEG:
 109         sp = subprocess.Popen( [ ffmpeg_path, '-i', '-', \
 110             '-map', vmap, '-map', amap,
 111             '-acodec', 'copy', '-vcodec', 'copy', '-f', 'vob', '-' ], \
 112             stdout = subprocess.PIPE, \
 113             stdin = subprocess.PIPE, \
 114             bufsize = BLOCKSIZE * MAXBLOCKS )
 115
 116         # Make an event to shutdown the thread
 117         sde = threading.Event()
 118         sde.clear()
 119
 120         # Stream data to the subprocess
 121         t = Thread( target=WriteSectorStreamToSubprocess, args=(ts,sp,sde,BLOCKSIZE) )
 122         t.start()
 123
 124     mpgcat_procs[inFile] = {'stream': ts, 'start': 0, 'end': 0, \
 125                             'thread': t, 'process':sp, 'event':sde, \
 126                             'last_read': time.time(), 'blocks': []}
 127
 128     reap_process(inFile)
 129     transfer_blocks(inFile, outFile)
 130
 131 def is_resumable(inFile, offset):
 132     if inFile in mpgcat_procs:
 133         proc = mpgcat_procs[inFile]
 134         if proc['start'] <= offset < proc['end']:
 135             return True
 136         else:
 137             cleanup(inFile)
 138     return False
 139
 140 def resume_transfer(inFile, outFile, offset):
 141     proc = mpgcat_procs[inFile]
 142     offset -= proc['start']
 143     try:
 144         for block in proc['blocks']:
 145             length = len(block)
 146             if offset < length:
 147                 if offset > 0:
 148                     block = block[offset:]
 149                 outFile.write('%x\r\n' % len(block))
 150                 outFile.write(block)
 151                 outFile.write('\r\n')
 152             offset -= length
 153         outFile.flush()
 154     except Exception, msg:
 155         logger.info(msg)
 156         return
 157     proc['start'] = proc['end']
 158     proc['blocks'] = []
 159
 160     transfer_blocks(inFile, outFile)
 161
 162 def transfer_blocks(inFile, outFile):
 163     proc = mpgcat_procs[inFile]
 164     blocks = proc['blocks']
 165
 166     while True:
 167         try:
 168             if USE_FFMPEG:
 169                 block = proc['process'].stdout.read(BLOCKSIZE)
 170             else:
 171                 block = proc['stream'].read(BLOCKSIZE)
 172             proc['last_read'] = time.time()
 173         except Exception, msg:
 174             logger.info(msg)
 175             cleanup(inFile)
 176             break
 177
 178         if not block or len(block) == 0:
 179             try:
 180                 outFile.flush()
 181                 proc['stream'].close()
 182             except Exception, msg:
 183                 logger.info(msg)
 184             else:
 185                 cleanup(inFile)
 186             break
 187
 188         blocks.append(block)
 189         proc['end'] += len(block)
 190         if len(blocks) > MAXBLOCKS:
 191             proc['start'] += len(blocks[0])
 192             blocks.pop(0)
 193
 194         try:
 195             outFile.write('%x\r\n' % len(block))
 196             outFile.write(block)
 197             outFile.write('\r\n')
 198         except Exception, msg:
 199             logger.info(msg)
 200             break
 201
 202 def reap_process(inFile):
 203     if inFile in mpgcat_procs:
 204         proc = mpgcat_procs[inFile]
 205         if proc['last_read'] + TIMEOUT < time.time():
 206             cleanup(inFile)
 207
 208         else:
 209             reaper = threading.Timer(TIMEOUT, reap_process, (inFile,))
 210             reapers[inFile] = reaper
 211             reaper.start()
 212
 213 def cleanup(inFile):
 214
 215     # Don't fear the reaper
 216     try:
 217         reapers[inFile].cancel()
 218         del reapers[inFile]
 219     except:
 220         pass
 221
 222     if USE_FFMPEG:
 223         kill(mpgcat_procs[inFile]['process'])
 224         mpgcat_procs[inFile]['process'].wait()
 225
 226     # Tell thread to break out of loop
 227     mpgcat_procs[inFile]['event'].set()
 228     mpgcat_procs[inFile]['thread'].join()
 229
 230     del mpgcat_procs[inFile]
 231
 232 def supported_format( inFile ):
 233     dvd = virtualdvd.VirtualDVD( inFile )
 234     return dvd.Valid() and dvd.file_id != -1
 235
 236 def size(inFile):
 237     try:
 238         dvd = virtualdvd.VirtualDVD( inFile )
 239         return dvd.FileTitle().Size()
 240     except:
 241         return 0
 242
 243 def video_info(inFile, audio_spec = "", cache=True):
 244     vInfo = dict()
 245     try:
 246         mtime = os.stat(inFile).st_mtime
 247     except:
 248         mtime = 0
 249
 250     if cache:
 251         if inFile in info_cache and info_cache[inFile][0] == mtime:
 252             debug('CACHE HIT! %s' % inFile)
 253             return info_cache[inFile][1]
 254
 255     dvd = virtualdvd.VirtualDVD( inFile )
 256     if not dvd.Valid() or dvd.file_id == -1:
 257         debug('Not a valid dvd file')
 258         return dict()
 259
 260     ffmpeg_path = config.get_bin('ffmpeg')
 261
 262     title = dvd.FileTitle()
 263     sid = title.FindBestAudioStreamID( audio_spec )
 264     ts = DVDTitleStream( title.Stream() )
 265     ts.seek(0)
 266
 267     # Make a subprocess to get the information from a stream
 268     proc = subprocess.Popen( [ ffmpeg_path, '-i', '-' ], \
 269         stdout=subprocess.PIPE, \
 270         stdin=subprocess.PIPE, \
 271         stderr=subprocess.STDOUT, \
 272         bufsize=BLOCKSIZE * MAXBLOCKS )
 273
 274     # Make an event to shutdown the thread
 275     sde = threading.Event()
 276     sde.clear()
 277
 278     # Stream data to the subprocess
 279     t = Thread( target=WriteSectorStreamToSubprocess, args=(ts,proc,sde,BLOCKSIZE) )
 280     t.start()
 281
 282     # Readin the output from the subprocess
 283     output = ""
 284     while 1:
 285
 286         # Don't throw on any IO errors
 287         try:
 288             data = proc.stdout.read( BLOCKSIZE )
 289         except IOError:
 290             break
 291
 292         # If we're blank, then the data stream is empty
 293         if len(data) == 0:
 294             break
 295
 296         # append the output
 297         output += data
 298
 299     # Shutdown the helper threads/processes
 300     sde.set()
 301     proc.wait()
 302     t.join()
 303
 304     # Close the title stream
 305     ts.close()
 306
 307     #print "VOB Info:", output
 308     vInfo['mapAudio'] = ''
 309
 310     attrs = {'container': r'Input #0, ([^,]+),',
 311              'vCodec': r'.*Video: ([^,]+),.*',             # video codec
 312              'aKbps': r'.*Audio: .+, (.+) (?:kb/s).*',     # audio bitrate
 313              'aCodec': r'.*Audio: ([^,]+),.*',             # audio codec
 314              'aFreq': r'.*Audio: .+, (.+) (?:Hz).*',       # audio frequency
 315              'mapVideo': r'([0-9]+\.[0-9]+).*: Video:.*',  # video mapping
 316              'mapAudio': r'([0-9]+\.[0-9]+)\[0x%02x\]: Audio:.*' % sid } # Audio mapping
 317
 318     for attr in attrs:
 319         rezre = re.compile(attrs[attr])
 320         x = rezre.search(output)
 321         if x:
 322             #print attr, attrs[attr], x.group(1)
 323             vInfo[attr] = x.group(1)
 324         else:
 325             #print attr, attrs[attr], '(None)'
 326             if attr in ['container', 'vCodec']:
 327                 vInfo[attr] = ''
 328                 vInfo['Supported'] = False
 329             else:
 330                 vInfo[attr] = None
 331             #print '***************** failed at ' + attr + ' : ' + attrs[attr]
 332             debug('failed at ' + attr)
 333
 334     # Get the Pixel Aspect Ratio
 335     rezre = re.compile(r'.*Video: .+PAR ([0-9]+):([0-9]+) DAR [0-9:]+.*')
 336     x = rezre.search(output)
 337     if x and x.group(1) != "0" and x.group(2) != "0":
 338         vInfo['par1'] = x.group(1) + ':' + x.group(2)
 339         vInfo['par2'] = float(x.group(1)) / float(x.group(2))
 340     else:
 341         vInfo['par1'], vInfo['par2'] = None, None
 342
 343     # Get the Display Aspect Ratio
 344     rezre = re.compile(r'.*Video: .+DAR ([0-9]+):([0-9]+).*')
 345     x = rezre.search(output)
 346     if x and x.group(1) != "0" and x.group(2) != "0":
 347         vInfo['dar1'] = x.group(1) + ':' + x.group(2)
 348     else:
 349         vInfo['dar1'] = None
 350
 351     # Get the video dimensions
 352     rezre = re.compile(r'.*Video: .+, (\d+)x(\d+)[, ].*')
 353     x = rezre.search(output)
 354     if x:
 355         vInfo['vWidth'] = int(x.group(1))
 356         vInfo['vHeight'] = int(x.group(2))
 357     else:
 358         vInfo['vWidth'] = ''
 359         vInfo['vHeight'] = ''
 360         vInfo['Supported'] = False
 361         debug('failed at vWidth/vHeight')
 362
 363     vInfo['millisecs'] = title.Time().MSecs()
 364     vInfo['Supported'] = True
 365
 366     if cache:
 367         info_cache[inFile] = (mtime, vInfo)
 368
 369     return vInfo
 370
 371 def kill(popen):
 372     debug('killing pid=%s' % str(popen.pid))
 373     if mswindows:
 374         win32kill(popen.pid)
 375     else:
 376         import os, signal
 377         for i in xrange(3):
 378             debug('sending SIGTERM to pid: %s' % popen.pid)
 379             os.kill(popen.pid, signal.SIGTERM)
 380             time.sleep(.5)
 381             if popen.poll() is not None:
 382                 debug('process %s has exited' % popen.pid)
 383                 break
 384         else:
 385             while popen.poll() is None:
 386                 debug('sending SIGKILL to pid: %s' % popen.pid)
 387                 os.kill(popen.pid, signal.SIGKILL)
 388                 time.sleep(.5)
 389
 390 def win32kill(pid):
 391     import ctypes
 392     handle = ctypes.windll.kernel32.OpenProcess(1, False, pid)
 393     ctypes.windll.kernel32.TerminateProcess(handle, -1)
 394     ctypes.windll.kernel32.CloseHandle(handle)