The official Python library for the Plapperi.ch API. Translate text to Swiss German dialects and synthesize natural-sounding speech.
- 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
Install the package using pip:
pip install plapperi- Python 3.9 or higher
- An API key from Plapperi.ch
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")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."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" |
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)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
)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)
)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)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']}")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']}")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 closedHandle 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}")The synthetization API is currently under development:
# Future API (not yet implemented)
audio = client.synthetization.synth(
text="Grüezi mitenand!",
voice="swiss-german-female",
)Initialize the Plapperi client.
Parameters:
api_key(str, optional): Your API key. If not provided, reads fromPLAPPERI_API_KEYenvironment variable.base_url(str): Base URL for the API. Default:https://api.plapperi.chtimeout(float): Request timeout in seconds. Default:30.0
Methods:
close(): Close the HTTP client- Can be used as a context manager with
withstatement
Translate text and wait for completion.
Parameters:
text(str): Text to translate to Swiss Germandialect(str): Dialect code (e.g., 'vs', 'be', 'zh')beam_size(int): Beam size for translation quality (1-8). Default:4poll_interval(float): Seconds between status checks. Default:1.0timeout(float): Maximum seconds to wait. Default:60.0
Returns: str - The translated text
Raises:
PlapperiTimeoutError: If job doesn't complete within timeoutApiError: If job fails or API error occurs
Start a translation job without waiting.
Parameters:
text(str): Text to translatedialect(str): Dialect codebeam_size(int): Beam size (1-8). Default:4
Returns: Job - Job information with job_id and status
Check the status of a translation job.
Parameters:
job_id(str): The job ID fromstart()
Returns: TranslationStatus - Status object with:
job_id(str): The job identifierstatus(JobStatus): Current status (PENDING, PROCESSING, COMPLETED, FAILED)result(TranslationResult | None): Translation result if completederror(str | None): Error message if failed- Properties:
is_completed,is_failed,is_pending,is_processing
Synthesize text and wait for completion.
Parameters:
text(str): Text to translate to Swiss Germanvoice(str): Voice identifier (e.g., 'aragon')poll_interval(float): Seconds between status checks. Default:1.0timeout(float): Maximum seconds to wait. Default:60.0
Returns: bytes - The synthetized audio
Raises:
PlapperiTimeoutError: If job doesn't complete within timeoutApiError: If job fails or API error occurs
Start a translation job without waiting.
Parameters:
text(str): Text to translatedialect(str): Voice identifier
Returns: Job - Job information with job_id and status
Check the status of a synthetization job.
Parameters:
job_id(str): The job ID fromstart()
Returns: TranslationStatus - Status object with:
job_id(str): The job identifierstatus(JobStatus): Current status (PENDING, PROCESSING, COMPLETED, FAILED)result(SynthetizationResult | None): Synthetization result if completederror(str | None): Error message if failed- Properties:
is_completed,is_failed,is_pending,is_processing
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 errorfrom 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"When processing multiple translations, consider these strategies:
-
Sequential with Manual Control (Simple, predictable):
- Start all jobs first
- Poll until all complete
- Good for small batches (<10 texts)
-
Concurrent Polling (Fast, efficient):
- Use ThreadPoolExecutor
- Each thread manages one translation
- Good for medium batches (10-100 texts)
-
Chunked Processing (Scalable):
- Process in chunks of 10-20
- Avoids overwhelming the API
- Good for large batches (100+ texts)
- 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
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')}")- Documentation: https://plapperi.ch/docs
- Issues: GitHub Issues
- Email: info@noxenum.io
This project is licensed under the Apache License 2.0 - see the LICENSE file for details.
- Synthetization API support (TTS)
- 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
- Streaming responses
- WebSocket support for real-time translation
Built with ❤️ by Noxenum for the Swiss German community.