New streaming module

go/audio/amplification.go

@@ -15,50 +15,31 @@ import (
 // (pcm_f32le, audioFormat=3). The write is atomic: samples are written to a
 // temp file in the same directory and then renamed over the original.
 func Gain(path string, offsetDB float64) error {
-	f, err := os.Open(path)
+	left, right, sampleRate, err := readSamples(path)
 	if err != nil {
-		return fmt.Errorf("opening %s: %w", path, err)
-	}
-
-	h, err := readWAVHeader(f)
-	if err != nil {
-		f.Close()
-		return fmt.Errorf("reading WAV header: %w", err)
-	}
-
-	raw := make([]byte, h.dataSize)
-	if _, err := io.ReadFull(f, raw); err != nil {
-		f.Close()
-		return fmt.Errorf("reading PCM data: %w", err)
-	}
-	f.Close()
-
-	var left, right []float64
-	switch {
-	case h.audioFormat == 3 && h.bitsPerSample == 32:
-		left, right = SamplesFromFloat32(raw)
-	case h.audioFormat == 1 && h.bitsPerSample == 24:
-		left, right = SamplesFromInt24(raw)
-	case h.audioFormat == 1 && h.bitsPerSample == 32:
-		left, right = SamplesFromInt32(raw)
-	case h.audioFormat == 1 && h.bitsPerSample == 16:
-		left, right = SamplesFromInt16(raw)
-	default:
-		return fmt.Errorf("unsupported format: audioformat=%d bits=%d", h.audioFormat, h.bitsPerSample)
+		return fmt.Errorf("gain: %w", err)
 	}
+	GainSamples(left, right, offsetDB)
+	return WriteFloat32WAV(path, sampleRate, left, right)
+}
 
+// GainSamples applies a linear gain of offsetDB decibels to both channels in
+// place. It is the in-memory core of Gain.
+func GainSamples(left, right []float64, offsetDB float64) {
 	gain := math.Pow(10, offsetDB/20)
 	for i := range left {
 		left[i] *= gain
+	}
+	for i := range right {
 		right[i] *= gain
 	}
-
-	return writeFloat32WAV(path, h.sampleRate, left, right)
 }
 
-// writeFloat32WAV writes stereo float64 samples to a 32-bit float WAV at path,
-// atomically via a temp file in the same directory followed by a rename.
-func writeFloat32WAV(path string, sampleRate uint32, left, right []float64) error {
+// WriteFloat32WAV writes stereo float64 samples to a 32-bit float WAV at path,
+// atomically via a temp file in the same directory followed by a fsync and
+// rename — the sync matters on servers, where a crash mid-pipeline must not
+// leave a truncated file behind a completed rename.
+func WriteFloat32WAV(path string, sampleRate uint32, left, right []float64) error {
 	dir := filepath.Dir(path)
 	tmp, err := os.CreateTemp(dir, ".gain-*.wav.tmp")
 	if err != nil {
@@ -71,6 +52,11 @@ func writeFloat32WAV(path string, sampleRate uint32, left, right []float64) erro
 		os.Remove(tmpName)
 		return err
 	}
+	if err := tmp.Sync(); err != nil {
+		tmp.Close()
+		os.Remove(tmpName)
+		return fmt.Errorf("syncing temp file: %w", err)
+	}
 	if err := tmp.Close(); err != nil {
 		os.Remove(tmpName)
 		return fmt.Errorf("closing temp file: %w", err)
@@ -93,7 +79,15 @@ func writeFloat32WAVTo(w io.Writer, sampleRate uint32, left, right []float64) er
 
 	blockAlign := uint16(numChannels * bitsPerSample / 8)
 	byteRate := sampleRate * uint32(blockAlign)
-	dataSize := uint32(nFrames) * uint32(blockAlign)
+
+	// RIFF sizes are uint32: past 4 GiB the header silently wraps and every
+	// downstream reader sees a corrupt file. Refuse instead — callers with
+	// longer material must split or lower the intermediate sample rate.
+	dataSize64 := uint64(nFrames) * uint64(blockAlign)
+	if dataSize64+36 > math.MaxUint32 {
+		return fmt.Errorf("WAV data would be %d bytes — over the 4 GiB RIFF limit", dataSize64)
+	}
+	dataSize := uint32(dataSize64)
 	riffSize := 36 + dataSize
 
 	// Buffer the writer: every header field and sample is emitted through bw as a

go/audio/dynamics.go

@@ -2,10 +2,10 @@ package audio
 
 import (
 	"fmt"
-	"io"
 	"math"
-	"os"
+	"runtime"
 	"sort"
+	"sync"
 )
 
 // This file holds three purely time-domain dynamics processors:
@@ -82,7 +82,7 @@ func LookaheadLimiter(path string, thresholdDb, lookaheadMs, releaseMs float64)
 		return fmt.Errorf("lookahead limiter: %w", err)
 	}
 	LookaheadLimitSamples(left, right, float64(sampleRate), thresholdDb, lookaheadMs, releaseMs)
-	return writeFloat32WAV(path, sampleRate, left, right)
+	return WriteFloat32WAV(path, sampleRate, left, right)
 }
 
 // LookaheadLimitSamples is the in-place core of LookaheadLimiter.
@@ -114,18 +114,15 @@ func LookaheadLimitSamples(left, right []float64, sampleRate, thresholdDb, looka
 	// reconstructed waveform — the real peaks between samples — not just the
 	// discrete sample values. That is what makes this a genuine true-peak limiter:
 	// the output's true peak is guaranteed under the ceiling at the working rate,
-	// no oversampled pipeline required to fake it.
-	tpL := truePeakEnvelope(left[:n])
-	tpR := truePeakEnvelope(right[:n])
+	// no oversampled pipeline required to fake it. The per-channel envelopes are
+	// folded straight into required[] rather than materialized — on multi-hour
+	// files the two extra float64 arrays were gigabytes of transient memory —
+	// and the scan is chunk-parallel: each output depends only on the 12
+	// preceding (read-only) input samples and the writes are disjoint.
 	required := make([]float64, n)
-	for i := range n {
-		tp := math.Max(tpL[i], tpR[i])
-		if tp > thresh {
-			required[i] = thresh / tp
-		} else {
-			required[i] = 1
-		}
-	}
+	chunkedParallel(n, func(start, end int) {
+		requiredGainRange(left, right, required, thresh, start, end)
+	})
 
 	// winMin[i] = min(required[i .. i+look]); the gain therefore starts dropping
 	// as soon as a peak enters the lookahead window, never after it has passed.
@@ -152,33 +149,67 @@ func LookaheadLimitSamples(left, right []float64, sampleRate, thresholdDb, looka
 	}
 }
 
-// truePeakEnvelope returns, per sample, the inter-sample (true) peak magnitude at
-// that position — the max over the 4 BS.1770 polyphase sub-samples of the
-// reconstructed waveform. It is the per-sample form of channelTruePeakLinear (same
-// tpFIR), and it drives the true-peak limiter so it tames the peaks BETWEEN
-// samples, not only those landing on them.
-func truePeakEnvelope(samples []float64) []float64 {
-	n := len(samples)
-	env := make([]float64, n)
+// requiredGainRange fills required[start:end] with the gain that tames the
+// stereo-linked true peak at each position to thresh (1.0 where no reduction
+// is needed). The FIR reads input back to start-11; zero before index 0.
+func requiredGainRange(left, right, required []float64, thresh float64, start, end int) {
 	const taps = 12
-	for i := range samples {
+	for i := start; i < end; i++ {
 		var mx float64
 		for phase := range 4 {
-			var acc float64
+			var accL, accR float64
 			for k := range taps {
 				idx := i - k
 				if idx < 0 {
 					break
 				}
-				acc += tpFIR[phase][k] * samples[idx]
+				accL += tpFIR[phase][k] * left[idx]
+				accR += tpFIR[phase][k] * right[idx]
+			}
+			if a := math.Abs(accL); a > mx {
+				mx = a
 			}
-			if a := math.Abs(acc); a > mx {
+			if a := math.Abs(accR); a > mx {
 				mx = a
 			}
 		}
-		env[i] = mx
+		if mx > thresh {
+			required[i] = thresh / mx
+		} else {
+			required[i] = 1
+		}
+	}
+}
+
+// chunkedParallel splits [0, n) into one contiguous range per worker and
+// runs fn over them concurrently. fn must be safe for disjoint ranges (the
+// envelope/peak scans here only read shared input and write disjoint output).
+// Small inputs run inline.
+func chunkedParallel(n int, fn func(start, end int)) {
+	workers := runtime.GOMAXPROCS(0)
+	if workers > 8 {
+		workers = 8
+	}
+	const minChunk = 1 << 16
+	if n < minChunk*2 || workers < 2 {
+		fn(0, n)
+		return
 	}
-	return env
+	chunk := (n + workers - 1) / workers
+	var wg sync.WaitGroup
+	for w := range workers {
+		start := w * chunk
+		if start >= n {
+			break
+		}
+		end := min(start+chunk, n)
+		wg.Add(1)
+		go func(start, end int) {
+			defer wg.Done()
+			fn(start, end)
+		}(start, end)
+	}
+	wg.Wait()
 }
 
 // slidingMin returns, for every index i, the minimum of x over the forward
@@ -225,6 +256,13 @@ func MeasureDynamicsScore(path string) (*DynamicsScoreAnalysis, error) {
 	return dynamicsScore(left, float64(sampleRate)), nil
 }
 
+// MeasureDynamicsScoreSamples computes the Dynamics Score from in-memory
+// channel-1 (left) samples — the buffer-level form of MeasureDynamicsScore for
+// callers that keep the whole file resident across processing stages.
+func MeasureDynamicsScoreSamples(left []float64, sampleRate float64) *DynamicsScoreAnalysis {
+	return dynamicsScore(left, sampleRate)
+}
+
 // dynamicsScore is the in-memory core of MeasureDynamicsScore, operating on the
 // channel-1 (left) samples — the channel astats' DS is parsed from.
 func dynamicsScore(x []float64, sampleRate float64) *DynamicsScoreAnalysis {
@@ -336,7 +374,7 @@ func LimitConforming(path string, strength int) error {
 		return fmt.Errorf("conformity limiter: %w", err)
 	}
 	LookaheadLimitSamples(left, right, float64(sampleRate), c.thresholdDb, c.attackMs, c.releaseMs)
-	return writeFloat32WAV(path, sampleRate, left, right)
+	return WriteFloat32WAV(path, sampleRate, left, right)
 }
 
 // LimitCharacter applies the soft-knee character limiter at the calibrated
@@ -348,7 +386,7 @@ func LimitCharacter(path string, strength int) error {
 		return fmt.Errorf("character limiter: %w", err)
 	}
 	CharacterLimitSamples(left, right, float64(sampleRate), c.thresholdDb, characterLimiterKnee, c.attackMs, c.releaseMs)
-	return writeFloat32WAV(path, sampleRate, left, right)
+	return WriteFloat32WAV(path, sampleRate, left, right)
 }
 
 // CharacterLimiter applies a soft-knee limiter at path in place, rewriting it as
@@ -363,7 +401,7 @@ func CharacterLimiter(path string, thresholdDb, kneeDb, attackMs, releaseMs floa
 		return fmt.Errorf("character limiter: %w", err)
 	}
 	CharacterLimitSamples(left, right, float64(sampleRate), thresholdDb, kneeDb, attackMs, releaseMs)
-	return writeFloat32WAV(path, sampleRate, left, right)
+	return WriteFloat32WAV(path, sampleRate, left, right)
 }
 
 // CharacterLimitSamples is the in-place core of CharacterLimiter: a feed-forward
@@ -383,7 +421,7 @@ func Compress(path string, thresholdDb, ratio, kneeDb, attackMs, releaseMs, make
 		return fmt.Errorf("compressor: %w", err)
 	}
 	CompressSamples(left, right, float64(sampleRate), thresholdDb, ratio, kneeDb, attackMs, releaseMs, makeupDb, rmsMs)
-	return writeFloat32WAV(path, sampleRate, left, right)
+	return WriteFloat32WAV(path, sampleRate, left, right)
 }
 
 // CompressSamples is the in-place core of Compress.
@@ -432,7 +470,7 @@ func UpwardCompress(path string, thresholdDb, ratio, kneeDb, attackMs, releaseMs
 		return fmt.Errorf("upward compressor: %w", err)
 	}
 	UpwardCompressSamples(left, right, float64(sampleRate), thresholdDb, ratio, kneeDb, attackMs, releaseMs, maxBoostDb, rmsMs)
-	return writeFloat32WAV(path, sampleRate, left, right)
+	return WriteFloat32WAV(path, sampleRate, left, right)
 }
 
 // UpwardCompressSamples is the in-place core of UpwardCompress.
@@ -519,37 +557,9 @@ func processDynamics(left, right []float64, sampleRate, thresholdDb, ratio, knee
 }
 
 // readStereo reads a WAV at path into stereo float64 buffers and also returns
-// its sample rate, decoding the same formats ReadWAV supports. ReadWAV itself
-// does not surface the sample rate, which the time-based processors here need to
-// turn millisecond time factors into per-sample coefficients.
+// its sample rate. It is a thin alias for the package-central readSamples,
+// kept for the time-based processors here that turn millisecond time factors
+// into per-sample coefficients.
 func readStereo(path string) (left, right []float64, sampleRate uint32, err error) {
-	f, err := os.Open(path)
-	if err != nil {
-		return nil, nil, 0, err
-	}
-	defer f.Close()
-
-	h, err := readWAVHeader(f)
-	if err != nil {
-		return nil, nil, 0, err
-	}
-
-	raw := make([]byte, h.dataSize)
-	if _, err := io.ReadFull(f, raw); err != nil {
-		return nil, nil, 0, err
-	}
-
-	switch {
-	case h.audioFormat == 3 && h.bitsPerSample == 32:
-		left, right = SamplesFromFloat32(raw)
-	case h.audioFormat == 1 && h.bitsPerSample == 24:
-		left, right = SamplesFromInt24(raw)
-	case h.audioFormat == 1 && h.bitsPerSample == 32:
-		left, right = SamplesFromInt32(raw)
-	case h.audioFormat == 1 && h.bitsPerSample == 16:
-		left, right = SamplesFromInt16(raw)
-	default:
-		return nil, nil, 0, fmt.Errorf("format not supported: audioformat: %d, bits per sample: %d", h.audioFormat, h.bitsPerSample)
-	}
-	return left, right, h.sampleRate, nil
+	return readSamples(path)
 }

go/audio/dynaudnorm.go

@@ -46,8 +46,10 @@ func BuildDynaudnormFilter(params *DynaudnormParams, isSpeech bool) string {
 		return ""
 	}
 
+	// gausssize must be odd — ffmpeg coerced the old 36 to 37 with a warning,
+	// so 37 is spelled out to keep the effective behaviour and lose the warning.
 	return fmt.Sprintf(
-		"dynaudnorm=framelen=650:gausssize=36:targetrms=%.6f:threshold=%.6f:altboundary=true:overlap=0.95",
+		"dynaudnorm=framelen=650:gausssize=37:targetrms=%.6f:threshold=%.6f:altboundary=true:overlap=0.95",
 		params.TargetRMS,
 		params.Threshold,
 	)