Metadata-Version: 2.5
Name: plapperi
Version: 0.0.4
Summary: The official Python library for the Plapperi API
Project-URL: Homepage, https://github.com/Plapperi/plapperi-python
Project-URL: Issues, https://github.com/Plapperi/plapperi-python/issues
Author-email: Noxenum <info@noxenum.io>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx<1,>=0.23.0
Requires-Dist: pydantic<3,>=2.0.0
Provides-Extra: test
Requires-Dist: pytest; extra == 'test'
Requires-Dist: pytest-env; extra == 'test'
Description-Content-Type: text/markdown

# Plapperi Python SDK

[![PyPI version](https://img.shields.io/pypi/v/plapperi.svg)](https://pypi.org/project/plapperi/)
[![Python support](https://img.shields.io/pypi/pyversions/plapperi.svg)](https://pypi.org/project/plapperi/)
[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)

The official Python library for the [Plapperi.ch](https://plapperi.ch/) API. Translate text to Swiss German dialects and synthesize natural-sounding speech.

## Features

- **Dialect Translation**: Translate High German to various Swiss German dialects
- **Speech Synthesis**: Generate natural-sounding audio from Swiss German text
- **Multi-Speaker Dialogues**: Generate one native WAV with two stable voices and dialects
- **Voice Catalogue**: Discover current voices, gender labels, and synthesis dialects
- **Async Support**: Both synchronous and asynchronous operations
- **Batch Processing**: Efficient handling of multiple translation jobs
- **Type Safety**: Full type hints with Pydantic models
- **Context Manager Support**: Clean resource management

## Installation

Install the package using pip:

```bash
pip install plapperi
```

## Requirements

- Python 3.9 or higher
- An API key from [Plapperi.ch](https://plapperi.ch/)

## Quick Start

### Setting up Authentication

Set your API key as an environment variable:

```bash
export PLAPPERI_API_KEY="your-api-key-here"
```

Or pass it directly when initializing the client:

```python
from plapperi import Plapperi

client = Plapperi(api_key="your-api-key-here")
```

### Available Voices

Fetch the current Studio voice roster instead of maintaining a local list:

```python
from plapperi import Plapperi

client = Plapperi()

for voice in client.voices.list():
    dialects = ", ".join(dialect.value for dialect in voice.supported_dialects)
    print(f"{voice.name} ({voice.gender.value}): {dialects}")
```

All current Studio voices support the `be`, `gr`, `lu`, and `zh` synthesis
dialects. Dialect is selected when synthesizing and is not permanently assigned
to a voice.

### Basic Translation

Translate text to a Swiss German dialect:

```python
from plapperi import Plapperi

client = Plapperi()

# Translate to Valais dialect
result = client.translation.translate(
    text="Die Bevölkerung hat genug von den vielen Touristen.",
    dialect="vs",
)

print(result)
# Output: "D'Bevölkrig het gnüeg va de viele Touristu."
```

### Supported Dialects

The following Swiss German dialects are currently supported:

| Dialect Code | Region | Example Usage |
|--------------|--------|---------------|
| `vs` | Valais (Wallis) | `dialect="vs"` |
| `bs` | Basel-Stadt | `dialect="bs"` |
| `ag` | Aargau | `dialect="ag"` |
| `be` | Bern | `dialect="be"` |
| `zh` | Zürich | `dialect="zh"` |
| `lu` | Luzern | `dialect="lu"` |
| `gr` | Graubünden | `dialect="gr"` |
| `sg` | St. Gallen | `dialect="sg"` |

### Basic Speech Synthetization

Synthesize speech:

```python
from plapperi import Dialect, Plapperi

client = Plapperi()

# Synthesize stable Züridütsch with voice wren
audio_bytes = client.synthetization.synth(
    text="D Bevölkerig het gnueg vode vellne Touriste.",
    voice="wren",
    dialect=Dialect.ZURICH,
)
```

Speech synthesis supports `be`, `gr`, `lu`, and `zh`. If `dialect` is omitted,
the SDK uses `zh` for backward compatibility.

Save audio to file:

```python
with open("output.wav", "wb") as f:
    f.write(audio_bytes)
```

### Native Multi-Speaker Dialogues

Generate one WAV containing a complete two-speaker Swiss German conversation:

```python
from plapperi import DialogueTurn, Plapperi, Speaker

client = Plapperi()
audio = client.multi_speaker.synth(
    speakers=[
        Speaker(id="speaker-1", voice="tavin", dialect="zh"),
        Speaker(id="speaker-2", voice="brisa", dialect="be"),
    ],
    turns=[
        DialogueTurn(speaker_id="speaker-1", text="Hoi mitenand."),
        DialogueTurn(speaker_id="speaker-2", text="Sali, wie geits?"),
    ],
    timeout=900,
)

with open("dialogue.wav", "wb") as output:
    output.write(audio)
```

For manual job control, use `client.multi_speaker.start(...)` and poll
`client.multi_speaker.status(job_id)`. Multi-speaker input requires exactly two
distinct voices, at least one turn per speaker, no more than 100 turns, and no
more than 5,000 spoken characters in total. Supported dialogue dialects are
`be`, `gr`, `lu`, and `zh`.

## Advanced Usage

### Custom Configuration

Configure the client with custom timeout and base URL:

```python
from plapperi import Plapperi

client = Plapperi(
    api_key="your-api-key-here",
    base_url="https://api.plapperi.ch",
    timeout=60.0,  # seconds
)
```

### Translation with Beam Search

Control the translation quality using beam search:

```python
result = client.translation.translate(
    text="Guten Morgen, wie geht es Ihnen?",
    dialect="be",
    beam_size=8,  # Higher values = better quality but slower (default=4)
)
```

### Manual Job Control (Translation)

For more control over the translation process, you can manage jobs manually:

```python
from plapperi import Plapperi

client = Plapperi()

# Start a translation job
job = client.translation.start(
    text="Das Wetter ist heute sehr schön.",
    dialect="zh",
    beam_size=4,
)

print(f"Job ID: {job.job_id}")
print(f"Status: {job.status}")

# Poll for completion
import time
while True:
    status = client.translation.status(job.job_id)
    
    if status.is_completed:
        print(f"Translation: {status.result.translation}")
        break
    elif status.is_failed:
        print(f"Job failed: {status.error}")
        break
    elif status.is_processing:
        print("Still processing...")
    
    time.sleep(1.0)
```

### Batch Translation

Process multiple texts efficiently by managing jobs manually:

```python
from plapperi import Plapperi
import time

client = Plapperi()

texts = [
    "Guten Morgen!",
    "Wie geht es dir?",
    "Das Wetter ist schön.",
    "Ich mag Schweizer Schokolade.",
    "Bis bald!",
]

# Start all jobs
jobs = []
for text in texts:
    job = client.translation.start(text=text, dialect="be")
    jobs.append((text, job.job_id))
    print(f"Started job {job.job_id} for: {text}")

# Poll all jobs until complete
results = []
pending_jobs = dict(jobs)

while pending_jobs:
    for original_text, job_id in list(pending_jobs.items()):
        status = client.translation.status(job_id)
        
        if status.is_completed:
            results.append({
                "original": original_text,
                "translation": status.result.translation,
                "dialect": "be",
            })
            del pending_jobs[original_text]
            print(f"✓ Completed: {original_text}")
        elif status.is_failed:
            print(f"✗ Failed: {original_text} - {status.error}")
            del pending_jobs[original_text]
    
    if pending_jobs:
        time.sleep(0.5)  # Poll every 500ms

# Display results
for result in results:
    print(f"{result['original']} -> {result['translation']}")
```

### Optimized Batch Processing with Concurrent Polling

For better performance with large batches, use concurrent polling:

```python
from plapperi import Plapperi
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

client = Plapperi()

def process_translation(text, dialect="be", beam_size=4):
    """Start and wait for a single translation job."""
    job = client.translation.start(
        text=text,
        dialect=dialect,
        beam_size=beam_size,
    )
    
    # Poll until complete
    while True:
        status = client.translation.status(job.job_id)
        
        if status.is_completed:
            return {
                "original": text,
                "translation": status.result.translation,
                "job_id": job.job_id,
            }
        elif status.is_failed:
            return {
                "original": text,
                "error": status.error,
                "job_id": job.job_id,
            }
        
        time.sleep(0.5)

texts = [
    "Guten Morgen!",
    "Wie geht es dir?",
    "Das Wetter ist schön.",
    "Ich mag Schweizer Schokolade.",
    "Bis bald!",
]

# Process all translations concurrently
with ThreadPoolExecutor(max_workers=5) as executor:
    # Submit all jobs
    future_to_text = {
        executor.submit(process_translation, text): text 
        for text in texts
    }
    
    # Collect results as they complete
    for future in as_completed(future_to_text):
        result = future.result()
        if "error" in result:
            print(f"✗ {result['original']}: {result['error']}")
        else:
            print(f"✓ {result['original']} -> {result['translation']}")
```

### Context Manager Usage

Use the client as a context manager for automatic cleanup:

```python
from plapperi import Plapperi

with Plapperi() as client:
    result = client.translation.translate(
        text="Herzlichen Glückwunsch!",
        dialect="zh",
    )
    print(result)
# Client is automatically closed
```

### Error Handling

Handle API errors gracefully:

```python
from plapperi import Plapperi
from plapperi.errors.api_error import ApiError
from plapperi.errors.timeout_error import PlapperiTimeoutError

client = Plapperi()

try:
    result = client.translation.translate(
        text="Ein sehr langer Text...",
        dialect="vs",
        timeout=30.0,
    )
    print(result)
    
except PlapperiTimeoutError as e:
    print(f"Translation timed out: {e}")
    
except ApiError as e:
    print(f"API error: {e.status_code} - {e.body}")
    
except Exception as e:
    print(f"Unexpected error: {e}")
```

## Speech Synthesis

Select a supported dialect with either a string code or the `Dialect` enum:

```python
from plapperi import Dialect

audio = client.synthetization.synth(
    text="Grüezi mitenand!",
    voice="wren",
    dialect=Dialect.ZURICH,
)
```

## API Reference

### Client

#### `Plapperi(api_key=None, base_url="https://api.plapperi.ch", timeout=30.0)`

Initialize the Plapperi client.

**Parameters:**
- `api_key` (str, optional): Your API key. If not provided, reads from `PLAPPERI_API_KEY` environment variable.
- `base_url` (str): Base URL for the API. Default: `https://api.plapperi.ch`
- `timeout` (float): Request timeout in seconds. Default: `30.0`

**Methods:**
- `close()`: Close the HTTP client
- Can be used as a context manager with `with` statement

### Translation

#### `client.translation.translate(text, dialect, beam_size=4, poll_interval=1.0, timeout=60.0)`

Translate text and wait for completion.

**Parameters:**
- `text` (str): Text to translate to Swiss German
- `dialect` (str): Dialect code (e.g., 'vs', 'be', 'zh')
- `beam_size` (int): Beam size for translation quality (1-8). Default: `4`
- `poll_interval` (float): Seconds between status checks. Default: `1.0`
- `timeout` (float): Maximum seconds to wait. Default: `60.0`

**Returns:** `str` - The translated text

**Raises:**
- `PlapperiTimeoutError`: If job doesn't complete within timeout
- `ApiError`: If job fails or API error occurs

#### `client.translation.start(text, dialect, beam_size=4)`

Start a translation job without waiting.

**Parameters:**
- `text` (str): Text to translate
- `dialect` (str): Dialect code
- `beam_size` (int): Beam size (1-8). Default: `4`

**Returns:** `Job` - Job information with `job_id` and `status`

#### `client.translation.status(job_id)`

Check the status of a translation job.

**Parameters:**
- `job_id` (str): The job ID from `start()`

**Returns:** `TranslationStatus` - Status object with:
- `job_id` (str): The job identifier
- `status` (JobStatus): Current status (PENDING, PROCESSING, COMPLETED, FAILED)
- `result` (TranslationResult | None): Translation result if completed
- `error` (str | None): Error message if failed
- Properties: `is_completed`, `is_failed`, `is_pending`, `is_processing`

### Synthetization

#### `client.synthetization.synth(text, voice, poll_interval=1.0, timeout=60.0, *, dialect=Dialect.ZURICH)`

 Synthesize text and wait for completion.

**Parameters:**
- `text` (str): Text to translate to Swiss German
- `voice` (str): Voice identifier (e.g., 'wren')
- `poll_interval` (float): Seconds between status checks. Default: `1.0`
- `timeout` (float): Maximum seconds to wait. Default: `60.0`
- `dialect` (str | Dialect): Speech dialect (`be`, `gr`, `lu`, or `zh`). Default: `Dialect.ZURICH`

**Returns:** `bytes` - The synthetized audio

**Raises:**
- `PlapperiTimeoutError`: If job doesn't complete within timeout
- `ApiError`: If job fails or API error occurs

#### `client.synthetization.start(text, voice, dialect=Dialect.ZURICH)`

Start a synthetization job without waiting.

**Parameters:**
- `text` (str): Swiss German text to synthesize
- `voice` (str): Voice identifier
- `dialect` (str | Dialect): Speech dialect (`be`, `gr`, `lu`, or `zh`). Default: `Dialect.ZURICH`

**Returns:** `Job` - Job information with `job_id` and `status`

#### `client.synthetization.status(job_id)`

Check the status of a synthetization job.

**Parameters:**
- `job_id` (str): The job ID from `start()`

**Returns:** `SynthetizationStatus` - Status object with:
- `job_id` (str): The job identifier
- `status` (JobStatus): Current status (PENDING, PROCESSING, COMPLETED, FAILED)
- `result` (SynthetizationResult | None): Synthetization result if completed
- `error` (str | None): Error message if failed
- Properties: `is_completed`, `is_failed`, `is_pending`, `is_processing`

### Voices

#### `client.voices.list()`

Return the current Studio voice catalogue.

**Returns:** `list[Voice]`, where each voice contains:

- `id` (str): Stable lowercase voice identifier
- `name` (str): Public display name
- `gender` (VoiceGender): `VoiceGender.FEMALE` or `VoiceGender.MALE`
- `supported_dialects` (list[Dialect]): Dialects supported by the voice

## Type Definitions

### Job Status Values

```python
from plapperi.types.job import JobStatus

JobStatus.PENDING      # Job is queued
JobStatus.PROCESSING   # Job is being processed
JobStatus.COMPLETED    # Job completed successfully
JobStatus.FAILED       # Job failed with error
```

### Dialect Enum

```python
from plapperi.types.dialect import Dialect

Dialect.VALAIS      # "vs"
Dialect.BASEL       # "bs"
Dialect.AARGAU      # "ag"
Dialect.BERN        # "be"
Dialect.ZURICH      # "zh"
Dialect.LUCERNE     # "lu"
Dialect.GRAUBUNDEN  # "gr"
Dialect.ST_GALLEN   # "sg"
```

## Best Practices

### Batch Processing Strategy

When processing multiple translations, consider these strategies:

1. **Sequential with Manual Control** (Simple, predictable):
   - Start all jobs first
   - Poll until all complete
   - Good for small batches (<10 texts)

2. **Concurrent Polling** (Fast, efficient):
   - Use ThreadPoolExecutor
   - Each thread manages one translation
   - Good for medium batches (10-100 texts)

3. **Chunked Processing** (Scalable):
   - Process in chunks of 10-20
   - Avoids overwhelming the API
   - Good for large batches (100+ texts)

### Performance Tips

- **Batch Size**: Start 10-20 jobs, then poll
- **Poll Interval**: Use 0.5-1.0 seconds between checks
- **Beam Size**: Use 4 for balanced quality/speed, 6-8 for best quality
- **Timeout**: Set based on text length (30-60s typical)
- **Error Recovery**: Implement retry logic for failed jobs

### Example: Production-Ready Batch Processor

```python
from plapperi import Plapperi
from plapperi.errors.api_error import ApiError
import time
from typing import List, Dict

def batch_translate(
    texts: List[str],
    dialect: str = "be",
    beam_size: int = 4,
    max_concurrent: int = 10,
    poll_interval: float = 0.5,
    max_retries: int = 3,
) -> List[Dict]:
    """
    Translate multiple texts with automatic retry and error handling.
    
    Args:
        texts: List of texts to translate
        dialect: Target dialect
        beam_size: Quality parameter (1-8)
        max_concurrent: Maximum concurrent jobs
        poll_interval: Seconds between status checks
        max_retries: Retries for failed jobs
        
    Returns:
        List of dictionaries with 'original', 'translation', and 'status'
    """
    client = Plapperi()
    results = []
    
    # Process in chunks
    for i in range(0, len(texts), max_concurrent):
        chunk = texts[i:i + max_concurrent]
        chunk_results = []
        
        # Start all jobs in chunk
        jobs = []
        for text in chunk:
            try:
                job = client.translation.start(
                    text=text,
                    dialect=dialect,
                    beam_size=beam_size,
                )
                jobs.append({
                    "text": text,
                    "job_id": job.job_id,
                    "retries": 0,
                })
            except ApiError as e:
                chunk_results.append({
                    "original": text,
                    "translation": None,
                    "status": "error",
                    "error": str(e),
                })
        
        # Poll until all complete
        pending = jobs.copy()
        while pending:
            for job_info in pending[:]:
                try:
                    status = client.translation.status(job_info["job_id"])
                    
                    if status.is_completed:
                        chunk_results.append({
                            "original": job_info["text"],
                            "translation": status.result.translation,
                            "status": "success",
                        })
                        pending.remove(job_info)
                        
                    elif status.is_failed:
                        if job_info["retries"] < max_retries:
                            # Retry failed job
                            new_job = client.translation.start(
                                text=job_info["text"],
                                dialect=dialect,
                                beam_size=beam_size,
                            )
                            job_info["job_id"] = new_job.job_id
                            job_info["retries"] += 1
                        else:
                            chunk_results.append({
                                "original": job_info["text"],
                                "translation": None,
                                "status": "failed",
                                "error": status.error,
                            })
                            pending.remove(job_info)
                            
                except ApiError as e:
                    print(f"Error checking status: {e}")
            
            if pending:
                time.sleep(poll_interval)
        
        results.extend(chunk_results)
        print(f"Completed chunk {i//max_concurrent + 1}/{(len(texts)-1)//max_concurrent + 1}")
    
    client.close()
    return results

# Usage
texts = ["Guten Tag"] * 50
results = batch_translate(texts, dialect="zh", max_concurrent=10)

for r in results:
    if r["status"] == "success":
        print(f"✓ {r['original']} -> {r['translation']}")
    else:
        print(f"✗ {r['original']}: {r.get('error', 'Unknown error')}")
```

## Support

- **Documentation**: [https://plapperi.ch/docs](https://plapperi.ch/docs)
- **Issues**: [GitHub Issues](https://github.com/Plapperi/plapperi-python/issues)
- **Email**: info@noxenum.io

## License

This project is licensed under the Apache License 2.0 - see the [LICENSE](LICENSE) file for details.

## Changelog

### 0.0.4

- Added the typed voice catalogue through `client.voices.list()`
- Added `Voice` and `VoiceGender` public types
- Fixed multi-speaker polling for pending and processing jobs

### 0.0.3

- Native two-speaker dialogue synthesis with typed speaker and turn models
- Dialect selection for single- and multi-speaker synthesis, including Graubünden (`gr`)

### 0.0.2

- Synthetization API support (TTS)

### 0.0.1 (Current)

- Initial release
- Translation API support with multiple Swiss German dialects
- Synchronous operations
- Manual job control for batch processing
- Type-safe Pydantic models
- Context manager support

### Upcoming Features

- Streaming responses
- WebSocket support for real-time translation

## Acknowledgments

Built with ❤️ by [Noxenum](https://noxenum.io) for the Swiss German community.
