Class: Musa::Series::Composer::Composer

Inherits:

Object

Object
Musa::Series::Composer::Composer

show all

Defined in:: lib/musa-dsl/series/series-composer.rb

Overview

Multi-input/output serie transformation pipeline system.

Composer enables building complex transformation graphs with multiple named inputs, outputs, and intermediate processing stages. It provides a declarative DSL for defining pipelines and routing data between them, similar to modular synthesis patch routing.

Architecture

The Composer system consists of:

Pipelines: Named transformation stages that process series
Routes: Connections between pipelines specifying data flow
Inputs: Named entry points (automatically proxied and buffered)
Outputs: Named exit points for consuming results
DSL Context: method_missing-based interface for pipeline definition

Pipeline Definition

Pipelines are defined using method calls in the DSL block. Each pipeline consists of:

Constructor (optional): Series constructor like S, H, etc.
Operations (one or more): Transformations like reverse, skip, etc.

Routing System

Routes connect pipelines using the route method:

route from_pipeline, to: to_pipeline, on: :source, as: :key

Routing modes:

Hash assignment (when as: provided):
- Data is assigned to: to_pipeline.input[on][as] = from_pipeline.output
- Used by operations accepting hash inputs (e.g., H constructor)
- Default on is :sources when as is provided
Setter method (when as: omitted):
- Calls: to_pipeline.input.source = from_pipeline.output
- Used by operations with explicit source setters
- Default on is :source when as is omitted

Commit System

Composer uses two-phase initialization:

Definition phase: Routes and pipelines are declared
Commit phase: All connections are resolved and finalized

Key behaviors:

Routes can be defined in any order (order-independent)
Output access is blocked until commit! is called
Commit resolves all routes and sets up buffering
auto_commit: true (default) commits automatically after DSL block

DSL Context

The DSL uses method_missing to enable natural pipeline definition:

Named access: input, output, pipeline names return symbols
Pipeline creation: name arg1, arg2, ... creates pipeline
Operation symbols: Series operations return as symbols for parsing
Constructor symbols: Series constructors return as symbols for parsing

The per-method examples further down are written against that first composer: a single reverse step from input to output.

Examples:

Pipeline with constructor

composer = Composer.new(inputs: nil) do
  my_pipeline ({ S: [1, 2, 3] }), reverse, { skip: 1 }
  route my_pipeline, to: output
end
composer.output.i.to_a  # => [2, 1]
# Creates: S([1,2,3]) → reverse → skip(1)

Pipeline with operations only

composer = Composer.new(input: S(1, 2, 3)) do
  my_pipeline reverse, { skip: 1 }
  route input, to: my_pipeline
  route my_pipeline, to: output
end
composer.output.i.to_a  # => [2, 1]
# Applies: input → reverse → skip(1)

Hash routing

composer = Composer.new(inputs: [:a, :b], auto_commit: false) do
  step1 reverse
  step2 reverse
  hash_merge ({ H: {} })

  route a, to: step1
  route b, to: step2
  route step1, to: hash_merge, as: :x    # hash_merge.input[:sources][:x] = step1.output
  route step2, to: hash_merge, as: :y    # hash_merge.input[:sources][:y] = step2.output
  route hash_merge, to: output
end

composer.input(:a).proxy_source = S(1, 2, 3)
composer.input(:b).proxy_source = S(10, 20, 30)
composer.commit!
composer.output.i.to_a  # => [{x: 3, y: 30}, {x: 2, y: 20}, {x: 1, y: 10}]

Setter routing

composer = Composer.new(input: S(1, 2, 3)) do
  step1 reverse
  route input, to: step1            # step1.input.source = input.output
  route step1, to: output
end

composer.output.i.to_a  # => [3, 2, 1]

Basic pipeline

composer = Composer.new(input: S(1, 2, 3)) do
  step1 reverse
  route input, to: step1
  route step1, to: output
end
composer.output.i.to_a  # => [3, 2, 1]

Multiple inputs merging

composer = Composer.new(inputs: { a: S(1, 2), b: S(10, 20) }) do
  hash_merge ({ H: {} })
  route a, to: hash_merge, as: :x
  route b, to: hash_merge, as: :y
  route hash_merge, to: output
end
composer.output.i.to_a  # => [{x: 1, y: 10}, {x: 2, y: 20}]

Multiple outputs

composer = Composer.new(input: S(1, 2, 3)) do
  doubled ({ eval: ->(v) { v * 2 } })
  tripled ({ eval: ->(v) { v * 3 } })

  route input, to: doubled
  route input, to: tripled
  route doubled, to: output
end
composer.output.i.to_a  # => [2, 4, 6]

Complex routing

composer = Composer.new(inputs: [:a, :b], auto_commit: false) do
  step1 reverse
  step2 ({ skip: 1 })
  hash_merge ({ H: {} })

  route a, to: step1
  route b, to: step2
  route step1, to: hash_merge, as: :x
  route step2, to: hash_merge, as: :y
  route hash_merge, to: output
end

composer.input(:a).proxy_source = S(1, 2, 3)
composer.input(:b).proxy_source = S(10, 20, 30)
composer.commit!

composer.output.i.to_a  # => [{x: 3, y: 20}, {x: 2, y: 30}]

Class Method Summary collapse

.[]=(name, pipeline) ⇒ Object

Instance Method Summary collapse

#commit! ⇒ void
Finalizes composer by resolving all routes and connections.
#initialize(input: nil, inputs: [:input], outputs: [:output], auto_commit: nil, &block) ⇒ Composer constructor
A new instance of Composer.
#input(name = nil) ⇒ ProxySerie
Accesses named input proxy for dynamic source assignment.
#output(name = nil) ⇒ Serie
Accesses named output series for consumption.
#pipeline(name, *elements) ⇒ void
Defines named pipeline with transformation operations.
#route(from, to:, on: nil, as: nil) ⇒ void
Defines routing connection between pipelines.
#update { ... } ⇒ void
Updates composer with additional DSL block.

Constructor Details

#initialize(input: nil, inputs: [:input], outputs: [:output], auto_commit: nil, &block) ⇒ `Composer`

Returns a new instance of Composer.

# File 'lib/musa-dsl/series/series-composer.rb', line 319

def initialize(input: nil, inputs: [:input], outputs: [:output], auto_commit: nil, &block)
  auto_commit = true if auto_commit.nil?

  inputs = case inputs
           when Array
             inputs.collect { |_| [_, nil] }.to_h
           when nil
             {}
           when Hash
             inputs
           else
             raise ArgumentError, "inputs: expected a Hash with input names and source series { name: serie, ... } or an Array with names [name, ...] but received #{inputs}"
           end

  inputs[:input] = input if input

  @pipelines = {}

  def @pipelines.[]=(name, pipeline)
    pipeline_to_add = @commited ? pipeline.commit! : pipeline
    super(name, pipeline_to_add)
  end

  @dsl = DSLContext.new(@pipelines)
  @inputs = {}
  @outputs = {}

  inputs.keys&.each do |input|
    p = Musa::Series::Constructors.PROXY(inputs[input])

    @inputs[input] = @pipelines[input] = Pipeline.new(input, input: p, output: p.buffered, pipelines: @pipelines)

    @dsl.define_singleton_method(input) { input }
  end

  outputs&.each do |output|
    p = Musa::Series::Constructors.PROXY()
    @outputs[output] = @pipelines[output] = Pipeline.new(output, is_output: true, input: p, output: p, pipelines: @pipelines)

    @dsl.define_singleton_method(output) { output }
  end

  @dsl.with &block if block
  commit! if auto_commit
end

Class Method Details

.[]=(name, pipeline) ⇒ `Object`

# File 'lib/musa-dsl/series/series-composer.rb', line 337

def @pipelines.[]=(name, pipeline)
  pipeline_to_add = @commited ? pipeline.commit! : pipeline
  super(name, pipeline_to_add)
end

Instance Method Details

#commit! ⇒ `void`

This method returns an undefined value.

Finalizes composer by resolving all routes and connections.

Commits the composer, resolving all route connections and setting up buffering. Must be called before accessing outputs. Cannot be called twice on same composer instance.

The commit process:

Recursively commits all output pipelines
Each pipeline commits its input routes
Connects all sources through buffers
Sets committed flag enabling output access

Examples:

Manual commit

deferred = Composer.new(auto_commit: false) do
  step1 reverse
  route input, to: step1
  route step1, to: output
end
deferred.input.proxy_source = S(1, 2, 3)
deferred.commit!
deferred.output.i.to_a  # => [3, 2, 1]

Raises:

(RuntimeError) —
if already committed

# File 'lib/musa-dsl/series/series-composer.rb', line 587

def commit!
  raise 'Already commited' if @commited

  @outputs.each_value do |pipeline|
    pipeline.commit!
  end

  @commited = true
end

#input(name = nil) ⇒ `ProxySerie`

Accesses named input proxy for dynamic source assignment.

Returns the proxy series for the specified input, allowing dynamic assignment of source series after composer creation. Used with auto_commit: false to set sources before manual commit.

Examples:

Set input source dynamically

composer = Composer.new(auto_commit: false) do
  step reverse
  route input, to: step
  route step, to: output
end

composer.input.proxy_source = S(1, 2, 3)
composer.commit!

Parameters:

name (Symbol, nil) (defaults to: nil) —
input name (defaults to :input)

Returns:

(ProxySerie) —
proxy series for source assignment

# File 'lib/musa-dsl/series/series-composer.rb', line 386

def input(name = nil)
  name ||= :input
  @inputs[name].input
end

#output(name = nil) ⇒ `Serie`

Accesses named output series for consumption.

Returns the output series for the specified output. Can only be called after commit! has been invoked. Raises error if composer is not committed.

Examples:

Access output

# Only after commit!; with the default auto_commit that has already
# happened by the time the constructor returns.
Composer.new(input: S(1, 2, 3)) do
  step1 reverse
  route input, to: step1
  route step1, to: output
end.output.i.to_a  # => [3, 2, 1]

Multiple outputs

two_ways = Composer.new(input: S(1, 2, 3), outputs: [:doubled, :tripled]) do
  twice({ map: ->(x) { x * 2 } })
  thrice({ map: ->(x) { x * 3 } })

  route input, to: twice
  route input, to: thrice
  route twice, to: doubled
  route thrice, to: tripled
end
two_ways.output(:doubled).i.to_a  # => [2, 4, 6]
two_ways.output(:tripled).i.to_a  # => [3, 6, 9]

Parameters:

name (Symbol, nil) (defaults to: nil) —
output name (defaults to :output)

Returns:

(Serie) —
output series

Raises:

(RuntimeError) —
if composer not yet committed

# File 'lib/musa-dsl/series/series-composer.rb', line 426

def output(name = nil)
  raise "Can't access output if the Composer is uncommited. Call '.commit' first." unless @commited

  name ||= :output
  @outputs[name].output
end

#pipeline(name, *elements) ⇒ `void`

This method returns an undefined value.

Defines named pipeline with transformation operations.

Creates a pipeline from constructor and/or operations. This method is typically called implicitly through the DSL's method_missing, but can be called directly for dynamic pipeline creation.

Examples:

Direct call (inside DSL block)

composer = Composer.new(inputs: nil) do
  pipeline(:my_step, [{ S: [1, 2, 3] }, :reverse])
  route my_step, to: output
end

composer.output.i.to_a  # => [3, 2, 1]

DSL equivalent (method_missing)

composer = Composer.new(inputs: nil) do
  my_step ({ S: [1, 2, 3] }), reverse
  route my_step, to: output
end

composer.output.i.to_a  # => [3, 2, 1]

Parameters:

name (Symbol) —
pipeline name
elements (Array) —
constructor and operations



530
531
532

# File 'lib/musa-dsl/series/series-composer.rb', line 530

def pipeline(name, *elements)
  @dsl.pipeline(name, elements)
end

#route(from, to:, on: nil, as: nil) ⇒ `void`

This method returns an undefined value.

Defines routing connection between pipelines.

Creates data flow route from source pipeline to destination pipeline. The routing behavior depends on whether as: parameter is provided.

With as: parameter (Hash routing):

Assigns to hash: to_pipeline.input[on][as] = from_pipeline.output
Default on is :sources
Used by operations accepting hash inputs (e.g., H constructor)

Without as: parameter (Setter routing):

Calls setter: to_pipeline.input.source = from_pipeline.output
Default on is :source
Used by operations with explicit source setter methods

Examples:

Hash routing (inside DSL block)

composer = Composer.new(inputs: [:a, :b], auto_commit: false) do
  step1 reverse
  step2 reverse
  hash_merge ({ H: {} })

  route a, to: step1
  route b, to: step2
  route step1, to: hash_merge, as: :x    # hash_merge.input[:sources][:x] = step1
  route step2, to: hash_merge, as: :y    # hash_merge.input[:sources][:y] = step2
  route hash_merge, to: output
end

composer.input(:a).proxy_source = S(1, 2)
composer.input(:b).proxy_source = S(10, 20)
composer.commit!
composer.output.i.to_a  # => [{x: 2, y: 20}, {x: 1, y: 10}]

Setter routing (inside DSL block)

composer = Composer.new(input: S(1, 2, 3)) do
  step reverse
  route input, to: step             # step.input.source = input
  route step, to: output
end

composer.output.i.to_a  # => [3, 2, 1]

Custom on parameter (inside DSL block)

composer = Composer.new(input: S(1, 2, 3), auto_commit: false) do
  step reverse
  hash_merge ({ H: {} })
  route input, to: step
  route step, to: hash_merge, on: :sources, as: :x
  route hash_merge, to: output
end

composer.commit!
composer.output.i.to_a  # => [{x: 3}, {x: 2}, {x: 1}]

Parameters:

from (Symbol) —
source pipeline name
to (Symbol) —
destination pipeline name
on (Symbol, nil) (defaults to: nil) —
attribute name (:source or :sources)
as (Symbol, nil) (defaults to: nil) —
hash key for assignment

Raises:

(ArgumentError) —
if pipeline names not found
(ArgumentError) —
if route already exists



498
499
500

# File 'lib/musa-dsl/series/series-composer.rb', line 498

def route(from, to:, on: nil, as: nil)
  @dsl.route(from, to: to, on: on, as: as)
end

#update { ... } ⇒ `void`

This method returns an undefined value.

Updates composer with additional DSL block.

Allows dynamic modification of composer after creation by executing additional DSL block in the composer's DSL context. Useful for progressive pipeline construction.

Examples:

Add routes dynamically

growing = Composer.new(input: S(1, 2, 3), auto_commit: false) do
  step1 reverse
  route input, to: step1
end
growing.update do
  route step1, to: output
end
growing.commit!
growing.output.i.to_a  # => [3, 2, 1]

Yields:

DSL block with additional pipeline definitions



556
557
558

# File 'lib/musa-dsl/series/series-composer.rb', line 556

def update(&block)
  @dsl.with &block
end

Class: Musa::Series::Composer::Composer

Overview

Architecture

Pipeline Definition

Routing System

Commit System

DSL Context

Examples:

Pipeline with constructor

Pipeline with operations only

Hash routing

Setter routing

Basic pipeline

Multiple inputs merging

Multiple outputs

Complex routing

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(input: nil, inputs: [:input], outputs: [:output], auto_commit: nil, &block) ⇒ Composer

Class Method Details

.[]=(name, pipeline) ⇒ Object

Instance Method Details

#commit! ⇒ void

Examples:

Manual commit

#input(name = nil) ⇒ ProxySerie

Examples:

Set input source dynamically

#output(name = nil) ⇒ Serie

Examples:

Access output

Multiple outputs

#pipeline(name, *elements) ⇒ void

Examples:

Direct call (inside DSL block)

DSL equivalent (method_missing)

#route(from, to:, on: nil, as: nil) ⇒ void

Examples:

Hash routing (inside DSL block)

Setter routing (inside DSL block)

Custom on parameter (inside DSL block)

#update { ... } ⇒ void

Examples:

Add routes dynamically

#initialize(input: nil, inputs: [:input], outputs: [:output], auto_commit: nil, &block) ⇒ `Composer`

.[]=(name, pipeline) ⇒ `Object`

#commit! ⇒ `void`

#input(name = nil) ⇒ `ProxySerie`

#output(name = nil) ⇒ `Serie`

#pipeline(name, *elements) ⇒ `void`

#route(from, to:, on: nil, as: nil) ⇒ `void`

#update { ... } ⇒ `void`