Realtime audio and video translation

import os
import time
import base64
import asyncio
import json
import websockets
import pyaudio
import queue
import threading
import traceback

class LiveTranslateClient:
  def __init__(self, api_key: str, target_language: str = "en", voice: str | None = "Cherry", *, audio_enabled: bool = True):
    if not api_key:
      raise ValueError("API key cannot be empty.")

    self.api_key = api_key
    self.target_language = target_language
    self.audio_enabled = audio_enabled
    self.voice = voice if audio_enabled else "Cherry"
    self.ws = None
    self.api_url = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-livetranslate-flash-realtime"

    # Audio input configuration (from microphone)
    self.input_rate = 16000
    self.input_chunk = 1600
    self.input_format = pyaudio.paInt16
    self.input_channels = 1

    # Audio output configuration (for playback)
    self.output_rate = 24000
    self.output_chunk = 2400
    self.output_format = pyaudio.paInt16
    self.output_channels = 1

    # State management
    self.is_connected = False
    self.audio_player_thread = None
    self.audio_playback_queue = queue.Queue()
    self.pyaudio_instance = pyaudio.PyAudio()

  async def connect(self):
    """Establish a WebSocket connection to the translation service."""
    headers = {"Authorization": f"Bearer {self.api_key}"}
    try:
      self.ws = await websockets.connect(self.api_url, additional_headers=headers)
      self.is_connected = True
      print(f"Successfully connected to the server: {self.api_url}")
      await self.configure_session()
    except Exception as e:
      print(f"Connection failed: {e}")
      self.is_connected = False
      raise

  async def configure_session(self):
    """Configure the translation session, setting the target language, voice, and other parameters."""
    config = {
      "event_id": f"event_{int(time.time() * 1000)}",
      "type": "session.update",
      "session": {
        # 'modalities' controls the output type.
        # ["text", "audio"]: Returns both translated text and synthesized audio (recommended).
        # ["text"]: Returns only the translated text.
        "modalities": ["text", "audio"] if self.audio_enabled else ["text"],
        **({"voice": self.voice} if self.audio_enabled and self.voice else {}),
        "input_audio_format": "pcm16",
        "output_audio_format": "pcm24",
        # 'input_audio_transcription' configures source language recognition.
        # Set 'model' to 'qwen3-asr-flash-realtime' to also output the source language recognition result.
        # "input_audio_transcription": {
        #     "model": "qwen3-asr-flash-realtime",
        #     "language": "zh"  # Source language, default is 'en'
        # },
        "translation": {
          "language": self.target_language
        }
      }
    }
    print(f"Sending session configuration: {json.dumps(config, indent=2, ensure_ascii=False)}")
    await self.ws.send(json.dumps(config))

  async def send_audio_chunk(self, audio_data: bytes):
    """Encode and send an audio data block to the server."""
    if not self.is_connected:
      return

    event = {
      "event_id": f"event_{int(time.time() * 1000)}",
      "type": "input_audio_buffer.append",
      "audio": base64.b64encode(audio_data).decode()
    }
    await self.ws.send(json.dumps(event))

  async def send_image_frame(self, image_bytes: bytes, *, event_id: str | None = None):
    # Send image data to the server
    if not self.is_connected:
      return

    if not image_bytes:
      raise ValueError("image_bytes cannot be empty")

    # Encode to Base64
    image_b64 = base64.b64encode(image_bytes).decode()

    event = {
      "event_id": event_id or f"event_{int(time.time() * 1000)}",
      "type": "input_image_buffer.append",
      "image": image_b64,
    }

    await self.ws.send(json.dumps(event))

  def _audio_player_task(self):
    stream = self.pyaudio_instance.open(
      format=self.output_format,
      channels=self.output_channels,
      rate=self.output_rate,
      output=True,
      frames_per_buffer=self.output_chunk,
    )
    try:
      while self.is_connected or not self.audio_playback_queue.empty():
        try:
          audio_chunk = self.audio_playback_queue.get(timeout=0.1)
          if audio_chunk is None: # Termination signal
            break
          stream.write(audio_chunk)
          self.audio_playback_queue.task_done()
        except queue.Empty:
          continue
    finally:
      stream.stop_stream()
      stream.close()

  def start_audio_player(self):
    """Start the audio player thread (only when audio output is enabled)."""
    if not self.audio_enabled:
      return
    if self.audio_player_thread is None or not self.audio_player_thread.is_alive():
      self.audio_player_thread = threading.Thread(target=self._audio_player_task, daemon=True)
      self.audio_player_thread.start()

  async def handle_server_messages(self, on_text_received):
    """Loop to process messages from the server."""
    try:
      async for message in self.ws:
        event = json.loads(message)
        event_type = event.get("type")
        if event_type == "response.audio.delta" and self.audio_enabled:
          audio_b64 = event.get("delta", "")
          if audio_b64:
            audio_data = base64.b64decode(audio_b64)
            self.audio_playback_queue.put(audio_data)

        elif event_type == "response.done":
          print("\n[INFO] A round of response is complete.")
          usage = event.get("response", {}).get("usage", {})
          if usage:
            print(f"[INFO] Token usage: {json.dumps(usage, indent=2, ensure_ascii=False)}")
        # Process source language recognition results (requires enabling input_audio_transcription.model)
        # elif event_type == "conversation.item.input_audio_transcription.text":
        #     stash = event.get("stash", "")  # Unconfirmed recognition text
        #     print(f"[Recognizing] {stash}")
        # elif event_type == "conversation.item.input_audio_transcription.completed":
        #     transcript = event.get("transcript", "")  # Complete recognition result
        #     print(f"[Source language] {transcript}")
        elif event_type == "response.audio_transcript.done":
          print("\n[INFO] Text translation complete.")
          text = event.get("transcript", "")
          if text:
            print(f"[INFO] Translated text: {text}")
        elif event_type == "response.text.done":
          print("\n[INFO] Text translation complete.")
          text = event.get("text", "")
          if text:
            print(f"[INFO] Translated text: {text}")

    except websockets.exceptions.ConnectionClosed as e:
      print(f"[WARNING] Connection closed: {e}")
      self.is_connected = False
    except Exception as e:
      print(f"[ERROR] An unknown error occurred while processing messages: {e}")
      traceback.print_exc()
      self.is_connected = False

  async def start_microphone_streaming(self):
    """Capture audio from the microphone and stream it to the server."""
    stream = self.pyaudio_instance.open(
      format=self.input_format,
      channels=self.input_channels,
      rate=self.input_rate,
      input=True,
      frames_per_buffer=self.input_chunk
    )
    print("Microphone is on. Start speaking...")
    try:
      while self.is_connected:
        audio_chunk = await asyncio.get_event_loop().run_in_executor(
          None, stream.read, self.input_chunk
        )
        await self.send_audio_chunk(audio_chunk)
    finally:
      stream.stop_stream()
      stream.close()

  async def close(self):
    """Gracefully close the connection and release resources."""
    self.is_connected = False
    if self.ws:
      await self.ws.close()
      print("WebSocket connection closed.")

    if self.audio_player_thread:
      self.audio_playback_queue.put(None) # Send termination signal
      self.audio_player_thread.join(timeout=1)
      print("Audio player thread stopped.")

    self.pyaudio_instance.terminate()
    print("PyAudio instance released.")

import os
import asyncio
from livetranslate_client import LiveTranslateClient

def print_banner():
  print("=" * 60)
  print("  Powered by Qwen qwen3-livetranslate-flash-realtime")
  print("=" * 60 + "\n")

def get_user_config():
  """Get user configuration"""
  print("Select a mode:")
  print("1. Voice + Text [Default] | 2. Text Only")
  mode_choice = input("Enter your choice (press Enter for Voice + Text): ").strip()
  audio_enabled = (mode_choice != "2")

  if audio_enabled:
    lang_map = {
      "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt",
      "7": "es", "8": "it", "9": "ko", "10": "ja", "11": "yue"
    }
    print("Select the target translation language (Voice + Text mode):")
    print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Korean | 10. Japanese | 11. Cantonese")
  else:
    lang_map = {
      "1": "en", "2": "zh", "3": "ru", "4": "fr", "5": "de", "6": "pt", "7": "es", "8": "it",
      "9": "id", "10": "ko", "11": "ja", "12": "vi", "13": "th", "14": "ar",
      "15": "yue", "16": "hi", "17": "el", "18": "tr"
    }
    print("Select the target translation language (Text Only mode):")
    print("1. English | 2. Chinese | 3. Russian | 4. French | 5. German | 6. Portuguese | 7. Spanish | 8. Italian | 9. Indonesian | 10. Korean | 11. Japanese | 12. Vietnamese | 13. Thai | 14. Arabic | 15. Cantonese | 16. Hindi | 17. Greek | 18. Turkish")

  choice = input("Enter your choice (defaults to the first option): ").strip()
  target_language = lang_map.get(choice, next(iter(lang_map.values())))

  voice = None
  if audio_enabled:
    print("\nSelect a speech synthesis voice:")
    voice_map = {"1": "Cherry", "2": "Nofish", "3": "Sunny", "4": "Jada", "5": "Dylan", "6": "Peter", "7": "Eric", "8": "Kiki", "9": "Ethan"}
    print("1. Cherry (Female) [Default] | 2. Nofish (Male) | 3. Sunny (Sichuan Female) | 4. Jada (Shanghai Female) | 5. Dylan (Beijing Male) | 6. Peter (Tianjin Male) | 7. Eric (Sichuan Male) | 8. Kiki (Cantonese Female) | 9. Ethan (Male)")
    voice_choice = input("Enter your choice (press Enter for Cherry): ").strip()
    voice = voice_map.get(voice_choice, "Cherry")
  return target_language, voice, audio_enabled

async def main():
  """Main program entry point"""
  print_banner()

  api_key = os.environ.get("DASHSCOPE_API_KEY")
  if not api_key:
    print("[ERROR] Set the DASHSCOPE_API_KEY environment variable.")
    print("  For example: export DASHSCOPE_API_KEY='YOUR_API_KEY'")
    return

  target_language, voice, audio_enabled = get_user_config()
  print("\nConfiguration complete:")
  print(f"  - Target language: {target_language}")
  if audio_enabled:
    print(f"  - Synthesized voice: {voice}")
  else:
    print("  - Output mode: Text Only")

  client = LiveTranslateClient(api_key=api_key, target_language=target_language, voice=voice, audio_enabled=audio_enabled)

  # Define the callback function
  def on_translation_text(text):
    print(text, end="", flush=True)

  try:
    print("Connecting to the translation service...")
    await client.connect()

    # Start audio playback based on the mode
    client.start_audio_player()

    print("\n" + "-" * 60)
    print("Connection successful! Speak into the microphone.")
    print("The program will translate your speech in real time and play the result. Press Ctrl+C to exit.")
    print("-" * 60 + "\n")

    # Run message handling and microphone recording concurrently
    message_handler = asyncio.create_task(client.handle_server_messages(on_translation_text))
    tasks = [message_handler]
    # Audio must be captured from the microphone for translation, regardless of whether audio output is enabled
    microphone_streamer = asyncio.create_task(client.start_microphone_streaming())
    tasks.append(microphone_streamer)

    await asyncio.gather(*tasks)

  except KeyboardInterrupt:
    print("\n\nUser interrupted. Exiting...")
  except Exception as e:
    print(f"\nA critical error occurred: {e}")
  finally:
    print("\nCleaning up resources...")
    await client.close()
    print("Program exited.")

if __name__ == "__main__":
  asyncio.run(main())

# pip install websocket-client
import json
import websocket
import os

API_KEY=os.getenv("DASHSCOPE_API_KEY")
API_URL = "wss://dashscope-intl.aliyuncs.com/api-ws/v1/realtime?model=qwen3-livetranslate-flash-realtime"

headers = [
  "Authorization: Bearer " + API_KEY
]

def on_open(ws):
  print(f"Connected to server: {API_URL}")
def on_message(ws, message):
  data = json.loads(message)
  print("Received event:", json.dumps(data, indent=2))
def on_error(ws, error):
  print("Error:", error)

ws = websocket.WebSocketApp(
  API_URL,
  header=headers,
  on_open=on_open,
  on_message=on_message,
  on_error=on_error
)

ws.run_forever()