# -*- coding: utf-8 -*-
-from timeside.analyze.api import *
from timeside.analyze.core import *
from timeside.analyze.channels import *
from timeside.analyze.format import *
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2007-2009 Guillaume Pellerin <yomguy@parisson.com>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Guillaume Pellerin <yomguy@parisson.com>
-
-from timeside.api import IProcessor
-
-class IAnalyzer(IProcessor):
- """Media item analyzer driver interface"""
-
- @staticmethod
- def name():
- """Return the analyzer name, such as "Mean Level", "Max level",
- "Total length, etc..
- """
-
- @staticmethod
- def unit():
- """Return the unit of the data such as "dB", "seconds", etc...
- """
-
- def render(self, media, options=None):
- """Return the result data of the process"""
-
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class ChannelAnalyser(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class MeanDCShiftAnalyser(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
import datetime
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class EncodingAnalyser(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class FormatAnalyser(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class MaxLevelAnalyzer(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class MeanLevelAnalyser(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class ResolutionAnalyser(AudioProcessor):
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.analyze.core import *
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
import numpy
class SampleRateAnalyzer(AudioProcessor):
from timeside.core import *
from tempfile import NamedTemporaryFile
-from timeside.analyze.api import IAnalyzer
+from timeside.api import IAnalyzer
+from timeside.exceptions import SubProcessError
import os
import random
import subprocess
stdout = subprocess.PIPE,
close_fds = True)
except:
- raise VampProcessError('Command failure:', command, proc)
+ raise SubProcessError('Command failure:', command, proc)
# Core processing
while True:
__chunk = proc.stdout.read(buffer_size)
status = proc.poll()
if status != None and status != 0:
- raise VampProcessError('Command failure:', command, proc)
+ raise SubProcessError('Command failure:', command, proc)
if len(__chunk) == 0:
break
yield __chunk
-class VampProcessError(TimeSideError):
-
- def __init__(self, message, command, subprocess):
- self.message = message
- self.command = str(command)
- self.subprocess = subprocess
-
- def __str__(self):
- if self.subprocess.stderr != None:
- error = self.subprocess.stderr.read()
- else:
- error = ''
- return "%s ; command: %s; error: %s" % (self.message,
- self.command,
- error)
# -*- coding: utf-8 -*-
#
-# Copyright (c) 2009 Olivier Guilyardi <olivier@samalyse.com>
+# Copyright (C) 2007-2009 Parisson
+# Copyright (c) 2007 Olivier Guilyardi <olivier@samalyse.com>
+# Copyright (c) 2007-2009 Guillaume Pellerin <pellerin@parisson.com>
#
# This file is part of TimeSide.
from timeside.component import Interface
class IProcessor(Interface):
+ """Base processor interface"""
@staticmethod
def id():
Typically this identifier is likely to be used during HTTP requests
and be passed as a GET parameter. Thus it should be as short as possible."""
+class IEncoder(IProcessor):
+ """Encoder driver interface"""
+
+ def __init__(self, output, nchannels, samplerate):
+ """The constructor must always accept the output, nchannels and samplerate
+ arguments. It may accepts extra arguments such as bitrate, depth, etc..,
+ but these must be optionnal, that is have a default value.
+
+ The output must either be a filepath or a callback function/method for
+ for the streaming mode. The callback must accept one argument which is
+ block of binary data.
+ """
+
+ @staticmethod
+ def format():
+ """Return the encode/encoding format as a short string
+ Example: "MP3", "OGG", "AVI", ...
+ """
+
+ @staticmethod
+ def description():
+ """Return a string describing what this encode format provides, is good
+ for, etc... The description is meant to help the end user decide what
+ format is good for him/her
+ """
+
+ @staticmethod
+ def file_extension():
+ """Return the filename extension corresponding to this encode format"""
+
+ @staticmethod
+ def mime_type():
+ """Return the mime type corresponding to this encode format"""
+
+ def set_metadata(self, metadata):
+ """metadata is a tuple containing tuples for each descriptor return by
+ the dc.Ressource of the item, in the model order :
+ ((name1, value1),(name2, value2),(name1, value3), ...)"""
+
+ def update(self):
+ """Updates the metadata into the file passed as the output argument
+ to the constructor. This method can't be called in streaming
+ mode."""
+
+ def process(self, frames):
+ """Encode the frames passed as a numpy array, where columns are channels.
+
+ In streaming mode the callback passed to the constructor is called whenever
+ a block of encoded data is ready."""
+
+ def finish(self):
+ """Flush the encoded data and close the output file/stream. Calling this method
+ may cause the streaming callback to be called if in streaming mode."""
+
+
+class IDecoder(IProcessor):
+ """Decoder driver interface"""
+
+ @staticmethod
+ def format():
+ """Return the decode/encoding format as a short string
+ Example: "MP3", "OGG", "AVI", ...
+ """
+
+ @staticmethod
+ def description():
+ """Return a string describing what this decode format provides, is good
+ for, etc... The description is meant to help the end user decide what
+ format is good for him/her
+ """
+
+ @staticmethod
+ def file_extension():
+ """Return the filename extension corresponding to this decode format"""
+
+ @staticmethod
+ def mime_type():
+ """Return the mime type corresponding to this decode format"""
+
+ def process(self, source, options=None):
+ """Perform the decoding process and stream the result through a generator
+
+ source is the audio/video source file absolute path.
+
+ It is highly recommended that decode drivers implement some sort of
+ cache instead of re-encoding each time process() is called.
+
+ It should be possible to make subsequent calls to process() with
+ different items, using the same driver instance.
+ """
+
+class IGrapher(IProcessor):
+ """Media item visualizer driver interface"""
+
+ @staticmethod
+ def name():
+ """Return the graph name, such as "Waveform", "Spectral view",
+ etc..
+ """
+
+ def set_colors(self, background=None, scheme=None):
+ """Set the colors used for image generation. background is a RGB tuple,
+ and scheme a a predefined color theme name"""
+ pass
+
+ def render(self, media_item, width=None, height=None, options=None):
+ """Generator that streams the graph output as a PNG image"""
+
+class IAnalyzer(IProcessor):
+ """Media item analyzer driver interface"""
+
+ @staticmethod
+ def name():
+ """Return the analyzer name, such as "Mean Level", "Max level",
+ "Total length, etc..
+ """
+
+ @staticmethod
+ def unit():
+ """Return the unit of the data such as "dB", "seconds", etc...
+ """
+
+ def render(self, media, options=None):
+ """Return the result data of the process"""
+
from timeside.component import *
from timeside.api import IProcessor
+from timeside.exceptions import Error, ApiError
import re
-__all__ = ['Processor', 'Component', 'implements', 'processors', 'TimeSideError']
-
-class TimeSideError(Exception):
- """Exception base class for errors in TimeSide."""
- # FIXME: is this redundant with Django's error handling ?
+__all__ = ['Processor', 'MetaProcessor', 'implements', 'processors',
+ 'get_processor']
_processors = {}
if new_class in implementations(IProcessor):
id = str(new_class.id())
if _processors.has_key(id):
- raise TimeSideError("%s and %s have the same id: '%s'"
+ raise ApiError("%s and %s have the same id: '%s'"
% (new_class.__name__, _processors[id].__name__, id))
if not MetaProcessor.valid_id.match(id):
- raise TimeSideError("%s has a malformed id: '%s'"
+ raise ApiError("%s has a malformed id: '%s'"
% (new_class.__name__, id))
-
_processors[id] = new_class
any of the descendants of this interface."""
return implementations(interface, recurse)
+
+def get_processor(processor_id):
+ """Return a processor by its id"""
+ if not _processors.has_key(processor_id):
+ raise Error("No processor registered with id: '%s'"
+ % processor_id)
+
+ return _processors[processor_id]
+
# -*- coding: utf-8 -*-
-from timeside.decode.api import *
from timeside.decode.core import *
from timeside.decode.ogg import *
from timeside.decode.flac import *
from timeside.decode.wav import *
-from timeside.decode.mp3 import *
\ No newline at end of file
+from timeside.decode.mp3 import *
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (C) 2007-2009 Parisson
-# Copyright (c) 2007 Olivier Guilyardi <olivier@samalyse.com>
-# Copyright (c) 2007-2009 Guillaume Pellerin <pellerin@parisson.com>
-#
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-from timeside.api import IProcessor
-from timeside.core import TimeSideError
-
-class IDecoder(IProcessor):
- """Decoder driver interface"""
-
- @staticmethod
- def format():
- """Return the decode/encoding format as a short string
- Example: "MP3", "OGG", "AVI", ...
- """
-
- @staticmethod
- def description():
- """Return a string describing what this decode format provides, is good
- for, etc... The description is meant to help the end user decide what
- format is good for him/her
- """
-
- @staticmethod
- def file_extension():
- """Return the filename extension corresponding to this decode format"""
-
- @staticmethod
- def mime_type():
- """Return the mime type corresponding to this decode format"""
-
- def process(self, source, options=None):
- """Perform the decoding process and stream the result through a generator
-
- source is the audio/video source file absolute path.
-
- It is highly recommended that decode drivers implement some sort of
- cache instead of re-encoding each time process() is called.
-
- It should be possible to make subsequent calls to process() with
- different items, using the same driver instance.
- """
-
-
-class DecodeProcessError(TimeSideError):
-
- def __init__(self, message, command, subprocess):
- self.message = message
- self.command = str(command)
- self.subprocess = subprocess
-
- def __str__(self):
- if self.subprocess.stderr != None:
- error = self.subprocess.stderr.read()
- else:
- error = ''
- return "%s ; command: %s; error: %s" % (self.message,
- self.command,
- error)
import subprocess
from timeside.decode.core import *
-from timeside.decode.api import IDecoder
+from timeside.api import IDecoder
from mutagen.flac import FLAC
from tempfile import NamedTemporaryFile
import subprocess
from timeside.decode.core import *
-from timeside.decode.api import IDecoder
+from timeside.api import IDecoder
class Mp3Decoder(DecoderCore):
import subprocess
from timeside.decode.core import *
-from timeside.decode.api import IDecoder
+from timeside.api import IDecoder
from mutagen.oggvorbis import OggVorbis
class OggDecoder(DecoderCore):
import subprocess
from timeside.decode.core import *
-from timeside.decode.api import IDecoder
+from timeside.api import IDecoder
class WavDecoder(DecoderCore):
"""Defines methods to decode from WAV"""
# -*- coding: utf-8 -*-
-from timeside.encode.api import *
from timeside.encode.core import *
from timeside.encode.ogg import *
from timeside.encode.wav import *
from timeside.encode.mp3 import *
-from timeside.encode.flac import *
\ No newline at end of file
+from timeside.encode.flac import *
+++ /dev/null
-# -*- coding: utf-8 -*-
-#
-# Copyright (C) 2007-2009 Parisson
-# Copyright (c) 2007 Olivier Guilyardi <olivier@samalyse.com>
-# Copyright (c) 2007-2009 Guillaume Pellerin <pellerin@parisson.com>
-#
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-from timeside.core import TimeSideError
-from timeside.api import IProcessor
-
-class IEncoder(IProcessor):
- """Encoder driver interface"""
-
- def __init__(self, output, nchannels, samplerate):
- """The constructor must always accept the output, nchannels and samplerate
- arguments. It may accepts extra arguments such as bitrate, depth, etc..,
- but these must be optionnal, that is have a default value.
-
- The output must either be a filepath or a callback function/method for
- for the streaming mode. The callback must accept one argument which is
- block of binary data.
- """
-
- @staticmethod
- def format():
- """Return the encode/encoding format as a short string
- Example: "MP3", "OGG", "AVI", ...
- """
-
- @staticmethod
- def description():
- """Return a string describing what this encode format provides, is good
- for, etc... The description is meant to help the end user decide what
- format is good for him/her
- """
-
- @staticmethod
- def file_extension():
- """Return the filename extension corresponding to this encode format"""
-
- @staticmethod
- def mime_type():
- """Return the mime type corresponding to this encode format"""
-
- def set_metadata(self, metadata):
- """metadata is a tuple containing tuples for each descriptor return by
- the dc.Ressource of the item, in the model order :
- ((name1, value1),(name2, value2),(name1, value3), ...)"""
-
- def update(self):
- """Updates the metadata into the file passed as the output argument
- to the constructor. This method can't be called in streaming
- mode."""
-
- def process(self, frames):
- """Encode the frames passed as a numpy array, where columns are channels.
-
- In streaming mode the callback passed to the constructor is called whenever
- a block of encoded data is ready."""
-
- def finish(self):
- """Flush the encoded data and close the output file/stream. Calling this method
- may cause the streaming callback to be called if in streaming mode."""
-
-class EncodeProcessError(TimeSideError):
-
- def __init__(self, message, command, subprocess):
- self.message = message
- self.command = str(command)
- self.subprocess = subprocess
-
- def __str__(self):
- if self.subprocess.stderr != None:
- error = self.subprocess.stderr.read()
- else:
- error = ''
- return "%s ; command: %s; error: %s" % (self.message,
- self.command,
- error)
-
import subprocess
from timeside.encode.core import *
-from timeside.encode.api import IEncoder
+from timeside.api import IEncoder
from tempfile import NamedTemporaryFile
class FlacEncoder(EncoderCore):
import subprocess
from timeside.encode.core import *
-from timeside.encode.api import IEncoder
+from timeside.api import IEncoder
class Mp3Encoder(EncoderCore):
import subprocess
from timeside.encode.core import *
-from timeside.encode.api import IEncoder
+from timeside.api import IEncoder
class OggVorbisEncoder(EncoderCore):
"""Defines methods to encode to OGG Vorbis"""
import string
from timeside.encode.core import *
-from timeside.encode.api import IEncoder
+from timeside.api import IEncoder
class WavEncoder(EncoderCore):
"""Defines methods to encode to WAV"""
--- /dev/null
+# -*- coding: utf-8 -*-
+#
+# Copyright (c) 2009 Olivier Guilyardi <olivier@samalyse.com>
+#
+# This file is part of TimeSide.
+
+# TimeSide is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+
+# TimeSide is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+
+# 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):
+ self.message = message
+ self.command = str(command)
+ self.subprocess = subprocess
+
+ def __str__(self):
+ if self.subprocess.stderr != None:
+ error = self.subprocess.stderr.read()
+ else:
+ error = ''
+ return "%s ; command: %s; error: %s" % (self.message,
+ self.command,
+ error)
# -*- coding: utf-8 -*-
-from timeside.graph.api import *
from timeside.graph.waveform_audiolab import *
from timeside.graph.spectrogram_audiolab import *
+++ /dev/null
-# -*- coding: utf-8 -*-
-# Copyright (C) 2007 Samalyse SARL#
-# Copyright (c) 2007-2009 Guillaume Pellerin <yomguy@parisson.com>
-
-# This file is part of TimeSide.
-
-# TimeSide is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-
-# TimeSide is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-
-# You should have received a copy of the GNU General Public License
-# along with TimeSide. If not, see <http://www.gnu.org/licenses/>.
-
-# Author: Guillaume Pellerin <yomguy@parisson.com>
-# Author: Olivier Guilyardi <olivier@samalyse.com>
-
-from timeside.api import IProcessor
-
-class IGrapher(IProcessor):
- """Media item visualizer driver interface"""
-
- @staticmethod
- def name():
- """Return the graph name, such as "Waveform", "Spectral view",
- etc..
- """
-
- def set_colors(self, background=None, scheme=None):
- """Set the colors used for image generation. background is a RGB tuple,
- and scheme a a predefined color theme name"""
- pass
-
- def render(self, media_item, width=None, height=None, options=None):
- """Generator that streams the graph output as a PNG image"""
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.core import *
-from timeside.graph.api import IGrapher
+from timeside.api import IGrapher
from tempfile import NamedTemporaryFile
from timeside.graph.wav2png import *
# Author: Guillaume Pellerin <yomguy@parisson.com>
from timeside.core import *
-from timeside.graph.api import IGrapher
+from timeside.api import IGrapher
from tempfile import NamedTemporaryFile
from timeside.graph.wav2png import *
from timeside.core import *
-class TestAnalyzers(Component):
- analyzers = processors(timeside.analyze.IAnalyzer)
+class TestAnalyzers:
+ analyzers = processors(timeside.api.IAnalyzer)
def list(self):
analyzers = []
print id + ' = ' + str(value) + ' ' + analyzer.unit()
-class TestDecoders(Component):
- decoders = processors(timeside.decode.IDecoder)
+class TestDecoders:
+ decoders = processors(timeside.api.IDecoder)
def list(self):
decoders_list = []
f.flush()
f.close()
-class TestEncoders(Component):
- encoders = processors(timeside.encode.IEncoder)
+class TestEncoders:
+ encoders = processors(timeside.api.IEncoder)
def list(self):
encoders = []
file.close()
-class TestGraphers(Component):
- graphers = processors(timeside.graph.IGrapher)
+class TestGraphers:
+ graphers = processors(timeside.api.IGrapher)
def list(self):
graphers = []
projects / timeside.git / commitdiff