Tradução em tempo real

1
2
3
4
5
6
7
8
9
10
11import os
import websocket

ws = websocket.WebSocket()
ws.connect(
    "wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate",
    header=[
        f"Authorization: Bearer {os.environ['OPENAI_API_KEY']}",
        "OpenAI-Safety-Identifier: hashed-user-id",
    ],
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
client := openai.NewClient()
session, err := client.Realtime.ConnectTranslation(ctx, "gpt-realtime-translate", realtime.TranslationConnectionOptions{}, option.WithHeader("OpenAI-Safety-Identifier", "hashed-user-id"))
if err != nil {
	return err
}
defer session.Abort()
created, err := session.Recv(ctx)
if err != nil {
	return err
}
fmt.Println(created.Type)
if created.Type != "session.created" {
	return fmt.Errorf("expected session.created, got %s", created.Type)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32import com.openai.client.okhttp.OkHttpClient;
import com.openai.core.ClientOptions;
import com.openai.helpers.TranslationConnection;
import com.openai.helpers.TranslationWebSocketOptions;
import com.openai.models.realtime.*;

var http = OkHttpClient.builder().build();
var options =
    ClientOptions.builder()
        .fromEnv()
        .httpClient(http)
        .putHeader("OpenAI-Safety-Identifier", "hashed-user-id")
        .build();
try (http;
    var connection =
        TranslationConnection.connect(
            options,
            TranslationWebSocketOptions.builder()
                .model("gpt-realtime-translate")
                .sendTimeout(java.time.Duration.ofSeconds(30))
                .build())) {
  while (true) {
    var event = connection.receive();
    if (event.error().isPresent())
      throw new IllegalStateException(event.error().orElseThrow().toString());
    if (event.sessionCreated().isPresent()) {
      System.out.println("Connected to translation session.");
      System.out.println(event);
      break;
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21require "async"
require "async/http/endpoint"
require "async/websocket/client"
require "json"

endpoint = Async::HTTP::Endpoint.parse("wss://api.openai.com/v1/realtime/translations?model=gpt-realtime-translate", timeout: 10, alpn_protocols: ["http/1.1"])
headers = {
  "Authorization" => "Bearer #{ENV.fetch("OPENAI_API_KEY")}",
  "OpenAI-Safety-Identifier" => "hashed-user-id"
}
Sync do |task|
  task.with_timeout(120) do
    Async::WebSocket::Client.connect(endpoint, headers: headers) do |connection|
      message = connection.read or raise "Connection closed before session creation"
      event = JSON.parse(message.to_str)
      raise "Expected session.created: #{event}" unless event["type"] == "session.created"

      puts(event.fetch("type"))
    end
  end
end

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16import json

ws.send(
    json.dumps(
        {
            "type": "session.update",
            "session": {
                "audio": {
                    "output": {
                        "language": "es",
                    },
                },
            },
        }
    )
)

1
2
3
4
5
6
7
8
9if err := session.Send(ctx, realtime.RealtimeTranslationClientEventParamOfSessionUpdate(realtime.RealtimeTranslationSessionUpdateRequestParam{
	Audio: realtime.RealtimeTranslationSessionUpdateRequestAudioParam{
		Output: realtime.RealtimeTranslationSessionUpdateRequestAudioOutputParam{
			Language: openai.String("es"),
		},
	},
})); err != nil {
	return err
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20import com.openai.client.okhttp.OkHttpClient;
import com.openai.core.ClientOptions;
import com.openai.helpers.TranslationConnection;
import com.openai.helpers.TranslationWebSocketOptions;
import com.openai.models.realtime.*;

connection.send(
    RealtimeTranslationClientEvent.ofSessionUpdate(
        RealtimeTranslationSessionUpdateEvent.builder()
            .session(
                RealtimeTranslationSessionUpdateRequest.builder()
                    .audio(
                        RealtimeTranslationSessionUpdateRequest.Audio.builder()
                            .output(
                                RealtimeTranslationSessionUpdateRequest.Audio.Output.builder()
                                    .language("es")
                                    .build())
                            .build())
                    .build())
            .build()));

1
2connection.write(JSON.generate(type: "session.update", session: { audio: { output: { language: "es" } } }))
connection.flush

1
2
3
4
5
6
7
8ws.send(
    json.dumps(
        {
            "type": "session.input_audio_buffer.append",
            "audio": base64_pcm16,
        }
    )
)

1
2
3
4
5
6
7// audio contains mono PCM16 samples at 24 kHz.
for start := 0; start < len(audio); start += 4800 {
	end := min(start+4800, len(audio))
	if err := session.Send(ctx, realtime.RealtimeTranslationClientEventParamOfSessionInputAudioBufferAppend(base64.StdEncoding.EncodeToString(audio[start:end]))); err != nil {
		return err
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14import com.openai.client.okhttp.OkHttpClient;
import com.openai.core.ClientOptions;
import com.openai.helpers.TranslationConnection;
import com.openai.helpers.TranslationWebSocketOptions;
import com.openai.models.realtime.*;
import java.util.Base64;

// Call for each PCM16, 24 kHz, mono chunk supplied by your audio source.
static void appendAudio(TranslationConnection connection, byte[] pcmAudio) {
  String base64Pcm16 = Base64.getEncoder().encodeToString(pcmAudio);
  connection.send(
      RealtimeTranslationClientEvent.ofSessionInputAudioBufferAppend(
          RealtimeTranslationInputAudioBufferAppendEvent.builder().audio(base64Pcm16).build()));
}

1
2
3
4
5
6
7
8require "base64"

File.open("speech.pcm", "rb") do |audio|
  while (chunk = audio.read(4_800))
    connection.write(JSON.generate(type: "session.input_audio_buffer.append", audio: Base64.strict_encode64(chunk)))
    connection.flush
  end
end

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50import base64
import json
from concurrent.futures import ThreadPoolExecutor
from contextlib import suppress
from queue import SimpleQueue

import sounddevice as sd

audio_queue = SimpleQueue()


def play_audio():
    try:
        # Closing the stream waits for PortAudio's pending buffers to finish.
        with sd.RawOutputStream(samplerate=24000, channels=1, dtype="int16") as stream:
            while (pcm := audio_queue.get()) is not None:
                stream.write(pcm)
    except Exception:
        # The receiver may have already closed the socket.
        with suppress(OSError):
            ws.abort()  # Wake recv() so the receiver can report the playback failure.
        raise


with ThreadPoolExecutor(max_workers=1) as executor:
    playback = executor.submit(play_audio)
    try:
        while True:
            event = json.loads(ws.recv())
            if playback.done():
                playback.result()  # Report a playback failure on the receiver.

            if event["type"] == "session.output_audio.delta":
                audio_queue.put(base64.b64decode(event["delta"]))

            if event["type"] in (
                "session.output_transcript.delta",
                "session.input_transcript.delta",
            ):
                print(event["delta"], end="", flush=True)

            if event["type"] == "error":
                raise RuntimeError(f"Translation error: {event['error']['code']}")

            if event["type"] == "session.closed":
                break
    finally:
        ws.close()
        audio_queue.put(None)
        playback.result()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24// Run this listener while another goroutine sends audio into the session.
func listen(ctx context.Context, session *realtime.TranslationSession) error {
	for {
		event, err := session.Recv(ctx)
		if err != nil {
			return err
		}
		switch event.Type {
		case "error":
			return fmt.Errorf("translation error: %s", event.OfRealtimeErrorEvent.Error.Message)
		case "session.output_transcript.delta":
			fmt.Print(event.OfRealtimeTranslationOutputTranscriptDeltaEvent.Delta)
		case "session.output_audio.delta":
			pcm, err := base64.StdEncoding.DecodeString(event.OfRealtimeTranslationOutputAudioDeltaEvent.Delta)
			if err != nil {
				return err
			}
			// Feed pcm into your application's 24 kHz audio player.
			fmt.Fprintln(log.Writer(), "Received audio bytes:", len(pcm))
		case "session.closed":
			return session.Close()
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76import base64
import json
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import suppress
from queue import SimpleQueue

import sounddevice as sd

audio_queue = SimpleQueue()


def play_audio():
    try:
        # Closing the stream waits for PortAudio's pending buffers to finish.
        with sd.RawOutputStream(samplerate=24000, channels=1, dtype="int16") as stream:
            while (pcm := audio_queue.get()) is not None:
                stream.write(pcm)
    except Exception:
        # The receiver may have already closed the socket.
        with suppress(OSError):
            ws.abort()  # Wake recv() so the receiver can report the playback failure.
        raise


translation_session_closing = False


def close_translation_session():
    global translation_session_closing
    if translation_session_closing:
        return

    translation_session_closing = True
    ws.send(json.dumps({"type": "session.close"}))


with ThreadPoolExecutor(max_workers=1) as executor:
    playback = executor.submit(play_audio)
    try:
        # Call this when the source stream ends.
        close_translation_session()
        # Adjust this deadline for your application's finalization requirements.
        drain_deadline = time.monotonic() + 15.0

        while True:
            remaining = drain_deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError("Translation finalization timed out")
            ws.settimeout(remaining)
            data = ws.recv()
            if not data:
                raise RuntimeError("Translation disconnected before session.closed")
            event = json.loads(data)
            if playback.done():
                playback.result()  # Report a playback failure on the receiver.

            if event["type"] == "error":
                raise RuntimeError(f"Translation error: {event['error']['code']}")

            if event["type"] == "session.output_audio.delta":
                audio_queue.put(base64.b64decode(event["delta"]))

            if event["type"] in (
                "session.output_transcript.delta",
                "session.input_transcript.delta",
            ):
                print(event["delta"], end="", flush=True)

            if event["type"] == "session.closed":
                break
    finally:
        # The server deadline covers receiving events, not local playback.
        ws.close()
        audio_queue.put(None)
        playback.result()

1
2
3
4
5
6// Stop sending audio first. The listener closes the transport after session.closed.
if err := session.Send(ctx, realtime.RealtimeTranslationClientEventUnionParam{
	OfSessionClose: &realtime.RealtimeTranslationSessionCloseEventParam{},
}); err != nil {
	return err
}