audiomangler/task.py

   1 # -*- coding: utf-8 -*-
   2 ###########################################################################
   3 #    Copyright (C) 2008 by Andrew Mahone
   4 #    <andrew.mahone@gmail.com>
   5 #
   6 # Copyright: See COPYING file that comes with this distribution
   7 #
   8 ###########################################################################
   9 import os
  10 import sys
  11 import atexit
  12 from types import FunctionType, GeneratorType
  13 from twisted.internet import threads, defer
  14 from twisted.python import failure
  15
  16 if 'twisted.internet.reactor' not in sys.modules:
  17     for reactor in ('kqreactor','epollreactor','pollreactor','selectreactor'):
  18         try:
  19             r = __import__('twisted.internet.' + reactor, fromlist=[reactor])
  20             r.install()
  21             break
  22         except ImportError: pass
  23
  24 from twisted.internet import reactor
  25
  26 def cleanup():
  27     if reactor.running: reactor.stop()
  28
  29 atexit.register(cleanup)
  30
  31 threadpool = None
  32
  33 class Clear(object):
  34     def __new__(cls,*args,**kw):
  35         return cls
  36
  37 class BaseTask(object):
  38     def __init__(self, target=None, args=(), kwargs=(), background=True):
  39         self.started = False
  40         self.target = target
  41         self.args=list(args)
  42         self.kwargs = dict(kwargs)
  43         self.background = background
  44
  45     def run(self):
  46         self.started = True
  47         if self.background:
  48             return self.runbg()
  49         else:
  50             return self.runfg()
  51
  52 class FuncTask(BaseTask):
  53     def __init__(self, target=None, args=(), kwargs=(), background=True):
  54         BaseTask.__init__(self, target, args, kwargs, background)
  55
  56     def runfg(self):
  57         try:
  58             self.target(*self.args, **self.kwargs)
  59             self.deferred =  defer.succeed(self)
  60             return self.deferred
  61         except:
  62             self.deferred = defer.failure(failure.Failure())
  63             return self.deferred
  64
  65     def runbg(self):
  66         self.deferred = threads.deferToThread(self.target, *self.args, **self.kwargs)
  67         return self.deferred
  68
  69 class CLITask(BaseTask):
  70     def __init__(self, target=None, args=(), stdin=None, stdout=None, stderr=None, kwargs=(), background=True):
  71         args = list(args)
  72         if target:
  73             args.insert(0,target)
  74         BaseTask.__init__(self, target, args, background=background)
  75         self.stdin = stdin
  76         self.stdout = stdout
  77         self.stderr = stderr
  78
  79     def runbg(self):
  80         if isinstance(self.stdin, basestring):
  81             self.stdin = file(self.stdin, 'rb')
  82         if isinstance(self.stdout, basestring):
  83             self.stdout = file(self.stdout, 'wb')
  84         if isinstance(self.stderr, basestring):
  85             self.stderr = file(self.stderr, 'wb')
  86         self.proc = Popen(executable=self.target, args=self.args, stdin=self.stdin, stderr=self.stderr, stdout=self.stdout)
  87
  88     def runfg(self):
  89         self.runbg()
  90         self.wait()
  91
  92     def wait(self):
  93         try:
  94             if not self.started:
  95                 return False
  96             if not hasattr(self,'proc'):
  97                 return False
  98             ret = self.proc.wait()
  99             if ret != 0:
 100                 raise CalledProcessError(ret,self.args)
 101         finally:
 102             for f in (self.stdin, self.stdout, self.stderr):
 103                 if isinstance(f,int) and f >2:
 104                     os.close(f)
 105                 if isinstance(f,file) and f not in (sys.stdin, sys.stdout, sys.stderr):
 106                     f.close()
 107
 108     def poll(self):
 109         if not self.started:
 110             return False
 111         if not hasattr(self,'proc'):
 112             return False
 113         ret = self.proc.poll()
 114         if ret is None:
 115             return False
 116         elif ret != 0:
 117             raise CalledProcessError(ret)
 118
 119 class TaskSet(FuncTask):
 120     def __init__(self, tasks=(), background=True):
 121         FuncTask.__init__(self, background=background)
 122         self.tasks = list(tasks)
 123
 124     def clear(self):
 125         while self.bgprocs:
 126             t = self.bgprocs.pop()
 127             t.wait()
 128
 129     def runfg(self):
 130         self.bgprocs = set()
 131         for t in self.tasks:
 132             if isinstance(t, BaseTask):
 133                 t = t.run()
 134                 if isinstance(t, BaseTask):
 135                     self.bgprocs.add(t)
 136             elif t is Clear:
 137                 self.clear()
 138         self.clear()
 139
 140 __all__ = ['FuncTask','CLITask','TaskSet','reactor']