### Exception Handling in NLS Python SDK Operations Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Provides a Python example demonstrating how to handle various exceptions defined by the NLS SDK during speech recognition operations. It covers common errors like invalid parameters, connection issues, and timeouts. ```python import nls from nls.exception import ( InvalidParameter, GetTokenFailed, ConnectionTimeout, ConnectionUnavailable, StartTimeoutException, StopTimeoutException, NotStartException, CompleteTimeoutException, WrongStateException ) def safe_recognition(audio_file): try: sr = nls.NlsSpeechRecognizer( url="wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1", token="your_token", appkey="your_appkey", on_completed=lambda msg, *args: print(msg) ) sr.start(aformat="pcm", sample_rate=16000) with open(audio_file, "rb") as f: data = f.read() sr.send_audio(data) sr.stop() except InvalidParameter as e: print(f"Invalid parameter: {e}") # Missing token/appkey, invalid format except GetTokenFailed as e: print(f"Token retrieval failed: {e}") except StartTimeoutException as e: print(f"Connection start timeout: {e}") except StopTimeoutException as e: print(f"Stop command timeout: {e}") except NotStartException as e: print(f"Operation before start: {e}") except CompleteTimeoutException as e: print(f"Completion timeout: {e}") except WrongStateException as e: print(f"Invalid state transition: {e}") except ConnectionResetError: print("Connection was reset by server") # Usage safe_recognition("audio.pcm") ``` -------------------------------- ### Short Sentence Speech Recognition with NlsSpeechRecognizer Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Implements one-shot speech recognition for short audio clips using the NlsSpeechRecognizer class. This snippet demonstrates how to initialize the recognizer with callbacks for different events (start, results, completion, error, close), send audio data in chunks, and process the transcribed text. It supports various audio formats and sample rates, with options for intermediate results and punctuation. ```python import nls import time URL = "wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1" TOKEN = "your_token" APPKEY = "your_appkey" class SpeechRecognitionDemo: def __init__(self): self.result = None def on_start(self, message, *args): print(f"Recognition started: {message}") def on_result_changed(self, message, *args): print(f"Intermediate result: {message}") def on_completed(self, message, *args): print(f"Final result: {message}") self.result = message def on_error(self, message, *args): print(f"Error occurred: {message}") def on_close(self, *args): print("Connection closed") def recognize_file(self, audio_file): # Load audio file with open(audio_file, "rb") as f: audio_data = f.read() # Initialize recognizer with callbacks sr = nls.NlsSpeechRecognizer( url=URL, token=TOKEN, appkey=APPKEY, on_start=self.on_start, on_result_changed=self.on_result_changed, on_completed=self.on_completed, on_error=self.on_error, on_close=self.on_close, callback_args=["session_1"] ) # Start recognition session sr.start( aformat="pcm", # Audio format: pcm, opus, opu, wav, mp3, speex, aac, amr sample_rate=16000, # Sample rate in Hz enable_intermediate_result=True, enable_punctuation_prediction=True, enable_inverse_text_normalization=True, timeout=10, ex={"vocabulary_id": "your_vocab_id"} # Optional extra parameters ) # Send audio data in chunks (640 bytes = 20ms of 16kHz mono PCM) chunk_size = 640 for i in range(0, len(audio_data), chunk_size): chunk = audio_data[i:i+chunk_size] sr.send_audio(chunk) time.sleep(0.02) # 20ms intervals # Stop recognition and wait for completion sr.stop(timeout=10) return self.result # Usage demo = SpeechRecognitionDemo() result = demo.recognize_file("audio.pcm") ``` -------------------------------- ### Get Alibaba Cloud NLS Access Token Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Retrieves an authentication token from Alibaba Cloud using access key ID and secret. This token is essential for initializing any NLS service component. It requires valid Alibaba Cloud credentials and specifies the region and NLS service version. ```python import nls # Get authentication token from Alibaba Cloud AKID = "your_access_key_id" AKSECRET = "your_access_key_secret" try: token = nls.getToken( akid=AKID, aksecret=AKSECRET, domain='cn-shanghai', version='2019-02-28', url='nls-meta.cn-shanghai.aliyuncs.com' ) print(f"Token obtained: {token}") except nls.exception.GetTokenFailed as e: print(f"Failed to get token: {e}") ``` -------------------------------- ### Implement Text-to-Speech Synthesis with NlsSpeechSynthesizer Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Demonstrates how to use the NlsSpeechSynthesizer class to convert text into speech audio. It includes configuration for voice selection, audio format, and handling the synthesis completion callback to save the resulting audio file. ```python import nls URL = "wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1" TOKEN = "your_token" APPKEY = "your_appkey" class TextToSpeechDemo: def __init__(self, output_file): self.output_file = output_file self.audio_data = b"" def on_metainfo(self, message, *args): print(f"Metainfo (subtitle): {message}") def on_data(self, data, *args): self.audio_data += data def on_completed(self, message, *args): print(f"Synthesis completed: {message}") with open(self.output_file, "wb") as f: f.write(self.audio_data) print(f"Audio saved to {self.output_file}") def on_error(self, message, *args): print(f"Error: {message}") def on_close(self, *args): print("Connection closed") def synthesize(self, text): tts = nls.NlsSpeechSynthesizer( url=URL, token=TOKEN, appkey=APPKEY, long_tts=False, on_metainfo=self.on_metainfo, on_data=self.on_data, on_completed=self.on_completed, on_error=self.on_error, on_close=self.on_close, callback_args=["tts_session"] ) tts.start( text=text, voice="xiaoyun", aformat="pcm", sample_rate=16000, volume=50, speech_rate=0, pitch_rate=0, wait_complete=True, start_timeout=10, completed_timeout=60, ex={"enable_subtitle": True} ) ``` -------------------------------- ### Implement Real-time Speech Transcription with NlsSpeechTranscriber Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Demonstrates how to use the NlsSpeechTranscriber class to perform continuous real-time transcription of audio streams. It covers setting up event callbacks, streaming audio chunks, and managing the transcription lifecycle. ```python import nls import time import threading URL = "wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1" TOKEN = "your_token" APPKEY = "your_appkey" class RealtimeTranscriptionDemo: def __init__(self): self.transcriber = None self.running = False def on_start(self, message, *args): print(f"Transcription started: {message}") def on_sentence_begin(self, message, *args): print(f"Sentence begin: {message}") def on_sentence_end(self, message, *args): print(f"Sentence end: {message}") def on_result_changed(self, message, *args): print(f"Result changed: {message}") def on_completed(self, message, *args): print(f"Transcription completed: {message}") def on_error(self, message, *args): print(f"Error: {message}") def on_close(self, *args): print("Connection closed") self.running = False def start_transcription(self, audio_file): with open(audio_file, "rb") as f: audio_data = f.read() self.transcriber = nls.NlsSpeechTranscriber( url=URL, token=TOKEN, appkey=APPKEY, on_start=self.on_start, on_sentence_begin=self.on_sentence_begin, on_sentence_end=self.on_sentence_end, on_result_changed=self.on_result_changed, on_completed=self.on_completed, on_error=self.on_error, on_close=self.on_close, callback_args=["transcribe_session"] ) self.transcriber.start( aformat="pcm", sample_rate=16000, enable_intermediate_result=True, enable_punctuation_prediction=True, enable_inverse_text_normalization=True, ping_interval=8, ping_timeout=None ) self.running = True chunk_size = 640 for i in range(0, len(audio_data), chunk_size): if not self.running: break chunk = audio_data[i:i+chunk_size] self.transcriber.send_audio(chunk) time.sleep(0.02) self.transcriber.ctrl(enable_voice_detection=True) self.transcriber.stop(timeout=10) def force_stop(self): if self.transcriber: self.transcriber.shutdown() demo = RealtimeTranscriptionDemo() demo.start_transcription("long_audio.pcm") ``` -------------------------------- ### Real-time Meeting Transcription with NlsRealtimeMeeting in Python Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Demonstrates how to use the NlsRealtimeMeeting class to perform real-time transcription for meetings. It includes joining a meeting, streaming audio, and handling transcription and translation results. Requires a meeting URL obtained from the CreateMeetingTask API and an audio file. ```python import nls import time class RealtimeMeetingDemo: def __init__(self): self.meeting = None self.running = False def on_start(self, message, *args): print(f"Meeting transcription started: {message}") def on_sentence_begin(self, message, *args): print(f"Sentence begin: {message}") def on_sentence_end(self, message, *args): print(f"Sentence end: {message}") def on_result_changed(self, message, *args): print(f"Transcription result: {message}") def on_result_translated(self, message, *args): print(f"Translation result: {message}") def on_completed(self, message, *args): print(f"Meeting transcription completed: {message}") def on_error(self, message, *args): print(f"Error: {message}") def on_close(self, *args): print("Meeting connection closed") self.running = False def join_meeting(self, meeting_url, audio_file): with open(audio_file, "rb") as f: audio_data = f.read() self.meeting = nls.NlsRealtimeMeeting( url=meeting_url, # Meeting join URL from CreateMeetingTask API on_start=self.on_start, on_sentence_begin=self.on_sentence_begin, on_sentence_end=self.on_sentence_end, on_result_changed=self.on_result_changed, on_result_translated=self.on_result_translated, on_completed=self.on_completed, on_error=self.on_error, on_close=self.on_close, callback_args=["meeting_session"] ) # Start meeting session self.meeting.start( timeout=10, ping_interval=8, ping_timeout=None, ex={"enable_translation": True} # Optional extra parameters ) self.running = True # Stream audio to meeting chunk_size = 640 for i in range(0, len(audio_data), chunk_size): if not self.running: break chunk = audio_data[i:i+chunk_size] self.meeting.send_audio(chunk) time.sleep(0.02) # Stop meeting transcription self.meeting.stop(timeout=10) def leave_meeting(self): if self.meeting: self.meeting.shutdown() # Usage demo = RealtimeMeetingDemo() demo.join_meeting("wss://meeting-url-from-api", "meeting_audio.pcm") ``` -------------------------------- ### Perform Streaming and One-Shot Text-to-Speech Synthesis with Python Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt This snippet demonstrates how to use the NlsStreamInputTtsSynthesizer class to handle both incremental text streaming and standard one-shot synthesis. It includes event callbacks for monitoring the synthesis lifecycle and saving the resulting audio data to a file. ```python import nls URL = "wss://nls-gateway.cn-shanghai.aliyuncs.com/ws/v1" TOKEN = "your_token" APPKEY = "your_appkey" class StreamingTTSDemo: def __init__(self, output_file): self.output_file = output_file self.audio_data = b"" def on_data(self, data, *args): self.audio_data += data def on_completed(self, message, *args): with open(self.output_file, "wb") as f: f.write(self.audio_data) def streaming_synthesis(self, text_chunks): synthesizer = nls.NlsStreamInputTtsSynthesizer( url=URL, token=TOKEN, appkey=APPKEY, on_data=self.on_data, on_completed=self.on_completed ) synthesizer.startStreamInputTts(voice="longxiaochun", aformat="pcm") for chunk in text_chunks: synthesizer.sendStreamInputTts(chunk) synthesizer.stopStreamInputTts() def simple_synthesis(self, text): synthesizer = nls.NlsStreamInputTtsSynthesizer( url=URL, token=TOKEN, appkey=APPKEY, on_data=self.on_data, on_completed=self.on_completed ) synthesizer.startTts(text=text, voice="longxiaochun", aformat="pcm") synthesizer.waitForComplete() ``` -------------------------------- ### SDK Logging and Debugging Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Utilities for enabling trace-level logging to troubleshoot connection and recognition issues within the SDK. ```APIDOC ## Enabling Debug Logging ### Description Use the nls.enableTrace method to toggle detailed logging for debugging purposes. ### Usage - `nls.enableTrace(True)`: Enable trace logging to console. - `nls.enableTrace(True, handler=file_handler)`: Enable trace logging to a specific file. - `nls.debug(msg)`: Log a debug message. - `nls.dump(tag, data)`: Log formatted data for inspection. ``` -------------------------------- ### Enabling Debug Logging for NLS Python SDK Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Shows how to enable and configure trace-level debug logging for the NLS Python SDK to assist in troubleshooting. It covers enabling logging to the console, to a file using a custom handler, and disabling logging. ```python import nls import logging # Enable trace logging to console (default) nls.enableTrace(True) # Enable trace logging with custom handler file_handler = logging.FileHandler("nls_debug.log") nls.enableTrace(True, handler=file_handler) # Disable trace logging nls.enableTrace(False) # Manual logging calls nls.debug("Debug message") nls.error("Error message") nls.warning("Warning message") nls.trace("Trace message - only shown when trace is enabled") nls.dump("Request", '{"header": {...}}') # Dump formatted data ``` -------------------------------- ### Exception Handling Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt Overview of the exception types defined in the SDK to handle various error conditions during speech processing. ```APIDOC ## Exception Handling ### Description The SDK provides specific exception classes to handle operational errors during speech recognition tasks. ### Common Exceptions - **InvalidParameter**: Raised when required parameters like token or appkey are missing or invalid. - **GetTokenFailed**: Raised when authentication token retrieval fails. - **StartTimeoutException**: Raised when the connection fails to establish within the timeout period. - **WrongStateException**: Raised when an operation is performed in an invalid state (e.g., calling stop before start). ``` -------------------------------- ### WebSocket Handshake Source: https://github.com/aliyun/alibabacloud-nls-python-sdk/blob/dev/nls/websocket/tests/data/header01.txt This snippet illustrates a successful WebSocket handshake response. ```APIDOC ## HTTP 101 WebSocket Protocol Handshake ### Description This response indicates that the server is switching protocols from HTTP to WebSocket, as requested by the client. ### Method N/A (This is a response, not a request) ### Endpoint N/A (This is a response, not a request) ### Parameters N/A ### Request Example N/A ### Response #### Success Response (101) - **Connection** (string) - Indicates the connection should be upgraded. - **Upgrade** (string) - Specifies the protocol to upgrade to (e.g., "WebSocket"). - **Sec-WebSocket-Accept** (string) - A base64 encoded string computed from the client's Sec-WebSocket-Key and the server's magic string. - **some_header** (string) - Example of an additional header that might be included. #### Response Example ``` HTTP/1.1 101 WebSocket Protocol Handshake Connection: Upgrade Upgrade: WebSocket Sec-WebSocket-Accept: Kxep+hNu9n51529fGidYu7a3wO0= some_header: something ``` ``` -------------------------------- ### NlsRealtimeMeeting - Real-time Meeting Transcription Source: https://context7.com/aliyun/alibabacloud-nls-python-sdk/llms.txt This class manages real-time transcription for meeting scenarios, including support for translation and audio streaming. ```APIDOC ## NlsRealtimeMeeting ### Description The NlsRealtimeMeeting class facilitates real-time transcription and translation for meeting sessions by connecting to a provided meeting URL. ### Method WebSocket (Streaming) ### Endpoint wss://[meeting-url-from-api] ### Parameters #### Request Body - **url** (string) - Required - The meeting join URL obtained from the CreateMeetingTask API. - **on_start** (callback) - Required - Callback triggered when transcription starts. - **on_result_changed** (callback) - Required - Callback triggered when partial transcription results are updated. - **on_result_translated** (callback) - Optional - Callback triggered when translation results are available. ### Request Example { "url": "wss://meeting-url-from-api", "ex": {"enable_translation": true} } ### Response #### Success Response (200) - **message** (string) - The transcribed or translated text content. ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.