Das deutsche Python-Forum

Code: Alles auswählen

from array import array
from io import BytesIO
import json
import os
from pyaudio import PyAudio, paInt16
import time

from pydub import *
from audioop import *

LANG = de-de

DEBUG = True

# Mic Settings
THRESHOLD = 5
RATE = 16000
TIMEOUT_LENGTH = 1

CHUNK = 4000
SHORT_NORMALIZE = (1.0 / 32768.0)
NORMALIZE = 1 / 32768


class AudioDetection:
    def __init__(self):
        p = PyAudio()
        self.stream = p.open(rate=RATE,
                             channels=1,
                             format=paInt16,
                             input=True,
                             output=True,
                             input_device_index=None,  # None use default
                             output_device_index=None,  # None use default
                             frames_per_buffer=8000)
        self.stream.start_stream()
        self.stt = Vosk(LANG)

    @staticmethod
    def rms(frame):
        shorts = array('h', frame)
        sum_squares = sum(
            (sample * NORMALIZE) ** 2
            for sample in shorts
        )
        return (sum_squares / len(shorts)) ** 0.5 * 1000

    def detect(self, frame):
        if DEBUG:
            print("Noise detected")
        rec = [frame]
        current = time.time()
        end = time.time() + TIMEOUT_LENGTH

        while current <= end:
            data = self.stream.read(CHUNK)
            if self.rms(data) >= THRESHOLD:
                end = time.time() + TIMEOUT_LENGTH
            current = time.time()
            rec.append(data)

        # rec = AudioSegment.silent(rec)

        if DEBUG:
            print("Play Stream")
            self.stream.write(b''.join(rec))
        text = self.stt.run(b''.join(rec))
        if DEBUG:
            print("Listened: ", text)
        print('Listening...')

    def run(self):
        print("Listening...")
        while True:
            mic_input = self.stream.read(CHUNK)
            rms_val = self.rms(mic_input)
            if rms_val >= THRESHOLD:
                self.detect(mic_input)


class Vosk:
    def __init__(self, language):
        from vosk import Model, KaldiRecognizer
        if not os.path.exists("models/vosk/" + language):
            print("Please download the model from https://alphacephei.com/vosk/models and unpack as "
                  "'" + language + "' in 'models/vosk'.")
            exit(1)

        model = Model("models/vosk/" + language)
        self.rec = KaldiRecognizer(model, RATE)

    def run(self, audiostream):
        audiostream = BytesIO(audiostream)
        while True:
            data = audiostream.read(4000)
            if len(data) == 0:
                break
            if self.rec.AcceptWaveform(data):
                if DEBUG:
                    res = json.loads(self.rec.Result())
                    print("DEBUG: " + res['text'])

        fres = json.loads(self.rec.FinalResult())
        return fres['text']


if __name__ == '__main__':
    recorder = AudioDetection()
    recorder.run()

Code: Alles auswählen

def echocancel(outputdata, inputdata):
    pos = audioop.findmax(outputdata, 800)    # one tenth second
    out_test = outputdata[pos*2:]
    in_test = inputdata[pos*2:]
    ipos, factor = audioop.findfit(in_test, out_test)
    # Optional (for better cancellation):
    # factor = audioop.findfactor(in_test[ipos*2:ipos*2+len(out_test)],
    #              out_test)
    prefill = '\0'*(pos+ipos)*2
    postfill = '\0'*(len(inputdata)-len(prefill)-len(outputdata))
    outputdata = prefill + audioop.mul(outputdata, 2, -factor) + postfill
    return audioop.add(inputdata, outputdata, 2)