"""TonePerfect pronunciation assessment API client. One file, standard library only. Wraps the public endpoints of https://api.toneperfect.app: GET /v1/capabilities -> languages served, limits, model version POST /v1/assess -> score one scripted recording POST /v1/lint -> check a reference text before recording (free, no key) GET /v1/status -> synthetic engine probe summary (free, no key) Command line: ``python -m toneperfect lint "我有三本书" --language zh-CN`` or ``python -m toneperfect lint --file lesson.csv`` (exit 1 when a text has an error). Accounts, API keys and billing are managed in the web dashboard (https://api.toneperfect.app/dashboard); those routes are session-based and are deliberately not wrapped here. Full reference: https://api.toneperfect.app/docs Quick start:: from toneperfect import TonePerfect client = TonePerfect(api_key="tp_live_...") result = client.assess("hello.wav", text="你好", language="zh-CN") print(result.assessable, result.scores, result.billing) """ from __future__ import annotations import io import json import os import re import secrets import http.client import socket import uuid from dataclasses import dataclass, field from typing import IO, Any, Dict, List, Mapping, Optional, Union from urllib import request as _request from urllib.error import HTTPError, URLError __all__ = ["TonePerfect", "TonePerfectError", "Assessment", "new_idempotency_key", "DEFAULT_BASE_URL", "TASKS", "DECISIONS", "RETRY_SAME_KEY_CODES", "TEST_OUTCOMES"] __version__ = "0.2.0" DEFAULT_BASE_URL = "https://api.toneperfect.app" TASKS = ("general", "connected_speech", "isolated_word") #: Outcomes a ``tp_test_`` key can rehearse (``test_outcome=`` on :meth:`TonePerfect.assess`). TEST_OUTCOMES = ("success", "unassessable", "invalid_assessment", "engine_unavailable", "subscription_required", "insufficient_credits", "account_busy", "rate_limit") #: Values of ``decision`` on words and phones. ``review`` / ``uncertain`` / ``no_correction`` #: are the feedback decisions; ``unassessable`` (no evidence for that unit), ``partial`` #: (word only partly covered by the phonetic inventory) and ``legacy_correction`` (Mandarin #: components decided by the legacy policy) also occur. Treat unknown values as "no decision". DECISIONS = ("review", "uncertain", "no_correction", "unassessable", "partial", "legacy_correction") #: Error codes after which the SAME Idempotency-Key must be retried: nothing was finalized. RETRY_SAME_KEY_CODES = frozenset({ "rate_limit", "account_busy", "gateway_busy", "decoder_busy", "audio_decoder_unavailable", "upload_timeout", "request_pending", "timeout", "network_error", }) #: Error codes that mean the key is spent: start a new attempt with a new key. NEW_KEY_CODES = frozenset({"idempotency_expired", "idempotency_conflict"}) _IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9_.:-]{8,128}$") AudioInput = Union[str, "os.PathLike[str]", bytes, bytearray, IO[bytes]] def new_idempotency_key(prefix: str = "tp") -> str: """A fresh key for one recording attempt (letters, digits, dots, dashes, underscores, colons; 8-128 chars). Keep the same key when the result is unknown or still pending (timeout, dropped connection, an HTML/empty 502/504 from a proxy, ``request_pending``); use a new key for a different attempt or after a definite failed response (``TonePerfectError.needs_new_key``). Never generate a new key inside an automatic retry loop. """ key = f"{prefix}-{uuid.uuid4().hex}-{secrets.token_hex(2)}" if not _IDEMPOTENCY_RE.match(key): raise ValueError("prefix must be letters, digits, dots, dashes, underscores or colons") return key class TonePerfectError(Exception): """Raised for any non-2xx response, and for transport failures (``status`` 0). ``code`` and ``message`` come from the API's ``error`` object (``{"error": {"code": ..., "message": ...}, "request_id": ...}``). A non-JSON body (for example an HTML 502 from a proxy) gets ``code="http_error"`` and the raw text as the message. What to do next is decided by the *body*, not the status: * ``finalized`` — the failure was stored under this Idempotency-Key (the body carries a ``req_…`` ``id`` and ``billing`` with ``credits: 0``; for example ``engine_unavailable`` or ``request_interrupted`` after the request was accepted). Retrying with the same key replays this same failure for 24 hours: ``needs_new_key`` is True and ``retryable`` is False. * ``retryable`` — nothing was finalized; retry later with the **same** key (``rate_limit``, ``account_busy``, ``gateway_busy``, ``decoder_busy``, ``audio_decoder_unavailable``, ``upload_timeout``, ``request_pending``, transport ``timeout`` / ``network_error``, and any non-finalized 5xx such as an HTML or empty 502/504 from a proxy). * Everything else (``invalid_api_key``, ``insufficient_credits``, ``invalid_audio``, ``unsupported_language``, …) needs a fix on your side. """ def __init__(self, status: int, code: str, message: str, request_id: Optional[str] = None, retry_after: Optional[float] = None, body: Optional[Dict[str, Any]] = None): super().__init__(f"{status} {code}: {message}") self.status = status self.code = code self.message = message self.request_id = request_id self.retry_after = retry_after self.body: Dict[str, Any] = body or {} #: ``req_…`` result id when the failure was finalized under your Idempotency-Key. self.id: Optional[str] = self.body.get("id") if isinstance(self.body.get("id"), str) else None self.billing: Optional[Dict[str, Any]] = self.body.get("billing") if isinstance(self.body.get("billing"), dict) else None @property def finalized(self) -> bool: """The failure is a stored result for this key (``id`` + ``billing`` present).""" return self.id is not None and self.billing is not None @property def needs_new_key(self) -> bool: """Start a new attempt with a new Idempotency-Key; the old key replays this response.""" return self.finalized or self.code in NEW_KEY_CODES @property def retryable(self) -> bool: """Retry later with the SAME Idempotency-Key: nothing was finalized.""" if self.finalized or self.code in NEW_KEY_CODES: return False return self.code in RETRY_SAME_KEY_CODES or self.status in (408, 429, 502, 503, 504) @dataclass class Assessment: """One ``POST /v1/assess`` result. ``data`` is the full JSON body exactly as returned by the API. The properties below are convenience accessors for its documented top-level fields; anything else (``syllables``, evidence, timings) is in ``data``. """ data: Dict[str, Any] status: int request_id: Optional[str] = None replayed: bool = False headers: Dict[str, str] = field(default_factory=dict) @property def assessable(self) -> bool: """False when the audio could not be meaningfully scored (silence, wrong sentence, unreadable). Never show an unassessable result as a score of zero; see ``reason``.""" return bool(self.data.get("assessable")) @property def reason(self) -> Optional[str]: return self.data.get("reason") @property def scores(self) -> Dict[str, Optional[float]]: """0-100 display grades per dimension; ``None`` means the dimension is not supported by this pack.""" return self.data.get("scores") or {} @property def overall(self) -> Optional[float]: return self.scores.get("overall") @property def words(self) -> List[Dict[str, Any]]: """Per word/character rows: grade, ``start_ms``/``end_ms`` and ``phones`` (expected, heard, score, decision; see :data:`DECISIONS`).""" return self.data.get("words") or [] @property def billing(self) -> Dict[str, Any]: """``{"credits": charged, "balance_credits": remaining}``; failed or unassessable requests charge 0.""" return self.data.get("billing") or {} @property def model_version(self) -> Optional[str]: return self.data.get("model_version") @property def id(self) -> Optional[str]: """Result id (``req_...``), also returned in the ``X-Request-Id`` header.""" return self.data.get("id") or self.request_id @property def test(self) -> bool: """True for a fixed example response from a ``tp_test_`` key (no audio scored, nothing charged).""" return self.data.get("test") is True class TonePerfect: """Client for https://api.toneperfect.app. :param api_key: secret key from the dashboard (``tp_live_...``). Defaults to ``$TONEPERFECT_API_KEY``. :param base_url: API origin. ``https://api.toneperfect.app`` goes through a proxy with a 26-second request timeout; if your assessments run longer than that, ask nik@toneperfect.app for the direct gateway address and pass it here. :param timeout: socket timeout in seconds for each request. """ def __init__(self, api_key: Optional[str] = None, base_url: str = DEFAULT_BASE_URL, timeout: float = 90.0, user_agent: Optional[str] = None): self.api_key = api_key or os.environ.get("TONEPERFECT_API_KEY", "") self.base_url = base_url.rstrip("/") self.timeout = timeout self.user_agent = user_agent or f"toneperfect-python/{__version__}" # ---- public endpoints ---- def capabilities(self) -> Dict[str, Any]: """``GET /v1/capabilities``: ``languages`` (locale, status, dimensions, limitations), ``limits``, ``model_version``. No API key required. Check ``status`` per language before sending it; every pack is currently labelled ``experimental``. """ status, headers, body = self._send("GET", "/v1/capabilities", None, {}) return self._json(status, headers, body) def lint(self, text: str, *, language: str = "zh-CN", pinyin: str = "", task: str = "general") -> Dict[str, Any]: """``POST /v1/lint``: check a reference text against the engine's rules before recording. Free, no API key. Returns ``{"ok", "language", "issues": [{code, message, field, severity}], "units"}``. A text that passes can still be refused by a newer engine; the assess response stays authoritative. """ body = json.dumps({"text": text or "", "language": language, "pinyin": pinyin or "", "task": task}).encode("utf-8") status, headers, raw = self._send("POST", "/v1/lint", body, {"Content-Type": "application/json"}) return self._json(status, headers, raw) def status(self) -> Dict[str, Any]: """``GET /v1/status``: the gateway's synthetic engine probe (status, last_check, last_24h, last_7d).""" status, headers, raw = self._send("GET", "/v1/status", None, {}) return self._json(status, headers, raw) def assess( self, audio: AudioInput, *, text: Optional[str] = None, language: str = "zh-CN", pinyin: Optional[str] = None, task: str = "general", idempotency_key: Optional[str] = None, filename: str = "recording.wav", content_type: str = "application/octet-stream", compact: bool = False, test_outcome: Optional[str] = None, ) -> Assessment: """``POST /v1/assess``: score one scripted recording against its reference text. :param audio: path, bytes, or a binary file object. Up to 30 s and 10 MB (WAV, MP3, M4A, OGG, WebM...). :param text: what the learner was asked to read. Required unless ``pinyin`` is given (Mandarin only). :param language: locale such as ``zh-CN``, ``en-US``, ``fr-FR`` — see :meth:`capabilities`. :param pinyin: optional explicit Mandarin reading with numbered tones, e.g. ``"ni3 hao3"``. :param task: ``general`` (default), ``connected_speech`` or ``isolated_word``. Mandarin per-component correction decisions apply only with ``connected_speech``. :param idempotency_key: unique per recording attempt; generated when omitted. If a network error prevents you from receiving a result, retry with the same key and identical inputs: a completed request is returned without charging again — including a completed failure (``Assessment.replayed`` / ``TonePerfectError.finalized``). Use a new key for a different attempt or after a definite failed response (``TonePerfectError.needs_new_key``). Keep the same key while the result is unknown or pending (``TonePerfectError.retryable``). :param compact: ask for the stable compact schema (``toneperfect.compact.v1``). :param test_outcome: ``tp_test_`` keys only: one of :data:`TEST_OUTCOMES`. :raises TonePerfectError: for every non-2xx status (``error.code`` is in ``.code``) and for transport failures (``status`` 0, code ``timeout`` or ``network_error``). """ if not self.api_key: raise TonePerfectError(401, "invalid_api_key", "Set api_key or the TONEPERFECT_API_KEY environment variable.") if task not in TASKS: raise ValueError(f"task must be one of {TASKS}") if not (text or pinyin): raise ValueError("Provide text (or pinyin for Mandarin).") key = idempotency_key or new_idempotency_key() if not _IDEMPOTENCY_RE.match(key): raise ValueError("idempotency_key must be 8-128 letters, digits, dots, dashes, underscores or colons") fields: Dict[str, str] = {"language": language, "task": task} if text: fields["text"] = text if pinyin: fields["pinyin"] = pinyin body, boundary = _multipart(fields, "audio", filename, content_type, _read_audio(audio)) headers = { "Authorization": f"Bearer {self.api_key}", "Idempotency-Key": key, "Content-Type": f"multipart/form-data; boundary={boundary}", } if compact: headers["TonePerfect-Response"] = "compact" if test_outcome: headers["TonePerfect-Test-Outcome"] = str(test_outcome) status, response_headers, raw = self._send("POST", "/v1/assess", body, headers) data = self._json(status, response_headers, raw) return Assessment( data=data, status=status, request_id=response_headers.get("x-request-id"), replayed=response_headers.get("idempotent-replayed", "").lower() == "true", headers=response_headers, ) # ---- transport ---- def _send(self, method: str, path: str, body: Optional[bytes], headers: Mapping[str, str]): req = _request.Request(self.base_url + path, data=body, method=method) req.add_header("Accept", "application/json") req.add_header("User-Agent", self.user_agent) for name, value in headers.items(): req.add_header(name, value) timeout_error = TonePerfectError(0, "timeout", f"No response within {self.timeout}s; the outcome is unknown, retry with the same Idempotency-Key.") try: try: with _request.urlopen(req, timeout=self.timeout) as response: # noqa: S310 - fixed https origin return response.status, {k.lower(): v for k, v in response.headers.items()}, response.read() except HTTPError as error: # a subclass of URLError: handle first return error.code, {k.lower(): v for k, v in error.headers.items()}, error.read() except URLError as error: # Timeouts while connecting or sending are wrapped in URLError by urllib. if isinstance(error.reason, (socket.timeout, TimeoutError)): raise timeout_error from error raise TonePerfectError(0, "network_error", f"Could not reach {self.base_url}: {error.reason}") from error except (socket.timeout, TimeoutError) as error: # Timeouts while waiting for or reading the response surface bare (OSError on 3.9). raise timeout_error from error except (OSError, http.client.HTTPException) as error: # reset connections, TLS failures, truncated bodies raise TonePerfectError(0, "network_error", f"Could not reach {self.base_url}: {error}") from error @staticmethod def _json(status: int, headers: Mapping[str, str], raw: bytes) -> Dict[str, Any]: try: data = json.loads(raw.decode("utf-8")) if raw else {} except ValueError: data = None if 200 <= status < 300 and isinstance(data, dict): return data retry_after = None try: retry_after = float(headers["retry-after"]) if "retry-after" in headers else None except ValueError: pass if isinstance(data, dict) and isinstance(data.get("error"), dict): err = data["error"] raise TonePerfectError(status, str(err.get("code", "request_error")), str(err.get("message", "")), data.get("request_id") or headers.get("x-request-id"), retry_after, data) snippet = raw[:200].decode("utf-8", "replace").strip() if raw else "" raise TonePerfectError(status, "http_error", snippet or f"HTTP {status}", headers.get("x-request-id"), retry_after) # ---- helpers ---- def _read_audio(audio: AudioInput) -> bytes: if isinstance(audio, (bytes, bytearray)): return bytes(audio) if hasattr(audio, "read"): data = audio.read() # type: ignore[union-attr] if isinstance(data, str): raise TypeError("Open audio files in binary mode ('rb').") return data with open(os.fspath(audio), "rb") as handle: # type: ignore[arg-type] return handle.read() def _multipart(fields: Mapping[str, str], file_field: str, filename: str, content_type: str, data: bytes): boundary = "----toneperfect" + secrets.token_hex(16) out = io.BytesIO() for name, value in fields.items(): out.write(f"--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n".encode("utf-8")) out.write(str(value).encode("utf-8")) out.write(b"\r\n") safe_name = filename.replace('"', "_").replace("\r", "").replace("\n", "") out.write(f"--{boundary}\r\nContent-Disposition: form-data; name=\"{file_field}\"; filename=\"{safe_name}\"\r\nContent-Type: {content_type}\r\n\r\n".encode("utf-8")) out.write(data) out.write(f"\r\n--{boundary}--\r\n".encode("utf-8")) return out.getvalue(), boundary # ---- command line: python -m toneperfect lint ... ---- def _lint_items(texts, path, language, pinyin): import csv items = [{"text": t, "language": language, "pinyin": pinyin} for t in texts] if path: with open(path, encoding="utf-8") as handle: lines = [line.strip() for line in handle if line.strip() and not line.lstrip().startswith("#")] if path.endswith(".jsonl"): for number, line in enumerate(lines, 1): try: row = json.loads(line) except ValueError: raise ValueError(f"line {number} is not valid JSON") from None if not isinstance(row, dict): raise ValueError(f"line {number} is not a JSON object") items.append({"text": row.get("text", ""), "language": row.get("language") or language, "pinyin": row.get("pinyin", "")}) elif path.endswith(".csv"): for row in csv.reader(lines): if not row or row[:2] == ["text", "language"]: continue items.append({"text": row[0].strip(), "language": (row[1].strip() if len(row) > 1 and row[1].strip() else language), "pinyin": row[2].strip() if len(row) > 2 else ""}) else: items.extend({"text": line, "language": language, "pinyin": ""} for line in lines) return items def main(argv=None, client: Optional["TonePerfect"] = None, out=None) -> int: """``python -m toneperfect lint TEXT... [--file F] [--language L] [--pinyin P] [--json]``. Exit 0 = every text passes, 1 = at least one error, 2 = usage or network problem. """ import argparse import sys out = out or sys.stdout parser = argparse.ArgumentParser(prog="python -m toneperfect", description="Check reference texts with POST /v1/lint (free, no key).") sub = parser.add_subparsers(dest="command", required=True) lint = sub.add_parser("lint", help="check reference texts before recording") lint.add_argument("texts", nargs="*") lint.add_argument("--file", "-f", help=".txt (one text per line), .csv (text,language[,pinyin]) or .jsonl") lint.add_argument("--language", "-l", default="zh-CN") lint.add_argument("--pinyin", default="") lint.add_argument("--task", default="general", choices=TASKS) lint.add_argument("--json", action="store_true") lint.add_argument("--base-url", default=DEFAULT_BASE_URL) try: args = parser.parse_args(argv) except SystemExit as exc: return 2 if exc.code else 0 try: items = _lint_items(args.texts, args.file, args.language, args.pinyin) except FileNotFoundError: print(f"Cannot read {args.file}: file not found", file=sys.stderr) return 2 except (OSError, UnicodeDecodeError, ValueError) as exc: print(f"Cannot read {args.file}: {exc}", file=sys.stderr) return 2 if not items: print("Give a text or --file.", file=sys.stderr) return 2 client = client or TonePerfect(api_key="unused-for-lint", base_url=args.base_url) failed = 0 for item in items: try: result = client.lint(item["text"], language=item["language"], pinyin=item["pinyin"], task=args.task) except TonePerfectError as exc: print(f"network/API error: {exc}", file=sys.stderr) return 2 failed += not result.get("ok") if args.json: print(json.dumps({**item, **result}, ensure_ascii=False), file=out) continue label = "ok " if result.get("ok") else "ERROR" print(f"{label} [{item['language']}] {item['text']}" + (f" ({item['pinyin']})" if item["pinyin"] else ""), file=out) for issue in result.get("issues") or []: print(f" {issue.get('severity')}: {issue.get('code')}: {issue.get('message')}", file=out) if not args.json: print(f"{len(items) - failed}/{len(items)} texts pass.", file=out) return 1 if failed else 0 def _lint_entry() -> int: """Console script ``toneperfect-lint ...`` (same as ``python -m toneperfect lint ...``).""" import sys return main(["lint", *sys.argv[1:]]) if __name__ == "__main__": raise SystemExit(main())