From 72decea86683d4ce9bcae4869f1944abc5071099 Mon Sep 17 00:00:00 2001 From: Dmitry Sergeev Date: Wed, 16 Sep 2026 23:50:11 +0300 Subject: [PATCH] audio works --- cmd/mxl-pattern/main.go | 240 ++++++++++++++++++++++++++++++++++- cmd/mxl-pattern/main_test.go | 93 ++++++++++++++ 2 files changed, 332 insertions(+), 1 deletion(-) diff --git a/cmd/mxl-pattern/main.go b/cmd/mxl-pattern/main.go index 7f4b1d9..21f7358 100644 --- a/cmd/mxl-pattern/main.go +++ b/cmd/mxl-pattern/main.go @@ -61,6 +61,16 @@ type parseResult struct { shouldRun bool } +type namedRunner struct { + name string + run func(context.Context) error +} + +type runnerResult struct { + name string + err error +} + var frameRates = map[string]mxl.Rational{ "23.97": {Num: 24000, Den: 1001}, "24": {Num: 24, Den: 1}, @@ -503,12 +513,21 @@ func run(ctx context.Context, args appArgs) (runErr error) { if err != nil { return fmt.Errorf("video configuration: %w", err) } + audioCfg, err := buildAudioConfig(args) + if err != nil { + return fmt.Errorf("audio configuration: %w", err) + } log.Printf("%s %s", APP_NAME, APP_VER) log.Printf("Domain: %s", args.domain) log.Printf("Video: %dx%d %d/%d", videoCfg.Width(), videoCfg.Height(), videoCfg.Rate().Num, videoCfg.Rate().Den) log.Printf("Video ID: %s", videoCfg.ID()) + if audioCfg != nil { + log.Printf("Audio: %d channels %d/%d Hz %.0f dBFS", + audioCfg.Channels(), audioCfg.Rate().Num, audioCfg.Rate().Den, audioCfg.LevelDBFS) + log.Printf("Audio ID: %s", audioCfg.ID()) + } inst, err := mxl.NewInstance(args.domain, "") if err != nil { @@ -520,7 +539,50 @@ func run(ctx context.Context, args appArgs) (runErr error) { } }() - return runVideo(ctx, inst, videoCfg) + runners := []namedRunner{ + { + name: "video", + run: func(ctx context.Context) error { + return runVideo(ctx, inst, videoCfg) + }, + }, + } + if audioCfg != nil { + runners = append(runners, namedRunner{ + name: "audio", + run: func(ctx context.Context) error { + return runAudio(ctx, inst, *audioCfg) + }, + }) + } + + return runConcurrent(ctx, runners...) +} + +func runConcurrent(ctx context.Context, runners ...namedRunner) error { + if len(runners) == 0 { + return nil + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + results := make(chan runnerResult, len(runners)) + for _, runner := range runners { + runner := runner + go func() { + results <- runnerResult{name: runner.name, err: runner.run(ctx)} + }() + } + + var resultErr error + for range runners { + result := <-results + if result.err != nil { + resultErr = errors.Join(resultErr, fmt.Errorf("%s flow: %w", result.name, result.err)) + cancel() + } + } + return resultErr } func runVideo(ctx context.Context, inst *mxl.Instance, cfg video.Config) (runErr error) { @@ -602,6 +664,157 @@ 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 @@ -643,3 +856,28 @@ 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 6b226cd..1e95be6 100644 --- a/cmd/mxl-pattern/main_test.go +++ b/cmd/mxl-pattern/main_test.go @@ -2,10 +2,15 @@ package main import ( "bytes" + "context" "encoding/json" + "errors" "os" "strings" "testing" + "time" + + "github.com/qvest-digital/go-mxl/mxl" "mxl-pattern-generator/internal/audio" "mxl-pattern-generator/internal/flowdef" @@ -250,3 +255,91 @@ func TestValidateAudioArgsRejectsUnknownLevelForFlowDefinition(t *testing.T) { t.Fatalf("error = %v, want unsupported audio level error", err) } } + +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{}) + peerStopped := make(chan struct{}) + + err := runConcurrent(context.Background(), + namedRunner{ + name: "video", + run: func(ctx context.Context) error { + <-peerStarted + return wantErr + }, + }, + namedRunner{ + name: "audio", + run: func(ctx context.Context) error { + close(peerStarted) + <-ctx.Done() + close(peerStopped) + return nil + }, + }, + ) + + if !errors.Is(err, wantErr) { + t.Fatalf("error = %v, want wrapped %v", err, wantErr) + } + if !strings.Contains(err.Error(), "video flow") { + t.Fatalf("error = %q, want runner name", err) + } + select { + case <-peerStopped: + default: + t.Fatal("runConcurrent returned before the sibling completed cleanup") + } +} + +func TestRunConcurrentParentCancellationIsGraceful(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + started := make(chan struct{}) + done := make(chan error, 1) + go func() { + done <- runConcurrent(ctx, namedRunner{ + name: "video", + run: func(ctx context.Context) error { + close(started) + <-ctx.Done() + return nil + }, + }) + }() + + <-started + cancel() + select { + case err := <-done: + if err != nil { + t.Fatalf("runConcurrent: %v", err) + } + case <-time.After(time.Second): + t.Fatal("runConcurrent did not stop after parent cancellation") + } +}