diff --git a/Sprachausgabe/osc_receiver.py b/Sprachausgabe/osc_receiver.py new file mode 100644 index 0000000..f75d29b --- /dev/null +++ b/Sprachausgabe/osc_receiver.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +""" +06_osc_receiver.py + +Very simple OSC receiver demo. + +It shows that different OSC addresses are mapped +to different handler functions. +""" + +from pathlib import Path + +from pythonosc.dispatcher import Dispatcher +from pythonosc.osc_server import BlockingOSCUDPServer +from piper_module import Narrator, Language, Quality, LLM + + +IP = "100.83.246.218" +PORT = 9001 + +def on_read(address, *args): + print(f"on_read() address={address} args={args}") + narrator.read(str(args[0])) + +def on_write_story(address, *args): + print(f"on_write_story() address={address} args={args}") + story = narrator.write_story(str(args[0]), str(args[1])) + print(story) + +def on_read_story(address, *args): + print(f"on_read_story() address={address} args={args}") + story = narrator.write_story(str(args[0]), str(args[1])) + narrator.read(story) + + +def main() -> None: + dispatcher = Dispatcher() + + dispatcher.map("/read", on_read) + dispatcher.map("/write_story", on_write_story) + dispatcher.map("/read_story", on_read_story) + + server = BlockingOSCUDPServer((IP, PORT), dispatcher) + + global narrator + llm = LLM("qwen2.5-7b-instruct-1m","http://100.83.153.234:1234/api/v1/chat" ) + narrator = Narrator(Language.de_DE, "thorsten", Path("C:\Interaktion\WerWolf\piper-voices"), llm) + + print(f"Listening for OSC on {IP}:{PORT}") + print("Try sending:") + print(" /read") + print(" /write_story") + print(" /read_story") + + + server.serve_forever() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/Sprachausgabe/output0.wav b/Sprachausgabe/output0.wav new file mode 100644 index 0000000..69ba7fb Binary files /dev/null and b/Sprachausgabe/output0.wav differ diff --git a/Sprachausgabe/piper_module.py b/Sprachausgabe/piper_module.py new file mode 100644 index 0000000..5368276 --- /dev/null +++ b/Sprachausgabe/piper_module.py @@ -0,0 +1,126 @@ +import wave +from piper import PiperVoice +import pygame +import time +import os +import sys +import pyaudio +import requests +from pathlib import Path +from enum import Enum + +class Language(Enum): + en_US = "en_US" + de_DE = "de_DE" + en_GB = "en_GB" + + +class Quality(Enum): + high = 3 + medium = 2 + low = 1 + x_low = 0 + +class LLM: + def __init__(self, model: str, url: str): + self.url = url + self.model = model + + +class Narrator: + def __init__(self, language: Language, name: str, repo: Path, llm: LLM): + + for quality in Quality: + filename = f"{language.name}-{name}-{quality.name}.onnx" + self.path = repo / language.name / name / quality.name / filename + + print("Teste:", self.path) + + if self.path.exists(): + print("Gefunden:", self.path) + break + else: + raise FileNotFoundError("Keine passende Voice-Datei gefunden") + + print("load voice") + + self.voice = PiperVoice.load(self.path) + + # for narrator texts -> connect to LLM + print(" create llm") + self.llm = llm + + + # Stream audio chunks and play them immediately + def read(self, text: str): + p = pyaudio.PyAudio() + + stream = None + + for audio_chunk in self.voice.synthesize(text): + + # Stream erst öffnen, wenn erster Chunk da ist + if stream is None: + stream = p.open( + format=p.get_format_from_width(audio_chunk.sample_width), + channels=audio_chunk.sample_channels, + rate=audio_chunk.sample_rate, + output=True + ) + + # Audio sofort abspielen + stream.write(audio_chunk.audio_int16_bytes) + + if stream: + stream.stop_stream() + stream.close() + + p.terminate() + + + # play selected audiofile and delete it after playing + def read_audio(self, number: int): + #setup + filename = f"output{number}.wav" + pygame.mixer.init() + + #play next audio + sound = pygame.mixer.Sound(filename) + channel = sound.play() + + # wait until the sound has finished playing + while channel.get_busy(): + time.sleep(0.1) + + # delete now obsolete file + os.remove(filename) + + pygame.mixer.quit() + + + # create wavefile from text + def create_audio(self, texts: list[str]): + + for i, text in enumerate(texts): + filename = f"output{i}.wav" + + with wave.open(f"output{i}.wav", "wb") as wav_file: + self.voice.synthesize_wav(text, wav_file) + + # create a story from a base text + def write_story(self, text, count): + + session = requests.Session() + session.trust_env = False + + payload = { + "model" : self.llm.model, + "input": "Formuliere:" + text +" zu einem erzählerischen Sprachtext um. Nicht mehr als " + count + " Sätze. Handlungsaufforderungen sollen klar heraushörbar sein." + } + + response = requests.post(self.llm.url, json=payload, timeout=30) + response.raise_for_status() + data = response.json() + story = data["output"][0]["content"] + + return story \ No newline at end of file