diff options
Diffstat (limited to 'server/internal')
| -rw-r--r-- | server/internal/api/api.go | 36 | ||||
| -rw-r--r-- | server/internal/decode/decode.go | 22 | ||||
| -rw-r--r-- | server/internal/decode/decode_test.go | 54 | ||||
| -rw-r--r-- | server/internal/detect/fft_bench_test.go | 36 | ||||
| -rw-r--r-- | server/internal/detect/lags.go | 229 | ||||
| -rw-r--r-- | server/internal/detect/lags_test.go | 116 | ||||
| -rw-r--r-- | server/internal/detect/runs_test.go | 105 | ||||
| -rw-r--r-- | server/internal/skips/ads.go | 236 | ||||
| -rw-r--r-- | server/internal/skips/ads_test.go | 316 | ||||
| -rw-r--r-- | server/internal/skips/skips.go | 6 | ||||
| -rw-r--r-- | server/internal/store/store.go | 23 |
11 files changed, 1169 insertions, 10 deletions
diff --git a/server/internal/api/api.go b/server/internal/api/api.go index b6f14ff..992f622 100644 --- a/server/internal/api/api.go +++ b/server/internal/api/api.go @@ -57,6 +57,22 @@ type episodeResponse struct { // "looked and found nothing" apart from "not looked at yet" -- a distinction the app got // wrong when the readout said "None" for both. Detected bool `json:"detected"` + + // Breaks are the stretches to skip: the advertising this episode shares with the others + // mirrored alongside it. + Breaks []breakRange `json:"breaks,omitempty"` + // FoundAdMs and InsertedAdMs let the app show what was accounted for against what the + // feed's declared length implies is there. + FoundAdMs int64 `json:"foundAdMs,omitempty"` + InsertedAdMs int64 `json:"insertedAdMs,omitempty"` + // AdsRefused: matches were found but were too extensive to be advertising, so none were + // published. Distinct from nothing having matched. + AdsRefused bool `json:"adsRefused,omitempty"` +} + +type breakRange struct { + StartMs int64 `json:"startMs"` + EndMs int64 `json:"endMs"` } // Handler returns the routed, authenticated handler. @@ -145,13 +161,21 @@ func (s *Server) episode(w http.ResponseWriter, r *http.Request) { return } + breaks := make([]breakRange, 0, len(found.Breaks)) + for _, br := range found.Breaks { + breaks = append(breaks, breakRange{StartMs: br.StartMs, EndMs: br.EndMs}) + } writeJSON(w, http.StatusOK, episodeResponse{ - Status: "ready", - AudioURL: "/audio/" + found.ID + "?token=" + url.QueryEscape(s.opts.Token), - DurationMs: found.DurationMs, - IntroMs: found.IntroMs, - OutroMs: found.OutroMs, - Detected: found.DetectedAtUnix != 0, + Status: "ready", + AudioURL: "/audio/" + found.ID + "?token=" + url.QueryEscape(s.opts.Token), + DurationMs: found.DurationMs, + IntroMs: found.IntroMs, + OutroMs: found.OutroMs, + Detected: found.DetectedAtUnix != 0, + Breaks: breaks, + FoundAdMs: found.FoundAdMs, + InsertedAdMs: found.InsertedAdMs, + AdsRefused: found.AdsRefused, }) } diff --git a/server/internal/decode/decode.go b/server/internal/decode/decode.go index 12ec204..85dde37 100644 --- a/server/internal/decode/decode.go +++ b/server/internal/decode/decode.go @@ -44,12 +44,26 @@ type Decoder struct { // the server has no equivalent of the ten-minute ceiling that killed this on the phone. const DefaultTimeout = 10 * time.Minute -// Range decodes durationMs of audio starting at startMs. A durationMs of zero decodes to the -// end of the file. +// Range decodes durationMs of audio starting at startMs, at [TargetSampleRate]. A durationMs of +// zero decodes to the end of the file. // // Decoding a range rather than always the whole file is what lets a caller verify one candidate // segment at full resolution without holding an entire episode in memory. func (d Decoder) Range(ctx context.Context, path string, startMs, durationMs int64) (*Clip, error) { + return d.RangeAt(ctx, path, startMs, durationMs, TargetSampleRate) +} + +// RangeAt is [Range] at a chosen sample rate. +// +// Whole-episode work needs a far lower rate than matching does: an hour at 8kHz is 29 million +// samples, and at 250Hz it is 900 thousand -- the difference between a correlation that is +// impossible and one that takes seconds. ffmpeg does the resampling, so the low rate costs +// nothing extra here rather than being decimated afterwards from audio that had to be decoded +// at full rate first. +func (d Decoder) RangeAt(ctx context.Context, path string, startMs, durationMs int64, sampleRate int) (*Clip, error) { + if sampleRate <= 0 { + sampleRate = TargetSampleRate + } args := []string{"-nostdin", "-hide_banner", "-loglevel", "error"} if startMs > 0 { args = append(args, "-ss", millisToTimestamp(startMs)) @@ -59,7 +73,7 @@ func (d Decoder) Range(ctx context.Context, path string, startMs, durationMs int args = append(args, "-t", millisToTimestamp(durationMs)) } // f64le straight out, so nothing downstream has to reinterpret or rescale it. - args = append(args, "-vn", "-ac", "1", "-ar", strconv.Itoa(TargetSampleRate), "-f", "f64le", "-") + args = append(args, "-vn", "-ac", "1", "-ar", strconv.Itoa(sampleRate), "-f", "f64le", "-") timeout := d.Timeout if timeout == 0 { @@ -80,7 +94,7 @@ func (d Decoder) Range(ctx context.Context, path string, startMs, durationMs int if err != nil { return nil, fmt.Errorf("decode %s: %w", path, err) } - return &Clip{Samples: samples, SampleRate: TargetSampleRate, StartMs: startMs}, nil + return &Clip{Samples: samples, SampleRate: sampleRate, StartMs: startMs}, nil } // DurationMs reports a file's real length, which is routinely not what its feed claims -- diff --git a/server/internal/decode/decode_test.go b/server/internal/decode/decode_test.go index fea4e85..e29a716 100644 --- a/server/internal/decode/decode_test.go +++ b/server/internal/decode/decode_test.go @@ -162,3 +162,57 @@ func TestAMissingFileIsAnError(t *testing.T) { t.Error("expected an error for a missing file") } } + +// Whole-episode work needs a much lower rate than matching does: an hour at 8kHz is 29 million +// samples and at 250Hz it is 900 thousand, which is the difference between a correlation that +// cannot be run and one that takes seconds. +func TestRangeAtDecodesToTheRateAsked(t *testing.T) { + requireFFmpeg(t) + path := writeTone(t, 100, 2) + + clip, err := Decoder{}.RangeAt(context.Background(), path, 0, 0, 250) + if err != nil { + t.Fatal(err) + } + if clip.SampleRate != 250 { + t.Errorf("sample rate %d, want 250", clip.SampleRate) + } + if len(clip.Samples) < 2*250-32 || len(clip.Samples) > 2*250+32 { + t.Errorf("got %d samples, want about %d for two seconds at 250Hz", len(clip.Samples), 2*250) + } +} + +// A 100Hz tone survives decimation to 250Hz (Nyquist 125Hz), so the structure is still there to +// correlate against -- which is the whole point of discovering at a low rate. +func TestRangeAtKeepsTheSignalItDecimates(t *testing.T) { + requireFFmpeg(t) + path := writeTone(t, 100, 2) + + clip, err := Decoder{}.RangeAt(context.Background(), path, 0, 0, 250) + if err != nil { + t.Fatal(err) + } + crossings := 0 + for i := 1; i < len(clip.Samples); i++ { + if (clip.Samples[i-1] < 0) != (clip.Samples[i] < 0) { + crossings++ + } + } + perSecond := float64(crossings) * 250 / float64(len(clip.Samples)) + if perSecond < 180 || perSecond > 220 { + t.Errorf("%.0f zero crossings per second, want about 200 for a 100Hz tone", perSecond) + } +} + +func TestRangeAtFallsBackToTheTargetRate(t *testing.T) { + requireFFmpeg(t) + path := writeTone(t, 440, 1) + + clip, err := Decoder{}.RangeAt(context.Background(), path, 0, 0, 0) + if err != nil { + t.Fatal(err) + } + if clip.SampleRate != TargetSampleRate { + t.Errorf("sample rate %d, want the default %d", clip.SampleRate, TargetSampleRate) + } +} diff --git a/server/internal/detect/fft_bench_test.go b/server/internal/detect/fft_bench_test.go new file mode 100644 index 0000000..be7f5ff --- /dev/null +++ b/server/internal/detect/fft_bench_test.go @@ -0,0 +1,36 @@ +package detect + +import ( + "math/rand" + "strconv" + "testing" +) + +// Sizing data for whole-episode work, kept because it is what the ad scan was designed +// against rather than guessed at. +// +// Measured here: 2^20 0.33s, 2^21 0.73s, 2^22 1.69s, 2^23 3.66s, 2^24 8.0s. A 70-minute +// episode at 8kHz is 33.6M samples, far too much to correlate whole; decimated to 250Hz it is +// 1.05M, so a full cross-correlation of two of them is 3 transforms of 2^22 -- about five +// seconds a pair, which is what makes scanning whole episodes possible at all. +func BenchmarkTransform(b *testing.B) { + for _, bits := range []uint{20, 21, 22, 23, 24} { + n := 1 << bits + b.Run(fmtSize(bits), func(b *testing.B) { + rng := rand.New(rand.NewSource(1)) + re := make([]float64, n) + im := make([]float64, n) + for i := range re { + re[i] = rng.NormFloat64() + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + Transform(re, im, false) + } + }) + } +} + +func fmtSize(bits uint) string { + return "2^" + strconv.Itoa(int(bits)) +} diff --git a/server/internal/detect/lags.go b/server/internal/detect/lags.go new file mode 100644 index 0000000..55d9544 --- /dev/null +++ b/server/internal/detect/lags.go @@ -0,0 +1,229 @@ +package detect + +import "math" + +// minLagSeparation keeps the shortlist to one entry per shared segment. +// +// A real match spreads its energy over neighbouring lags, so without this a single +// advertisement fills every slot on the shortlist and the second one is never looked at. +const minLagSeparation = 64 + +type scoredLag struct { + lag int + score float64 +} + +// BestLags shortlists the alignments at which a and b most strongly share audio. +// +// One cross-correlation over every lag, not a correlation per window position. That is the +// whole reason whole-episode scanning is possible: the sliding search it replaces cost a +// full-length correlation for each of 117 window positions -- 199 seconds for a 240-second +// clip, measured -- while this costs one FFT pair for the entire pair of recordings. A +// 70-minute episode decimated to 250Hz is three transforms of 2^22, about five seconds. +// +// A shortlist only. The scores here are unnormalised and favour loud or long overlaps, so they +// rank candidates and decide nothing: [MatchedRuns] establishes whether a lag means anything +// and how far the agreement actually extends. +// +// Lag is defined so that a[i] lines up with b[i-lag]. +func BestLags(a, b []float64, keep int) []int { + if len(a) == 0 || len(b) == 0 || keep < 1 { + return nil + } + + // Mean-centred, so two recordings that merely share a DC offset do not outscore two that + // share actual audio. + ca, cb := centred(a), centred(b) + + // Correlation is convolution with one input reversed. + n := NextPowerOfTwo(len(ca) + len(cb)) + aRe := make([]float64, n) + copy(aRe, ca) + aIm := make([]float64, n) + bRe := make([]float64, n) + for i, v := range cb { + bRe[len(cb)-1-i] = v + } + bIm := make([]float64, n) + + Transform(aRe, aIm, false) + Transform(bRe, bIm, false) + for i := 0; i < n; i++ { + re := aRe[i]*bRe[i] - aIm[i]*bIm[i] + im := aRe[i]*bIm[i] + aIm[i]*bRe[i] + aRe[i], aIm[i] = re, im + } + Transform(aRe, aIm, true) + + // Index k of the convolution corresponds to lag k - (len(b) - 1). + offset := len(cb) - 1 + best := make([]scoredLag, 0, keep) + for k := 0; k < len(ca)+len(cb)-1; k++ { + best = offer(best, scoredLag{lag: k - offset, score: math.Abs(aRe[k])}, keep) + } + + // Strongest first, by insertion sort: the shortlist is a handful of entries. + for i := 1; i < len(best); i++ { + for j := i; j > 0 && best[j].score > best[j-1].score; j-- { + best[j], best[j-1] = best[j-1], best[j] + } + } + out := make([]int, 0, len(best)) + for _, s := range best { + out = append(out, s.lag) + } + return out +} + +// offer adds a candidate to the shortlist, keeping one entry per neighbourhood of lags and at +// most keep entries overall. +func offer(best []scoredLag, candidate scoredLag, keep int) []scoredLag { + for i := range best { + if abs(best[i].lag-candidate.lag) <= minLagSeparation { + if candidate.score > best[i].score { + best[i] = candidate + } + return best + } + } + if len(best) < keep { + return append(best, candidate) + } + weakest := 0 + for i := range best { + if best[i].score < best[weakest].score { + weakest = i + } + } + if candidate.score > best[weakest].score { + best[weakest] = candidate + } + return best +} + +// Run is a stretch over which two recordings carry the same audio, in a's own sample positions. +type Run struct { + StartA int + EndA int // exclusive + // StartB is where the same audio begins in b. + StartB int + // Score is the weakest window correlation inside the run, so it describes the whole run + // rather than its best moment. + Score float64 +} + +// Length is the run's duration in samples. +func (r Run) Length() int { return r.EndA - r.StartA } + +// RunOptions governs what counts as agreement. +type RunOptions struct { + // WindowSamples is the granularity the comparison is made at. + WindowSamples int + // MinScore is the correlation a window must reach to count as the same audio. + MinScore float64 + // MinRunSamples discards agreement too short to be a real inserted segment. Two + // recordings will agree briefly by chance, and on silence they will agree completely. + MinRunSamples int +} + +// DefaultRunOptions is sized for audio decimated to 250Hz: one-second windows, and nothing +// shorter than four seconds taken seriously. +func DefaultRunOptions() RunOptions { + return RunOptions{WindowSamples: 250, MinScore: 0.75, MinRunSamples: 1000} +} + +// MatchedRuns reports the stretches over which a and b carry the same audio at the given lag. +// +// Direct, window by window, with no transform: once the alignment is known the comparison is a +// Pearson correlation over short windows, which is cheap and gives the extent of the agreement +// rather than only its strength. The extent is the point -- an advertisement has to be skipped +// in full, so where it stops matters as much as that it is there. +// +// Silent windows are not counted as agreement. Two recordings agree perfectly on silence, and +// treating that as shared audio would join unrelated segments across the gaps between them. +func MatchedRuns(a, b []float64, lag int, opts RunOptions) []Run { + if opts.WindowSamples <= 0 || len(a) == 0 || len(b) == 0 { + return nil + } + + var runs []Run + runStart := -1 + runScore := 1.0 + + for start := 0; start+opts.WindowSamples <= len(a); start += opts.WindowSamples { + bStart := start - lag + score, ok := windowAgreement(a, b, start, bStart, opts.WindowSamples) + + if ok && score >= opts.MinScore { + if runStart < 0 { + runStart, runScore = start, score + } else if score < runScore { + runScore = score + } + continue + } + runs = closeRun(runs, runStart, start, lag, runScore, opts) + runStart, runScore = -1, 1.0 + } + end := (len(a) / opts.WindowSamples) * opts.WindowSamples + return closeRun(runs, runStart, end, lag, runScore, opts) +} + +func closeRun(runs []Run, runStart, end, lag int, score float64, opts RunOptions) []Run { + if runStart < 0 || end-runStart < opts.MinRunSamples { + return runs + } + return append(runs, Run{StartA: runStart, EndA: end, StartB: runStart - lag, Score: score}) +} + +// windowAgreement is the Pearson correlation of one aligned window, and whether it carries +// enough signal to mean anything. +func windowAgreement(a, b []float64, aStart, bStart, win int) (float64, bool) { + if aStart < 0 || bStart < 0 || aStart+win > len(a) || bStart+win > len(b) { + return 0, false + } + + var sumA, sumB float64 + for i := 0; i < win; i++ { + sumA += a[aStart+i] + sumB += b[bStart+i] + } + meanA, meanB := sumA/float64(win), sumB/float64(win) + + var cov, varA, varB float64 + for i := 0; i < win; i++ { + da, db := a[aStart+i]-meanA, b[bStart+i]-meanB + cov += da * db + varA += da * da + varB += db * db + } + // Near-silence in either window: perfectly correlated and meaningless. + if varA < silenceFloor || varB < silenceFloor { + return 0, false + } + return cov / math.Sqrt(varA*varB), true +} + +// silenceFloor is the variance below which a window carries no signal worth comparing. Scaled +// for 16-bit-derived samples, where ordinary speech sits in the hundreds. +const silenceFloor = 1e-6 + +func centred(x []float64) []float64 { + mean := 0.0 + for _, v := range x { + mean += v + } + mean /= float64(len(x)) + out := make([]float64, len(x)) + for i, v := range x { + out[i] = v - mean + } + return out +} + +func abs(n int) int { + if n < 0 { + return -n + } + return n +} diff --git a/server/internal/detect/lags_test.go b/server/internal/detect/lags_test.go new file mode 100644 index 0000000..406a5d8 --- /dev/null +++ b/server/internal/detect/lags_test.go @@ -0,0 +1,116 @@ +package detect + +import ( + "math" + "math/rand" + "testing" +) + +// Finding where two recordings share audio, without sliding a window. +// +// The sliding search this replaces costs one full-length correlation per window position, which +// is why it could never run over whole episodes -- 199 seconds for a 240-second clip, measured. +// A single cross-correlation over every lag costs one FFT pair for the whole pair of episodes, +// and tells you *where* they align; the extent of the match is then cheap to walk out directly. + +func randomSignal(seed int64, n int) []float64 { + rng := rand.New(rand.NewSource(seed)) + out := make([]float64, n) + for i := range out { + out[i] = rng.NormFloat64() * 50 + } + return out +} + +// place copies segment into a copy of base at the given offset. +func place(base, segment []float64, at int) []float64 { + out := append([]float64(nil), base...) + copy(out[at:], segment) + return out +} + +func TestFindsTheLagOfASharedSegment(t *testing.T) { + shared := randomSignal(99, 4000) + // The same audio 1000 samples into a, and 5000 into b: a relative lag of -4000. + a := place(randomSignal(1, 20000), shared, 1000) + b := place(randomSignal(2, 20000), shared, 5000) + + lags := BestLags(a, b, 3) + + if len(lags) == 0 { + t.Fatal("no lags found") + } + // The segment sits at the same place in both once b is shifted by -4000. + found := false + for _, lag := range lags { + if math.Abs(float64(lag-(-4000))) <= 2 { + found = true + } + } + if !found { + t.Errorf("lags %v, expected one at about -4000", lags) + } +} + +func TestFindsTheLagWhenTheSegmentIsEarlierInTheSecond(t *testing.T) { + shared := randomSignal(98, 4000) + a := place(randomSignal(3, 20000), shared, 9000) + b := place(randomSignal(4, 20000), shared, 2000) + + lags := BestLags(a, b, 3) + + found := false + for _, lag := range lags { + if math.Abs(float64(lag-7000)) <= 2 { + found = true + } + } + if !found { + t.Errorf("lags %v, expected one at about 7000", lags) + } +} + +// Several ads in one episode means several lags, and all of them matter. +func TestFindsSeveralLagsWhenSeveralSegmentsAreShared(t *testing.T) { + first := randomSignal(97, 3000) + second := randomSignal(96, 3000) + a := place(place(randomSignal(5, 40000), first, 2000), second, 25000) + b := place(place(randomSignal(6, 40000), first, 8000), second, 18000) + + lags := BestLags(a, b, 6) + + wantFirst, wantSecond := -6000, 7000 + var gotFirst, gotSecond bool + for _, lag := range lags { + if math.Abs(float64(lag-wantFirst)) <= 2 { + gotFirst = true + } + if math.Abs(float64(lag-wantSecond)) <= 2 { + gotSecond = true + } + } + if !gotFirst || !gotSecond { + t.Errorf("lags %v, expected both %d and %d", lags, wantFirst, wantSecond) + } +} + +func TestUnrelatedRecordingsProduceNoStrongLag(t *testing.T) { + a := randomSignal(7, 20000) + b := randomSignal(8, 20000) + + // Whatever comes back must not survive verification; BestLags only shortlists, so this + // checks the shortlist is not absurdly confident rather than that it is empty. + lags := BestLags(a, b, 3) + for _, lag := range lags { + runs := MatchedRuns(a, b, lag, DefaultRunOptions()) + if len(runs) > 0 { + t.Errorf("lag %d produced runs %v on unrelated audio", lag, runs) + } + } +} + +func TestEmptyInputHasNoLags(t *testing.T) { + if lags := BestLags(nil, randomSignal(9, 100), 3); len(lags) != 0 { + t.Errorf("got %v", lags) + } +} diff --git a/server/internal/detect/runs_test.go b/server/internal/detect/runs_test.go new file mode 100644 index 0000000..adb47c0 --- /dev/null +++ b/server/internal/detect/runs_test.go @@ -0,0 +1,105 @@ +package detect + +import "testing" + +// How far the agreement extends, which is what actually gets skipped. An advertisement has to +// be skipped in full: where it stops matters as much as that it is there. + +func TestFindsTheExtentOfASharedStretch(t *testing.T) { + shared := randomSignal(50, 5000) + a := place(randomSignal(11, 30000), shared, 4000) + b := place(randomSignal(12, 30000), shared, 11000) + + runs := MatchedRuns(a, b, -7000, DefaultRunOptions()) + + if len(runs) != 1 { + t.Fatalf("got %d runs, want 1: %+v", len(runs), runs) + } + run := runs[0] + // Window-quantised, so allow a window either side of the true 4000..9000. + if run.StartA < 3750 || run.StartA > 4250 { + t.Errorf("StartA %d, want about 4000", run.StartA) + } + if run.EndA < 8750 || run.EndA > 9250 { + t.Errorf("EndA %d, want about 9000", run.EndA) + } + if run.StartB < 10750 || run.StartB > 11250 { + t.Errorf("StartB %d, want about 11000", run.StartB) + } +} + +func TestReportsTheWeakestWindowAsTheRunScore(t *testing.T) { + shared := randomSignal(51, 5000) + a := place(randomSignal(13, 20000), shared, 2000) + b := place(randomSignal(14, 20000), shared, 6000) + + runs := MatchedRuns(a, b, -4000, DefaultRunOptions()) + + if len(runs) == 0 { + t.Fatal("no runs") + } + // Identical audio, so even the weakest window should be near-perfect. + if runs[0].Score < 0.99 { + t.Errorf("score %v, want near 1 for identical audio", runs[0].Score) + } +} + +/* +Two recordings agree briefly by chance. An advertisement is seconds long, so anything shorter +than the floor is noise and keeping it would pepper an episode with one-second skips. +*/ +func TestDiscardsAgreementTooShortToBeReal(t *testing.T) { + shared := randomSignal(52, 400) + a := place(randomSignal(15, 20000), shared, 2000) + b := place(randomSignal(16, 20000), shared, 6000) + + runs := MatchedRuns(a, b, -4000, DefaultRunOptions()) + + for _, r := range runs { + if r.Length() < DefaultRunOptions().MinRunSamples { + t.Errorf("kept a %d-sample run, under the %d floor", r.Length(), DefaultRunOptions().MinRunSamples) + } + } +} + +/* +Silence is the trap. Two recordings agree perfectly on it, so counting silent windows as shared +audio would weld separate advertisements together across the pauses between them -- and in the +worst case declare a whole quiet episode shared. +*/ +func TestSilenceIsNotAgreement(t *testing.T) { + silent := make([]float64, 20000) + + runs := MatchedRuns(silent, silent, 0, DefaultRunOptions()) + + if len(runs) != 0 { + t.Errorf("silence produced runs: %+v", runs) + } +} + +func TestSilenceDoesNotJoinTwoSeparateSegments(t *testing.T) { + first := randomSignal(53, 3000) + second := randomSignal(54, 3000) + base := randomSignal(17, 40000) + // Two shared segments with four seconds of silence between them, at the same lag. + a := place(place(base, first, 5000), second, 12000) + b := place(place(randomSignal(18, 40000), first, 5000), second, 12000) + for i := 8000; i < 12000; i++ { + a[i], b[i] = 0, 0 + } + + runs := MatchedRuns(a, b, 0, DefaultRunOptions()) + + if len(runs) < 2 { + t.Errorf("got %d runs, want the two segments kept apart: %+v", len(runs), runs) + } +} + +func TestALagThatPushesTheWindowOutOfRangeYieldsNothing(t *testing.T) { + a := randomSignal(19, 5000) + b := randomSignal(20, 5000) + + if runs := MatchedRuns(a, b, 100000, DefaultRunOptions()); len(runs) != 0 { + t.Errorf("got %+v", runs) + } +} diff --git a/server/internal/skips/ads.go b/server/internal/skips/ads.go new file mode 100644 index 0000000..b3c1f3b --- /dev/null +++ b/server/internal/skips/ads.go @@ -0,0 +1,236 @@ +package skips + +import ( + "context" + "fmt" + "sort" + + "github.com/terst/listen/server/internal/decode" + "github.com/terst/listen/server/internal/detect" +) + +/* +Detecting the advertising rather than the intro. + +The inversion the data forced. Intro detection reported a segment confirmed on 8 of 8 Radiolab +episodes, passed its own self-check, and was wrong: searched for in an ad-free copy of the same +episode fetched hours earlier it scored r = 0.0621. It was in the ad break. + +Mirroring 8 episodes within one minute makes them maximally dependent -- the ad server inserts +whatever campaign is running into all of them, identically -- and that destroys the independence +the confirmation threshold needs. No amount of self-checking recovers it, because the segments +genuinely are the same audio. + +Turned around, the same property is a reliable advertisement detector: what simultaneous fetches +share *is* the inserted advertising. And unlike intro detection this can be checked against +something the detector did not produce -- the feed declares the ad-free length, the server was +served something longer, and the difference is how much advertising there should be. +*/ + +// AudioSource is the decoding ad detection needs. +type AudioSource interface { + // RangeAt decodes mono audio at a chosen sample rate. Zero durationMs means to the end. + RangeAt(ctx context.Context, path string, startMs, durationMs int64, sampleRate int) (*decode.Clip, error) + // DurationMs is the file's real length. + DurationMs(ctx context.Context, path string) (int64, error) +} + +// AdBreak is a stretch of one episode that other episodes also contain. +type AdBreak struct { + StartMs int64 + EndMs int64 + // Partners is how many other episodes were found to carry this audio. + Partners int + // Score is the weakest window agreement across the match, so it describes the whole break. + Score float64 +} + +// DurationMs is the break's length. +func (b AdBreak) DurationMs() int64 { return b.EndMs - b.StartMs } + +// Accounting is what the scan found against what the length gap says should be there. +// +// The gap is the point: it comes from the feed and the file, not from the detector, so it is the +// one number here that can contradict the scan. Far short means breaks are being missed; far +// over means programme is being matched. +type Accounting struct { + FoundMs int64 + // InsertedMs is served minus declared, or zero when the feed did not say. + InsertedMs int64 +} + +// AdResult is a batch's outcome. +type AdResult struct { + Breaks map[string][]AdBreak + Accounting map[string]Accounting + // Refused records episodes whose matches were too extensive to be advertising, and which + // therefore publish nothing. + Refused map[string]bool +} + +// AdConfig governs the scan. +type AdConfig struct { + // DiscoveryRate is the sample rate whole episodes are compared at. 250Hz turns an hour of + // audio from 29 million samples into 900 thousand, which is what makes a single + // cross-correlation over a whole pair of episodes possible. + DiscoveryRate int + // LagsPerPair is how many candidate alignments are examined per pair of episodes. An + // episode carries a handful of ad breaks, and each one shows up as its own alignment. + LagsPerPair int + // MaxSkipFraction refuses an episode whose matches cover more than this much of it. + // + // The guard that matters most: two episodes of a show can legitimately share a lot -- a + // re-aired episode, a two-parter with a long recap -- and skipping all of it would eat the + // programme rather than the advertising. + MaxSkipFraction float64 + // MaxBreakMs caps a single break. Longer than any plausible ad pod means shared programme. + MaxBreakMs int64 + // Runs is what counts as agreement within an alignment. + Runs detect.RunOptions +} + +// DefaultAdConfig is the production configuration. +func DefaultAdConfig() AdConfig { + return AdConfig{ + DiscoveryRate: 250, + LagsPerPair: 8, + MaxSkipFraction: 0.35, + MaxBreakMs: 240_000, + Runs: detect.DefaultRunOptions(), + } +} + +// DetectAds finds what a batch of episodes share, which for simultaneously-mirrored episodes is +// the advertising inserted into them. +// +// Pairwise, because an ad need not be in every episode -- campaigns rotate within a batch, and a +// break present in two episodes is still a break. Every pair costs one cross-correlation over +// the whole decimated episode, not a correlation per window position: the sliding search that +// replaces measured 199 seconds for a 240-second clip and could never have run over whole +// episodes. +func DetectAds(ctx context.Context, src AudioSource, episodes []Episode, cfg AdConfig) (*AdResult, error) { + result := &AdResult{ + Breaks: map[string][]AdBreak{}, + Accounting: map[string]Accounting{}, + Refused: map[string]bool{}, + } + if len(episodes) < 2 { + return result, nil + } + + rate := cfg.DiscoveryRate + if rate <= 0 { + rate = DefaultAdConfig().DiscoveryRate + } + + type loaded struct { + episode Episode + samples []float64 + servedMs int64 + } + clips := make([]loaded, 0, len(episodes)) + for _, ep := range episodes { + servedMs, err := src.DurationMs(ctx, ep.Path) + if err != nil { + return nil, fmt.Errorf("episode %s: %w", ep.ID, err) + } + clip, err := src.RangeAt(ctx, ep.Path, 0, 0, rate) + if err != nil { + return nil, fmt.Errorf("episode %s: %w", ep.ID, err) + } + clips = append(clips, loaded{episode: ep, samples: clip.Samples, servedMs: servedMs}) + } + + // found[episode index] accumulates every match from every partner, before merging. + found := make([][]AdBreak, len(clips)) + for i := 0; i < len(clips); i++ { + for j := i + 1; j < len(clips); j++ { + a, b := clips[i].samples, clips[j].samples + for _, lag := range detect.BestLags(a, b, cfg.LagsPerPair) { + for _, run := range detect.MatchedRuns(a, b, lag, cfg.Runs) { + found[i] = append(found[i], breakOf(run.StartA, run.EndA, run.Score, rate)) + found[j] = append(found[j], breakOf(run.StartB, run.StartB+run.Length(), run.Score, rate)) + } + } + } + } + + for i, clip := range clips { + // Merged without judgement first, so the accounting describes what was actually found + // and a refusal is recorded rather than silently leaving an episode with nothing. An + // episode that publishes no breaks because its matches were implausible is a different + // state from one where nothing matched, and the two looking identical is the mistake + // the app's own skip readout already made once. + matched := merge(found[i]) + total := int64(0) + for _, br := range matched { + total += br.DurationMs() + } + + inserted := int64(0) + if clip.episode.DeclaredDurationMs > 0 && clip.servedMs > clip.episode.DeclaredDurationMs { + inserted = clip.servedMs - clip.episode.DeclaredDurationMs + } + result.Accounting[clip.episode.ID] = Accounting{FoundMs: total, InsertedMs: inserted} + + // Too much of the episode matched to be advertising at all. + if clip.servedMs > 0 && float64(total) > cfg.MaxSkipFraction*float64(clip.servedMs) { + result.Refused[clip.episode.ID] = true + continue + } + + breaks := make([]AdBreak, 0, len(matched)) + for _, br := range matched { + // A single match longer than any real ad pod is shared programme -- a recap, or a + // repeat -- and cutting it would remove the episode rather than the advertising. + if cfg.MaxBreakMs > 0 && br.DurationMs() > cfg.MaxBreakMs { + result.Refused[clip.episode.ID] = true + continue + } + breaks = append(breaks, br) + } + if len(breaks) > 0 { + result.Breaks[clip.episode.ID] = breaks + } + } + return result, nil +} + +func breakOf(startSamples, endSamples int, score float64, rate int) AdBreak { + return AdBreak{ + StartMs: int64(startSamples) * 1000 / int64(rate), + EndMs: int64(endSamples) * 1000 / int64(rate), + Partners: 1, + Score: score, + } +} + +// merge folds overlapping and touching matches into single breaks, chronologically. +// +// The same advertisement is found once per partner that carries it, so without merging an ad in +// three episodes becomes three identical breaks. The partner count is what survives: it says how +// much agreement a break rests on. +func merge(raw []AdBreak) []AdBreak { + if len(raw) == 0 { + return nil + } + sort.Slice(raw, func(i, j int) bool { return raw[i].StartMs < raw[j].StartMs }) + + out := make([]AdBreak, 0, len(raw)) + current := raw[0] + for _, next := range raw[1:] { + if next.StartMs <= current.EndMs { + if next.EndMs > current.EndMs { + current.EndMs = next.EndMs + } + current.Partners++ + if next.Score < current.Score { + current.Score = next.Score + } + continue + } + out = append(out, current) + current = next + } + return append(out, current) +} diff --git a/server/internal/skips/ads_test.go b/server/internal/skips/ads_test.go new file mode 100644 index 0000000..476b819 --- /dev/null +++ b/server/internal/skips/ads_test.go @@ -0,0 +1,316 @@ +package skips + +import ( + "context" + "math/rand" + "testing" + + "github.com/terst/listen/server/internal/decode" +) + +/* +Detecting the advertising rather than the intro. + +This is the inversion the data forced. Intro detection on this server reported a segment +confirmed on 8 of 8 Radiolab episodes, passed its own self-check, and was wrong: the segment was +searched for in an ad-free copy of the same episode fetched hours earlier and scored r = 0.0621. +It was in the ad break. + +The cause is that mirroring 8 episodes within one minute makes them maximally dependent -- the +ad server inserts whatever campaign is running into all of them, identically. That destroys the +independence the confirmation threshold needs, and no amount of self-checking recovers it, +because the segments genuinely are the same audio. + +Turned around, the same property is a reliable advertisement detector: what simultaneous fetches +share *is* the inserted advertising. And the declared-versus-served length gap is a free, +independent measure of how much of it there should be -- so unlike intro detection, this can be +checked against something the detector did not produce. +*/ + +const adRate = 250 + +type adStub struct { + // whole holds each episode's audio at adRate, keyed by path. + whole map[string][]float64 + durations map[string]int64 + requests []int +} + +func (s *adStub) RangeAt(_ context.Context, path string, startMs, durationMs int64, rate int) (*decode.Clip, error) { + s.requests = append(s.requests, rate) + all := s.whole[path] + from := int(startMs) * adRate / 1000 + to := len(all) + if durationMs > 0 { + to = from + int(durationMs)*adRate/1000 + } + if from > len(all) { + from = len(all) + } + if to > len(all) { + to = len(all) + } + return &decode.Clip{Samples: all[from:to], SampleRate: adRate, StartMs: startMs}, nil +} + +func (s *adStub) DurationMs(_ context.Context, path string) (int64, error) { + return s.durations[path], nil +} + +func adNoise(seed int64, n int) []float64 { + rng := rand.New(rand.NewSource(seed)) + out := make([]float64, n) + for i := range out { + out[i] = rng.NormFloat64() * 50 + } + return out +} + +func put(base, segment []float64, at int) []float64 { + out := append([]float64(nil), base...) + copy(out[at:], segment) + return out +} + +// adConfig matches the stub's rate and keeps the thresholds production uses. +func adConfig() AdConfig { + cfg := DefaultAdConfig() + cfg.DiscoveryRate = adRate + return cfg +} + +func TestFindsAnAdSharedByTwoEpisodes(t *testing.T) { + ad := adNoise(900, 10*adRate) // ten seconds + stub := &adStub{ + whole: map[string][]float64{ + "a": put(adNoise(1, 300*adRate), ad, 20*adRate), + "b": put(adNoise(2, 300*adRate), ad, 50*adRate), + "c": adNoise(3, 300*adRate), + }, + durations: map[string]int64{"a": 300_000, "b": 300_000, "c": 300_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}, {ID: "e3", Path: "c"}} + + got, err := DetectAds(context.Background(), stub, eps, adConfig()) + if err != nil { + t.Fatal(err) + } + + if len(got.Breaks["e1"]) != 1 { + t.Fatalf("e1 breaks %+v, want 1", got.Breaks["e1"]) + } + br := got.Breaks["e1"][0] + if br.StartMs < 19_000 || br.StartMs > 21_000 { + t.Errorf("e1 break starts at %dms, want about 20000", br.StartMs) + } + if br.EndMs < 29_000 || br.EndMs > 31_000 { + t.Errorf("e1 break ends at %dms, want about 30000", br.EndMs) + } + // And the same audio located in the other episode, at its own position. + if len(got.Breaks["e2"]) != 1 { + t.Fatalf("e2 breaks %+v, want 1", got.Breaks["e2"]) + } + if s := got.Breaks["e2"][0].StartMs; s < 49_000 || s > 51_000 { + t.Errorf("e2 break starts at %dms, want about 50000", s) + } + // The episode sharing nothing gets nothing. + if len(got.Breaks["e3"]) != 0 { + t.Errorf("e3 breaks %+v, want none", got.Breaks["e3"]) + } +} + +func TestFindsSeveralAdBreaksInOneEpisode(t *testing.T) { + first := adNoise(901, 10*adRate) + second := adNoise(902, 12*adRate) + stub := &adStub{ + whole: map[string][]float64{ + "a": put(put(adNoise(11, 400*adRate), first, 15*adRate), second, 200*adRate), + "b": put(put(adNoise(12, 400*adRate), first, 60*adRate), second, 300*adRate), + }, + durations: map[string]int64{"a": 400_000, "b": 400_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}} + + got, err := DetectAds(context.Background(), stub, eps, adConfig()) + if err != nil { + t.Fatal(err) + } + + if len(got.Breaks["e1"]) != 2 { + t.Fatalf("e1 breaks %+v, want 2", got.Breaks["e1"]) + } + // Earliest first, because that is the order playback consumes them in. + if got.Breaks["e1"][0].StartMs > got.Breaks["e1"][1].StartMs { + t.Error("breaks are not in chronological order") + } +} + +/* +The free ground truth, and the reason this is checkable where intro detection was not. The feed +declares the ad-free length and the server was served something longer; the difference is the +advertising. Whatever the scan finds should add up to roughly that much. +*/ +func TestReportsFoundAgainstTheLengthGap(t *testing.T) { + ad := adNoise(903, 10*adRate) + stub := &adStub{ + whole: map[string][]float64{ + "a": put(adNoise(21, 300*adRate), ad, 20*adRate), + "b": put(adNoise(22, 300*adRate), ad, 50*adRate), + }, + durations: map[string]int64{"a": 300_000, "b": 300_000}, + } + eps := []Episode{ + {ID: "e1", Path: "a", DeclaredDurationMs: 290_000}, + {ID: "e2", Path: "b", DeclaredDurationMs: 290_000}, + } + + got, err := DetectAds(context.Background(), stub, eps, adConfig()) + if err != nil { + t.Fatal(err) + } + + acc := got.Accounting["e1"] + if acc.InsertedMs != 10_000 { + t.Errorf("inserted %dms, want 10000 (300s served, 290s declared)", acc.InsertedMs) + } + if acc.FoundMs < 9_000 || acc.FoundMs > 11_000 { + t.Errorf("found %dms, want about 10000", acc.FoundMs) + } +} + +// Nothing declared means nothing to check against, which must not read as "found nothing". +func TestAnUnknownDeclaredLengthReportsNoInsertedFigure(t *testing.T) { + ad := adNoise(904, 10*adRate) + stub := &adStub{ + whole: map[string][]float64{ + "a": put(adNoise(31, 300*adRate), ad, 20*adRate), + "b": put(adNoise(32, 300*adRate), ad, 50*adRate), + }, + durations: map[string]int64{"a": 300_000, "b": 300_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}} + + got, err := DetectAds(context.Background(), stub, eps, adConfig()) + if err != nil { + t.Fatal(err) + } + if got.Accounting["e1"].InsertedMs != 0 { + t.Errorf("inserted %dms, want 0 for an unknown declared length", got.Accounting["e1"].InsertedMs) + } +} + +/* +The guard that matters most. Two episodes of a show can legitimately share a great deal -- a +re-aired episode, a two-parter with a long recap -- and skipping all of it would eat the +programme. An episode whose matches exceed the ceiling is reported and published as nothing. +*/ +func TestRefusesToSkipMostOfAnEpisode(t *testing.T) { + // Two episodes that are the same recording apart from the first ten seconds. + body := adNoise(905, 300*adRate) + stub := &adStub{ + whole: map[string][]float64{ + "a": put(append([]float64(nil), body...), adNoise(41, 10*adRate), 0), + "b": put(append([]float64(nil), body...), adNoise(42, 10*adRate), 0), + }, + durations: map[string]int64{"a": 300_000, "b": 300_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}} + + got, err := DetectAds(context.Background(), stub, eps, adConfig()) + if err != nil { + t.Fatal(err) + } + + if len(got.Breaks["e1"]) != 0 { + t.Errorf("published %+v for an episode that is almost entirely shared", got.Breaks["e1"]) + } + if !got.Refused["e1"] { + t.Error("did not record that e1 was refused") + } +} + +// A single match longer than any plausible ad pod is a sign of shared programme, not advertising. +func TestRefusesASingleImplausiblyLongMatch(t *testing.T) { + shared := adNoise(906, 200*adRate) // over three minutes + stub := &adStub{ + whole: map[string][]float64{ + "a": put(adNoise(51, 600*adRate), shared, 50*adRate), + "b": put(adNoise(52, 600*adRate), shared, 300*adRate), + }, + durations: map[string]int64{"a": 600_000, "b": 600_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}} + + cfg := adConfig() + cfg.MaxBreakMs = 180_000 + got, err := DetectAds(context.Background(), stub, eps, cfg) + if err != nil { + t.Fatal(err) + } + + for _, br := range got.Breaks["e1"] { + if br.EndMs-br.StartMs > cfg.MaxBreakMs { + t.Errorf("published a %dms break over the %dms cap", br.EndMs-br.StartMs, cfg.MaxBreakMs) + } + } +} + +func TestOverlappingMatchesAreMergedIntoOneBreak(t *testing.T) { + ad := adNoise(907, 20*adRate) + // Three episodes all carrying the same ad: e1 will match it via both partners, and the two + // findings must not become two breaks. + stub := &adStub{ + whole: map[string][]float64{ + "a": put(adNoise(61, 300*adRate), ad, 30*adRate), + "b": put(adNoise(62, 300*adRate), ad, 80*adRate), + "c": put(adNoise(63, 300*adRate), ad, 150*adRate), + }, + durations: map[string]int64{"a": 300_000, "b": 300_000, "c": 300_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}, {ID: "e3", Path: "c"}} + + got, err := DetectAds(context.Background(), stub, eps, adConfig()) + if err != nil { + t.Fatal(err) + } + + if len(got.Breaks["e1"]) != 1 { + t.Errorf("e1 breaks %+v, want them merged into 1", got.Breaks["e1"]) + } + if got.Breaks["e1"][0].Partners != 2 { + t.Errorf("partners %d, want 2", got.Breaks["e1"][0].Partners) + } +} + +// Discovery happens at the low rate, which is what makes whole episodes affordable at all. +func TestDiscoversAtTheLowRate(t *testing.T) { + stub := &adStub{ + whole: map[string][]float64{"a": adNoise(71, 300*adRate), "b": adNoise(72, 300*adRate)}, + durations: map[string]int64{"a": 300_000, "b": 300_000}, + } + eps := []Episode{{ID: "e1", Path: "a"}, {ID: "e2", Path: "b"}} + + if _, err := DetectAds(context.Background(), stub, eps, adConfig()); err != nil { + t.Fatal(err) + } + for _, rate := range stub.requests { + if rate != adRate { + t.Errorf("decoded at %dHz, want everything at the discovery rate %d", rate, adRate) + } + } +} + +func TestOneEpisodeHasNothingToCompareAgainst(t *testing.T) { + stub := &adStub{ + whole: map[string][]float64{"a": adNoise(81, 300*adRate)}, + durations: map[string]int64{"a": 300_000}, + } + + got, err := DetectAds(context.Background(), stub, []Episode{{ID: "e1", Path: "a"}}, adConfig()) + if err != nil { + t.Fatal(err) + } + if len(got.Breaks) != 0 { + t.Errorf("found breaks with a single episode: %+v", got.Breaks) + } +} diff --git a/server/internal/skips/skips.go b/server/internal/skips/skips.go index a7ae808..4e644b0 100644 --- a/server/internal/skips/skips.go +++ b/server/internal/skips/skips.go @@ -76,6 +76,12 @@ type Decoder interface { type Episode struct { ID string Path string + // DeclaredDurationMs is what the feed claims the episode is. + // + // Carried because the difference between it and the file's real length is how much + // advertising was inserted -- a measure that comes from the feed and the file rather than + // from the detector, and so the only number here that can contradict a scan. + DeclaredDurationMs int64 } // Offsets is what playback needs for one episode: how much to skip at each end. diff --git a/server/internal/store/store.go b/server/internal/store/store.go index cfe6f40..a67e637 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -34,11 +34,34 @@ type Episode struct { IntroMs *int64 `json:"introMs,omitempty"` OutroMs *int64 `json:"outroMs,omitempty"` + // Breaks are the stretches of this episode that other episodes in the batch also carry, + // which for simultaneously-mirrored episodes is the advertising inserted into them. + Breaks []Break `json:"breaks,omitempty"` + + // FoundAdMs and InsertedAdMs are what the scan accounted for against what the feed's + // declared length implies is there. Kept because the second number does not come from the + // detector and so is the only thing able to contradict it. + FoundAdMs int64 `json:"foundAdMs,omitempty"` + InsertedAdMs int64 `json:"insertedAdMs,omitempty"` + + // AdsRefused records that matches were found but were too extensive to be advertising, so + // nothing was published. A different state from nothing having matched, and worth keeping + // distinct -- conflating the two is the mistake the app's skip readout already made. + AdsRefused bool `json:"adsRefused,omitempty"` + // DetectedAtUnix records when detection last ran over the show this belongs to, so a // result can be told apart from never having looked. DetectedAtUnix int64 `json:"detectedAtUnix,omitempty"` } +// Break is one stretch of an episode to skip. +type Break struct { + StartMs int64 `json:"startMs"` + EndMs int64 `json:"endMs"` + // Partners is how many other episodes were found to carry this audio. + Partners int `json:"partners,omitempty"` +} + // Mirrored reports whether this episode's audio is on disk. func (e *Episode) Mirrored() bool { return e.Path != "" } |