run audio extraction

cmd/mxl-pattern/main.go

@@ -551,7 +551,7 @@ func run(ctx context.Context, args appArgs) (runErr error) {
 		runners = append(runners, namedRunner{
 			name: "audio",
 			run: func(ctx context.Context) error {
-				return runAudio(ctx, inst, *audioCfg)
+				return audio.Run(ctx, inst, *audioCfg)
 			},
 		})
 	}
@@ -664,157 +664,6 @@ func runVideo(ctx context.Context, inst *mxl.Instance, cfg video.Config) (runErr
 	}
 }
 
-func runAudio(
-	ctx context.Context,
-	inst *mxl.Instance,
-	cfg audio.Config,
-) (runErr error) {
-	flowJSON, err := json.Marshal(cfg.Definition)
-	if err != nil {
-		return fmt.Errorf("marshal audio flow definition: %w", err)
-	}
-
-	writer, isCreated, err := inst.NewWriter(string(flowJSON))
-	if err != nil {
-		return fmt.Errorf("create audio writer: %w", err)
-	}
-	defer func() {
-		if err := writer.Close(); err != nil {
-			runErr = errors.Join(
-				runErr,
-				fmt.Errorf("close audio writer: %w", err),
-			)
-		}
-	}()
-	if !isCreated {
-		log.Printf("reusing existing audio flow: %s", cfg.ID())
-	}
-
-	writerCfg := writer.Config()
-	if writerCfg.Common.Format != mxl.FormatAudio {
-		return fmt.Errorf(
-			"audio writer has format %s, want audio",
-			writerCfg.Common.Format,
-		)
-	}
-	if writerCfg.Continuous.ChannelCount != uint32(cfg.Channels()) {
-		return fmt.Errorf(
-			"audio writer has %d channels, configured generator expects %d",
-			writerCfg.Continuous.ChannelCount,
-			cfg.Channels(),
-		)
-	}
-
-	rate := writerCfg.Common.GrainRate
-	if rate != cfg.Rate() {
-		return fmt.Errorf(
-			"audio writer has sample rate %d/%d, configured generator expects %d/%d",
-			rate.Num,
-			rate.Den,
-			cfg.Rate().Num,
-			cfg.Rate().Den,
-		)
-	}
-
-	const baseFrequency = 1000.0
-
-	gen, err := audio.NewSineGenerator(cfg, baseFrequency)
-	if err != nil {
-		return fmt.Errorf("initialize audio generator: %w", err)
-	}
-
-	batch := audioBatchSize(rate)
-	maxBatch, err := writer.GetMaxWriteLengthSamples()
-	if err != nil {
-		return fmt.Errorf("get maximum audio write length: %w", err)
-	}
-	if maxBatch == 0 {
-		return fmt.Errorf("audio writer reported a maximum write length of zero samples")
-	}
-	if batch > maxBatch {
-		batch = maxBatch
-	}
-	index := mxl.CurrentIndex(rate)
-	if index < batch-1 {
-		return fmt.Errorf("current audio index %d is too small for batch size %d", index, batch)
-	}
-
-	log.Printf(
-		"writing audio flow sampleRate=%d/%d channels=%d batch=%d starting at idx=%d",
-		rate.Num,
-		rate.Den,
-		cfg.Channels(),
-		batch,
-		index,
-	)
-
-	var samplesWritten uint64
-
-	for {
-		select {
-		case <-ctx.Done():
-			log.Printf("stopping audio after %d samples", samplesWritten)
-			return nil
-		default:
-		}
-
-		access, err := writer.OpenSamples(index, int(batch))
-		if err != nil {
-			return fmt.Errorf(
-				"open %d audio samples at index %d: %w",
-				batch,
-				index,
-				err,
-			)
-		}
-
-		firstSample := index - batch + 1
-		for channel := uint64(0); channel < access.ChannelCount; channel++ {
-			first, second, err := access.ChannelFragments(channel)
-			if err != nil {
-				return cancelAudioSamples(
-					access,
-					fmt.Errorf(
-						"get fragments for audio channel %d at index %d: %w",
-						channel,
-						index,
-						err,
-					),
-				)
-			}
-
-			if err := gen.Generate(
-				uint(channel),
-				firstSample,
-				first,
-				second,
-			); err != nil {
-				return cancelAudioSamples(
-					access, fmt.Errorf(
-						"generate audio channel %d at index %d: %w",
-						channel,
-						index,
-						err,
-					),
-				)
-			}
-		}
-
-		if err := access.Commit(); err != nil {
-			return fmt.Errorf(
-				"commit %d audio samples at index %d: %w",
-				batch,
-				index,
-				err,
-			)
-		}
-
-		samplesWritten += batch
-		index += batch
-		mxl.SleepNs(mxl.NsUntilIndex(index, rate))
-	}
-}
-
 func buildTextOverlay(cfg video.Config) (overlay *generator.TextOverlay, resultErr error) {
 	if cfg.Overlay.Text == "" {
 		return nil, nil
@@ -856,28 +705,3 @@ func cancelVideoGrain(grain *mxl.GrainWriteAccess, cause error) error {
 	}
 	return cause
 }
-
-func audioBatchSize(rate mxl.Rational) uint64 {
-	if rate.Num <= 0 || rate.Den <= 0 {
-		return 1
-	}
-
-	samples := rate.Num / (100 * rate.Den)
-	if samples < 1 {
-		return 1
-	}
-	return uint64(samples)
-}
-
-func cancelAudioSamples(
-	access *mxl.SamplesWriteAccess,
-	cause error,
-) error {
-	if err := access.Cancel(); err != nil {
-		return errors.Join(
-			cause,
-			fmt.Errorf("cancel audio samples: %w", err),
-		)
-	}
-	return cause
-}

cmd/mxl-pattern/main_test.go

@@ -10,8 +10,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/qvest-digital/go-mxl/mxl"
-
 	"mxl-pattern-generator/internal/audio"
 	"mxl-pattern-generator/internal/flowdef"
 )
@@ -256,30 +254,6 @@ func TestValidateAudioArgsRejectsUnknownLevelForFlowDefinition(t *testing.T) {
 	}
 }
 
-func TestAudioBatchSize(t *testing.T) {
-	tests := []struct {
-		name string
-		rate mxl.Rational
-		want uint64
-	}{
-		{name: "44.1 kHz", rate: mxl.Rational{Num: 44100, Den: 1}, want: 441},
-		{name: "48 kHz", rate: mxl.Rational{Num: 48000, Den: 1}, want: 480},
-		{name: "96 kHz", rate: mxl.Rational{Num: 96000, Den: 1}, want: 960},
-		{name: "192 kHz", rate: mxl.Rational{Num: 192000, Den: 1}, want: 1920},
-		{name: "minimum", rate: mxl.Rational{Num: 1, Den: 1}, want: 1},
-		{name: "zero numerator", rate: mxl.Rational{Num: 0, Den: 1}, want: 1},
-		{name: "zero denominator", rate: mxl.Rational{Num: 48000, Den: 0}, want: 1},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			if got := audioBatchSize(tc.rate); got != tc.want {
-				t.Fatalf("audioBatchSize(%d/%d) = %d, want %d", tc.rate.Num, tc.rate.Den, got, tc.want)
-			}
-		})
-	}
-}
-
 func TestRunConcurrentCancelsSiblingAndWaitsForCleanup(t *testing.T) {
 	wantErr := errors.New("writer failed")
 	peerStarted := make(chan struct{})

internal/audio/runner.go

@@ -0,0 +1,187 @@
+package audio
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+
+	"github.com/qvest-digital/go-mxl/mxl"
+)
+
+func Run(
+	ctx context.Context,
+	inst *mxl.Instance,
+	cfg Config,
+) (runErr error) {
+	flowJSON, err := json.Marshal(cfg.Definition)
+	if err != nil {
+		return fmt.Errorf("marshal audio flow definition: %w", err)
+	}
+
+	writer, isCreated, err := inst.NewWriter(string(flowJSON))
+	if err != nil {
+		return fmt.Errorf("create audio writer: %w", err)
+	}
+	defer func() {
+		if err := writer.Close(); err != nil {
+			runErr = errors.Join(
+				runErr,
+				fmt.Errorf("close audio writer: %w", err),
+			)
+		}
+	}()
+	if !isCreated {
+		log.Printf("reusing existing audio flow: %s", cfg.ID())
+	}
+
+	writerCfg := writer.Config()
+	if writerCfg.Common.Format != mxl.FormatAudio {
+		return fmt.Errorf(
+			"audio writer has format %s, want audio",
+			writerCfg.Common.Format,
+		)
+	}
+	if writerCfg.Continuous.ChannelCount != uint32(cfg.Channels()) {
+		return fmt.Errorf(
+			"audio writer has %d channels, configured generator expects %d",
+			writerCfg.Continuous.ChannelCount,
+			cfg.Channels(),
+		)
+	}
+
+	rate := writerCfg.Common.GrainRate
+	if rate != cfg.Rate() {
+		return fmt.Errorf(
+			"audio writer has sample rate %d/%d, configured generator expects %d/%d",
+			rate.Num,
+			rate.Den,
+			cfg.Rate().Num,
+			cfg.Rate().Den,
+		)
+	}
+
+	const baseFrequency = 1000.0
+
+	gen, err := NewSineGenerator(cfg, baseFrequency)
+	if err != nil {
+		return fmt.Errorf("initialize audio generator: %w", err)
+	}
+
+	batch := batchSize(rate)
+	maxBatch, err := writer.GetMaxWriteLengthSamples()
+	if err != nil {
+		return fmt.Errorf("get maximum audio write length: %w", err)
+	}
+	if maxBatch == 0 {
+		return fmt.Errorf("audio writer reported a maximum write length of zero samples")
+	}
+	if batch > maxBatch {
+		batch = maxBatch
+	}
+	index := mxl.CurrentIndex(rate)
+	if index < batch-1 {
+		return fmt.Errorf("current audio index %d is too small for batch size %d", index, batch)
+	}
+
+	log.Printf(
+		"writing audio flow sampleRate=%d/%d channels=%d batch=%d starting at idx=%d",
+		rate.Num,
+		rate.Den,
+		cfg.Channels(),
+		batch,
+		index,
+	)
+
+	var samplesWritten uint64
+
+	for {
+		select {
+		case <-ctx.Done():
+			log.Printf("stopping audio after %d samples", samplesWritten)
+			return nil
+		default:
+		}
+
+		access, err := writer.OpenSamples(index, int(batch))
+		if err != nil {
+			return fmt.Errorf(
+				"open %d audio samples at index %d: %w",
+				batch,
+				index,
+				err,
+			)
+		}
+
+		firstSample := index - batch + 1
+		for channel := uint64(0); channel < access.ChannelCount; channel++ {
+			first, second, err := access.ChannelFragments(channel)
+			if err != nil {
+				return cancelSamples(
+					access,
+					fmt.Errorf(
+						"get fragments for audio channel %d at index %d: %w",
+						channel,
+						index,
+						err,
+					),
+				)
+			}
+
+			if err := gen.Generate(
+				uint(channel),
+				firstSample,
+				first,
+				second,
+			); err != nil {
+				return cancelSamples(
+					access, fmt.Errorf(
+						"generate audio channel %d at index %d: %w",
+						channel,
+						index,
+						err,
+					),
+				)
+			}
+		}
+
+		if err := access.Commit(); err != nil {
+			return fmt.Errorf(
+				"commit %d audio samples at index %d: %w",
+				batch,
+				index,
+				err,
+			)
+		}
+
+		samplesWritten += batch
+		index += batch
+		mxl.SleepNs(mxl.NsUntilIndex(index, rate))
+	}
+}
+
+func cancelSamples(
+	access *mxl.SamplesWriteAccess,
+	cause error,
+) error {
+	if err := access.Cancel(); err != nil {
+		return errors.Join(
+			cause,
+			fmt.Errorf("cancel audio samples: %w", err),
+		)
+	}
+	return cause
+}
+
+func batchSize(rate mxl.Rational) uint64 {
+	if rate.Num <= 0 || rate.Den <= 0 {
+		return 1
+	}
+
+	samples := rate.Num / (100 * rate.Den)
+	if samples < 1 {
+		return 1
+	}
+	return uint64(samples)
+}

internal/audio/runner_test.go

@@ -0,0 +1,31 @@
+package audio
+
+import (
+	"testing"
+
+	"github.com/qvest-digital/go-mxl/mxl"
+)
+
+func TestBatchSize(t *testing.T) {
+	tests := []struct {
+		name string
+		rate mxl.Rational
+		want uint64
+	}{
+		{name: "44.1 kHz", rate: mxl.Rational{Num: 44100, Den: 1}, want: 441},
+		{name: "48 kHz", rate: mxl.Rational{Num: 48000, Den: 1}, want: 480},
+		{name: "96 kHz", rate: mxl.Rational{Num: 96000, Den: 1}, want: 960},
+		{name: "192 kHz", rate: mxl.Rational{Num: 192000, Den: 1}, want: 1920},
+		{name: "minimum", rate: mxl.Rational{Num: 1, Den: 1}, want: 1},
+		{name: "zero numerator", rate: mxl.Rational{Num: 0, Den: 1}, want: 1},
+		{name: "zero denominator", rate: mxl.Rational{Num: 48000, Den: 0}, want: 1},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := batchSize(tc.rate); got != tc.want {
+				t.Fatalf("batchSize(%d/%d) = %d, want %d", tc.rate.Num, tc.rate.Den, got, tc.want)
+			}
+		})
+	}
+}