waveforms of first played track now generating, but still bugs with ending the task

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()
+			_ = e
-			_ = os.Remove(a.pathForID(id))
+			//e.cancel()
-			delete(a.entries, id)
+			//_ = os.Remove(a.pathForID(id))
+			//delete(a.entries, id)
 		}
 	}
 

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")
 			}

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()
 			}

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)
+}

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)
+			fifoPath := filepath.Join(filepath.Dir(path), filepath.Base(path)+"_fifo")
-			if err != nil {
+			copyFileToFifo(ctx, job, path, fifoPath, fileDone)
-				job.setError(err)
+			path = fifoPath // MPV will read from the FIFO
-				return
-			}
-			path = srv.Addr()
-			log.Println("streaming file to MPV at ", path)
-			go srv.Serve()
 		}
 
 		// 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() {
+			if fileIsDone {
-				break
+				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 v > peak {
-		if abs > peak {
+			peak = v
-			peak = abs
+		} 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
 			}
 		}

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() {