Source code for psychopy.voicekey

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""voicekey: A toolkit for programming virtual voice-keys.

Copyright (c) Jeremy R. Gray, 2015
License: Distributed under the terms of the GPLv3
Dev status: beta. Can work well in some circumstances, not widely tested.

_BaseVoiceKey is the main abstract class. Subclass and override the detect()
method. See SimpleThresholdVoiceKey or OnsetVoiceKey for examples.

__version__ = 0.5

# pyo: see
    import pyo64 as pyo
    have_pyo64 = True
except Exception:
    import pyo
    have_pyo64 = False

# pyo_server will point to a booted pyo server once pyo_init() is called:
pyo_server = None

# helper functions for time, signal processing, and file I/O:
from . vk_tools import *

# Constants:
T_BASELINE_PERIOD = 0.200  # sec; time assumed not to contain any speech
T_BASELINE_ON = 0.035  # sec; self.baseline is between T_BASELINE_ON ..OFF
T_BASELINE_OFF = 0.180  # sec
TOO_LOUD = 0.01
TOO_QUIET = 10 ** -7
RATE = 44100  # default sampling rate

# max recording time 30 minutes; longer is ok but not tested, lots of lag:

class VoiceKeyException(Exception):

class _BaseVoiceKey:
    """Abstract base class for virtual voice-keys.

    Accepts data as real-time input (from a microphone by default) or off-line
    (if `file_in` is a valid file).
    Over-ride detect() and other methods as needed. See examples.

    def __init__(self, sec=0, file_out='', file_in='', **config):

                duration to record in seconds

                name for output filename (for microphone input)

                name of input file for sound source (not microphone)

            config:  kwargs dict of parameters for configuration. defaults are:

                'msPerChunk': 2; duration of each real-time analysis chunk, in ms

                'signaler': default None

                'autosave': True; False means manual saving to a file is still
                    possible (by calling .save() but not called automatically upon

                'chnl_in' : microphone channel;
                    see psychopy.sound.backend.get_input_devices()

                'chnl_out': not implemented; output device to use

                'start': 0, select section from a file based on (start, stop) time

                'stop': -1, end of file (default)

                'vol': 0.99, volume 0..1

                'low': 100, Hz, low end of bandpass; can vary for M/F speakers

                'high': 3000, Hz, high end of bandpass

                'threshold': 10

                'baseline': 0; 0 = auto-detect; give a non-zero value to use that

                'more_processing': True; compute more stats per chunk including
                    bandpass; try False if 32-bit python can't keep up

                'zero_crossings': True
        if not (pyo_server and pyo_server.getIsBooted() and
            msg = 'Need a running pyo server: call voicekey.pyo_init()'
            raise VoiceKeyException(msg)
        self.rate = pyo_server.getSamplingRate()  # pyo_init enforces 16000+ Hz
        self.sec = float(sec)
        if self.sec > MAX_RECORDING_SEC:
            msg = 'for recording, time in seconds cannot be longer than {0}'
            raise VoiceKeyException(msg.format(MAX_RECORDING_SEC))

        # detect whether given a numpy array directly
        # TO-DO: self.array_in handling needs code review
        source = file_in
        self.array_in = []
        if type(source) in [np.ndarray]:
            self.array_in = source
            file_in = '<array len={0}>'.format(len(source))
        self.file_in, self.file_out = file_in, file_out

        # Configuration defaults:
        self.config = {'msPerChunk': 2,
                       'signaler': None,
                       'autosave': True,
                       'chnl_in': 0,  # pyo.pa_get_default_input()
                       # 'chnl_out': 2,  # pyo.pa_get_default_output() no go
                       'start': 0,
                       'stop': -1,
                       'vol': 0.99,
                       'low': 100,
                       'high': 3000,
                       'threshold': 10,
                       'baseline': 0,
                       'more_processing': True,
                       'zero_crossings': True}
        self.baseline = self.config['baseline']
        self.bad_baseline = False
        self.stopped = False
        self.msPerChunk = float(self.config['msPerChunk'])
        if not 0.65 <= self.msPerChunk <= 32:
            msg = 'msPerChunk should be 0.65 to 32; suggested = 2'
            raise ValueError(msg)


    def _set_source(self):
        """Data source: file_in, array, or microphone
        if os.path.isfile(self.file_in):
            _rate, self._sndTable = table_from_file(self.file_in,
            if _rate != self.rate:
                print('file sample rate differs from the voice-key rate.')
            self._source = pyo.TableRead(self._sndTable,
            self.sec = self._sndTable.getDur()
        elif len(self.array_in):
            self._sndTable = table_from_samples(self.array_in,
            self._source = pyo.TableRead(self._sndTable,
            self.sec = self._sndTable.size / self.rate
            # fall through to source = microphone
            ch = self.config['chnl_in']
            self._source = pyo.Input(chnl=ch, mul=self.config['vol'])

    def _set_defaults(self):
        """Set remaining defaults, initialize lists to hold summary stats
        # adjust self.sec based on start, stop times:
        if (self.config['start'], self.config['stop']) != (0, -1):
            if self.config['stop'] > self.config['start']:
                self.sec = self.config['stop'] - self.config['start']
            elif self.config['start'] <= self.sec:
                self.sec = self.sec - self.config['start']
        self.chunks = int(self.sec * 1000. / self.msPerChunk)  # ideal no slip
        # total chunk count and current-chunk index:
        self.count = 0

        self.filename = self.file_out or 'rec.wav'
        self.filesize = None

        # timing data for diagnostics
        self.elapsed = 0
        self.t_enter = []  # time at chunk entry
        self.t_exit = []  # time at chunk exit
        self.t_proc = []  # proportion of chunk-time spent doing _do_chunk

        # data cache: = []  # raw unprocessed data, in chunks
        self.power = []
        self.power_bp = []
        self.power_above = []
        self.zcross = []
        self.max_bp = 0
        self.max_bp_chunk = None
        bandpass_pre_cache(rate=self.rate)  # for faster bandpass filtering

        # default event parameters:
        self.event_detected = False
        self.event_lag = 0  # lag required to detect the event prior to trip()
        self.event_time = 0  # becomes time of detected event = time at trip()
        self.event_onset = 0  # best estimate of the onset of the event

    def _set_signaler(self):
        """Set the signaler to be called by trip()
        if not self.config['signaler']:
            self.config['signaler'] = None  # _BaseVoiceKeySignal()
        self.event_signaler = self.config['signaler']

    def _set_tables(self):
        """Set up the pyo tables (allocate memory, etc).

        One source -> three pyo tables: chunk=short, whole=all, baseline.
        triggers fill tables from self._source; make triggers in .start()
        sec_per_chunk = self.msPerChunk / 1000.
        self._chunktable = pyo.NewTable(length=sec_per_chunk)
        self._wholetable = pyo.NewTable(length=self.sec)
        if self.baseline < TOO_QUIET:
            self._baselinetable = pyo.NewTable(length=T_BASELINE_OFF)

    def _set_baseline(self):
        """Set self.baseline = rms(silent period) using _baselinetable data.

        Called automatically (via pyo trigger) when the baseline table
        is full. This is better than using chunks (which have gaps between
        them) or the whole table (which can be very large = slow to work
        data = np.array(self._baselinetable.getTable())
        tstart = int(T_BASELINE_ON * self.rate)
        segment_power = rms(data[tstart:])

        # Look for bad baseline period:
        if self.baseline > TOO_LOUD:
            self.bad_baseline = True

        # Dubiously quiet is bad too:
        if segment_power < TOO_QUIET:
            msg = ('Baseline period is TOO quiet\nwrong input '
                   'channel selected? device-related initial delay?')
            raise ValueError(msg)

        self.baseline = max(segment_power, 1)

    def _process(self, chunk):
        """Calculate and store basic stats about the current chunk.

        This gets called every chunk -- keep it efficient, esp 32-bit python
        # band-pass filtering:
        if self.config['more_processing']:
            bp_chunk = bandpass(chunk, self.config['low'],
                                self.config['high'], self.rate)
            bp_chunk = chunk

        # loudness after bandpass filtering:

        _mx = max(bp_chunk)
        if _mx > self.max_bp:
            self.max_bp = _mx
            self.max_bp_chunk = self.count  # chunk containing the max

        if self.config['more_processing']:
            # more bandpass
            bp3k_chunk = bandpass(chunk, self.config['low'], 3000, self.rate)
            bp8k_chunk = bandpass(chunk, self.config['low'], 8000, self.rate)
            # "content filtered speech" (~ affect only):
            bp2k8k_chunk = bandpass(chunk, 2000, 8000, self.rate)

            # basic loudness:

            # above a threshold or not:
            above_01 = int(self.power[self.count] > self.config['threshold'])

        if self.config['zero_crossings']:
            # zero-crossings per ms:
            zx = zero_crossings(bp_chunk)
            self.zcross.append(np.sum(zx) / self.msPerChunk)

    def detect(self):
        """Override to define a detection algorithm.
            if condition:

        See SimpleThresholdVoiceKey for a minimal usage example, or
        VoicelessPlosiveVoiceKey for a more involved one.
        raise NotImplementedError('override; see SimpleThresholdVoiceKey')

    def trip(self):
        """Trip the voice-key; does not stop recording.
        # calls .start() on the event-signaler thread. Only `detect()` should
        # call `trip()`. Customize `.detect()` rather than the logic here.

        self.event_detected = True
        self.event_time = self.elapsed
        if hasattr(self, 'event_signaler') and self.event_signaler:

    def _do_chunk(self):
        """Core function to handle a chunk (= a few ms) of input.

        There can be small temporal gaps between or within chunks, i.e.,
        `slippage`. Adjust several parameters until this is small: msPerChunk,
        and what processing is done within ._process().

        A trigger (`_chunktrig`) signals that `_chunktable` has been filled
        and has set `_do_chunk` as the function to call upon triggering.
        `.play()` the trigger again to start recording the next chunk.
        if self.stopped:

        self.elapsed = self.t_enter[-1] - self.t_enter[0]
        self.t_baseline_has_elapsed = bool(self.elapsed > T_BASELINE_PERIOD)

        # Get the table content as np.array
        chunk = np.asarray(self._chunktable.getTable())
        chunk = np.int16(chunk * 2 ** 15)

        # Calc basic stats, then use to detect features
        self.detect()  # conditionally call trip()

        # Trigger a new chunk recording, or stop if stopped or time is up:
        t_end = get_time()
        if t_end - self.t_enter[0] < self.sec:
            if not self.stopped:
        # *** triggers the next chunk ***
                self.count += 1

    def start(self, silent=False):
        """Start reading and processing audio data from a file or microphone.
        if self.stopped:
            raise VoiceKeyException('cannot start a stopped recording')
        self.t_start = get_time()

        # triggers: fill tables, call _do_chunk & _set_baseline:
        self._chunktrig = pyo.Trig()
        self._chunkrec = pyo.TrigTableRec(self._source, self._chunktrig,
        self._chunklooper = pyo.TrigFunc(self._chunkrec["trig"],
        self._wholetrig = pyo.Trig()
        self._wholerec = pyo.TrigTableRec(self._source, self._wholetrig,
        self._wholestopper = pyo.TrigFunc(self._wholerec["trig"], self.stop)

        # skip if a baseline value was given in config:
        if not self.baseline:
            self._baselinetrig = pyo.Trig()
            self._baselinerec = pyo.TrigTableRec(self._source,
            self._calc_baseline = pyo.TrigFunc(self._baselinerec["trig"],

        # send _source to sound-output (speakers etc) as well:
        if self.file_in and not silent:

        # start calling self._do_chunk by flipping its trigger;
        # _do_chunk then triggers itself via _chunktrigger until done:

        return self

    def slippage(self):
        """Diagnostic: Ratio of the actual (elapsed) time to the ideal time.

        Ideal ratio = 1 = sample-perfect acquisition of msPerChunk, without
        any gaps between or within chunks. 1. / slippage is the proportion of
        samples contributing to chunk stats.
        if len(self.t_enter) > 1:
            diffs = np.array(self.t_enter[1:]) - np.array(self.t_enter[:-1])
            ratio = np.mean(diffs) * 1000. / self.msPerChunk
            ratio = 0
        return ratio

    def started(self):
        """Boolean property, whether `.start()` has been called.
        return bool(hasattr(self, '_chunklooper'))  # .start() has been called

    def stop(self):
        """Stop a voice-key in progress.

        Ends and saves the recording if using microphone input.
        # Will be stopped at self.count (= the chunk index), but that is less
        # reliable than self.elapsed due to any slippage.

        if self.stopped:
        self.stopped = True
        self.t_stop = get_time()

        if self.config['autosave']:

        # Calc the proportion of the available time spent doing _do_chunk:
        for ch in range(len(self.t_exit)):
            t_diff = self.t_exit[ch] - self.t_enter[ch]
            self.t_proc.append(t_diff * 1000 / self.msPerChunk)

    def join(self, sec=None):
        """Sleep for `sec` or until end-of-input, and then call stop().
        sleep(sec or self.sec - self.elapsed)

    def wait_for_event(self, plus=0):
        """Start, join, and wait until the voice-key trips, or it times out.

        Optionally wait for some extra time, `plus`, before calling `stop()`.
        if not self.started:
        while not self.event_time and not self.stopped:
            sleep(self.msPerChunk / 1000.)
        if not self.stopped:
            naptime = min(plus, self.sec - self.elapsed)  # approx...
            if naptime > 0:
        # next sleep() helps avoid pyo error:
        #    "ReferenceError: weakly-referenced object no longer exists"
        sleep(1.5 * self.msPerChunk / 1000.)

        return self.elapsed

    def save(self, ftype='', dtype='int16'):
        """Save new data to file, return the size of the saved file (or None).

        The file format is inferred from the filename extension, e.g., `flac`.
        This will be overridden by the `ftype` if one is provided; defaults to
        `wav` if nothing else seems reasonable. The optional `dtype` (e.g.,
        `int16`) can be any of the sample types supported by `pyo`.
        if self.file_in or not self.count:

        self.save_fmt = os.path.splitext(self.filename)[1].lstrip('.')
        fmt = ftype or self.save_fmt or 'wav'
        if not self.filename.endswith('.' + fmt):
            self.filename += '.' + fmt

        # Save the recording (continuous, non-chunked):
        end_index = int(self.elapsed * self.rate)  # ~samples
        if end_index < self._wholetable.size:
            dataf = np.asarray(self._wholetable.getTable()[:end_index])
            samples_to_file(dataf, self.rate, self.filename,
                            fmt=fmt, dtype=dtype)
            self.sec = pyo.sndinfo(self.filename)[1]
            table_to_file(self._wholetable, self.filename,
                          fmt=fmt, dtype=dtype)
        self.filesize = os.path.getsize(self.filename)
        return self.filesize

class SimpleThresholdVoiceKey(_BaseVoiceKey):
    """Class for simple threshold voice key (loudness-based onset detection).

    The "hello world" of voice-keys.

    def detect(self):
        """Trip if the current chunk's audio power > 10 * baseline loudness.
        if self.event_detected or not self.baseline:
        current = self.power[-1]
        threshold = 10 * self.baseline
        if current > threshold:

[docs]class OnsetVoiceKey(_BaseVoiceKey): """Class for speech onset detection. Uses bandpass-filtered signal (100-3000Hz). When the voice key trips, the best voice-onset RT estimate is saved as `self.event_onset`, in sec. """
[docs] def detect(self): """Trip if recent audio power is greater than the baseline. """ if self.event_detected or not self.baseline: return window = 5 # recent hold duration window, in chunks threshold = 10 * self.baseline conditions = all([x > threshold for x in self.power_bp[-window:]]) if conditions: self.event_lag = window * self.msPerChunk / 1000. self.event_onset = self.elapsed - self.event_lag self.trip() self.event_time = self.event_onset
[docs]class OffsetVoiceKey(_BaseVoiceKey): """Class to detect the offset of a single-word utterance. """ def __init__(self, sec=10, file_out='', file_in='', delay=0.3, **kwargs): """Record and ends the recording after speech offset. When the voice key trips, the best voice-offset RT estimate is saved as `self.event_offset`, in seconds. :Parameters: `sec`: duration of recording in the absence of speech or other sounds. `delay`: extra time to record after speech offset, default 0.3s. The same methods are available as for class OnsetVoiceKey. """ config = {'sec': sec, 'file_out': file_out, 'file_in': file_in, 'delay': delay} kwargs.update(config) super(OffsetVoiceKey, self).__init__(**kwargs) def detect(self): """Listen for onset, offset, delay, then end the recording. """ if self.event_detected or not self.baseline: return if not self.event_onset: window = 5 # chunks threshold = 10 * self.baseline conditions = all([x > threshold for x in self.power_bp[-window:]]) if conditions: self.event_lag = window * self.msPerChunk / 1000. self.event_onset = self.elapsed - self.event_lag self.event_offset = 0 elif not self.event_offset: window = 25 threshold = 10 * self.baseline # segment = np.array(self.power_bp[-window:]) conditions = all([x < threshold for x in self.power_bp[-window:]]) # conditions = np.all(segment < threshold) if conditions: self.event_lag = window * self.msPerChunk / 1000. self.event_offset = self.elapsed - self.event_lag self.event_time = self.event_offset # for plotting elif self.elapsed > self.event_offset + self.config['delay']: self.trip() self.stop()
# ----- Convenience classes ------------------------------------------------- class Recorder(_BaseVoiceKey): """Convenience class: microphone input only (no real-time analysis). Using `record()` is like `.join()`: it will block execution. But it will also try to save the recording automatically even if interrupted (whereas `.start().join()` will not do so). This might be especially useful when making long recordings. """ def __init__(self, sec=2, filename='rec.wav'): super(Recorder, self).__init__(sec, file_out=filename) # def _set_defaults(self): # pass def __del__(self): if hasattr(self, 'filename') and not os.path.isfile(self.filename): def _set_baseline(self): pass def detect(self): pass def _process(self, *args, **kwargs): pass def record(self, sec=None): try: self.start().join(sec) except Exception: raise class Player(_BaseVoiceKey): """Convenience class: sound output only (no real-time analysis). """ def __init__(self, sec=None, source='rec.wav', start=0, stop=-1, rate=44100): if type(source) in [np.ndarray]: sec = len(source) / rate elif os.path.isfile(source): sec = pyo.sndinfo(source)[1] config = {'start': start, 'stop': stop} super(Player, self).__init__(sec, file_in=source, **config) # def _set_defaults(self): # ideally override but need more refactoring # pass def _set_baseline(self): pass def detect(self): pass def _process(self, *args, **kwargs): pass def play(self, sec=None): self.start().join(sec) # ----- pyo initialization (essential) ------------------------------------- def pyo_init(rate=44100, nchnls=1, buffersize=32, duplex=1): """Start and boot a global pyo server, restarting if needed. """ global pyo_server if rate < 16000: raise ValueError('sample rate must be 16000 or higher') # re-init if hasattr(pyo_server, 'shutdown'): pyo_server.stop() sleep(0.25) # make sure enough time passes for the server to shutdown pyo_server.shutdown() sleep(0.25) pyo_server.reinit(sr=rate, nchnls=nchnls, buffersize=buffersize, duplex=duplex) else: pyo_server = pyo.Server(sr=rate, nchnls=nchnls, # 1 = mono buffersize=buffersize, # ideal = 64 or higher duplex=duplex) # 1 = input + output pyo_server.boot().start() # avoid mac issue of losing first 0.5s if no sound played for ~1 minute: if sys.platform == 'darwin': z2 = np.zeros(2) _sndTable = pyo.DataTable(size=2, init=z2.T.tolist(), chnls=nchnls) _snd = pyo.TableRead(_sndTable, freq=rate, mul=0) time.sleep(0.510)

Back to top