diff options
| -rw-r--r-- | server/cmd/listend/main.go | 12 | ||||
| -rw-r--r-- | server/cmd/listend/shadow.go | 13 | ||||
| -rw-r--r-- | server/internal/signature/shadow.go | 33 | ||||
| -rw-r--r-- | server/internal/signature/shadow_test.go | 52 |
4 files changed, 88 insertions, 22 deletions
diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go index a556f5e..a4400d2 100644 --- a/server/cmd/listend/main.go +++ b/server/cmd/listend/main.go @@ -77,7 +77,6 @@ func main() { keep: *keep, mirror: &mirror.Mirror{AudioDir: audioDir, Probe: decoder}, decoder: decoder, - audioDir: audioDir, requests: make(chan request, 64), wakeups: make(chan struct{}, 1), } @@ -154,12 +153,11 @@ type worker struct { // signatures is the library the shadow run learns into. Nothing it holds is published yet. signatures *signature.Library signaturePath string - keep int - mirror *mirror.Mirror - decoder decode.Decoder - audioDir string - requests chan request - wakeups chan struct{} + keep int + mirror *mirror.Mirror + decoder decode.Decoder + requests chan request + wakeups chan struct{} } // wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll. diff --git a/server/cmd/listend/shadow.go b/server/cmd/listend/shadow.go index e9376f1..7531115 100644 --- a/server/cmd/listend/shadow.go +++ b/server/cmd/listend/shadow.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "path/filepath" "sort" "time" @@ -68,7 +69,7 @@ func (w *worker) runShadow(ctx context.Context, show *store.Show, episodes []*st batch.Episodes = append(batch.Episodes, ep) } - report, err := signature.NewShadow(w.signatures, audioFor(w.decoder, episodes)).Run(ctx, batch) + report, err := signature.NewShadow(w.signatures, audioFor(w.decoder, w.mirror.AudioDir, episodes)).Run(ctx, batch) if err != nil { log.Printf("%s: shadow run failed: %v", show.Title, err) return @@ -118,10 +119,16 @@ const shadowMinOccasions = 2 // // The paths come from the batch rather than from a lookup: these are the episodes detection // just finished with, so their files are the ones it read and are certainly still there. -func audioFor(decoder decode.Decoder, episodes []*store.Episode) signature.Audio { +// +// store.Episode.Path is relative to the audio directory, which is what detection joins it with +// before handing it to the decoder. Passing it raw sent ffmpeg looking in the working directory, +// and the first real shadow run died on it. +func audioFor(decoder decode.Decoder, audioDir string, episodes []*store.Episode) signature.Audio { paths := make(map[string]string, len(episodes)) for _, e := range episodes { - paths[e.ID] = e.Path + if e.Path != "" { + paths[e.ID] = filepath.Join(audioDir, e.Path) + } } return episodeAudio{decoder: decoder, pathOf: func(id string) string { return paths[id] }} } diff --git a/server/internal/signature/shadow.go b/server/internal/signature/shadow.go index 9c2fe6b..412597c 100644 --- a/server/internal/signature/shadow.go +++ b/server/internal/signature/shadow.go @@ -1,9 +1,6 @@ package signature -import ( - "context" - "fmt" -) +import "context" // Audio supplies decoded audio at [SampleRate]. type Audio interface { @@ -89,13 +86,17 @@ func NewShadow(library *Library, audio Audio) *Shadow { // they came from, and every run would report itself as a triumph. func (s *Shadow) Run(ctx context.Context, batch ShadowBatch) (Report, error) { var report Report + // Counted per episode, not per failure: an episode that could not be read fails both the + // matching pass and the harvesting pass, and reporting it twice would overstate the damage. + unreadable := map[string]bool{} for _, ep := range batch.Episodes { if ctx.Err() != nil { + report.Skipped = len(unreadable) return report, ctx.Err() } samples, err := s.audio.Episode(ctx, ep.ID) if err != nil || len(samples) == 0 { - report.Skipped++ + unreadable[ep.ID] = true continue } for _, m := range s.library.Match(Take(samples), s.MinSightings) { @@ -112,29 +113,37 @@ func (s *Shadow) Run(ctx context.Context, batch ShadowBatch) (Report, error) { for _, ep := range batch.Episodes { if ctx.Err() != nil { + report.Skipped = len(unreadable) return report, ctx.Err() } for _, br := range ep.Breaks { - if err := s.harvest(ctx, batch, ep, br, &report); err != nil { - return report, err + // A break this cannot read is counted and stepped over. Every other per-item loop + // in this server isolates a single failure the same way, and the first real run + // lost a whole show's learning to one episode whose path was wrong. + if !s.harvest(ctx, batch, ep, br, &report) { + unreadable[ep.ID] = true + break } } } + report.Skipped = len(unreadable) return report, nil } -func (s *Shadow) harvest(ctx context.Context, batch ShadowBatch, ep ShadowEpisode, br ShadowBreak, report *Report) error { +// harvest stores one break's audio as a signature, and reports whether the episode could be +// read at all. A false means give up on this episode, not on the batch. +func (s *Shadow) harvest(ctx context.Context, batch ShadowBatch, ep ShadowEpisode, br ShadowBreak, report *Report) bool { duration := br.EndMs - br.StartMs if duration <= 0 { - return nil + return true } samples, err := s.audio.Range(ctx, ep.ID, br.StartMs, duration) if err != nil { - return fmt.Errorf("decode %s %d-%dms: %w", ep.ID, br.StartMs, br.EndMs, err) + return false } fp := Take(samples) if len(fp.Hashes) == 0 { - return nil + return true } _, isNew := s.library.Observe(Observation{ Fingerprint: fp, @@ -149,7 +158,7 @@ func (s *Shadow) harvest(ctx context.Context, batch ShadowBatch, ep ShadowEpisod } else { report.Reinforced++ } - return nil + return true } // kindFor maps what detection called a break onto what the library stores. diff --git a/server/internal/signature/shadow_test.go b/server/internal/signature/shadow_test.go index 3e6336f..f6c6676 100644 --- a/server/internal/signature/shadow_test.go +++ b/server/internal/signature/shadow_test.go @@ -2,6 +2,7 @@ package signature_test import ( "context" + "errors" "testing" "github.com/terst/listen/server/internal/signature" @@ -14,6 +15,27 @@ func (f fakeAudio) Episode(_ context.Context, id string) ([]float64, error) { return f[id], nil } +// failingAudio serves one episode and refuses the rest, which is what an unreadable mirrored +// file looks like. +type failingAudio struct { + fakeAudio + refuse map[string]bool +} + +func (f failingAudio) Episode(ctx context.Context, id string) ([]float64, error) { + if f.refuse[id] { + return nil, errors.New("no such file") + } + return f.fakeAudio.Episode(ctx, id) +} + +func (f failingAudio) Range(ctx context.Context, id string, startMs, durMs int64) ([]float64, error) { + if f.refuse[id] { + return nil, errors.New("no such file") + } + return f.fakeAudio.Range(ctx, id, startMs, durMs) +} + func (f fakeAudio) Range(_ context.Context, id string, startMs, durMs int64) ([]float64, error) { all := f[id] from := int(startMs) * signature.SampleRate / 1000 @@ -181,3 +203,33 @@ func TestAShadowRunPublishesNothing(t *testing.T) { t.Fatal("the shadow run changed the breaks it was given") } } + +// One unreadable file must not cost the batch its learning. Every other per-item loop in this +// server isolates a single failure, and the first real shadow run aborted a whole show's pass +// because one episode's path was wrong. +func TestOneUnreadableEpisodeDoesNotCostTheBatchItsLearning(t *testing.T) { + ad := tone(42, seconds(20)) + cue := tone(7, seconds(8)) + audio := failingAudio{ + fakeAudio: fakeAudio{"good": planted(1, ad, 300, 60), "bad": planted(2, cue, 300, 30)}, + refuse: map[string]bool{"bad": true}, + } + l := signature.NewLibrary() + + report, err := signature.NewShadow(l, audio).Run(context.Background(), batch( + signature.ShadowEpisode{ID: "bad", + Breaks: []signature.ShadowBreak{{StartMs: 30_000, EndMs: 38_000, Kind: "ad", Partners: 3}}}, + signature.ShadowEpisode{ID: "good", + Breaks: []signature.ShadowBreak{{StartMs: 60_000, EndMs: 80_000, Kind: "ad", Partners: 4}}}, + )) + + if err != nil { + t.Fatalf("one bad file failed the whole run: %v", err) + } + if report.Harvested != 1 { + t.Fatalf("harvested %d, want the one episode that could be read", report.Harvested) + } + if report.Skipped != 1 { + t.Fatalf("skipped %d, want 1 -- an unreadable file should be counted, not silent", report.Skipped) + } +} |