Upload files to "Sprachausgabe"

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()

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