ich beschäftige mich seit einiger Zeit mit der Manipulation von Audioframes um dessen Qualität steigern zu können um in meinem Fall die Erkennung von Sprache-zu-Text (Vosk) zu verbessern.
Geplant ist:
- Rauschen entfernen
- Stille anhängen um das zerhacken des streams zu vermeiden
- Komressor Funktion um die stimme anheben zu können
- Weitere verbesserungen
Mein Scrit sieht derzeit so aus:
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()
Mein Grundgedanke ist das ich ersmal die byteorder berücksichten sollte So das man mit Little Endian arbeitet doch wie gehts weiter?
Mir gehen langsam die Ideen aus und hoffe ihr könnt mir dabei weiterhelfen.
MfG
Toaster1337
