plapperi 0.0.2

The official Python library for the Plapperi API

Plapperi Python SDK

The official Python library for the 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
• 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:

pip install plapperi

Requirements

• Python 3.9 or higher
• An API key from Plapperi.ch

Quick Start

Setting up Authentication

Set your API key as an environment variable:

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

Or pass it directly when initializing the client:

from plapperi import Plapperi

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

Basic Translation

Translate text to a Swiss German dialect:

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:

from plapperi import Plapperi

client = Plapperi()

# Synthesize with voice aragon
audio_bytes = client.synthetization.synth(
    text="D Bevölkerig het gnueg vode vellne Touriste.",
    voice="aragon",
)

Save audio to file:

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

Advanced Usage

Custom Configuration

Configure the client with custom timeout and base URL:

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:

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:

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:

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:

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:

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:

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

The synthetization API is currently under development:

# Future API (not yet implemented)
audio = client.synthetization.synth(
    text="Grüezi mitenand!",
    voice="swiss-german-female",
)

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)

Synthesize text and wait for completion.

Parameters:

• text (str): Text to translate to Swiss German
• voice (str): Voice identifier (e.g., 'aragon')
• poll_interval (float): Seconds between status checks. Default: 1.0
• timeout (float): Maximum seconds to wait. Default: 60.0

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, beam_size=4)

Start a translation job without waiting.

Parameters:

• text (str): Text to translate
• dialect (str): Voice identifier

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: TranslationStatus - 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

Type Definitions

Job Status Values

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

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

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
• Issues: GitHub Issues
• Email: [e-mail]

License

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

Changelog

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 for the Swiss German community.