Class: Wavify::Core::Stream

Inherits:
Object
  • Object
show all
Includes:
Enumerable, Enumerable[SampleBuffer]
Defined in:
lib/wavify/core/stream.rb,
sig/stream.rbs

Overview

Lazy streaming pipeline for chunk-based audio processing.

Instances are typically created via Audio.stream.

Defined Under Namespace

Classes: MeterProcessor, ProgressProcessor, UserCodeError

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(source, codec:, format:, chunk_size: 4096, codec_read_options: {}) ⇒ Stream

Returns a new instance of Stream.

Parameters:

  • source (String, IO) —

    input path or IO

  • codec (Class) —

    codec class implementing stream_read/stream_write

  • format (Format, nil) —

    source format (may be inferred later)

  • chunk_size (Integer) (defaults to: 4096) —

    chunk size in frames

  • codec_read_options (Hash) (defaults to: {}) —

    codec-specific options forwarded to stream_read

  • codec: (Object)
  • format: (Format, nil)
  • chunk_size: (Integer) (defaults to: 4096)
  • codec_read_options: (Hash[Symbol, untyped]) (defaults to: {}) 


20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/wavify/core/stream.rb', line 20

def initialize(source, codec:, format:, chunk_size: 4096, codec_read_options: {})
  @source = source
  @codec = codec
  @format = format
  @chunk_size = validate_chunk_size!(chunk_size)
  @codec_read_options = validate_codec_read_options!(codec_read_options)
  @pipeline = []
  @pipeline_names = []
  @tee_targets = []
  @take_duration_seconds = nil
  @drop_duration_seconds = nil
  @enumerated = false
  @source_start_position = source_position(source)
  @stage_input_formats = []
end

Instance Attribute Details

#chunk_size ⇒ Integer (readonly)

Returns the value of attribute chunk_size.

Returns:

  • (Integer) 


13
14
15
 
# File 'lib/wavify/core/stream.rb', line 13

def chunk_size
  @chunk_size
end

Instance Method Details

#drop_duration(duration) ⇒ Stream

Drops an initial duration from the source stream.

Parameters:

  • duration (Numeric, Duration) —

    seconds or duration object

Returns:



96
97
98
99
 
# File 'lib/wavify/core/stream.rb', line 96

def drop_duration(duration)
  @drop_duration_seconds = coerce_duration_seconds!(duration, "duration")
  self
end

#dry_run(format: nil) ⇒ Hash

Reads and processes the stream without writing output. User processors and callbacks are still executed and may have side effects.

Parameters:

  • format (Format, nil) (defaults to: nil) —

    optional output conversion to validate

  • format: (Format, nil) (defaults to: nil)

Returns:

  • (Hash) 


197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/wavify/core/stream.rb', line 197

def dry_run(format: nil)
  raise InvalidFormatError, "format must be Core::Format" if format && !format.is_a?(Format)

  tee_targets = @tee_targets
  @tee_targets = []
  stats = {
    chunks: 0,
    sample_frame_count: 0,
    format: format,
    pipeline: pipeline_steps,
    latency: latency,
    lookahead: lookahead,
    tail_duration: pipeline_tail_duration
  }

  each_chunk do |chunk|
    output_chunk = format ? chunk.convert(format) : chunk
    stats[:chunks] += 1
    stats[:sample_frame_count] += output_chunk.sample_frame_count
    stats[:format] ||= output_chunk.format
  end
  stats[:duration] = stats[:format] ? Duration.from_samples(stats[:sample_frame_count], stats[:format].sample_rate) : nil
  stats
ensure
  @tee_targets = tee_targets if defined?(tee_targets)
end

#each_chunk ⇒ Stream #each_chunk ⇒ Enumerator[SampleBuffer, Stream] Also known as: each

Iterates processed chunks.

Overloads:

Yields:

  • (chunk)

Yield Parameters:

Returns:

  • (Enumerator) 


229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
 
# File 'lib/wavify/core/stream.rb', line 229

def each_chunk
  return enum_for(:each_chunk) unless block_given?

  prepare_source_for_enumeration!
  reset_pipeline!
  @last_output_format = nil
  drop_frames = nil
  take_frames = nil
  input_taken_frames = 0
  output_drop_frames = nil
  output_take_frames = nil
  output_taken_frames = 0
  @stage_input_formats = []

  with_tee_writers do |tee_writers|
    with_stream_context("stream read", codec: @codec, target: @source) do
      take_complete = Object.new
      catch(take_complete) do
        @codec.stream_read(@source, chunk_size: @chunk_size, **@codec_read_options) do |chunk|
          with_stream_context("stream processing", codec: @codec, target: @source) do
            @format ||= chunk.format
            drop_frames ||= duration_frames(@drop_duration_seconds, chunk.format) || 0
            take_frames = duration_frames(@take_duration_seconds, chunk.format) if take_frames.nil? && @take_duration_seconds
            input_chunk, drop_frames, input_taken_frames = apply_duration_window(
              chunk,
              drop_frames: drop_frames,
              take_frames: take_frames,
              taken_frames: input_taken_frames
            )
            throw(take_complete) if take_frames && input_taken_frames >= take_frames && input_chunk.nil?
            next unless input_chunk

            output_chunk = apply_pipeline(input_chunk)
            output_drop_frames, output_take_frames, output_taken_frames = emit_output_chunk(
              output_chunk,
              tee_writers: tee_writers,
              drop_frames: output_drop_frames,
              take_frames: output_take_frames,
              taken_frames: output_taken_frames
            ) { |windowed| yield windowed }
            throw(take_complete) if take_frames && input_taken_frames >= take_frames
          end
        end
      end
    end

    with_stream_context("stream flush", codec: @codec, target: @source) do
      flush_pipeline do |chunk|
        output_drop_frames, output_take_frames, output_taken_frames = emit_output_chunk(
          chunk,
          tee_writers: tee_writers,
          drop_frames: output_drop_frames,
          take_frames: output_take_frames,
          taken_frames: output_taken_frames
        ) { |windowed| yield windowed }
      end
    end
  end
rescue UserCodeError => e
  raise e.original
end

#format ⇒ Format

Returns a known source format. Access before enumeration may perform a metadata probe; normal decoding learns the format from the first chunk.

Returns:



38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/wavify/core/stream.rb', line 38

def format
  return @format if @format

  position = source_position(@source)
  metadata = @codec.metadata(@source, **metadata_probe_options)
  @format = metadata.fetch(:format)
ensure
  if position && @source.respond_to?(:seek)
    @source.seek(position, IO::SEEK_SET)
  end
end

#latency ⇒ Float

Returns summed processor latency in seconds.

Returns:

  • (Float) —

    summed processor latency in seconds



183
184
185
 
# File 'lib/wavify/core/stream.rb', line 183

def latency
  @pipeline.sum { |processor| processor_duration(processor, :latency) }
end

#lookahead ⇒ Float

Returns summed processor lookahead in seconds.

Returns:

  • (Float) —

    summed processor lookahead in seconds



188
189
190
 
# File 'lib/wavify/core/stream.rb', line 188

def lookahead
  @pipeline.sum { |processor| processor_duration(processor, :lookahead) }
end

#map_chunks(name: nil) {|arg0| ... } ⇒ Stream

Adds a block processor for chunk-to-chunk transforms.

Parameters:

  • name (String, Symbol, nil) (defaults to: nil) —

    optional display name for inspection

  • name: (String, Symbol, nil) (defaults to: nil)

Yields:

Yield Parameters:

Yield Returns:

  • (Object)

Returns:

Raises:



77
78
79
80
81
 
# File 'lib/wavify/core/stream.rb', line 77

def map_chunks(name: nil, &block)
  raise InvalidParameterError, "map_chunks requires a block" unless block

  pipe(block, name: name || :map_chunks)
end

#meter {|stats| ... } ⇒ Stream

Installs a passive per-chunk level meter.

Yields:

  • (stats)

Yield Parameters:

  • stats (Hash)

Returns:

Raises:



106
107
108
109
110
 
# File 'lib/wavify/core/stream.rb', line 106

def meter(&block)
  raise InvalidParameterError, "meter requires a block" unless block

  pipe(MeterProcessor.new(block), name: :meter)
end

#pipe(processor = nil, name: nil, &block) ⇒ Stream

Adds a processor to the stream pipeline.

Processors may respond to #process, #call, or #apply. Stateful processors may also expose #reset and #flush.

Parameters:

  • processor (#call, #process, #apply, nil) (defaults to: nil)
  • name (String, Symbol, nil) (defaults to: nil) —

    optional display name for inspection

  • name: (String, Symbol, nil) (defaults to: nil)

Returns:



58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/wavify/core/stream.rb', line 58

def pipe(processor = nil, name: nil, &block)
  if processor && block
    raise InvalidParameterError, "pipe accepts either a processor or a block, not both"
  end

  candidate = processor || block
  unless candidate.respond_to?(:call) || candidate.respond_to?(:process) || candidate.respond_to?(:apply)
    raise InvalidParameterError, "processor must respond to :call, :process, or :apply"
  end

  @pipeline << candidate
  @pipeline_names << validate_pipeline_name!(name)
  self
end

#pipeline ⇒ Array<Object>

Returns registered processors in execution order.

Returns:

  • (Array<Object>) —

    registered processors in execution order



165
166
167
 
# File 'lib/wavify/core/stream.rb', line 165

def pipeline
  @pipeline.dup
end

#pipeline_steps ⇒ Array<Hash>

Returns processor names and objects in execution order.

Returns:

  • (Array<Hash>) —

    processor names and objects in execution order



170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/wavify/core/stream.rb', line 170

def pipeline_steps
  @pipeline.map.with_index do |processor, index|
    {
      name: @pipeline_names.fetch(index),
      processor: processor,
      latency: processor_duration(processor, :latency),
      lookahead: processor_duration(processor, :lookahead),
      tail_duration: processor_duration(processor, :tail_duration)
    }
  end
end

#progress(total_frames: nil) {|stats| ... } ⇒ Stream

Installs a passive progress callback based on processed output chunks.

Parameters:

  • total_frames (Integer, nil) (defaults to: nil) —

    optional expected output frame count

  • total_frames: (Integer, nil) (defaults to: nil)

Yields:

  • (stats)

Yield Parameters:

  • stats (Hash)

Returns:

Raises:



118
119
120
121
122
123
124
125
 
# File 'lib/wavify/core/stream.rb', line 118

def progress(total_frames: nil, &block)
  raise InvalidParameterError, "progress requires a block" unless block
  if total_frames && (!total_frames.is_a?(Integer) || total_frames.negative?)
    raise InvalidParameterError, "total_frames must be a non-negative Integer"
  end

  pipe(ProgressProcessor.new(block, total_frames: total_frames), name: :progress)
end

#take_duration(duration) ⇒ Stream

Limits the source stream to the first duration after any dropped prefix.

Parameters:

  • duration (Numeric, Duration) —

    seconds or duration object

Returns:



87
88
89
90
 
# File 'lib/wavify/core/stream.rb', line 87

def take_duration(duration)
  @take_duration_seconds = coerce_duration_seconds!(duration, "duration")
  self
end

#tee(path_or_io, format: nil, codec: nil, filename: nil, codec_options: nil) ⇒ Stream

Writes processed chunks to an additional output while the stream is consumed.

Parameters:

  • path_or_io (String, IO)
  • format (Format, nil) (defaults to: nil) —

    optional output format

  • codec_options (Hash, nil) (defaults to: nil) —

    codec-specific options forwarded to stream_write

  • format: (Format, nil) (defaults to: nil)
  • codec: (Symbol, untyped) (defaults to: nil)
  • filename: (String, nil) (defaults to: nil)
  • codec_options: (Hash[Symbol, untyped], nil) (defaults to: nil)

Returns:

Raises:



133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/wavify/core/stream.rb', line 133

def tee(path_or_io, format: nil, codec: nil, filename: nil, codec_options: nil)
  output_codec = detect_output_codec(path_or_io, codec: codec, filename: filename)
  target_format = resolve_target_format(format, output_codec)
  raise InvalidFormatError, "format is required when teeing stream output" unless target_format.is_a?(Format)

  @tee_targets << {
    target: path_or_io,
    codec: output_codec,
    format: target_format,
    codec_options: validate_codec_options!(codec_options, "codec_options")
  }
  self
end

#to_audio ⇒ Audio

Materializes the processed stream into an Audio object.

Returns:

Raises:



150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/wavify/core/stream.rb', line 150

def to_audio
  output_format = nil
  samples = []
  each_chunk do |chunk|
    output_format ||= chunk.format
    converted = chunk.format == output_format ? chunk : chunk.convert(output_format)
    samples.concat(converted.samples)
  end
  output_format ||= @last_output_format || @format
  raise InvalidFormatError, "stream format is unknown" unless output_format.is_a?(Format)

  Audio.new(SampleBuffer.new(samples, output_format))
end

#write_to(path_or_io, format: nil, codec: nil, filename: nil, codec_options: nil, overwrite: true) ⇒ String, IO

Writes the processed stream to a path or writable IO.

Parameters:

  • path_or_io (String, IO)
  • format (Format, nil) (defaults to: nil) —

    output format (required for raw output if unknown)

  • codec (Symbol, Class, nil) (defaults to: nil) —

    explicit output codec for generic IO targets

  • filename (String, nil) (defaults to: nil) —

    filename hint used for codec selection

  • codec_options (Hash) (defaults to: nil) —

    codec-specific options forwarded to stream_write

  • overwrite (Boolean) (defaults to: true) —

    whether existing path output may be replaced

  • format: (Format, nil) (defaults to: nil)
  • codec: (Symbol, untyped) (defaults to: nil)
  • filename: (String, nil) (defaults to: nil)
  • codec_options: (Hash[Symbol, untyped], nil) (defaults to: nil)
  • overwrite: (Boolean) (defaults to: true)

Returns:

  • (String, IO) —

    the same target argument



302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
 
# File 'lib/wavify/core/stream.rb', line 302

def write_to(path_or_io, format: nil, codec: nil, filename: nil, codec_options: nil, overwrite: true)
  validate_overwrite!(path_or_io, overwrite)
  output_codec = detect_output_codec(path_or_io, codec: codec, filename: filename)
  target_format = resolve_target_format(format, output_codec)
  options = validate_codec_options!(codec_options, "codec_options")

  with_output_target(path_or_io, overwrite: overwrite) do |target|
    with_stream_context("stream write", codec: output_codec, target: path_or_io) do
      output_codec.stream_write(target, format: target_format, **options) do |writer|
        each_chunk do |chunk|
          output_chunk = target_format ? chunk.convert(target_format) : chunk
          with_stream_context("stream write", codec: output_codec, target: path_or_io) do
            writer.call(output_chunk)
          end
        end
      end
    end
  end

  path_or_io
end