diff options
Diffstat (limited to 'server/cmd')
| -rw-r--r-- | server/cmd/listend/main.go | 34 | ||||
| -rw-r--r-- | server/cmd/listend/shadow.go | 127 |
2 files changed, 154 insertions, 7 deletions
diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go index c500a63..a556f5e 100644 --- a/server/cmd/listend/main.go +++ b/server/cmd/listend/main.go @@ -31,6 +31,7 @@ import ( "github.com/terst/listen/server/internal/decode" "github.com/terst/listen/server/internal/feed" "github.com/terst/listen/server/internal/mirror" + "github.com/terst/listen/server/internal/signature" "github.com/terst/listen/server/internal/skips" "github.com/terst/listen/server/internal/store" ) @@ -58,15 +59,27 @@ func main() { log.Fatalf("state: %v", err) } + signaturePath := filepath.Join(*dataDir, "signatures.json") + // A damaged library is not worth stopping for: it is evidence that can be gathered again, + // unlike the state file, which holds the only record of where mirrored audio is. + signatures, err := signature.Load(signaturePath) + if err != nil { + log.Printf("signatures: %v -- starting an empty library", err) + signatures = signature.NewLibrary() + } + decoder := decode.Decoder{} worker := &worker{ - state: state, - statePath: statePath, - keep: *keep, - mirror: &mirror.Mirror{AudioDir: audioDir, Probe: decoder}, - decoder: decoder, - requests: make(chan request, 64), - wakeups: make(chan struct{}, 1), + state: state, + statePath: statePath, + signatures: signatures, + signaturePath: signaturePath, + keep: *keep, + mirror: &mirror.Mirror{AudioDir: audioDir, Probe: decoder}, + decoder: decoder, + audioDir: audioDir, + requests: make(chan request, 64), + wakeups: make(chan struct{}, 1), } server := api.New(state, worker, api.Options{ @@ -138,9 +151,13 @@ type request struct { type worker struct { state *store.State statePath string + // 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{} } @@ -358,6 +375,9 @@ func (w *worker) detect(ctx context.Context, show *store.Show, candidates []*sto if w.detectAds(ctx, show, episodes, byID) { show.RecordDetected(ids) + // After the batch is recorded, so the shadow run's occasion names the batch it + // learned from. It publishes nothing -- see runShadow. + w.runShadow(ctx, show, candidates) } } diff --git a/server/cmd/listend/shadow.go b/server/cmd/listend/shadow.go new file mode 100644 index 0000000..e9376f1 --- /dev/null +++ b/server/cmd/listend/shadow.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "fmt" + "log" + "sort" + "time" + + "github.com/terst/listen/server/internal/decode" + "github.com/terst/listen/server/internal/signature" + "github.com/terst/listen/server/internal/store" +) + +// episodeAudio decodes episodes for fingerprinting, at the rate the signature package wants. +type episodeAudio struct { + decoder decode.Decoder + pathOf func(id string) string +} + +func (a episodeAudio) Episode(ctx context.Context, id string) ([]float64, error) { + return a.decode(ctx, id, 0, 0) +} + +func (a episodeAudio) Range(ctx context.Context, id string, startMs, durationMs int64) ([]float64, error) { + return a.decode(ctx, id, startMs, durationMs) +} + +func (a episodeAudio) decode(ctx context.Context, id string, startMs, durationMs int64) ([]float64, error) { + path := a.pathOf(id) + if path == "" { + return nil, nil + } + clip, err := a.decoder.RangeAt(ctx, path, startMs, durationMs, signature.SampleRate) + if err != nil { + return nil, err + } + return clip.Samples, nil +} + +// runShadow learns from a finished detection pass and logs what the library would have cut. +// +// It publishes nothing. The library is new and its proposals have never been checked against +// what a listener actually hears, so for now it watches a few weeks of real batches and says +// what it would have done. The log is the deliverable. +func (w *worker) runShadow(ctx context.Context, show *store.Show, episodes []*store.Episode) { + if w.signatures == nil || len(episodes) == 0 { + return + } + started := time.Now() + + batch := signature.ShadowBatch{ + // The detection batch is the occasion: these episodes were mirrored within minutes of + // each other and carry whatever campaign was running then, so they are one observation. + Occasion: show.FeedURL + "@" + show.DetectedBatch, + AtUnix: time.Now().Unix(), + } + for _, e := range episodes { + ep := signature.ShadowEpisode{ID: e.ID, FeedURL: show.FeedURL} + for _, br := range e.Breaks { + ep.Breaks = append(ep.Breaks, signature.ShadowBreak{ + StartMs: br.StartMs, + EndMs: br.EndMs, + Kind: br.Kind, + Partners: br.Partners, + }) + } + batch.Episodes = append(batch.Episodes, ep) + } + + report, err := signature.NewShadow(w.signatures, audioFor(w.decoder, episodes)).Run(ctx, batch) + if err != nil { + log.Printf("%s: shadow run failed: %v", show.Title, err) + return + } + + novel := 0 + for _, p := range report.WouldCut { + if p.Novel { + novel++ + } + } + log.Printf("%s: shadow -- learned %d, reinforced %d, library %d (%d worth matching); "+ + "would cut %d, %d of them new%s, took %s", + show.Title, report.Harvested, report.Reinforced, w.signatures.Len(), + len(w.signatures.Trusted(shadowMinOccasions)), len(report.WouldCut), novel, + skippedNote(report.Skipped), time.Since(started).Round(time.Second)) + + // The novel ones are the whole point: what the library would add to what detection already + // publishes. Logged individually so they can be listened to and judged. + sort.Slice(report.WouldCut, func(i, j int) bool { return report.WouldCut[i].StartMs < report.WouldCut[j].StartMs }) + for _, p := range report.WouldCut { + if !p.Novel { + continue + } + log.Printf(" would cut %s %s-%s (%s, %d votes, seen on %d occasion(s), %d partners)", + p.EpisodeID, clock(p.StartMs), clock(p.EndMs), p.Entry.Kind, p.Votes, + p.Entry.Confidence(), p.Entry.Partners) + } + + if err := w.signatures.Save(w.signaturePath); err != nil { + log.Printf("save signatures: %v", err) + } +} + +func skippedNote(n int) string { + if n == 0 { + return "" + } + return fmt.Sprintf(", %d unreadable", n) +} + +// shadowMinOccasions is how many separate weeks a signature needs before the library would act +// on it, except for advertising detection has already corroborated. See signature.Entry.Trusted. +const shadowMinOccasions = 2 + +// audioFor supplies decoded episodes to the shadow run. +// +// 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 { + paths := make(map[string]string, len(episodes)) + for _, e := range episodes { + paths[e.ID] = e.Path + } + return episodeAudio{decoder: decoder, pathOf: func(id string) string { return paths[id] }} +} |