tingwu-sdk-python-realtime

class TingWuRealtimeCallback:
    """An interface that defines callback methods for getting TingWu results.
       Derive from this class and implement its function to provide your own data.
    """

    def on_open(self) -> None:
        pass # connected callback

    def on_started(self, task_id: str) -> None:
        pass # protocal handshake over

    def on_speech_listen(self, result: dict):
        pass # callback speech listen event

    def on_recognize_result(self, result: dict):
        pass # callback recognize result

    def on_ai_result(self, result: dict):
        pass # callback ai result

    def on_stopped(self) -> None:
        pass # recognize over

    def on_error(self, error_code: str, error_msg: str) -> None:
        pass

    def on_close(self, close_status_code, close_msg):
        pass

{
  "header": {
    "event":"result-generated",
    "task_id": "f2E3zvK*******wp"
  },
  "payload": {
    "output": {
      "action": "recognize-result",
      "transcription": {
        "sentenceId": 0,
        "beginTime": 100,
        "endTime": 2720,
        "sentenceEnd": true,
        "text": "This is a sentence for testing.",
        "words": [
          {
            "beginTime": 100,
            "endTime": 427,
            "text": "This"
          },
          {
            "beginTime": 427,
            "endTime": 755,
            "text": "is a"
          },
          {
            "beginTime": 755,
            "endTime": 1082,
            "text": "sentence"
          },
          {
            "beginTime": 1082,
            "endTime": 1410,
            "text": "used for"
          },
          {
            "beginTime": 1410,
            "endTime": 1737,
            "text": "testing"
          },
          {
            "beginTime": 1737,
            "endTime": 2065,
            "text": "of"
          },
          {
            "beginTime": 2065,
            "endTime": 2392,
            "text": "text"
          },
          {
            "beginTime": 2392,
            "endTime": 2720,
            "text": "."
          }
        ]
      },
      "translations": {
        "sentenceEnd": true,
        "translations": {
          "zh": {
            "sentenceId": 0,
            "beginTime": 100,
            "endTime": 2720,
            "text": "This is a sentence for testing.",
            "lang": "zh",
            "words": [
              {
                "beginTime": 100,
                "endTime": 427,
                "text": "This"
              },
              {
                "beginTime": 427,
                "endTime": 755,
                "text": "is a"
              },
              {
                "beginTime": 755,
                "endTime": 1082,
                "text": "sentence"
              },
              {
                "beginTime": 1082,
                "endTime": 1410,
                "text": "used for"
              },
              {
                "beginTime": 1410,
                "endTime": 1737,
                "text": "testing"
              },
              {
                "beginTime": 1737,
                "endTime": 2065,
                "text": "of"
              },
              {
                "beginTime": 2065,
                "endTime": 2392,
                "text": "text"
              },
              {
                "beginTime": 2392,
                "endTime": 2720,
                "text": "."
              }
            ],
            "sentenceEnd": true
          }
        }
      }
    }
  }
}

import logging
import time
import sys
from dashscope.multimodal.tingwu.tingwu_realtime import TingWuRealtime, TingWuRealtimeCallback

# Configure logging - a key improvement
logger = logging.getLogger('dashscope')
logger.setLevel(logging.DEBUG)

# Create a console handler and set the level to debug
console_handler = logging.StreamHandler(sys.stdout)  # Explicitly specify output to stdout
console_handler.setLevel(logging.DEBUG)

# Create a formatter
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# Add the formatter to the handler
console_handler.setFormatter(formatter)

# Add the handler to the logger
logger.addHandler(console_handler)

# Force flush the log output
logger.propagate = False

class TestCallback(TingWuRealtimeCallback):
    def __init__(self):  # Fix: __init__ method name
        super().__init__()
        self.can_send_audio = False
        print("TestCallback initialized")  # Add debug output
        sys.stdout.flush()  # Force flush

    def on_open(self) -> None:
        print('TingWuClient:: on websocket open.')  # Use print to ensure output
        logger.info('TingWuClient:: on websocket open.')
        sys.stdout.flush()

    def on_started(self, task_id: str) -> None:
        print(f'TingWuClient:: on task started. task_id: {task_id}')
        logger.info('TingWuClient:: on task started. task_id: %s', task_id)
        sys.stdout.flush()

    def on_speech_listen(self, result: dict):
        print(f'TingWuClient:: on speech listen. result: {result}')
        logger.info('TingWuClient:: on speech listen. result: %s', result)
        self.can_send_audio = True  # Mark that audio data can be sent
        sys.stdout.flush()

    def on_recognize_result(self, result: dict):
        print(f'TingWuClient:: on recognize result. result: {result}')
        logger.info('TingWuClient:: on recognize result. result: %s', result)
        sys.stdout.flush()

    def on_ai_result(self, result: dict):
        print(f'TingWuClient:: on ai result. result: {result}')
        logger.info('TingWuClient:: on ai result. result: %s', result)
        sys.stdout.flush()

    def on_stopped(self) -> None:
        print('TingWuClient:: on task stopped.')
        logger.info('TingWuClient:: on task stopped.')
        self.can_send_audio = False  # Mark that audio data cannot be sent
        sys.stdout.flush()

    def on_error(self, error_code: str, error_msg: str) -> None:
        print(f'TingWuClient:: on error. error_code: {error_code}, error_msg: {error_msg}')
        logger.info('TingWuClient:: on error. error_code: %s, error_msg: %s',
                    error_code, error_msg)
        sys.stdout.flush()

    def on_close(self, close_status_code, close_msg):
        print(f'TingWuClient:: on websocket close. close_status_code: {close_status_code}, close_msg: {close_msg}')
        logger.info('TingWuClient:: on websocket close. close_status_code: %s, close_msg: %s',
                    close_status_code, close_msg)
        sys.stdout.flush()

class RunTingWuRealtime:
    @classmethod
    def setup_class(cls):
        cls.model = 'tingwu-industrial-instruction'
        cls.format = 'pcm'
        cls.sample_rate = 16000
        cls.file = './data/meeting_audio.pcm'
        cls.appId = 'your-app-id'
        cls.base_address = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference'
        cls.api_key = 'Your-dashscope-api-key'
        cls.terminology = 'your-terminology-id'

    def start_with_stream(self):
        print("Starting test...")
        sys.stdout.flush()

        callback = TestCallback()

        print("Creating TingWuRealtime instance...")
        sys.stdout.flush()

        tingwu_realtime = TingWuRealtime(
            model=self.model,
            audio_format=self.format,
            sample_rate=self.sample_rate,
            app_id=self.appId,
            base_address=self.base_address,
            api_key=self.api_key,
            terminology=self.terminology,
            callback=callback,
            max_end_silence=3000
        )

        print("Starting TingWu connection...")
        sys.stdout.flush()
        tingwu_realtime.start()

        # Wait for the connection to be established
        time.sleep(2)
        # Wait for the system state to transition to speech_listen
        try:
            print(f"Opening file: {self.file}")
            sys.stdout.flush()

            with open(self.file, 'rb') as f:
                chunk_count = 0
                while True:
                    chunk = f.read(3200)
                    if not chunk:
                        print("File reading complete")
                        sys.stdout.flush()
                        break

                    # Fix logic error: Send only when can_send_audio is True
                    if callback.can_send_audio:
                        tingwu_realtime.send_audio_frame(chunk)
                        time.sleep(0.1)
                        chunk_count += 1
                        if chunk_count % 10 == 0:  # Print every 10 chunks
                            print(f"Sent {chunk_count} audio chunks")
                            sys.stdout.flush()
                    else:
                        print("Waiting for permission to send audio...")
                        sys.stdout.flush()
                        time.sleep(0.1)  # Wait for a short period
            logger.info('============= File reading complete')
            time.sleep(100)  # Wait for a period of time
        except FileNotFoundError:
            print(f"File not found: {self.file}")
            sys.stdout.flush()
        except Exception as e:
            print(f"An error occurred: {e}")
            sys.stdout.flush()
        finally:
            print("Sending stop instruction...")
            sys.stdout.flush()
            tingwu_realtime.stop()
            time.sleep(10)  # Wait for a period of time
            tingwu_realtime.close()
            print("TingWu connection closed")

if __name__ == '__main__':
    print('Starting test_tingwu_realtime test')
    sys.stdout.flush()

    logger.debug('Start test_tingwu_realtime.')

    test_env = RunTingWuRealtime()
    test_env.setup_class()  # Set up the test environment
    test_env.start_with_stream()  # Call the test method

    print('Test complete')
    sys.stdout.flush()