interpretation — end-to-end simultaneous interpretation
An end-to-end real-time SI engine: it consumes 16 kHz PCM audio frames directly and, inside one long-lived session, emits the source text (interim/final), the translation (interim/final) and optional synthesized speech as it goes.
Plugins taking this route bypass the host's ASR → MT → TTS cascade (typical example: iFlytek real-time interpretation).
Declaration
{
"id": "community.cloud.xunfei_si",
"name": "科大讯飞实时同传",
"version": "1.0.0",
"target_app_version": "0.2",
"type": "interpretation",
"entrypoint": "xunfei_si.py:XunfeiInterpretationProvider",
"tags": ["cloud", "streaming", "low-latency"],
"requires_hf": false,
"supported_langs": ["en", "zh"],
"ui_extensions": { "settings_view": "SettingsView.qml" },
"config_schema": [
{ "key": "appid", "label": "AppID", "type": "string", "default": "" },
{ "key": "api_key", "label": "API Key", "type": "string", "default": "" }
]
}After enabling it, select it as the active SI engine on the settings page; when the user switches into SI mode the host starts streaming.
Base class contract
Must implement (abstract methods):
def start_session(self, session_id="peer", source_lang="zh", target_lang="en",
voice=None, tts_enabled=False) -> bool
"""Open an SI session (establish the long-lived connection and get ready to stream)."""
def feed_audio(self, session_id: str, pcm_chunk: np.ndarray) -> None
"""Feed 16 kHz mono PCM frames (float32 or int16)."""
def stop_session(self, session_id: str = "peer") -> None
"""Send the end frame and close the session gracefully."""
def is_active(self, session_id: str = "peer") -> bool
"""Whether the session is connected and streaming."""Emitting results upwards (already implemented in the base class, call them directly):
self.emit_transcript(text, is_final=False, session_id="peer") # source text
self.emit_translation(src, dst, is_final=False, session_id="peer") # translation
self.emit_tts_audio(pcm_bytes, sample_rate=16000, session_id="peer") # synthesized audio
self.emit_status(status, session_id="peer") # 'idle'|'connecting'|'streaming'|'error'
self.emit_error(message, session_id="peer")is_final=False goes through the typewriter interim path (the host displays it as it arrives); True means final.
Compatibility aliases:
dispatch_transcript/dispatch_translation/dispatch_tts_audio/dispatch_status/dispatch_errorare equivalent toemit_*.
Dual-track sessions session_id
| Value | Meaning |
|---|---|
"peer" | Upper track: the other party / system sound (inbound SI) |
"self" | Lower track: my microphone (outbound SI) |
The host may open two sessions at the same time. If your underlying protocol does not support multiple sessions, reuse the connection inside a single session, or establish separate connections per session_id.
Voices
def list_voices(self) -> list[dict]: ... # SI speaker list
def voice_config_key(self, target_lang: str) -> str
"""Return the speaker config key for the target language: Chinese → 'voice_zh', everything else → 'voice_en'."""
def get_voice_for(self, target_lang: str) -> str
"""Get the speaker id configured for the current language direction (empty string when unconfigured)."""Speakers are split by language (don't use a Chinese voice on an English translation, it produces weird audio); the config keys are the two shown above.
Notes
- Streaming is a real-time hot path:
feed_audiomust return fast; do not do blocking IO / heavy computation inside it. stop_sessionmust really close the connection and release threads; the host calls it when SI is stopped, the engine is switched, or the app exits.- Report session state truthfully through
emit_status; the user interface uses it to show "connecting / interpreting / error".
Reference implementation
extensions/community_cloud_xunfei_si/ (includes a custom SettingsView.qml).