Class: Ultrasafeai::Audio::Stream::StreamClient

Inherits:
Object
  • Object
show all
Defined in:
lib/ultrasafeai/audio/stream/stream_client.rb

Overview

WebSocket client for the UltrasafeAI live ASR stream.

Usually reached via client.audio.stream, but can be constructed directly.

Examples:

session = client.audio.stream.connect(
  ConnectOptions.new(model: "usf-mini-asr", sample_rate: 16_000, partial_results: true)
)
session.on("transcript") { |event| puts event["full_text"] }
session.send(pcm_chunk)
session.close

Instance Method Summary collapse

Constructor Details

#initialize(api_key:, base_url: "wss://api.us.tech/v1") ⇒ StreamClient

Returns a new instance of StreamClient.



23
24
25
26
 
# File 'lib/ultrasafeai/audio/stream/stream_client.rb', line 23

def initialize(api_key:, base_url: "wss://api.us.tech/v1")
  @api_key = api_key
  @base_url = base_url.sub(%r{/$}, "")
end

Instance Method Details

#connect(opts = nil) ⇒ AudioStreamSession

Open a live ASR WebSocket session.

Parameters:

  • opts (ConnectOptions, nil) (defaults to: nil) —

    connection options

Returns:



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
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/ultrasafeai/audio/stream/stream_client.rb', line 32

def connect(opts = nil)
  opts ||= ConnectOptions.new
  url = build_url(opts)
  max_retries = opts.max_retries || 3
  backoff_ms  = opts.backoff_ms  || 500.0

  last_exc = nil
  (max_retries + 1).times do |attempt|
    if attempt > 0
      delay = backoff_ms * (2**(attempt - 1)) / 1000.0
      sleep(delay)
    end

    begin
      session = nil
      latch = Mutex.new
      cond = ConditionVariable.new

      ws = WebSocket::Client::Simple.connect(url)

      ws.on(:open) do
        latch.synchronize { session = AudioStreamSession.new(ws); cond.signal }
      end

      ws.on(:message) do |msg|
        next unless session
        if msg.type == :text
          begin
            event = JSON.parse(msg.data)
            type  = event[AudioStreamSession::DISCRIMINANT]
            session.emit(type, event) if type
          rescue JSON::ParserError => e
            session.emit("parse_error", e, msg.data)
          end
        end
      end

      ws.on(:close) do |e|
        session&.emit("close", e.code, e.reason.to_s)
      end

      ws.on(:error) do |e|
        session&.emit("ws_error", e)
      end

      latch.synchronize { cond.wait(latch, 10) }
      return session if session
      raise "WebSocket open timed out"
    rescue => e
      last_exc = e
    end
  end

  raise last_exc || RuntimeError.new("ASR WebSocket connection failed")
end