Transcribe streaming audio

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
    StreamingSessionParameters,
13
    TerminationEvent,
14
    TurnEvent,
15
)
16
17
api_key = "<YOUR_API_KEY>"
18
19
logging.basicConfig(level=logging.INFO)
20
logger = logging.getLogger(__name__)
21
22
23
def on_begin(self: Type[StreamingClient], event: BeginEvent):
24
    print(f"Session started: {event.id}")
25
26
27
def on_turn(self: Type[StreamingClient], event: TurnEvent):
28
    print(f"{event.transcript} ({event.end_of_turn})")
29
30
    if event.end_of_turn and not event.turn_is_formatted:
31
        params = StreamingSessionParameters(
32
            format_turns=True,
33
        )
34
35
        self.set_params(params)
36
37
38
def on_terminated(self: Type[StreamingClient], event: TerminationEvent):
39
    print(
40
        f"Session terminated: {event.audio_duration_seconds} seconds of audio processed"
41
    )
42
43
44
def on_error(self: Type[StreamingClient], error: StreamingError):
45
    print(f"Error occurred: {error}")
46
47
48
def main():
49
    client = StreamingClient(
50
        StreamingClientOptions(
51
            api_key=api_key,
52
            api_host="streaming.assemblyai.com",
53
        )
54
    )
55
56
    client.on(StreamingEvents.Begin, on_begin)
57
    client.on(StreamingEvents.Turn, on_turn)
58
    client.on(StreamingEvents.Termination, on_terminated)
59
    client.on(StreamingEvents.Error, on_error)
60
61
    client.connect(
62
        StreamingParameters(
63
            sample_rate=16000,
64
            format_turns=True,
65
        )
66
    )
67
68
    try:
69
        client.stream(
70
          aai.extras.MicrophoneStream(sample_rate=16000)
71
        )
72
    finally:
73
        client.disconnect(terminate=True)
74
75
76
if __name__ == "__main__":
77
    main()

1
logging.basicConfig(level=logging.INFO)
2
logger = logging.getLogger(__name__)
3
4
def on_begin(self: Type[StreamingClient], event: BeginEvent):
5
    print(f"Session started: {event.id}")
6
7
8
def on_turn(self: Type[StreamingClient], event: TurnEvent):
9
    print(f"{event.transcript} ({event.end_of_turn})")
10
11
    if event.end_of_turn and not event.turn_is_formatted:
12
        params = StreamingSessionParameters(
13
            format_turns=True,
14
        )
15
16
        self.set_params(params)
17
18
19
def on_terminated(self: Type[StreamingClient], event: TerminationEvent):
20
    print(
21
        f"Session terminated: {event.audio_duration_seconds} seconds of audio processed"
22
    )
23
24
25
def on_error(self: Type[StreamingClient], error: StreamingError):
26
    print(f"Error occurred: {error}")

1
    finally:
2
        client.disconnect(terminate=True)