tap detector

mic_feed.py

import pyaudio
import numpy as np
from collections import namedtuple

CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 44100

Chunk = namedtuple('Chunk', 'data time')


class MicrophoneFeed(object):
    def __init__(self):
        self.enabled = True
        self.p = pyaudio.PyAudio()
        self.stream = self.p.open(format=FORMAT, channels=CHANNELS, rate=RATE,
                                  input=True, frames_per_buffer=CHUNK)
        self.t = 0.0

    def __iter__(self):
        while self.enabled:
            data = self.stream.read(CHUNK)
            chunk = np.fromstring(data, 'int16')
            yield Chunk(chunk, self.t)
            self.t += CHUNK / RATE

    def close(self):
        self.enabled = False

tap_detect.py

from mic_feed import MicrophoneFeed
from collections import namedtuple

THRESHOLD = 19000000

Tap = namedtuple('Tap', 'time')


def _print_tap_timestamp(tap):
    print(tap.time)


class AmplitudeDetector(object):
    """Call a sufficiently noisy event a tap."""

    def __init__(self, feed, threshold=THRESHOLD):
        self.feed = feed
        self.threshold = threshold

    def detect(self, chunk):
        if abs(chunk.data).sum() > self.threshold:
            return True

    def __iter__(self):
        for tap in self.feed:
            if self.detect(tap):
                yield Tap(tap.time)


class RateLimitedDetector(AmplitudeDetector):
    """ detect taps with a minimum number of seconds in between"""
    def __init__(self, d, rate_limit=1, threshold=THRESHOLD):
        AmplitudeDetector.__init__(self, d, threshold)
        self.feed = d
        self.last_tap = -rate_limit
        self.rate_limit = rate_limit

    def __iter__(self):
        for tap in self.feed:
            if self.detect(tap):
                if tap.time - self.last_tap > self.rate_limit:
                    self.last_tap = tap.time
                    yield tap


class MultiTapDetector(AmplitudeDetector):
    """ detect N sequential taps """
    def __init__(self, d, rate_limit=1, num_of_taps=2, threshold=THRESHOLD):
        AmplitudeDetector.__init__(self, d, threshold)
        self.feed = d
        self.last_tap = -rate_limit
        self.rate_limit = rate_limit
        self.num_of_taps = num_of_taps
        self.taps_per_rate = []

    def __iter__(self):
        for tap in self.feed:
            if self.detect(tap):
                if tap.time - self.last_tap > self.rate_limit:
                    self.last_tap = tap.time
                    self.taps_per_rate = []
                else:
                    self.taps_per_rate.append(tap.time)
                    if len(self.taps_per_rate) == self.num_of_taps:
                        yield tap


def tap_detect(callback=None):
    callback = callback or _print_tap_timestamp
    feed = MicrophoneFeed()
    detector = AmplitudeDetector(feed, threshold=10000000)
    for tap in detector:
        callback(tap)


def rate_limited_tap_detect(rate=1, callback=None):
    callback = callback or _print_tap_timestamp
    feed = MicrophoneFeed()
    detector = RateLimitedDetector(feed, rate_limit=rate, threshold=10000000)
    for tap in detector:
        callback(tap)


def multi_tap_detect(num=2, rate=1, callback=None):
    callback = callback or _print_tap_timestamp
    feed = MicrophoneFeed()
    detector = MultiTapDetector(feed, rate, num)
    for tap in detector:
        callback(tap)