Source Code for Module timeside.grapher.core

  1  #!/usr/bin/env python 
  2  # -*- coding: utf-8 -*- 
  3   
  4  # Copyright (C) 2008 MUSIC TECHNOLOGY GROUP (MTG) 
  5  #                    UNIVERSITAT POMPEU FABRA 
  6  # 
  7  # This program is free software: you can redistribute it and/or modify 
  8  # it under the terms of the GNU Affero General Public License as 
  9  # published by the Free Software Foundation, either version 3 of the 
 10  # License, or (at your option) any later version. 
 11  # 
 12  # This program is distributed in the hope that it will be useful, 
 13  # but WITHOUT ANY WARRANTY; without even the implied warranty of 
 14  # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 15  # GNU Affero General Public License for more details. 
 16  # 
 17  # You should have received a copy of the GNU Affero General Public License 
 18  # along with this program.  If not, see <http://www.gnu.org/licenses/>. 
 19  # 
 20  # Authors: 
 21  #   Bram de Jong <bram.dejong at domain.com where domain in gmail> 
 22  #   Guillaume Pellerin <yomguy@parisson.com> 
 23   
 24   
 25  import optparse, math, sys, numpy 
 26   
 27  try: 
 28      from PIL import ImageFilter, ImageChops, Image, ImageDraw, ImageColor, ImageEnhance 
 29  except ImportError: 
 30      import ImageFilter, ImageChops, Image, ImageDraw, ImageColor, ImageEnhance 
 31   
 32  from timeside.core import * 
 33  from timeside.api import IGrapher 
 34  from timeside.grapher.color_schemes import default_color_schemes 
 35  from utils import * 
 36   
 37   
 38 -class Spectrum(object): 
 39      """ FFT based frequency analysis of audio frames.""" 
 40   
 41 -    def __init__(self, fft_size, samplerate, blocksize, totalframes, lower, higher, window_function=None): 
 42          self.fft_size = fft_size 
 43          self.window = window_function(self.fft_size) 
 44          self.window_function = window_function 
 45          self.spectrum_range = None 
 46          self.lower = lower 
 47          self.higher = higher 
 48          self.blocksize = blocksize 
 49          self.lower_log = math.log10(self.lower) 
 50          self.higher_log = math.log10(self.higher) 
 51          self.clip = lambda val, low, high: min(high, max(low, val)) 
 52          self.totalframes = totalframes 
 53          self.samplerate = samplerate 
 54          self.window_function = window_function 
 55          self.window = self.window_function(self.blocksize) 
 56          # Hanning window by default 
 57          if self.window_function: 
 58              self.window = self.window_function(self.blocksize) 
 59          else: 
 60              self.window_function = numpy.hanning 
 61              self.window = self.window_function(self.blocksize) 
 62   
 63   
 64 -    def process(self, frames, eod, spec_range=120.0): 
 65          """ Returns a tuple containing the spectral centroid and the spectrum (dB scales) of the input audio frames. 
 66          FFT window sizes are adatable to the input frame size.""" 
 67   
 68          samples = frames[:,0] 
 69          nsamples = len(frames[:,0]) 
 70          if nsamples != self.blocksize: 
 71              self.window = self.window_function(nsamples) 
 72          samples *= self.window 
 73   
 74          while nsamples > self.fft_size: 
 75              self.fft_size = 2 * self.fft_size 
 76   
 77          zeros_p = numpy.zeros(self.fft_size/2-int(nsamples/2)) 
 78          if nsamples % 2: 
 79              zeros_n = numpy.zeros(self.fft_size/2-int(nsamples/2)-1) 
 80          else: 
 81              zeros_n = numpy.zeros(self.fft_size/2-int(nsamples/2)) 
 82          samples = numpy.concatenate((zeros_p, samples, zeros_n), axis=0) 
 83   
 84          fft = numpy.fft.fft(samples) 
 85          # normalized abs(FFT) between 0 and 1 
 86          spectrum = numpy.abs(fft[:fft.shape[0] / 2 + 1]) / float(nsamples) 
 87          length = numpy.float64(spectrum.shape[0]) 
 88   
 89          # scale the db spectrum from [- spec_range db ... 0 db] > [0..1] 
 90          db_spectrum = ((20*(numpy.log10(spectrum + 1e-30))).clip(-spec_range, 0.0) + spec_range)/spec_range 
 91          energy = spectrum.sum() 
 92          spectral_centroid = 0 
 93   
 94          if energy > 1e-20: 
 95              # calculate the spectral centroid 
 96              if self.spectrum_range == None: 
 97                  self.spectrum_range = numpy.arange(length) 
 98              spectral_centroid = (spectrum * self.spectrum_range).sum() / (energy * (length - 1)) * self.samplerate * 0.5 
 99              # clip > log10 > scale between 0 and 1 
100              spectral_centroid = (math.log10(self.clip(spectral_centroid, self.lower, self.higher)) - \ 
101                                  self.lower_log) / (self.higher_log - self.lower_log) 
102   
103          return (spectral_centroid, db_spectrum) 
104   
105   
106 -class Grapher(Processor): 
107      ''' 
108      Generic abstract class for the graphers 
109      ''' 
110   
111      fft_size = 0x1000 
112      frame_cursor = 0 
113      pixel_cursor = 0 
114      lower_freq = 20 
115   
116      implements(IGrapher) 
117      abstract() 
118   
119 -    def __init__(self, width=1024, height=256, bg_color=None, color_scheme='default'): 
120          super(Grapher, self).__init__() 
121          self.bg_color = bg_color 
122          self.color_scheme = color_scheme 
123          self.graph = None 
124          self.image_width = width 
125          self.image_height = height 
126          self.bg_color = bg_color 
127          self.color_scheme = color_scheme 
128          self.previous_x, self.previous_y = None, None 
129   
130      @staticmethod 
131 -    def id(): 
132          return "generic_grapher" 
133   
134      @staticmethod 
135 -    def name(): 
136          return "Generic grapher" 
137   
138 -    def set_colors(self, bg_color, color_scheme): 
139          self.bg_color = bg_color 
140          self.color_color_scheme = color_scheme 
141   
142 -    def setup(self, channels=None, samplerate=None, blocksize=None, totalframes=None): 
143          super(Grapher, self).setup(channels, samplerate, blocksize, totalframes) 
144          self.sample_rate = samplerate 
145          self.higher_freq = self.sample_rate/2 
146          self.block_size = blocksize 
147          self.total_frames = totalframes 
148          self.image = Image.new("RGBA", (self.image_width, self.image_height), self.bg_color) 
149          self.samples_per_pixel = self.total_frames / float(self.image_width) 
150          self.buffer_size = int(round(self.samples_per_pixel, 0)) 
151          self.pixels_adapter = FixedSizeInputAdapter(self.buffer_size, 1, pad=False) 
152          self.pixels_adapter_totalframes = self.pixels_adapter.blocksize(self.total_frames) 
153          self.spectrum = Spectrum(self.fft_size, self.sample_rate, self.block_size, self.total_frames, 
154                                   self.lower_freq, self.higher_freq, numpy.hanning) 
155          self.pixel = self.image.load() 
156          self.draw = ImageDraw.Draw(self.image) 
157   
158      @interfacedoc 
159 -    def render(self, output=None): 
160          if output: 
161              self.image.save(output) 
162              return 
163          return self.image 
164   
165 -    def watermark(self, text, font=None, color=(255, 255, 255), opacity=.6, margin=(5,5)): 
166          self.image = im_watermark(self.image, text, color=color, opacity=opacity, margin=margin) 
167   
168 -    def draw_peaks(self, x, peaks, line_color): 
169          """Draw 2 peaks at x""" 
170   
171          y1 = self.image_height * 0.5 - peaks[0] * (self.image_height - 4) * 0.5 
172          y2 = self.image_height * 0.5 - peaks[1] * (self.image_height - 4) * 0.5 
173   
174          if self.previous_y: 
175              self.draw.line([self.previous_x, self.previous_y, x, y1, x, y2], line_color) 
176          else: 
177              self.draw.line([x, y1, x, y2], line_color) 
178   
179          self.draw_anti_aliased_pixels(x, y1, y2, line_color) 
180          self.previous_x, self.previous_y = x, y2 
181   
182 -    def draw_peaks_inverted(self, x, peaks, line_color): 
183          """Draw 2 inverted peaks at x""" 
184   
185          y1 = self.image_height * 0.5 - peaks[0] * (self.image_height - 4) * 0.5 
186          y2 = self.image_height * 0.5 - peaks[1] * (self.image_height - 4) * 0.5 
187   
188          if self.previous_y and x < self.image_width-1: 
189              if y1 < y2: 
190                  self.draw.line((x, 0, x, y1), line_color) 
191                  self.draw.line((x, self.image_height , x, y2), line_color) 
192              else: 
193                  self.draw.line((x, 0, x, y2), line_color) 
194                  self.draw.line((x, self.image_height , x, y1), line_color) 
195          else: 
196              self.draw.line((x, 0, x, self.image_height), line_color) 
197          self.draw_anti_aliased_pixels(x, y1, y2, line_color) 
198          self.previous_x, self.previous_y = x, y1 
199   
200 -    def draw_anti_aliased_pixels(self, x, y1, y2, color): 
201          """ vertical anti-aliasing at y1 and y2 """ 
202   
203          y_max = max(y1, y2) 
204          y_max_int = int(y_max) 
205          alpha = y_max - y_max_int 
206   
207          if alpha > 0.0 and alpha < 1.0 and y_max_int + 1 < self.image_height: 
208              current_pix = self.pixel[int(x), y_max_int + 1] 
209              r = int((1-alpha)*current_pix[0] + alpha*color[0]) 
210              g = int((1-alpha)*current_pix[1] + alpha*color[1]) 
211              b = int((1-alpha)*current_pix[2] + alpha*color[2]) 
212              self.pixel[x, y_max_int + 1] = (r,g,b) 
213   
214          y_min = min(y1, y2) 
215          y_min_int = int(y_min) 
216          alpha = 1.0 - (y_min - y_min_int) 
217   
218          if alpha > 0.0 and alpha < 1.0 and y_min_int - 1 >= 0: 
219              current_pix = self.pixel[x, y_min_int - 1] 
220              r = int((1-alpha)*current_pix[0] + alpha*color[0]) 
221              g = int((1-alpha)*current_pix[1] + alpha*color[1]) 
222              b = int((1-alpha)*current_pix[2] + alpha*color[2]) 
223              self.pixel[x, y_min_int - 1] = (r,g,b) 
224   
225 -    def draw_peaks_contour(self): 
226          contour = self.contour.copy() 
227          contour = smooth(contour, window_len=16) 
228          contour = normalize(contour) 
229   
230          # Scaling 
231          #ratio = numpy.mean(contour)/numpy.sqrt(2) 
232          ratio = 1 
233          contour = normalize(numpy.expm1(contour/ratio))*(1-10**-6) 
234   
235          # Spline 
236          #contour = cspline1d(contour) 
237          #contour = cspline1d_eval(contour, self.x, dx=self.dx1, x0=self.x[0]) 
238   
239          if self.symetry: 
240              height = int(self.image_height/2) 
241          else: 
242              height = self.image_height 
243   
244          # Multicurve rotating 
245          for i in range(0,self.ndiv): 
246              self.previous_x, self.previous_y = None, None 
247   
248              bright_color = int(255*(1-float(i)/(self.ndiv*2))) 
249              bright_color = 255-bright_color+self.color_offset 
250              #line_color = self.color_lookup[int(self.centroids[j]*255.0)] 
251              line_color = (bright_color,bright_color,bright_color) 
252   
253              # Linear 
254              #contour = contour*(1.0-float(i)/self.ndiv) 
255              #contour = contour*(1-float(i)/self.ndiv) 
256   
257              # Cosinus 
258              contour = contour*numpy.arccos(float(i)/self.ndiv)*2/numpy.pi 
259              #contour = self.contour*(1-float(i)*numpy.arccos(float(i)/self.ndiv)*2/numpy.pi/self.ndiv) 
260              #contour = contour + ((1-contour)*2/numpy.pi*numpy.arcsin(float(i)/self.ndiv)) 
261   
262              curve = (height-1)*contour 
263              #curve = contour*(height-2)/2+height/2 
264   
265              for x in self.x: 
266                  x = int(x) 
267                  y = curve[x] 
268                  if not x == 0: 
269                      if not self.symetry: 
270                          self.draw.line([self.previous_x, self.previous_y, x, y], line_color) 
271                          self.draw_anti_aliased_pixels(x, y, y, line_color) 
272                      else: 
273                          self.draw.line([self.previous_x, self.previous_y+height, x, y+height], line_color) 
274                          self.draw_anti_aliased_pixels(x, y+height, y+height, line_color) 
275                          self.draw.line([self.previous_x, -self.previous_y+height, x, -y+height], line_color) 
276                          self.draw_anti_aliased_pixels(x, -y+height, -y+height, line_color) 
277                  else: 
278                      if not self.symetry: 
279                          self.draw.point((x, y), line_color) 
280                      else: 
281                          self.draw.point((x, y+height), line_color) 
282                  self.previous_x, self.previous_y = x, y 
283   
284   
285   
286  if __name__ == "__main__": 
287      import doctest 
288      doctest.testmod() 
289