timeside.core

1 # -*- coding: utf-8 -*- 2 # 3 # Copyright (c) 2009-2013 Parisson SARL 4 # Copyright (c) 2009 Olivier Guilyardi <olivier@samalyse.com> 5 # 6 # This file is part of TimeSide. 7 8 # TimeSide is free software: you can redistribute it and/or modify 9 # it under the terms of the GNU General Public License as published by 10 # the Free Software Foundation, either version 2 of the License, or 11 # (at your option) any later version. 12 13 # TimeSide is distributed in the hope that it will be useful, 14 # but WITHOUT ANY WARRANTY; without even the implied warranty of 15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 16 # GNU General Public License for more details. 17 18 # You should have received a copy of the GNU General Public License 19 # along with TimeSide. If not, see <http://www.gnu.org/licenses/>. 20 21 from timeside.component import * 22 from timeside.api import IProcessor 23 from timeside.exceptions import Error, ApiError 24 25 26 import re 27 import time 28 import numpy 29 import uuid 30 31 __all__ = ['Processor', 'MetaProcessor', 'implements', 'abstract', 32 'interfacedoc', 'processors', 'get_processor', 'ProcessPipe', 33 'FixedSizeInputAdapter'] 34 35 _processors = {}

36 37 38 -class MetaProcessor(MetaComponent):

39 """Metaclass of the Processor class, used mainly for ensuring that processor 40 id's are wellformed and unique""" 41 42 valid_id = re.compile("^[a-z][_a-z0-9]*$") 43

44 - def __new__(cls, name, bases, d):

45 new_class = MetaComponent.__new__(cls, name, bases, d) 46 if new_class in implementations(IProcessor): 47 id = str(new_class.id()) 48 if _processors.has_key(id): 49 # Doctest test can duplicate a processor 50 # This can be identify by the conditon "module == '__main__'" 51 if new_class.__module__ == '__main__': 52 new_class = _processors[id] 53 elif _processors[id].__module__ == '__main__': 54 pass 55 else: 56 raise ApiError("%s and %s have the same id: '%s'" 57 % (new_class.__name__, _processors[id].__name__, id)) 58 if not MetaProcessor.valid_id.match(id): 59 raise ApiError("%s has a malformed id: '%s'" 60 % (new_class.__name__, id)) 61 62 _processors[id] = new_class 63 64 return new_class

65

66 67 -class Processor(Component):

68 """Base component class of all processors 69 70 71 Attributes: 72 parents : List of parent Processors that must be processed 73 before the current Processor 74 pipe : The current ProcessPipe in which the Processor will run 75 """ 76 __metaclass__ = MetaProcessor 77 78 abstract() 79 implements(IProcessor) 80

81 - def __init__(self):

82 super(Processor, self).__init__() 83 84 self.parents = [] 85 self.source_mediainfo = None 86 self.pipe = None 87 self.UUID = uuid.uuid4()

88 89 @interfacedoc

90 - def setup(self, channels=None, samplerate=None, blocksize=None, 91 totalframes=None):

92 self.source_channels = channels 93 self.source_samplerate = samplerate 94 self.source_blocksize = blocksize 95 self.source_totalframes = totalframes 96 97 # If empty Set default values for input_* attributes 98 # may be setted by the processor during __init__() 99 if not hasattr(self, 'input_channels'): 100 self.input_channels = self.source_channels 101 if not hasattr(self, 'input_samplerate'): 102 self.input_samplerate = self.source_samplerate 103 if not hasattr(self, 'input_blocksize'): 104 self.input_blocksize = self.source_blocksize 105 if not hasattr(self, 'input_stepsize'): 106 self.input_stepsize = self.source_blocksize

107 108 109 # default channels(), samplerate() and blocksize() implementations returns 110 # the source characteristics, but processors may change this behaviour by 111 # overloading those methods 112 @interfacedoc

113 - def channels(self):

114 return self.source_channels

115 116 @interfacedoc

117 - def samplerate(self):

118 return self.source_samplerate

119 120 @interfacedoc

121 - def blocksize(self):

122 return self.source_blocksize

123 124 @interfacedoc

125 - def totalframes(self):

126 return self.source_totalframes

127 128 @interfacedoc

129 - def process(self, frames, eod):

130 return frames, eod

131 132 @interfacedoc

133 - def post_process(self):

134 pass

135 136 @interfacedoc

137 - def release(self):

138 pass

139 140 @interfacedoc

141 - def mediainfo(self):

142 return self.source_mediainfo

143 144 @interfacedoc

145 - def uuid(self):

146 return str(self.UUID)

147

148 - def __del__(self):

149 self.release()

150

151 - def __or__(self, other):

152 return ProcessPipe(self, other)

153

154 155 -class FixedSizeInputAdapter(object):

156 """Utility to make it easier to write processors which require fixed-sized 157 input buffers.""" 158

159 - def __init__(self, buffer_size, channels, pad=False):

160 """Construct a new adapter: buffer_size is the desired buffer size in frames, 161 channels the number of channels, and pad indicates whether the last block should 162 be padded with zeros.""" 163 164 self.buffer = numpy.empty((buffer_size, channels)) 165 self.buffer_size = buffer_size 166 self.len = 0 167 self.pad = pad

168

169 - def blocksize(self, input_totalframes):

170 """Return the total number of frames that this adapter will output according to the 171 input_totalframes argument""" 172 173 blocksize = input_totalframes 174 if self.pad: 175 mod = input_totalframes % self.buffer_size 176 if mod: 177 blocksize += self.buffer_size - mod 178 179 return blocksize

180

181 - def process(self, frames, eod):

182 """Returns an iterator over tuples of the form (buffer, eod) where buffer is a 183 fixed-sized block of data, and eod indicates whether this is the last block. 184 In case padding is deactivated the last block may be smaller than the buffer size. 185 """ 186 src_index = 0 187 remaining = len(frames) 188 189 while remaining: 190 space = self.buffer_size - self.len 191 copylen = remaining < space and remaining or space 192 src = frames[src_index:src_index + copylen] 193 if self.len == 0 and copylen == self.buffer_size: 194 # avoid unnecessary copy 195 buffer = src 196 else: 197 buffer = self.buffer 198 buffer[self.len:self.len + copylen] = src 199 200 remaining -= copylen 201 src_index += copylen 202 self.len += copylen 203 204 if self.len == self.buffer_size: 205 yield buffer, (eod and not remaining) 206 self.len = 0 207 208 if eod and self.len: 209 block = self.buffer 210 if self.pad: 211 self.buffer[self.len:self.buffer_size] = 0 212 else: 213 block = self.buffer[0:self.len] 214 215 yield block, True 216 self.len = 0

217

218 219 -def processors(interface=IProcessor, recurse=True):

220 """Returns the processors implementing a given interface and, if recurse, 221 any of the descendants of this interface.""" 222 return implementations(interface, recurse)

223

224 225 -def get_processor(processor_id):

226 """Return a processor by its id""" 227 if not _processors.has_key(processor_id): 228 raise Error("No processor registered with id: '%s'" 229 % processor_id) 230 231 return _processors[processor_id]

232

233 234 -class ProcessPipe(object):

235 """Handle a pipe of processors 236 237 Attributes: 238 processor: List of all processors in the Process pipe 239 results : Results Container for all the analyzers of the Pipe process 240 """ 241

242 - def __init__(self, *others):

243 self.processors = [] 244 self |= others 245 246 from timeside.analyzer.core import AnalyzerResultContainer 247 self.results = AnalyzerResultContainer()

248

249 - def __or__(self, other):

250 return ProcessPipe(self, other)

251

252 - def __ior__(self, other):

253 if isinstance(other, Processor): 254 for parent in other.parents: 255 self |= parent 256 self.processors.append(other) 257 other.process_pipe = self 258 elif isinstance(other, ProcessPipe): 259 self.processors.extend(other.processors) 260 else: 261 try: 262 iter(other) 263 except TypeError: 264 raise Error("Can not add this type of object to a pipe: %s", str(other)) 265 266 for item in other: 267 self |= item 268 269 return self

270

271 - def __repr__(self):

272 pipe = '' 273 for item in self.processors: 274 pipe += item.id() 275 if item != self.processors[-1]: 276 pipe += ' | ' 277 return pipe

278

279 - def run(self, channels=None, samplerate=None, blocksize=None, stack=None):

280 """Setup/reset all processors in cascade and stream audio data along 281 the pipe. Also returns the pipe itself.""" 282 283 source = self.processors[0] 284 items = self.processors[1:] 285 source.setup(channels=channels, samplerate=samplerate, 286 blocksize=blocksize) 287 288 if stack is None: 289 self.stack = False 290 else: 291 self.stack = stack 292 293 if self.stack: 294 self.frames_stack = [] 295 296 last = source 297 298 # setup/reset processors and configure properties throughout the pipe 299 for item in items: 300 item.source_mediainfo = source.mediainfo() 301 item.setup(channels=last.channels(), 302 samplerate=last.samplerate(), 303 blocksize=last.blocksize(), 304 totalframes=last.totalframes()) 305 last = item 306 307 # now stream audio data along the pipe 308 eod = False 309 while not eod: 310 frames, eod = source.process() 311 if self.stack: 312 self.frames_stack.append(frames) 313 for item in items: 314 frames, eod = item.process(frames, eod) 315 316 # Post-processing 317 for item in items: 318 item.post_process() 319 320 # Release processors 321 if self.stack: 322 if not isinstance(self.frames_stack, numpy.ndarray): 323 self.frames_stack = numpy.vstack(self.frames_stack) 324 from timeside.decoder.core import ArrayDecoder 325 new_source = ArrayDecoder(samples=self.frames_stack, 326 samplerate=source.samplerate()) 327 new_source.setup(channels=source.channels(), 328 samplerate=source.samplerate(), 329 blocksize=source.blocksize()) 330 self.processors[0] = new_source 331 332 for item in items: 333 item.release() 334 self.processors.remove(item)

335

Source Code for Module timeside.core