1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 '''
22 Collections of preprocessors to use as decorators for the analyzers process
23
24 Preprocessors process the (frame, eod) arguments in order to handle various
25 preprocessing such as :
26 - Downmixing to mono
27 - Adapt the frames to match the input_blocksize and input_stepsize
28 of the analyzer
29 '''
32 '''
33 Pre-processing decorator that downmixes frames from multi-channel to mono
34
35 Downmix is achieved by averaging all channels
36
37 >>> from timeside.analyzer.preprocessors import downmix_to_mono
38 >>> @downmix_to_mono
39 ... def process(analyzer,frames,eod):
40 ... print 'Frames, eod inside process :'
41 ... print frames, eod
42 ... return frames, eod
43 ...
44 >>> import numpy as np
45 >>> frames = np.asarray([[1,2],[3,4],[5,6],[7,8],[9,10]])
46 >>> eod = False
47 >>> frames_, eod_ = process(object(),frames,eod)
48 Frames, eod inside process :
49 [ 1.5 3.5 5.5 7.5 9.5] False
50
51 Outside Process frames and eod are preserved :
52
53 >>> frames_
54 array([[ 1, 2],
55 [ 3, 4],
56 [ 5, 6],
57 [ 7, 8],
58 [ 9, 10]])
59 >>> eod_
60 False
61 '''
62
63 import functools
64
65 @functools.wraps(process_func)
66 def wrapper(analyzer, frames, eod):
67
68 if frames.ndim > 1:
69 downmix_frames = frames.mean(axis=-1)
70 else:
71 downmix_frames = frames
72
73 process_func(analyzer, downmix_frames, eod)
74
75 return frames, eod
76 return wrapper
77
80 '''
81 Pre-processing decorator that adapt frames to match input_blocksize and
82 input_stepsize of the decorated analyzer
83
84 >>> from timeside.analyzer.preprocessors import frames_adapter
85 >>> @frames_adapter
86 ... def process(analyzer,frames,eod):
87 ... analyzer.frames.append(frames)
88 ... return frames, eod
89 ...
90 >>> class Fake_Analyzer(object):
91 ... def __init__(self):
92 ... self.input_blocksize = 4
93 ... self.input_stepsize = 3
94 ... self.frames = [] # Container for the frame as viewed by process
95 >>> import numpy as np
96 >>> analyzer = Fake_Analyzer()
97 >>> frames = np.asarray(range(0,12))
98 >>> eod = False
99 >>> frames_, eod_ = process(analyzer,frames,eod)
100
101 Inside the process the frames have been adapted to match input_blocksize
102 and input_stepsize
103
104 >>> analyzer.frames
105 [array([0, 1, 2, 3]), array([3, 4, 5, 6]), array([6, 7, 8, 9])]
106
107 Outside the process, the original frames and eod are preserved:
108
109 >>> frames_
110 array([ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
111 >>> eod_
112 False
113
114 Releasing the process with eod=True will zeropad the last frame if necessary
115
116 >>> frames = np.asarray(range(12,14))
117 >>> eod = True
118 >>> frames_, eod_ = process(analyzer,frames,eod)
119 >>> analyzer.frames
120 [array([0, 1, 2, 3]), array([3, 4, 5, 6]), array([6, 7, 8, 9]), array([ 9, 10, 11, 12]), array([12, 13, 0, 0])]
121 '''
122
123 import functools
124 import numpy as np
125
126 class framesBuffer(object):
127
128 def __init__(self, blocksize, stepsize):
129 self.blocksize = blocksize
130 self.stepsize = stepsize
131 self.buffer = None
132
133 def frames(self, frames, eod):
134 if self.buffer is not None:
135 stack = np.concatenate([self.buffer, frames])
136 else:
137 stack = frames.copy()
138
139 stack_length = len(stack)
140
141 nb_frames = (
142 stack_length - self.blocksize + self.stepsize) // self.stepsize
143 nb_frames = max(nb_frames, 0)
144 frames_length = nb_frames * self.stepsize + \
145 self.blocksize - self.stepsize
146 last_block_size = stack_length - frames_length
147
148 if eod:
149
150 pad_shape = tuple(
151 self.blocksize - last_block_size if i == 0 else x
152 for i, x in enumerate(frames.shape))
153 stack = np.concatenate([stack, np.zeros(pad_shape,
154 dtype=frames.dtype)])
155 nb_frames += 1
156
157 self.buffer = stack[nb_frames * self.stepsize:]
158
159 eod_list = np.repeat(False, nb_frames)
160 if eod and len(eod_list):
161 eod_list[-1] = eod
162
163 for index, eod in zip(xrange(0, nb_frames*self.stepsize, self.stepsize), eod_list):
164 yield (stack[index:index + self.blocksize],eod)
165
166 @functools.wraps(process_func)
167 def wrapper(analyzer, frames, eod):
168
169 if not hasattr(analyzer, 'frames_buffer'):
170 analyzer.frames_buffer = framesBuffer(analyzer.input_blocksize,
171 analyzer.input_stepsize)
172
173
174 for adapted_frames, adapted_eod in analyzer.frames_buffer.frames(frames, eod):
175 process_func(analyzer, adapted_frames, adapted_eod)
176
177 return frames, eod
178 return wrapper
179
180
181 if __name__ == "__main__":
182
183 from tests import test_analyzer_preprocessors
184 from tests.unit_timeside import run_test_module
185 run_test_module(test_analyzer_preprocessors)
186