class IProcessor(Interface):
+
"""Common processor interface"""
@staticmethod
class IEncoder(IProcessor):
+
"""Encoder driver interface. Each encoder is expected to support a specific
format."""
True when end-of-data is reached."""
# implementation: the constructor must always accept the output argument. It may
- # accept extra arguments such as bitrate, depth, etc.., but these must be optionnal
+ # accept extra arguments such as bitrate, depth, etc.., but these must
+ # be optionnal
@staticmethod
def format():
class IDecoder(IProcessor):
+
"""Decoder driver interface. Decoders are different of encoders in that
a given driver may support several input formats, hence this interface doesn't
export any static method, all informations are dynamic."""
def mime_type():
"""Return the mime type corresponding to this decoded format"""
+
class IGrapher(IProcessor):
+
"""Media item visualizer driver interface"""
# implementation: graphers which need to know the total number of frames
class IAnalyzer(IProcessor):
+
"""Media item analyzer driver interface. This interface is abstract, it doesn't
describe a particular type of analyzer but is rather meant to group analyzers.
In particular, the way the result is returned may greatly vary from sub-interface
class IValueAnalyzer(IAnalyzer):
+
"""Interface for analyzers which return a single numeric value from result()"""
def result():
class IEffect(IProcessor):
+
"""Effect processor interface"""
def __init__(self):
@staticmethod
def name():
"""Return the effect name"""
-
__all__ = ['Component', 'MetaComponent', 'implements', 'abstract',
'interfacedoc', 'Interface', 'implementations', 'ComponentError']
+
class Interface(object):
+
"""Marker base class for interfaces."""
+
def implements(*interfaces):
"""Registers the interfaces implemented by a component when placed in the
class header"""
MetaComponent.implements.extend(interfaces)
+
def abstract():
"""Declare a component as abstract when placed in the class header"""
MetaComponent.abstract = True
+
def implementations(interface, recurse=True, abstract=False):
"""Returns the components implementing interface, and if recurse, any of
the descendants of interface. If abstract is True, also return the
find_implementations(interface, recurse, abstract, result)
return result
+
def interfacedoc(func):
if isinstance(func, staticmethod):
- raise ComponentError("@interfacedoc can't handle staticmethod (try to put @staticmethod above @interfacedoc)")
+ raise ComponentError(
+ "@interfacedoc can't handle staticmethod (try to put @staticmethod above @interfacedoc)")
if not func.__doc__:
func.__doc__ = "@interfacedoc"
func._interfacedoc = True
return func
+
class MetaComponent(type):
+
"""Metaclass of the Component class, used mainly to register the interface
declared to be implemented by a component."""
- implementations = []
- implements = []
- abstract = False
+ implementations = []
+ implements = []
+ abstract = False
def __new__(cls, name, bases, d):
new_class = type.__new__(cls, name, bases, d)
member.__doc__ = if_member.__doc__
MetaComponent.implements = []
- MetaComponent.abstract = False
+ MetaComponent.abstract = False
return new_class
+
class Component(object):
+
"""Base class of all components"""
__metaclass__ = MetaComponent
+
def extend_unique(list1, list2):
"""Extend list1 with list2 as list.extend(), but doesn't append duplicates
to list1"""
if item not in list1:
list1.append(item)
+
def find_implementations(interface, recurse, abstract, result):
"""Find implementations of an interface or of one of its descendants and
extend result with the classes found."""
for i in subinterfaces:
find_implementations(i, recurse, abstract, result)
+
class ComponentError(Exception):
pass
class MetaProcessor(MetaComponent):
+
"""Metaclass of the Processor class, used mainly for ensuring that processor
id's are wellformed and unique"""
pass
else:
raise ApiError("%s and %s have the same id: '%s'"
- % (new_class.__name__, _processors[id].__name__, id))
+ % (new_class.__name__, _processors[id].__name__, id))
if not MetaProcessor.valid_id.match(id):
raise ApiError("%s has a malformed id: '%s'"
- % (new_class.__name__, id))
+ % (new_class.__name__, id))
_processors[id] = new_class
class Processor(Component):
+
"""Base component class of all processors
@interfacedoc
def setup(self, channels=None, samplerate=None, blocksize=None,
totalframes=None):
- self.source_channels = channels
- self.source_samplerate = samplerate
- self.source_blocksize = blocksize
- self.source_totalframes = totalframes
+ self.source_channels = channels
+ self.source_samplerate = samplerate
+ self.source_blocksize = blocksize
+ self.source_totalframes = totalframes
# If empty Set default values for input_* attributes
# may be setted by the processor during __init__()
if not hasattr(self, 'input_stepsize'):
self.input_stepsize = self.source_blocksize
-
# default channels(), samplerate() and blocksize() implementations returns
# the source characteristics, but processors may change this behaviour by
# overloading those methods
class FixedSizeInputAdapter(object):
+
"""Utility to make it easier to write processors which require fixed-sized
input buffers."""
channels the number of channels, and pad indicates whether the last block should
be padded with zeros."""
- self.buffer = numpy.empty((buffer_size, channels))
+ self.buffer = numpy.empty((buffer_size, channels))
self.buffer_size = buffer_size
- self.len = 0
- self.pad = pad
+ self.len = 0
+ self.pad = pad
def blocksize(self, input_totalframes):
"""Return the total number of frames that this adapter will output according to the
remaining = len(frames)
while remaining:
- space = self.buffer_size - self.len
+ space = self.buffer_size - self.len
copylen = remaining < space and remaining or space
- src = frames[src_index:src_index + copylen]
+ src = frames[src_index:src_index + copylen]
if self.len == 0 and copylen == self.buffer_size:
# avoid unnecessary copy
buffer = src
remaining -= copylen
src_index += copylen
- self.len += copylen
+ self.len += copylen
if self.len == self.buffer_size:
yield buffer, (eod and not remaining)
"""Return a processor by its id"""
if not _processors.has_key(processor_id):
raise Error("No processor registered with id: '%s'"
- % processor_id)
+ % processor_id)
return _processors[processor_id]
class ProcessPipe(object):
+
"""Handle a pipe of processors
Attributes:
try:
iter(other)
except TypeError:
- raise Error("Can not add this type of object to a pipe: %s", str(other))
+ raise Error(
+ "Can not add this type of object to a pipe: %s", str(other))
for item in other:
self |= item
import threading
class PipeThread(threading.Thread):
+
def __init__(self, process_pipe):
super(PipeThread, self).__init__(name='pipe_thread')
self.process_pipe = process_pipe
raise TypeError('Function only available in streaming mode')
while pipe_thread.is_alive():
- #yield count
+ # yield count
chunk = self._streamer.get_stream_chunk()
if chunk is not None:
yield chunk
# You should have received a copy of the GNU General Public License
# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
class Error(Exception):
+
"""Exception base class for errors in TimeSide."""
+
class ApiError(Exception):
+
"""Exception base class for errors in TimeSide."""
+
class SubProcessError(Error):
+
"""Exception for reporting errors from a subprocess"""
def __init__(self, message, command, subprocess):
# You should have received a copy of the GNU General Public License
# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
+
class Metadata(object):
pass
-
-
projects / timeside.git / commitdiff