#! /usr/bin/python # # Author: Arjun Sarwal arjun@laptop.org # Copyright (C) 2007, Arjun Sarwal # Copyright (C) 2009, Walter Bender # Copyright (C) 2009, Benjamin Berg, Sebastian Berg # Copyright (C) 2009, Sayamindu Dasgupta # # This program 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. # # This program 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 this program; if not, write to the Free Software # Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. import pygst pygst.require("0.10") import gst import gst.interfaces import gobject import os import subprocess import numpy as np from string import find import config #This has all the golabals # Initialize logging. import logging log = logging.getLogger('Measure') log.setLevel(logging.DEBUG) logging.basicConfig() class AudioGrab: def __init__(self, callable1, journal): self.callable1 = callable1 self.ji = journal self.sensor = None self.temp_buffer = [0] self.picture_buffer = [] self.draw_graph_status = False self.f = None self.logging_status = False self.screenshot = True self.rate = 48000 self.final_count = 0 self.count_temp = 0 self.entry_count = 0 self.waveform_id = 1 self.logging_state = False self.buffer_interval_logging = 0 self.counter_buffer = 0 self.pipeline = gst.Pipeline("pipeline") self.alsasrc = gst.element_factory_make("alsasrc", "alsa-source") self.pipeline.add(self.alsasrc) self.caps1 = gst.element_factory_make("capsfilter", "caps1") self.pipeline.add(self.caps1) caps_str = "audio/x-raw-int,rate=%d,channels=1,depth=16" % \ (config.RATE, ) self.caps1.set_property("caps", gst.caps_from_string(caps_str) ) self.fakesink = gst.element_factory_make("fakesink", "fsink") self.pipeline.add(self.fakesink) self.fakesink.connect("handoff", self.on_buffer) self.fakesink.set_property("signal-handoffs",True) gst.element_link_many(self.alsasrc, self.caps1, self.fakesink) self.dont_queue_the_buffer = False self._mixer = gst.element_factory_make('alsamixer') rc = self._mixer.set_state(gst.STATE_PAUSED) assert rc == gst.STATE_CHANGE_SUCCESS log.debug('controls: %r', [t.props.untranslated_label for t in self._mixer.list_tracks()]) self._dc_control = self._find_control(['dc mode']) self._mic_bias_control = self._find_control(['mic bias', 'dc input bias', 'V_REFOUT']) self._mic_boost_control = self._find_control(['mic boost', 'analog mic boost']) self._mic_gain_control = self._find_control(['mic']) self._capture_control = self._find_control(['capture']) self._master_control = self._find_control(['master']) ####Variables for saving and resuming state of sound device###### self.master = self.get_master() self.bias = config.BIAS self.dcmode = config.DC_MODE_ENABLE self.capture_gain = config.CAPTURE_GAIN self.mic_boost = config.MIC_BOOST self.mic = self.get_mic_gain() ################################################################# def set_handoff_signal(self, handoff_state): """Sets whether the handoff signal would generate an interrupt or not""" self.fakesink.set_property("signal-handoffs",handoff_state) def _new_buffer(self, buf): if self.dont_queue_the_buffer == False: self.temp_buffer = buf self.callable1(buf) else: pass def on_buffer(self, element, buffer, pad): """The function that is called whenever new data is available This is the signal handler for the handoff signal""" temp_buffer = np.fromstring(buffer, 'int16') if self.dont_queue_the_buffer == False: self._new_buffer(temp_buffer) else: pass if self.logging_state==True: if self.waveform_id == config.SOUND_MAX_WAVE_LOGS: self.waveform_id = 1 self.logging_state = False self.ji.stop_session() else: if self.counter_buffer == self.buffer_interval_logging: self._emit_for_logging(temp_buffer) self.counter_buffer=0 self.counter_buffer+=1 # If a record is to be written, thats all for the logging session if self.buffer_interval_logging ==0: self.logging_state = False self.ji.stop_session() self.waveform_id = 1 return False def set_freeze_the_display(self, freeze = False): """Useful when just the display is needed to be frozen, but logging should continue""" self.dont_queue_the_buffer = not freeze def get_freeze_the_display(self): """Returns state of queueing the buffer""" return not self.dont_queue_the_buffer def set_sensor(self, sensor): self.sensor = sensor def _emit_for_logging(self, buf): """Sends the data for logging""" if self.buffer_interval_logging==0: self.ji.take_screenshot() else: if self.screenshot == True: self.ji.take_screenshot(self.waveform_id) self.waveform_id+=1 else: # save value to Journal self.ji.write_value(buf[0]) # display value on Sensor toolbar try: self.sensor.set_sample_value(str(buf[0])) except: pass def start_sound_device(self): """Start or Restart grabbing data from the audio capture""" gst.event_new_flush_start() self.pipeline.set_state(gst.STATE_PLAYING) def stop_sound_device(self): """Stop grabbing data from capture device""" gst.event_new_flush_stop() self.pipeline.set_state(gst.STATE_NULL) def set_logging_params(self, start_stop=False, interval=0, screenshot=True): """Configures for logging of data i.e. starts or stops a session Sets an interval if logging interval is to be started Sets if screenshot of waveform is to be taken or values need to be written""" self.logging_state = start_stop self.set_buffer_interval_logging(interval) #if interval==0: # self.take_picture() self.reset_counter_buffer() self.screenshot = screenshot def take_picture(self): """Used to grab and temporarily store the current buffer""" self.picture_buffer = self.temp_buffer.copy() def set_logging_state(self, start_stop=False): """Sets whether buffer is to be emited for logging (True) or not (False)""" self.logging_state = start_stop def set_buffer_interval_logging(self, interval=0): """Sets the number of buffers after which a buffer needs to be emitted""" self.buffer_interval_logging = interval def reset_counter_buffer(self): """Resets the counter buffer used to keep track of after how many buffers to emit a buffer for logging""" self.counter_buffer = 0 def set_sampling_rate(self, sr): """Sets the sampling rate of the capture device Sampling rate must be given as an integer for example 16000 for setting 16Khz sampling rate The sampling rate would be set in the device to the nearest available""" self.pause_grabbing() caps_str = "audio/x-raw-int,rate=%d,channels=1,depth=16" % (sr, ) self.caps1.set_property("caps", gst.caps_from_string(caps_str) ) self.resume_grabbing() def get_sampling_rate(self): """Gets the sampling rate of the capture device""" return int(self.caps1.get_property("caps")[0]['rate'] ) def set_callable1(self, callable1): """Sets the callable to the drawing function for giving the data at the end of idle-add""" self.callable1 = callable1 def start_grabbing(self): """Called right at the start of the Activity""" self.start_sound_device() def pause_grabbing(self): """When Activity goes into background""" self.save_state() self.stop_sound_device() def resume_grabbing(self): """When Activity becomes active after going to background""" self.start_sound_device() self.resume_state() def stop_grabbing(self): self.stop_sound_device() self.set_handoff_signal(False) def _find_control(self, prefixes): """Try to find a mixer control matching one of the prefixes. The control with the best match (smallest difference in length between label and prefix) will be returned. If no match is found, None is returned. """ def best_prefix(label, prefixes): matches = [len(label)-len(p) for p in prefixes if label.startswith(p)] if not matches: return None matches.sort() return matches[0] controls = [] for track in self._mixer.list_tracks(): label = track.props.untranslated_label.lower() diff = best_prefix(label, prefixes) if diff is not None: controls.append((track, diff)) controls.sort(key=lambda e: e[1]) if controls: return controls[0][0] return None def save_state(self): """Saves the state of all audio controls""" self.master = self.get_master() self.bias = self.get_bias() self.dcmode = self.get_dc_mode() self.capture_gain = self.get_capture_gain() self.mic_boost = self.get_mic_boost() def resume_state(self): """Put back all audio control settings from the saved state""" self.set_master(self.master) self.set_bias(self.bias) self.set_dc_mode(self.dcmode) self.set_capture_gain(self.capture_gain) self.set_mic_boost(self.mic_boost) def _get_mute(self, control, name, default): if not control: log.warning('No %s control, returning constant mute status', name) return default value = bool(control.flags & gst.interfaces.MIXER_TRACK_MUTE) log.debug('Getting %s (%s) mute status: %r', name, control.props.untranslated_label, value) return value def _set_mute(self, control, name, value): if not control: log.warning('No %s control, not setting mute', name) return self._mixer.set_mute(control, value) log.debug('Set mute for %s (%s) to %r', name, control.props.untranslated_label, value) def _get_volume(self, control, name): if not control: log.warning('No %s control, returning constant volume', name) return 100 hw_volume = self._mixer.get_volume(control)[0] min_vol = control.min_volume max_vol = control.max_volume percent = (hw_volume - min_vol)*100//(max_vol - min_vol) log.debug('Getting %s (%s) volume: %d (%d)', name, control.props.untranslated_label, percent, hw_volume) return percent def _set_volume(self, control, name, value): if not control: log.warning('No %s control, not setting volume', name) return min_vol = control.min_volume max_vol = control.max_volume hw_volume = value*(max_vol - min_vol)//100 + min_vol self._mixer.set_volume(control, (hw_volume,)*control.num_channels) log.debug('Set volume of %s (%s) to %d (%d)', name, control.props.untranslated_label, value, hw_volume) def mute_master(self): """Mutes the Master Control""" self._set_mute(self._master_control, 'Master', True) def unmute_master(self): """Unmutes the Master Control""" self._set_mute(self._master_control, 'Master', False) def set_master(self, master_val): """Sets the Master gain slider settings master_val must be given as an integer between 0 and 100 indicating the percentage of the slider to be set""" self._set_volume(self._master_control, 'Master', master_val) def get_master(self): """Gets the Master gain slider settings. The value returned is an integer between 0-100 and is an indicative of the percentage 0 - 100%""" return self._get_volume(self._master_control, 'master') def set_bias(self,bias_state=False): """Enables / disables bias voltage. On XO-1.5 it uses the 80% setting. """ if not isinstance(self._mic_bias_control, gst.interfaces.MixerOptions): return self._set_mute(self._mic_bias_control, 'Mic Bias', not bias_state) values = self._mic_bias_control.get_values() # We assume that the values are sorted from lowest (=off) to highest. # Since they are mixed strings ("Off", "50%", etc.), we cannot easily # ensure this ourselves by sorting with the default sort order. if bias_state: self._mixer.set_option(self._mic_bias_control, values[-1]) else: self._mixer.set_option(self._mic_bias_control, values[0]) def get_bias(self): """Check whether bias voltage is enabled.""" if not isinstance(self._mic_bias_control, gst.interfaces.MixerOptions): return not self._get_mute(self._mic_bias_control, 'Mic Bias', False) values = self._mic_bias_control.get_values() current = self._mixer.get_option(self._mic_bias_control) # same ordering assertion as in set_bias() applies if current == values[0]: return False return True def set_dc_mode(self, dc_mode = False): """Sets the DC Mode Enable control pass False to mute and True to unmute""" self._set_mute(self._dc_control, 'DC mode', not dc_mode) def get_dc_mode(self): """Returns the setting of DC Mode Enable control i .e. True: Unmuted and False: Muted""" return not self._get_mute(self._dc_control, 'DC mode', False) def set_mic_boost(self, mic_boost=False): """Set Mic Boost. True = +20dB, False = 0dB""" if not isinstance(self._mic_boost_control, gst.interfaces.MixerOptions): return self._set_mute(self._mic_boost_control, 'Mic Boost', mic_boost) values = self._mic_boost_control.get_values() if '20dB' not in values or '0dB' not in values: logging.error("Mic Boost (%s) is an option list, but doesn't " "contain 0dB and 20dB settings", self._mic_boost_control.props.label) return if mic_boost: self._mixer.set_option(self._mic_boost_control, '20dB') else: self._mixer.set_option(self._mic_boost_control, '0dB') def get_mic_boost(self): """Return Mic Boost setting. True = +20dB, False = 0dB""" if not isinstance(self._mic_boost_control, gst.interfaces.MixerOptions): return self._get_mute(self._mic_boost_control, 'Mic Boost', False) values = self._mic_boost_control.get_values() if '20dB' not in values or '0dB' not in values: logging.error("Mic Boost (%s) is an option list, but doesn't " "contain 0dB and 20dB settings", self._mic_boost_control.props.label) return False current = self._mixer.get_option(self._mic_boost_control) if current == '20dB': return True return False def set_capture_gain(self, capture_val): """Sets the Capture gain slider settings capture_val must be given as an integer between 0 and 100 indicating the percentage of the slider to be set""" self._set_volume(self._capture_control, 'Capture', capture_val) def get_capture_gain(self): """Gets the Capture gain slider settings. The value returned is an integer between 0-100 and is an indicative of the percentage 0 - 100%""" return self._get_volume(self._capture_control, 'Capture') def set_mic_gain(self, mic_val): """Sets the MIC gain slider settings mic_val must be given as an integer between 0 and 100 indicating the percentage of the slider to be set""" self._set_volume(self._mic_gain_control, 'Mic', mic_val) def get_mic_gain(self): """Gets the MIC gain slider settings. The value returned is an integer between 0-100 and is an indicative of the percentage 0 - 100%""" return self._get_volume(self._mic_gain_control, 'Mic') def mute_mic(self): """Mutes the Mic Control""" self._set_mute(self._mic_mute_control, 'Mic', True) def unmute_mic(self): """Unmutes the Mic Control""" self._set_mute(self._mic_mute_control, 'Mic', False) def set_sensor_type(self, sensor_type=1): """Set the type of sensor you want to use. Set sensor_type according to the following 0 - AC coupling with Bias Off --> Very rarely used. Use when connecting a dynamic microphone externally 1 - AC coupling with Bias On --> The default settings. The internal MIC uses these 2 - DC coupling with Bias Off --> Used when using a voltage output sensor. For example LM35 which gives output proportional to temperature 3 - DC coupling with Bias On --> Used with resistive sensors. For example""" if sensor_type==0: self.set_dc_mode(False) self.set_bias(False) # XO-1.5 self.set_capture_gain(80) # XO-1 self.set_capture_gain(50) self.set_mic_boost(True) elif sensor_type==1: self.set_dc_mode(False) self.set_bias(True) # XO-1.5 self.set_capture_gain(80) # XO-1 self.set_capture_gain(40) self.set_mic_boost(True) elif sensor_type==2: self.set_dc_mode(True) self.set_bias(False) self.set_capture_gain(0) self.set_mic_boost(False) elif sensor_type==3: self.set_dc_mode(True) self.set_bias(True) self.set_capture_gain(0) self.set_mic_boost(False) def on_activity_quit(self): """When Activity quits""" self.set_mic_boost(config.QUIT_MIC_BOOST) self.set_dc_mode(config.QUIT_DC_MODE_ENABLE) self.set_capture_gain(config.QUIT_CAPTURE_GAIN) self.set_bias(config.QUIT_BIAS) self.stop_sound_device()