Class: Cohere::Transcribe::Runtime::WordPipeline::Coordinator

Inherits:
Object
  • Object
show all
Defined in:
lib/cohere/transcribe/runtime/word_pipeline.rb

Overview

Executes an ordered two-phase batch. The ASR callback returns either AlignmentWork or Final. Alignment errors intended to be per-file must likewise be converted to a final result by the alignment callback; fatal/cancellation exceptions intentionally cross this boundary.

Instance Method Summary collapse

Constructor Details

#initialize(total:, evict_asr:, align:, fixed_results: {}, completed: nil, gc_interval: 8, reload: nil, memory_byte_limit: nil, start_alignment: nil) ⇒ Coordinator

Returns a new instance of Coordinator.

Raises:

  • (ArgumentError) 


97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/cohere/transcribe/runtime/word_pipeline.rb', line 97

def initialize(total:, evict_asr:, align:, fixed_results: {}, completed: nil, gc_interval: 8,
               reload: nil, memory_byte_limit: nil, start_alignment: nil)
  @total = Integer(total)
  raise ArgumentError, "word-pipeline total must be nonnegative" if @total.negative?
  raise ArgumentError, "evict_asr must respond to call" unless evict_asr.respond_to?(:call)
  raise ArgumentError, "align must respond to call" unless align.respond_to?(:call)
  raise ArgumentError, "completed must respond to call" if completed && !completed.respond_to?(:call)
  raise ArgumentError, "start_alignment must respond to call" if
    start_alignment && !start_alignment.respond_to?(:call)

  @fixed_results = fixed_results.to_h.freeze
  @evict_asr = evict_asr
  @align = align
  @completed = completed
  @gc_interval = Integer(gc_interval)
  raise ArgumentError, "gc_interval must be positive" unless @gc_interval.positive?
  raise ArgumentError, "reload and memory_byte_limit must be provided together" unless
    reload.nil? == memory_byte_limit.nil?
  raise ArgumentError, "reload must respond to call" if reload && !reload.respond_to?(:call)

  @reload = reload
  @start_alignment = start_alignment
  @memory_byte_limit = Integer(memory_byte_limit) if memory_byte_limit
  raise ArgumentError, "memory_byte_limit must be positive" if
    @memory_byte_limit && !@memory_byte_limit.positive?
end

Instance Method Details

#run(prepared, &asr) ⇒ Object

Raises:

  • (ArgumentError) 


124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/cohere/transcribe/runtime/word_pipeline.rb', line 124

def run(prepared, &asr)
  raise ArgumentError, "an ASR callback is required" unless asr

  slots = Array.new(@total, UNSET)
  @fixed_results.each do |index, result|
    outcome = if result.is_a?(AlignmentWork) || result.is_a?(Final)
                result
              else
                Final.new(index: index, result: result)
              end
    install!(slots, outcome)
  end
  transitioned = false
  begin
    collect_asr!(slots, prepared, &asr)
    missing = slots.each_index.select { |index| slots.fetch(index).equal?(UNSET) }
    unless missing.empty?
      raise TranscriptionRuntimeError,
            "Word ASR phase omitted input index(es): #{missing.join(", ")}"
    end

    # Set the flag before calling so an eviction exception is not
    # followed by a second, potentially destructive close attempt.
    transitioned = true
    alignment_required = slots.grep(AlignmentWork).any?(&:audio_required)
    @evict_asr.call if alignment_required

    align_slots!(slots)
    completed_count = 0
    slots.each_with_index do |outcome, index|
      result = outcome.is_a?(Final) ? outcome.result : outcome
      raise TranscriptionRuntimeError, "Word pipeline returned no result for input #{index}" if result.nil?

      slots[index] = result
      completed_count += 1
      @completed&.call(index, completed_count, @total, result)
      GC.start(full_mark: false, immediate_mark: true, immediate_sweep: true) if
        (completed_count % @gc_interval).zero?
    end
    slots.freeze
  ensure
    @evict_asr.call unless transitioned
  end
end