diff --git a/backend/audiocache.go b/backend/audiocache.go index ee9141d..fb5ba4a 100644 --- a/backend/audiocache.go +++ b/backend/audiocache.go @@ -53,6 +53,9 @@ func NewAudioCache(ctx context.Context, s *ServerManager, baseCacheDir string) ( // PathForCachedFile returns the local filesystem path for a cached track, // if the file has finished downloading. If not cached, it returns an empty string. func (a *AudioCache) PathForCachedFile(id string) string { + a.mutex.Lock() + defer a.mutex.Unlock() + if item, ok := a.entries[id]; ok && item.done { return a.pathForID(id) } @@ -63,20 +66,29 @@ func (a *AudioCache) PathForCachedFile(id string) string { // including one that is in the process of downloading. // If it is not cached or downloading, it returns an empty string. func (a *AudioCache) PathForCachedOrDownloadingFile(id string) string { + a.mutex.Lock() + defer a.mutex.Unlock() + if _, ok := a.entries[id]; ok { return a.pathForID(id) } return "" } -// cacheFile begins downloading a file (if not already downloading) and stores it +// CacheFile begins downloading a file (if not already downloading) and stores it // to the cache directory under its ID as filename. The download is asynchronous. -func (a *AudioCache) cacheFile(id, dlURL string) { +func (a *AudioCache) CacheFile(id, dlURL string) { s := a.s.Server if s == nil { return } + a.mutex.Lock() + defer a.mutex.Unlock() + a.cacheFile(id, dlURL) +} + +func (a *AudioCache) cacheFile(id, dlURL string) { if _, ok := a.entries[id]; !ok { ctx, cancel := context.WithCancel(a.rootCtx) a.entries[id] = &cacheEntry{cancel: cancel} @@ -107,9 +119,10 @@ func (a *AudioCache) CacheOnly(keep string, fetch []AudioCacheRequest) { if id != keep && !slices.ContainsFunc(fetch, func(a AudioCacheRequest) bool { return a.ID == id }) { - e.cancel() - _ = os.Remove(a.pathForID(id)) - delete(a.entries, id) + _ = e + //e.cancel() + //_ = os.Remove(a.pathForID(id)) + //delete(a.entries, id) } } diff --git a/backend/playbackengine.go b/backend/playbackengine.go index 943c36c..1d5b109 100644 --- a/backend/playbackengine.go +++ b/backend/playbackengine.go @@ -565,7 +565,10 @@ func (p *playbackEngine) cacheNextTracks() { if p.audiocache != nil { // fetch up to the 2 next tracks in the queue to the cache fetch := make([]AudioCacheRequest, 0, 3) - for _, idx := range [3]int{p.nowPlayingIdx, p.nowPlayingIdx + 1, p.nowPlayingIdx + 2} { + // if nothing is playing (index = -1), treat the beginning of the queue as + // the "currently" playing track, since we're probably about to play it + npI := max(p.nowPlayingIdx, 0) + for _, idx := range [3]int{npI, npI + 1, npI + 2} { if idx > 0 && idx < len(p.playQueue) { item := p.playQueue[idx] if item.Metadata().Type == mediaprovider.MediaItemTypeTrack { @@ -581,6 +584,7 @@ func (p *playbackEngine) cacheNextTracks() { id = np.Metadata().ID } p.audiocache.CacheOnly(id, fetch) + log.Println("fetching files", sharedutil.MapSlice(fetch, func(a AudioCacheRequest) string { return a.ID })) } } @@ -663,12 +667,20 @@ func (p *playbackEngine) nextPlayingIndex() int { } func (p *playbackEngine) setTrack(idx int, next bool, startTime float64) error { + var item mediaprovider.MediaItem + var url string + if idx >= 0 { + item = p.playQueue[idx] + url = p.getMediaURLForIdx(idx) + } + track, isTrack := item.(*mediaprovider.Track) + if p.audiocache != nil && isTrack { + p.audiocache.CacheFile(item.Metadata().ID, p.getMediaURLForIdx(idx)) + } + if urlP, ok := p.player.(player.URLPlayer); ok { - url := "" var meta mediaprovider.MediaItemMetadata if idx >= 0 { - item := p.playQueue[idx] - track, isTrack := item.(*mediaprovider.Track) meta = item.Metadata() if isTrack && p.audiocache != nil { if filepath := p.audiocache.PathForCachedFile(track.ID); filepath != "" { @@ -676,9 +688,6 @@ func (p *playbackEngine) setTrack(idx int, next bool, startTime float64) error { log.Println("playing file from cache") } } - if url == "" { - url = p.getMediaURLForIdx(idx) - } if url == "" { return errors.New("no stream URL") } diff --git a/backend/playbackmanager.go b/backend/playbackmanager.go index d18f520..a1a3a7a 100644 --- a/backend/playbackmanager.go +++ b/backend/playbackmanager.go @@ -112,7 +112,6 @@ func (p *PlaybackManager) addOnTrackChangeHook() { updateUnfinishedJob := func(job *WaveformImageJob) { ctx, c := context.WithCancel(p.cache.rootCtx) refreshCancel = c - log.Println("starting img update func") go func(ctx context.Context, job *WaveformImageJob) { for { time.Sleep(333 * time.Millisecond) @@ -120,7 +119,7 @@ func (p *PlaybackManager) addOnTrackChangeHook() { case <-ctx.Done(): return default: - log.Println("updating waveform img") + log.Println("updating waveform img, job step", job.step) img := job.Get() for _, cb := range p.onWaveformImgUpdate { cb(img) @@ -138,6 +137,7 @@ func (p *PlaybackManager) addOnTrackChangeHook() { var im *WaveformImage done := false if nextWaveformJob.ItemID == item.Metadata().ID { + log.Println("Have waveform in progress for", item.Metadata().ID) done = nextWaveformJob.Done() im = nextWaveformJob.Get() } diff --git a/backend/util/filestreamer.go b/backend/util/filestreamer.go index 545ed51..8f015fc 100644 --- a/backend/util/filestreamer.go +++ b/backend/util/filestreamer.go @@ -32,11 +32,8 @@ func NewFileStreamerServer(path string, isComplete func() bool) (*FileStreamerSe done: make(chan struct{}), } - mux := http.NewServeMux() - mux.HandleFunc("/stream", fs.streamHandler) - fs.server = &http.Server{ - Handler: mux, + Handler: handler{fs}, } return fs, nil @@ -45,7 +42,7 @@ func NewFileStreamerServer(path string, isComplete func() bool) (*FileStreamerSe // Addr returns the server address (host:port). func (fs *FileStreamerServer) Addr() string { _, port, _ := net.SplitHostPort(fs.listener.Addr().String()) - return "http://localhost:" + port + "/stream" + return "http://127.0.0.1:" + port + "/" } // Serve starts serving and waits for a single request to complete. @@ -54,6 +51,7 @@ func (fs *FileStreamerServer) Serve() error { _ = fs.server.Serve(fs.listener) }() + log.Println("Serving and WAITING for done") <-fs.done // wait for the handler to finish ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) @@ -64,10 +62,15 @@ func (fs *FileStreamerServer) Serve() error { // Handler that streams the file using chunked transfer encoding. func (fs *FileStreamerServer) streamHandler(w http.ResponseWriter, r *http.Request) { + log.Println("FILE STREAMER REQUEST") defer close(fs.done) // signal Serve() to shut down after this request + totalWrote := 0 + defer log.Println("File streamer wrote", totalWrote, "bytes") + file, err := os.Open(fs.Path) if err != nil { + log.Println("File streamer failed to open source file") http.Error(w, "could not open file", http.StatusInternalServerError) return } @@ -81,6 +84,7 @@ func (fs *FileStreamerServer) streamHandler(w http.ResponseWriter, r *http.Reque buf := make([]byte, 4096) for { + complete := fs.IsComplete() n, err := file.Read(buf) if err != nil && err != io.EOF { log.Printf("read error: %v", err) @@ -88,7 +92,7 @@ func (fs *FileStreamerServer) streamHandler(w http.ResponseWriter, r *http.Reque } if n > 0 { - _, err := w.Write(buf[:n]) + written, err := w.Write(buf[:n]) if err != nil { log.Printf("client write error: %v", err) break @@ -96,9 +100,10 @@ func (fs *FileStreamerServer) streamHandler(w http.ResponseWriter, r *http.Reque if canFlush { flusher.Flush() } + totalWrote += written } - if n == 0 && fs.IsComplete() { + if n == 0 && complete { break } @@ -108,3 +113,13 @@ func (fs *FileStreamerServer) streamHandler(w http.ResponseWriter, r *http.Reque } } } + +type handler struct { + fs *FileStreamerServer +} + +var _ http.Handler = handler{} + +func (h handler) ServeHTTP(w http.ResponseWriter, req *http.Request) { + h.fs.streamHandler(w, req) +} diff --git a/backend/waveformimage.go b/backend/waveformimage.go index e4dd45b..82bddb6 100644 --- a/backend/waveformimage.go +++ b/backend/waveformimage.go @@ -15,10 +15,10 @@ import ( "time" "github.com/dweymouth/supersonic/backend/mediaprovider" - "github.com/dweymouth/supersonic/backend/util" "github.com/go-audio/audio" "github.com/go-audio/wav" "github.com/supersonic-app/go-mpv" + "golang.org/x/sys/unix" ) type WaveformImageGenerator struct { @@ -39,10 +39,13 @@ type WaveformImageJob struct { progress int // first invalid pixel in X direction done bool cancel func() + + step int } func (w *WaveformImageJob) Cancel() { if w != nil && w.cancel != nil { + w.step = -1 w.cancel() } } @@ -101,13 +104,25 @@ func (w *WaveformImageGenerator) StartWaveformGeneration(item *mediaprovider.Tra path := w.audioCache.PathForCachedOrDownloadingFile(job.ItemID) // wait for file to begin downloading if not already for path == "" { - time.Sleep(10 * time.Millisecond) + time.Sleep(50 * time.Millisecond) if e := ctx.Err(); e != nil { job.setError(e) return } path = w.audioCache.PathForCachedOrDownloadingFile(job.ItemID) } + // and wait for content to begin being written + for { + if s, err := os.Stat(path); err == nil && s.Size() > 0 { + break + } + time.Sleep(50 * time.Millisecond) + if e := ctx.Err(); e != nil { + job.setError(e) + return + } + } + job.step = 1 dir := filepath.Dir(path) transcodeFile := filepath.Join(dir, filepath.Base(path)+"_waveform.wav") @@ -117,25 +132,22 @@ func (w *WaveformImageGenerator) StartWaveformGeneration(item *mediaprovider.Tra } // If file isn't fully downloaded from server, - // stream it to MPV via HTTP so it doesn't possibly + // stream it to MPV via fifo so it doesn't possibly // terminate the conversion to WAV early encountering EOF if !fileDone() { - srv, err := util.NewFileStreamerServer(path, fileDone) - if err != nil { - job.setError(err) - return - } - path = srv.Addr() - log.Println("streaming file to MPV at ", path) - go srv.Serve() + fifoPath := filepath.Join(filepath.Dir(path), filepath.Base(path)+"_fifo") + copyFileToFifo(ctx, job, path, fifoPath, fileDone) + path = fifoPath // MPV will read from the FIFO } // Start converting the file to WAV for analysis var wavConvertDone bool go func() { + job.step = 2 err := convertToWav(ctx, path, transcodeFile) wavConvertDone = true if err != nil { + log.Println("Error converting to wav", err) job.setError(err) } }() @@ -145,12 +157,13 @@ func (w *WaveformImageGenerator) StartWaveformGeneration(item *mediaprovider.Tra if s, err := os.Stat(transcodeFile); err == nil && s.Size() > 0 { break } - time.Sleep(10 * time.Millisecond) + time.Sleep(50 * time.Millisecond) if e := ctx.Err(); e != nil { job.setError(e) return } } + job.step = 3 // Start analyzing the converted wav file data := &waveformData{} @@ -160,10 +173,14 @@ func (w *WaveformImageGenerator) StartWaveformGeneration(item *mediaprovider.Tra log.Println("error analyzing wav", err.Error()) job.setError(err) } + data.done = true }() // Start generating the waveform image - go generateWaveformImage(ctx, data, job) + go func() { + generateWaveformImage(ctx, data, job) + job.done = true + }() }() return job } @@ -177,8 +194,6 @@ type waveformData struct { } func generateWaveformImage(ctx context.Context, data *waveformData, job *WaveformImageJob) { - defer func() { job.done = true }() - centerY := job.img.Rect.Dy() / 2 // 16 top := centerY - 1 bottom := centerY @@ -194,7 +209,7 @@ func generateWaveformImage(ctx context.Context, data *waveformData, job *Wavefor if ctx.Err() != nil { return // expired } - time.Sleep(10 * time.Millisecond) + time.Sleep(50 * time.Millisecond) } rms := float64(data.RMS[x]) / 255.0 @@ -222,10 +237,73 @@ func generateWaveformImage(ctx context.Context, data *waveformData, job *Wavefor } } +func copyFileToFifo(ctx context.Context, job *WaveformImageJob, filePath, fifoPath string, fileDone func() bool) { + if err := unix.Mkfifo(fifoPath, 0600); err != nil { + job.setError(err) + return + } + + go func() { + fifo, err := os.OpenFile(fifoPath, os.O_WRONLY, os.ModeNamedPipe) + if err != nil { + job.setError(err) + return + } + file, err := os.Open(filePath) + if err != nil { + job.setError(err) + return + } + defer file.Close() + defer fifo.Close() + + buf := make([]byte, 4096) + offset := int64(0) + + for { + if e := ctx.Err(); e != nil { + job.setError(e) + return + } + + done := fileDone() + info, err := os.Stat(filePath) + if err != nil { + job.setError(err) + return + } + currentSize := info.Size() + + // If we've read everything available so far + if offset >= currentSize { + if done { + return // done + } + // wait a bit for more data to be written to file + time.Sleep(50 * time.Millisecond) + continue + } + + // Calculate how much is safe to read + toRead := currentSize - offset - 1 + if toRead > int64(cap(buf)) { + toRead = int64(cap(buf)) + } + + n, err := file.ReadAt(buf[:toRead], offset) + if n > 0 { + if _, werr := fifo.Write(buf[:n]); werr != nil { + job.setError(werr) + return + } + offset += int64(n) + } + } + }() +} + // assumes mono, 16 bit func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformData, millisecs int64, fileDone func() bool) error { - defer func() { data.done = true }() - f, err := os.Open(transcodeFile) if err != nil { log.Println("error opening transcoded file") @@ -254,14 +332,16 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat bytesPerSample := int64(2 * format.NumChannels) // 16-bit = 2 bytes per channel // file read loop - for { + doneReading := false + for !doneReading { select { case <-ctx.Done(): return ctx.Err() default: } + fileIsDone := fileDone() - if !fileDone() { + if !fileIsDone { // Check how many samples we can safely read without encountering EOF // and adjust read buffer size accordingly @@ -272,14 +352,14 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat currentSize := stat.Size() // how many bytes can we read without nearing EOF - readableBytes := currentSize - int64(samplesPerChunk)*int64(curChunk)*bytesPerSample - 16384 //buffer for safety + readableBytes := currentSize - int64(samplesPerChunk)*int64(curChunk)*bytesPerSample - 8192 //buffer for safety // Estimate how many samples we can read maxSamples := int(readableBytes / bytesPerSample) if maxSamples <= 0 { // Wait for more data to be written to file - time.Sleep(10 * time.Millisecond) + time.Sleep(50 * time.Millisecond) continue } @@ -293,13 +373,14 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat n, err := decoder.PCMBuffer(buf) if n == 0 || err == io.EOF { - if fileDone() { - break + if fileIsDone { + doneReading = true } if err == io.EOF && !fileDone() { return errors.New("WAV read got premature EOF") } - continue + } else if fileIsDone { + log.Println("read samples on done file") } if err != nil { return err @@ -320,6 +401,7 @@ func analyzeWavFile(ctx context.Context, transcodeFile string, data *waveformDat data.progress = curChunk chunkSamples = chunkSamples[:0] if curChunk >= 1024 { + doneReading = true break } } @@ -348,9 +430,10 @@ func computePeakAndRMS(chunk []float32) (peak float32, rms float32) { var sumSquares float64 peak = 0.0 for _, v := range chunk { - abs := float32(math.Abs(float64(v))) - if abs > peak { - peak = abs + if v > peak { + peak = v + } else if v < -peak { + peak = -v } sumSquares += float64(v * v) } @@ -394,14 +477,20 @@ func convertToWav(ctx context.Context, inPath, outPath string) error { case <-ctx.Done(): return ctx.Err() default: - ia := m.GetPropertyString("idle-active") - if ia == "yes" || ia == "true" { - return nil - } // use small timeout to allow detecting ctx expiry // without too much delay e := m.WaitEvent(0.05 /*timeout seconds*/) if e.Event_Id == mpv.EVENT_IDLE { + if _, err := os.Stat(outPath); os.IsNotExist(err) { + log.Printf("WARNING! file %s does not exist after MPV convert", outPath) + } + return nil + } + ia := m.GetPropertyString("idle-active") + if ia == "yes" || ia == "true" { + if _, err := os.Stat(outPath); os.IsNotExist(err) { + log.Printf("WARNING! file %s does not exist after MPV convert", outPath) + } return nil } } diff --git a/ui/widgets/waveformseekbar.go b/ui/widgets/waveformseekbar.go index 52bd07f..470fccf 100644 --- a/ui/widgets/waveformseekbar.go +++ b/ui/widgets/waveformseekbar.go @@ -45,7 +45,7 @@ func (w *WaveformSeekbar) UpdateImage(img *backend.WaveformImage) { w.img.Image = img prm, fg := w.getThemeColors() w.recolorImage(prm, fg, w.imgProgressPixel) - w.Refresh() + w.img.Refresh() } func (w *WaveformSeekbar) Refresh() {