Qwen3-TTS-streaming/api/examples.py

"""
Examples demonstrating the extended Qwen3-TTS API endpoints.
"""

import requests
import json
from pathlib import Path


BASE_URL = "http://localhost:8000"


def example_text_validation():
    """Example: Validate text before TTS."""
    print("\n=== Text Validation Example ===")
    
    response = requests.post(
        f"{BASE_URL}/v1/text/validate",
        json={
            "text": "Hello, this is a test of the text-to-speech system.",
            "language": "English"
        }
    )
    
    data = response.json()
    print(f"Valid: {data['valid']}")
    print(f"Characters: {data['character_count']}")
    print(f"Estimated duration: {data['estimated_duration']:.2f}s")
    if data['warnings']:
        print(f"Warnings: {data['warnings']}")


def example_batch_processing():
    """Example: Create and monitor batch job."""
    print("\n=== Batch Processing Example ===")
    
    # Create batch job
    response = requests.post(
        f"{BASE_URL}/v1/batch/create",
        json={
            "items": [
                {"text": "Hello world", "language": "English"},
                {"text": "Bonjour le monde", "language": "English"},
                {"text": "Hola mundo", "language": "English"},
            ],
            "model": "Qwen/Qwen3-TTS-12Hz-1.7B-Base"
        }
    )
    
    job = response.json()
    job_id = job['id']
    print(f"Created batch job: {job_id}")
    print(f"Status: {job['status']}")
    print(f"Total items: {job['request_counts']['total']}")
    
    # Check job status
    response = requests.get(f"{BASE_URL}/v1/batch/{job_id}")
    job = response.json()
    print(f"Updated status: {job['status']}")


def example_voice_management():
    """Example: Upload and manage custom voices."""
    print("\n=== Voice Management Example ===")
    
    # List existing voices
    response = requests.get(f"{BASE_URL}/v1/voices")
    voices = response.json()
    print(f"Voices available: {len(voices['data'])}")
    
    for voice in voices['data']:
        print(f"  - {voice['name']} ({voice['language']}): {voice['id']}")
    
    # To upload a voice:
    # response = requests.post(
    #     f"{BASE_URL}/v1/voices/upload",
    #     data={
    #         "name": "Alice",
    #         "language": "English",
    #         "ref_text": "This is a reference audio for voice cloning."
    #     },
    #     files={"audio": open("ref_audio.wav", "rb")}
    # )
    # voice = response.json()
    # print(f"Uploaded voice: {voice['id']}")


def example_usage_tracking():
    """Example: Check API usage and quota."""
    print("\n=== Usage & Quota Example ===")
    
    # Get usage statistics
    response = requests.get(f"{BASE_URL}/v1/usage")
    usage = response.json()
    print(f"Total requests: {usage['requests_made']}")
    print(f"Audio generated: {usage['audio_generated_minutes']:.2f} minutes")
    print(f"Requests by model:")
    for model, count in usage['requests_by_model'].items():
        print(f"  - {model}: {count}")
    
    # Get quota info
    response = requests.get(f"{BASE_URL}/v1/quota")
    quota = response.json()
    print(f"\nRate limit: {quota['requests_per_minute']} requests/minute")
    print(f"Remaining: {quota['remaining_requests']} requests this minute")


def example_audio_conversion():
    """Example: Convert audio format/sample rate."""
    print("\n=== Audio Conversion Example ===")
    
    # First, generate some audio
    response = requests.post(
        f"{BASE_URL}/v1/audio/speech",
        json={
            "text": "This is audio to convert.",
            "language": "English"
        }
    )
    audio_data = response.json()
    
    # Convert format (resample to 16kHz)
    response = requests.post(
        f"{BASE_URL}/v1/audio/convert",
        json={
            "audio_base64": audio_data['audio_base64'],
            "target_format": "wav",
            "target_sample_rate": 16000
        }
    )
    
    converted = response.json()
    print(f"Original sample rate: {audio_data['sample_rate']} Hz")
    print(f"Converted to: {converted['sample_rate']} Hz")
    print(f"Format: {converted['format']}")


def example_model_info():
    """Example: Get model configuration and languages."""
    print("\n=== Model Information Example ===")
    
    # Get model config
    response = requests.get(
        f"{BASE_URL}/v1/models/Qwen/Qwen3-TTS-12Hz-1.7B-Base/config"
    )
    config = response.json()
    print(f"Model: {config['id']}")
    print(f"Size: {config['size']}")
    print(f"Max text length: {config['max_text_length']}")
    print(f"Output sample rate: {config['output_sample_rate']} Hz")
    print(f"Languages: {', '.join(config['languages'])}")
    
    # Get supported languages with details
    response = requests.get(
        f"{BASE_URL}/v1/models/Qwen/Qwen3-TTS-12Hz-1.7B-Base/languages"
    )
    languages = response.json()
    print(f"\nSupported languages:")
    for lang in languages['languages']:
        print(f"  - {lang['name']} ({lang['code']})")


def example_health_and_models():
    """Example: Check API health and list models."""
    print("\n=== Health & Models Example ===")
    
    # Health check
    response = requests.get(f"{BASE_URL}/v1/health")
    health = response.json()
    print(f"API Status: {health['status']}")
    print(f"Model: {health['model']}")
    print(f"Device: {health['device']}")
    print(f"Ready: {health['ready']}")
    
    # List models
    response = requests.get(f"{BASE_URL}/v1/models")
    models = response.json()
    print(f"\nAvailable models: {len(models['data'])}")
    for model in models['data']:
        print(f"  - {model['id']}")
        print(f"    Streaming: {model['supports_streaming']}")
        print(f"    Voice cloning: {model['supports_voice_clone']}")


if __name__ == "__main__":
    print("Qwen3-TTS Extended API Examples")
    print("=" * 50)
    print(f"API Server: {BASE_URL}")
    
    try:
        # Run examples
        example_health_and_models()
        example_text_validation()
        example_model_info()
        example_usage_tracking()
        example_voice_management()
        example_batch_processing()
        example_audio_conversion()
        
        print("\n" + "=" * 50)
        print("✅ All examples completed!")
        print("For more details, visit: http://localhost:8000/docs")
        
    except requests.exceptions.ConnectionError:
        print("\n❌ Error: Could not connect to API server")
        print(f"Make sure the server is running at {BASE_URL}")
    except Exception as e:
        print(f"\n❌ Error: {e}")