Supported Languages

1
import logging
2
from typing import Type
3
4
import assemblyai as aai
5
from assemblyai.streaming.v3 import (
6
    BeginEvent,
7
    StreamingClient,
8
    StreamingClientOptions,
9
    StreamingError,
10
    StreamingEvents,
11
    StreamingParameters,
12
    TurnEvent,
13
    TerminationEvent,
14
)
15
16
api_key = "<YOUR_API_KEY>"
17
18
logging.basicConfig(level=logging.INFO)
19
logger = logging.getLogger(__name__)
20
21
def on_begin(self: Type[StreamingClient], event: BeginEvent):
22
    print(f"Session started: {event.id}")
23
24
def on_turn(self: Type[StreamingClient], event: TurnEvent):
25
    print(f"{event.transcript} ({event.end_of_turn})")
26
27
def on_terminated(self: Type[StreamingClient], event: TerminationEvent):
28
    print(
29
        f"Session terminated: {event.audio_duration_seconds} seconds of audio processed"
30
    )
31
32
def on_error(self: Type[StreamingClient], error: StreamingError):
33
    print(f"Error occurred: {error}")
34
35
def main():
36
    client = StreamingClient(
37
        StreamingClientOptions(
38
            api_key=api_key,
39
            api_host="streaming.assemblyai.com",
40
        )
41
    )
42
43
    client.on(StreamingEvents.Begin, on_begin)
44
    client.on(StreamingEvents.Turn, on_turn)
45
    client.on(StreamingEvents.Termination, on_terminated)
46
    client.on(StreamingEvents.Error, on_error)
47
48
    client.connect(
49
        StreamingParameters(
50
            sample_rate=16000,
51
            speech_model="u3-rt-pro",
52
        )
53
    )
54
55
    try:
56
        client.stream(
57
            aai.extras.MicrophoneStream(sample_rate=16000)
58
        )
59
    finally:
60
        client.disconnect(terminate=True)
61
62
if __name__ == "__main__":
63
    main()