From 07769859f7d64f36a5473238937f021b2cbecb68 Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Thu, 17 Sep 2026 20:14:41 +0300 Subject: [PATCH] run audio extraction --- cmd/mxl-pattern/main.go | 178 +------------------------------- cmd/mxl-pattern/main_test.go | 26 ----- internal/audio/runner.go | 187 ++++++++++++++++++++++++++++++++++ internal/audio/runner_test.go | 31 ++++++ 4 files changed, 219 insertions(+), 203 deletions(-) create mode 100644 internal/audio/runner.go create mode 100644 internal/audio/runner_test.go diff --git a/cmd/mxl-pattern/main.go b/cmd/mxl-pattern/main.go index 5ca954f..6d99599 100644 --- a/cmd/mxl-pattern/main.go +++ b/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 -} diff --git a/cmd/mxl-pattern/main_test.go b/cmd/mxl-pattern/main_test.go index 1e95be6..639b6c5 100644 --- a/cmd/mxl-pattern/main_test.go +++ b/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{}) diff --git a/internal/audio/runner.go b/internal/audio/runner.go new file mode 100644 index 0000000..e49bf7c --- /dev/null +++ b/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) +} diff --git a/internal/audio/runner_test.go b/internal/audio/runner_test.go new file mode 100644 index 0000000..2bb6c22 --- /dev/null +++ b/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) + } + }) + } +}