Make sure to include `【message_idx†source】` markers to provide citations based on this file, where [message_idx] is provided at the beginning of this message#!/usr/bin/env python3 """ Ulrike - Lokaler Sprachassistent Hotword Detection (Vosk) → Audio Recording → Whisper STT → OpenHAB """ import os import queue import wave import json import logging import subprocess import sounddevice as sd from vosk import Model, KaldiRecognizer from pathlib import Path # === Konfiguration === HOTWORD = "ulrike" MODEL_PATH = "/opt/whisper/vosk-model-small-de-0.15" WHISPER_MODEL = "/opt/whisper/models/ggml-base.bin" WHISPER_BIN = "/usr/local/bin/whisper" AUDIO_FILE = "/opt/whisper/aufnahme.wav" RECORD_SECONDS = 5 SAMPLE_RATE = 16000 CHANNELS = 1 BLOCKSIZE = 8000 LOGFILE = "/opt/whisper/ulrike.log" # === Logging einrichten === logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(LOGFILE), logging.StreamHandler() # Auch in Console ausgeben ] ) # === Audio-Queue === audio_queue = queue.Queue() def audio_callback(indata, frames, time, status): """Callback für Audio-Input-Stream""" if status: logging.warning(f"Audio-Status: {status}") audio_queue.put(bytes(indata)) def save_audio_buffer(buffer_data, filename): """Speichert Audio-Puffer als WAV-Datei""" try: with wave.open(filename, 'wb') as wf: wf.setnchannels(CHANNELS) wf.setsampwidth(2) # int16 = 2 bytes wf.setframerate(SAMPLE_RATE) wf.writeframes(b''.join(buffer_data)) return True except Exception as e: logging.error(f"Fehler beim Speichern der Audio-Datei: {e}") return False def transcribe_with_whisper(audio_file): """Führt Whisper-Transkription durch und gibt Text zurück""" try: result = subprocess.run( [WHISPER_BIN, "-m", WHISPER_MODEL, "-f", audio_file, "-l", "de"], capture_output=True, text=True, timeout=30 ) # Whisper gibt Format aus: [00:00.000 --> 00:05.000] Text hier # Wir extrahieren nur den Text lines = result.stdout.strip().split('\n') transcript_lines = [line for line in lines if '-->' in line or (line and not line.startswith('[') and not line.startswith('whisper'))] # Bereinige Zeitstempel transcript = ' '.join(transcript_lines) if '-->' in transcript: transcript = transcript.split(']')[-1].strip() return transcript.strip() if transcript else None except subprocess.TimeoutExpired: logging.error("Whisper Timeout nach 30 Sekunden") return None except Exception as e: logging.error(f"Fehler bei Whisper-Transkription: {e}") return None def send_to_openhab(text): """Sendet Transkript an OpenHAB (Platzhalter)""" # TODO: Implementiere REST API Call zu OpenHAB logging.info(f"[OpenHAB] Würde senden: {text}") # Beispiel: # import requests # requests.post("http://openhab-pi:8080/rest/items/VoiceCommand", # json={"value": text}) def main(): """Hauptprogramm""" # Validierung if not os.path.exists(MODEL_PATH): logging.error(f"Vosk-Modell nicht gefunden: {MODEL_PATH}") return 1 if not os.path.exists(WHISPER_BIN): logging.error(f"Whisper Binary nicht gefunden: {WHISPER_BIN}") return 1 # Vosk initialisieren logging.info("Lade Vosk-Modell...") model = Model(MODEL_PATH) recognizer = KaldiRecognizer(model, SAMPLE_RATE) recognizer.SetWords(True) logging.info(f"🎤 Ulrike lauscht... Hotword: '{HOTWORD}'") logging.info(f"📊 Sample Rate: {SAMPLE_RATE} Hz, Aufnahmedauer: {RECORD_SECONDS}s") try: with sd.RawInputStream( samplerate=SAMPLE_RATE, blocksize=BLOCKSIZE, dtype='int16', channels=CHANNELS, callback=audio_callback ): while True: # Warte auf Audio-Daten data = audio_queue.get() # Vosk-Erkennung if recognizer.AcceptWaveform(data): result = json.loads(recognizer.Result()) text = result.get("text", "").lower() if not text: continue # Debug: Zeige was erkannt wurde logging.debug(f"Erkannt: {text}") # Hotword-Check if HOTWORD in text: logging.info(f"✨ Hotword erkannt: '{text}'") # Sammle Audio für Aufnahme logging.info(f"📼 Nehme {RECORD_SECONDS} Sekunden auf...") audio_buffer = [] frames_needed = int((SAMPLE_RATE * RECORD_SECONDS) / BLOCKSIZE) for _ in range(frames_needed): audio_buffer.append(audio_queue.get()) # Speichere WAV if save_audio_buffer(audio_buffer, AUDIO_FILE): logging.info("💾 Aufnahme gespeichert") # Transkribiere mit Whisper logging.info("🔄 Starte Whisper-Transkription...") transcript = transcribe_with_whisper(AUDIO_FILE) if transcript: logging.info(f"📝 Transkript: '{transcript}'") send_to_openhab(transcript) else: logging.warning("⚠️ Keine Transkription erhalten") # Zurück zum Lauschen logging.info("👂 Lausche weiter auf Hotword...") except KeyboardInterrupt: logging.info("\n🛑 Ulrike wurde gestoppt") return 0 except Exception as e: logging.error(f"❌ Kritischer Fehler: {e}", exc_info=True) return 1 if __name__ == "__main__": exit(main())

All the files uploaded by the user have been fully loaded. Searching won't provide additional information.

Make sure to include `【message_idx†source】` markers to provide citations based on this file, where [message_idx] is provided at the beginning of this message#!/usr/bin/env python3 """ Ulrike - Lokaler Sprachassistent (Version mit Pre-Buffer) Hotword Detection (Vosk) -> Audio Recording -> Whisper STT -> OpenHAB """ import os import queue import wave import json import logging import subprocess import sounddevice as sd import re from vosk import Model, KaldiRecognizer from collections import deque # === Konfiguration === HOTWORD = "ulrike" MODEL_PATH = "/opt/whisper/vosk-model-small-de-0.15" WHISPER_MODEL = "/opt/whisper/models/ggml-base.bin" WHISPER_BIN = "/usr/local/bin/whisper" AUDIO_FILE = "/opt/whisper/aufnahme.wav" # Timings RECORD_SECONDS = 4 # Aufnahmezeit NACH dem Hotword PRE_BUFFER_SECONDS = 2.5 # Zeit VOR der Hotword-Erkennung (um Satzanfang zu retten) SAMPLE_RATE = 16000 CHANNELS = 1 BLOCKSIZE = 8000 LOGFILE = "/opt/whisper/ulrike.log" # === Logging === logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[logging.FileHandler(LOGFILE), logging.StreamHandler()] ) # === Audio-Handling === audio_queue = queue.Queue() # Ringpuffer für die "Vergangenheit" pre_buffer_size = int((SAMPLE_RATE * PRE_BUFFER_SECONDS) / BLOCKSIZE) pre_buffer = deque(maxlen=pre_buffer_size) def audio_callback(indata, frames, time, status): """Füllt die Queue für Vosk und den Puffer für die Aufnahme""" if status: logging.warning(f"Audio-Status: {status}") data = bytes(indata) audio_queue.put(data) pre_buffer.append(data) def save_audio_buffer(buffer_data, filename): """Speichert die Liste von Audio-Chunks als WAV""" try: with wave.open(filename, 'wb') as wf: wf.setnchannels(CHANNELS) wf.setsampwidth(2) wf.setframerate(SAMPLE_RATE) wf.writeframes(b''.join(buffer_data)) return True except Exception as e: logging.error(f"Speicherfehler: {e}") return False def transcribe_with_whisper(audio_file): """Ruft Whisper auf und filtert die Ausgabe""" try: # --no-timestamps sorgt für saubere Textausgabe result = subprocess.run( [WHISPER_BIN, "-m", WHISPER_MODEL, "-f", audio_file, "-l", "de", "--no-timestamps"], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: logging.error(f"Whisper Fehler: {result.stderr}") return None output = result.stdout.strip() # Entferne Whisper-Halluzinationen und Metadaten lines = [] for line in output.splitlines(): line = re.sub(r"\[.*?\]", "", line).strip() # Entfernt [Musik] etc. if line and not line.startswith("whisper_"): lines.append(line) return " ".join(lines) if lines else None except Exception as e: logging.error(f"Whisper-Prozessfehler: {e}") return None def send_to_openhab(text): """Platzhalter für deine OpenHAB-Anbindung""" logging.info(f"[OpenHAB] Sende: {text}") # Hier kommt dein requests.post hin def main(): if not os.path.exists(MODEL_PATH) or not os.path.exists(WHISPER_BIN): logging.error("Pfade prüfen! Modell oder Whisper-Binary fehlt.") return 1 logging.info("Lade Vosk-Modell...") model = Model(MODEL_PATH) recognizer = KaldiRecognizer(model, SAMPLE_RATE) logging.info(f"🎤 Ulrike bereit (Puffer: {PRE_BUFFER_SECONDS}s)") try: with sd.RawInputStream( samplerate=SAMPLE_RATE, blocksize=BLOCKSIZE, dtype='int16', channels=CHANNELS, callback=audio_callback ): while True: data = audio_queue.get() if recognizer.AcceptWaveform(data): result = json.loads(recognizer.Result()) text = result.get("text", "").lower() if HOTWORD in text: logging.info(f"✨ Hotword: '{text}'") # Kombiniere Vergangenheit (Puffer) mit der Zukunft audio_buffer = list(pre_buffer) logging.info(f"📼 Aufnahme läuft...") frames_needed = int((SAMPLE_RATE * RECORD_SECONDS) / BLOCKSIZE) for _ in range(frames_needed): audio_buffer.append(audio_queue.get()) if save_audio_buffer(audio_buffer, AUDIO_FILE): logging.info("🔄 Transkription...") transcript = transcribe_with_whisper(AUDIO_FILE) if transcript: # Optional: Hotword vom Anfang des Textes löschen clean_text = re.sub(rf"^{HOTWORD}\s*", "", transcript, flags=re.IGNORECASE) logging.info(f"📝 Befehl: '{clean_text}'") send_to_openhab(clean_text) # Queue leeren, um 'input overflow' nach der Rechenpause zu vermeiden while not audio_queue.empty(): audio_queue.get_nowait() logging.info("👂 Lausche weiter...") except KeyboardInterrupt: logging.info("🛑 Stopp") except Exception as e: logging.error(f"❌ Fehler: {e}", exc_info=True) if __name__ == "__main__": main()

All the files uploaded by the user have been fully loaded. Searching won't provide additional information.