joke_bot.py

#!/usr/bin/env python3
"""
Witz-Bot — Reachy Mini Joke Bot

Schläft per Default. Wacht bei Geräuschen auf, gähnt, streckt sich,
schaut sich um, erzählt einen deutschen Witz und lacht — dann schläft er wieder.

Voraussetzungen (einmalig installieren):
    pip install edge-tts miniaudio
"""

import asyncio
import time
import random

import miniaudio
import numpy as np
import edge_tts

from reachy_mini import ReachyMini
from reachy_mini.utils import create_head_pose

try:
    from reachy_mini.motion.recorded_move import RecordedMoves
    _emotions = RecordedMoves("pollen-robotics/reachy-mini-emotions-library")
except Exception:
    _emotions = None

# ── Konfiguration ─────────────────────────────────────────────────────────────

TTS_VOICE      = "de-DE-KatjaNeural"   # Deutsche Stimme (Microsoft Neural TTS)
AUDIO_RATE     = 16_000                 # Hz — Roboter-Lautsprecher-Samplerate
WAKE_THRESHOLD = 4                      # Aufeinanderfolgende Erkennungen zum Aufwachen
POLL_INTERVAL  = 0.15                   # Sekunden zwischen Mikrofon-Polls

# Antennen-Positionen in Radiant [rechts, links]
ANT_NORMAL = [-0.1745,  0.1745]   # ≈ ±10°, Standardposition
ANT_DROOP  = [-2.2,     2.2   ]   # ≈ ±126°, hängend (müde)
ANT_ALERT  = [-0.4,     0.4   ]   # ≈ ±23°, aufmerksam

JOKES: list[tuple[str, str]] = [
    ("Warum können Geister so schlecht lügen?",
     "Weil man durch sie hindurchsehen kann!"),
    ("Wie nennt man einen Bumerang, der nicht zurückkommt?",
     "Einen Stock!"),
    ("Was macht ein Pirat am Computer?",
     "Er drückt die Enter-Taste!"),
    ("Warum gehen Skelette nicht in den Krieg?",
     "Weil sie keinen Mumm haben!"),
    ("Warum ist die Sechs so traurig?",
     "Weil Sieben Acht Neun!"),
    ("Wie heißt der Bruder von Robocop?",
     "Robocup!"),
    ("Was sagt ein Vampir, wenn er telefoniert?",
     "Auf Wieder-beißen! Äh … ich meine: Auf Wiederhören!"),
    ("Was ist der Unterschied zwischen einem Snowboarder und einem Schachspieler?",
     "Einer macht das Brett — der andere brettert!"),
    ("Warum macht ein Besen so gute Arbeit?",
     "Er ist eben gut im Kehren!"),
    ("Was sagt ein Hai, wenn er gegen etwas stößt?",
     "Haihai!"),
    ("Was ist groß, grün und geht rauf und runter?",
     "Ein Kaktus im Fahrstuhl!"),
    ("Treffen sich zwei Jäger. Beide tot.",
     "Nein, Spaß — das ist kein Witz, das ist Tragödie!"),
]


# ── TTS ───────────────────────────────────────────────────────────────────────

async def _fetch_mp3(text: str) -> bytes:
    chunks: list[bytes] = []
    async for chunk in edge_tts.Communicate(text, TTS_VOICE).stream():
        if chunk["type"] == "audio":
            chunks.append(chunk["data"])
    return b"".join(chunks)


def synth(text: str) -> np.ndarray:
    """Text → float32-Array (N, 1) bei AUDIO_RATE Hz für den Roboter-Lautsprecher."""
    mp3_bytes = asyncio.run(_fetch_mp3(text))
    decoded = miniaudio.decode(
        mp3_bytes,
        nchannels=1,
        sample_rate=AUDIO_RATE,
        output_format=miniaudio.SampleFormat.FLOAT32,
    )
    samples = np.frombuffer(decoded.samples, dtype=np.float32).copy()
    return samples.reshape(-1, 1)


def play(mini: ReachyMini, samples: np.ndarray) -> None:
    """Audio zum Roboter-Lautsprecher schicken und auf Ende warten."""
    mini.media.push_audio_sample(samples)
    time.sleep(len(samples) / AUDIO_RATE + 0.3)


# ── Gestiken ──────────────────────────────────────────────────────────────────

def do_yawn(mini: ReachyMini) -> None:
    """Gähnen: Kopf lehnt zurück, Antennen hängen schlaff, kehrt zurück."""
    # Mund auf — Kopf nach hinten kippen
    mini.goto_target(
        create_head_pose(pitch=-22, degrees=True),
        antennas=ANT_DROOP,
        duration=1.5,
    )
    time.sleep(2.0)
    # Aufrecht
    mini.goto_target(create_head_pose(), antennas=ANT_NORMAL, duration=1.0)
    time.sleep(1.1)


def do_stretch(mini: ReachyMini) -> None:
    """Strecken: Körper dreht sich links und rechts (Hals/Wirbelsäule lockern)."""
    mini.goto_target(create_head_pose(), body_yaw=np.deg2rad(40), duration=1.5)
    time.sleep(1.6)
    mini.goto_target(create_head_pose(), body_yaw=np.deg2rad(-40), duration=2.0)
    time.sleep(2.1)
    mini.goto_target(create_head_pose(), body_yaw=0.0, duration=1.0)
    time.sleep(1.1)


def look_around(mini: ReachyMini) -> None:
    """Schaut sich neugierig um: links → rechts → Mitte."""
    mini.goto_target(
        create_head_pose(yaw=50, degrees=True),
        antennas=ANT_ALERT,
        duration=0.8,
    )
    time.sleep(1.0)
    mini.goto_target(create_head_pose(yaw=-50, degrees=True), duration=1.2)
    time.sleep(1.3)
    mini.goto_target(create_head_pose(), antennas=ANT_NORMAL, duration=0.6)
    time.sleep(0.7)


def do_laugh(mini: ReachyMini) -> None:
    """Lachen: Emotions-Library wenn verfügbar, sonst schnelles Kopfnicken."""
    if _emotions is not None:
        try:
            mini.play_move(_emotions.get("happy"), initial_goto_duration=0.4)
            return
        except Exception:
            pass

    # Fallback: Kopfnicken + Antennen-Wackeln
    for i in range(7):
        pitch = 20 if i % 2 == 0 else -5
        ant   = ANT_ALERT if i % 2 == 0 else ANT_NORMAL
        mini.goto_target(
            create_head_pose(pitch=pitch, degrees=True),
            antennas=ant,
            duration=0.18,
        )
        time.sleep(0.2)
    mini.goto_target(create_head_pose(), antennas=ANT_NORMAL, duration=0.5)
    time.sleep(0.6)


# ── Schlaferkennung ───────────────────────────────────────────────────────────

def wait_for_sound(mini: ReachyMini) -> None:
    """Blockiert, bis Sprache/Geräusch zuverlässig erkannt wird."""
    print("  Warte auf Geräusch …", end="", flush=True)
    consecutive = 0
    while consecutive < WAKE_THRESHOLD:
        _, is_speech = mini.media.get_DoA()
        if is_speech:
            consecutive += 1
        else:
            consecutive = max(0, consecutive - 1)
        time.sleep(POLL_INTERVAL)
    print(" Geräusch erkannt!")


# ── Hauptschleife ─────────────────────────────────────────────────────────────

def main() -> None:
    print("Witz-Bot startet …")
    print("  Synthetisiere Witze (benötigt Internet) …")

    jokes_audio: list[tuple[np.ndarray, np.ndarray]] = []
    for i, (setup, punchline) in enumerate(JOKES, 1):
        print(f"    [{i}/{len(JOKES)}] {setup[:50]}…")
        jokes_audio.append((synth(setup), synth(punchline)))

    print(f"  {len(jokes_audio)} Witze bereit. Verbinde mit Roboter …")

    with ReachyMini(media_backend="default") as mini:
        mini.media.start_recording()
        mini.media.start_playing()
        print("  Verbunden! Drücke Strg+C zum Beenden.\n")

        try:
            while True:
                print("Schlafen …")
                mini.goto_sleep()

                wait_for_sound(mini)

                mini.wake_up()

                do_yawn(mini)
                do_stretch(mini)
                look_around(mini)

                idx = random.randrange(len(JOKES))
                setup_text, punchline_text = JOKES[idx]
                setup_audio, punchline_audio = jokes_audio[idx]

                print(f"\nWitz #{idx + 1}: {setup_text}")
                play(mini, setup_audio)

                time.sleep(1.5)  # dramatische Pause

                print(f"          → {punchline_text}\n")
                play(mini, punchline_audio)

                time.sleep(0.5)
                do_laugh(mini)
                time.sleep(1.0)

        except KeyboardInterrupt:
            print("\nBeende Witz-Bot …")
            mini.goto_sleep()
        finally:
            mini.media.stop_recording()
            mini.media.stop_playing()
            print("Tschüss!")


if __name__ == "__main__":
    main()