3cx-telefonie-middleware/tools/test_routepoint90_media.py

#!/usr/bin/env python3

import asyncio
import json
import sys

from app.config import Settings
from app.media import ThreeCXMediaClient
from app.media_session import MediaSession
from app.whisper import WhisperWorker


async def main():
    settings = Settings()
    media = ThreeCXMediaClient(settings)

    print("=== ROUTEPOINT 90 MEDIA TEST ===")
    print("RoutePoint:", settings.threecx_media_routepoint_dn)
    print()
    print("Suche aktiven Participant ...")

    participants = await media.participants()

    print("Participants:", len(participants))

    if not participants:
        print()
        print("KEIN AKTIVER PARTICIPANT.")
        print("Jetzt DN 90 anrufen und den Anruf annehmen.")
        return 2

    print(json.dumps(
        participants,
        indent=2,
        ensure_ascii=False,
    ))

    connected = [
        p for p in participants
        if str(p.get("status", "")).lower() == "connected"
    ]

    if not connected:
        print()
        print("Kein CONNECTED Participant gefunden.")
        return 3

    participant = connected[0]

    print()
    print("=== VERWENDE PARTICIPANT ===")
    print("ID:", participant.get("id"))
    print("Call-ID:", participant.get("callid"))
    print("Leg-ID:", participant.get("legid"))
    print("Status:", participant.get("status"))
    print()

    whisper = WhisperWorker(settings)

    session = MediaSession(
        settings,
        media,
        whisper,
    )

    print("Starte MediaSession ...")
    print("Jetzt ca. 10 Sekunden sprechen.")
    print()

    result = await session.process_participant(
        participant,
        max_bytes=160000,
    )

    print()
    print("=== MEDIA ERGEBNIS ===")
    print("Call-ID:", result["callid"])
    print("Leg-ID:", result["legid"])
    print("Participant:", result["participant_id"])
    print("Audio:", result["audio_bytes"], "Bytes")
    print("WAV:", result["audio_file"])
    print()

    transcription = result["transcription"]

    print("=== WHISPER ===")
    print("Sprache:", transcription["language"])
    print(
        "Wahrscheinlichkeit:",
        transcription["language_probability"],
    )
    print("Text:", transcription["text"])
    print()
    print("Segmente:", len(transcription["segments"]))

    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))