wehub-resource-sync
523 lines
16 KiB
Go
523 lines
16 KiB
Go
package player
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"encoding/binary"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"math"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/gopxl/beep/v2"
|
||
)
|
||
|
||
// pcmFrameSize16 is the byte size of one stereo s16le sample frame (2 channels × 2 bytes).
|
||
const pcmFrameSize16 = 4
|
||
|
||
// pcmFrameSize32 is the byte size of one stereo f32le sample frame (2 channels × 4 bytes).
|
||
const pcmFrameSize32 = 8
|
||
|
||
// ffmpegPipeTimeout limits how long URL streams may take to produce initial PCM.
|
||
const ffmpegPipeTimeout = 15 * time.Second
|
||
|
||
// pcmFrameSize returns the byte size of one stereo sample frame for the given format.
|
||
func pcmFrameSize(f32 bool) int {
|
||
if f32 {
|
||
return pcmFrameSize32
|
||
}
|
||
return pcmFrameSize16
|
||
}
|
||
|
||
// decodePCMFrame decodes one stereo sample frame from buf into a [2]float64.
|
||
func decodePCMFrame(buf []byte, f32 bool) [2]float64 {
|
||
if f32 {
|
||
return [2]float64{
|
||
float64(math.Float32frombits(binary.LittleEndian.Uint32(buf[0:4]))),
|
||
float64(math.Float32frombits(binary.LittleEndian.Uint32(buf[4:8]))),
|
||
}
|
||
}
|
||
left := int16(binary.LittleEndian.Uint16(buf[0:2]))
|
||
right := int16(binary.LittleEndian.Uint16(buf[2:4]))
|
||
return [2]float64{float64(left) / 32768, float64(right) / 32768}
|
||
}
|
||
|
||
// streamFromReader is the shared Stream() implementation for all pipe-based
|
||
// PCM streamers. It reads frames from a buffered reader, decodes them, and
|
||
// records the first non-EOF error.
|
||
func streamFromReader(reader *bufio.Reader, samples [][2]float64, buf []byte, f32 bool, errp *error) (int, bool) {
|
||
fs := pcmFrameSize(f32)
|
||
n := 0
|
||
for i := range samples {
|
||
_, err := io.ReadFull(reader, buf[:fs])
|
||
if err != nil {
|
||
if err != io.EOF && err != io.ErrUnexpectedEOF {
|
||
*errp = err
|
||
}
|
||
break
|
||
}
|
||
samples[i] = decodePCMFrame(buf[:fs], f32)
|
||
n++
|
||
}
|
||
return n, n > 0
|
||
}
|
||
|
||
// decodeFFmpeg uses ffmpeg to decode any audio file into raw PCM,
|
||
// returning a seekable beep.StreamSeekCloser.
|
||
// bitDepth selects the output format: 16 (s16le) or 32 (f32le, lossless).
|
||
func decodeFFmpeg(path string, sr beep.SampleRate, bitDepth int) (beep.StreamSeekCloser, beep.Format, error) {
|
||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||
ext := filepath.Ext(path)
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to play %s files — install it with your package manager", ext)
|
||
}
|
||
|
||
pcmFmt, codec, precision := ffmpegPCMArgs(bitDepth)
|
||
cmd := exec.Command("ffmpeg",
|
||
"-i", path,
|
||
"-f", pcmFmt,
|
||
"-acodec", codec,
|
||
"-ar", strconv.Itoa(int(sr)),
|
||
"-ac", "2",
|
||
"-loglevel", "error",
|
||
"pipe:1",
|
||
)
|
||
|
||
out, err := cmd.Output()
|
||
if err != nil {
|
||
// cmd.Output captures stderr into ExitError.Stderr; surface it since
|
||
// ffmpeg writes the actual failure reason there (-loglevel error).
|
||
var ee *exec.ExitError
|
||
if errors.As(err, &ee) && len(ee.Stderr) > 0 {
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg decode: %w: %s", err, bytes.TrimSpace(ee.Stderr))
|
||
}
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg decode: %w", err)
|
||
}
|
||
|
||
format := beep.Format{
|
||
SampleRate: sr,
|
||
NumChannels: 2,
|
||
Precision: precision,
|
||
}
|
||
|
||
return &pcmStreamer{data: out, f32: bitDepth == 32}, format, nil
|
||
}
|
||
|
||
// pcmStreamer wraps raw stereo PCM data (s16le or f32le) as a beep.StreamSeekCloser.
|
||
type pcmStreamer struct {
|
||
data []byte
|
||
pos int // current sample frame index
|
||
f32 bool // true = f32le (32-bit float), false = s16le (16-bit int)
|
||
}
|
||
|
||
func (p *pcmStreamer) Stream(samples [][2]float64) (int, bool) {
|
||
fs := pcmFrameSize(p.f32)
|
||
totalFrames := len(p.data) / fs
|
||
|
||
if p.pos >= totalFrames {
|
||
return 0, false
|
||
}
|
||
|
||
n := 0
|
||
for i := range samples {
|
||
if p.pos >= totalFrames {
|
||
break
|
||
}
|
||
off := p.pos * fs
|
||
samples[i] = decodePCMFrame(p.data[off:off+fs], p.f32)
|
||
p.pos++
|
||
n++
|
||
}
|
||
return n, true
|
||
}
|
||
|
||
func (p *pcmStreamer) Err() error { return nil }
|
||
|
||
func (p *pcmStreamer) Len() int {
|
||
return len(p.data) / pcmFrameSize(p.f32)
|
||
}
|
||
|
||
func (p *pcmStreamer) Position() int {
|
||
return p.pos
|
||
}
|
||
|
||
func (p *pcmStreamer) Seek(pos int) error {
|
||
if pos < 0 || pos > p.Len() {
|
||
return fmt.Errorf("seek position %d out of range [0, %d]", pos, p.Len())
|
||
}
|
||
p.pos = pos
|
||
return nil
|
||
}
|
||
|
||
func (p *pcmStreamer) Close() error {
|
||
p.data = nil
|
||
return nil
|
||
}
|
||
|
||
// decodeFFmpegStream starts ffmpeg as a subprocess and streams PCM data
|
||
// incrementally from its stdout pipe. Unlike decodeFFmpeg, this does not
|
||
// wait for the entire input to be read — suitable for live/infinite streams.
|
||
func decodeFFmpegStream(path string, sr beep.SampleRate, bitDepth int) (*ffmpegPipeStreamer, beep.Format, error) {
|
||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||
ext := filepath.Ext(path)
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to play %s files — install it with your package manager", ext)
|
||
}
|
||
fp, format, err := startFFmpegPipe(path, nil, sr, bitDepth)
|
||
if err != nil {
|
||
return nil, beep.Format{}, err
|
||
}
|
||
return &ffmpegPipeStreamer{ffmpegPipe: fp}, format, nil
|
||
}
|
||
|
||
// startFFmpegPipe launches ffmpeg transcoding input to raw PCM on stdout and
|
||
// returns an ffmpegPipe reading that stdout. input is the ffmpeg -i argument
|
||
// (a URL/path, or "pipe:0" when feeding via stdin); stdin, when non-nil, is
|
||
// wired to the process. Callers add the concrete Seek behavior by embedding the
|
||
// returned ffmpegPipe in a streamer type.
|
||
func startFFmpegPipe(input string, stdin io.ReadCloser, sr beep.SampleRate, bitDepth int) (ffmpegPipe, beep.Format, error) {
|
||
pcmFmt, codec, precision := ffmpegPCMArgs(bitDepth)
|
||
cmd := exec.Command("ffmpeg",
|
||
"-i", input,
|
||
"-f", pcmFmt,
|
||
"-acodec", codec,
|
||
"-ar", strconv.Itoa(int(sr)),
|
||
"-ac", "2",
|
||
"-loglevel", "error",
|
||
"pipe:1",
|
||
)
|
||
cmd.Stdin = stdin
|
||
|
||
pipe, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
return ffmpegPipe{}, beep.Format{}, fmt.Errorf("ffmpeg stdout pipe: %w", err)
|
||
}
|
||
if err := cmd.Start(); err != nil {
|
||
return ffmpegPipe{}, beep.Format{}, fmt.Errorf("ffmpeg start: %w", err)
|
||
}
|
||
|
||
fp := ffmpegPipe{cmd: cmd, reader: bufio.NewReaderSize(pipe, pipeBufSize), pipe: pipe, src: stdin, f32: bitDepth == 32}
|
||
format := beep.Format{SampleRate: sr, NumChannels: 2, Precision: precision}
|
||
return fp, format, nil
|
||
}
|
||
|
||
// ffmpegPipe holds the common state and methods shared by all pipe-based
|
||
// ffmpeg streamers. Each concrete streamer embeds this and adds its own
|
||
// Seek (and optionally start) implementation.
|
||
type ffmpegPipe struct {
|
||
cmd *exec.Cmd
|
||
reader *bufio.Reader
|
||
pipe io.ReadCloser
|
||
src io.ReadCloser // optional stdin source (e.g. ICY-wrapped body); closed on stop
|
||
buf [pcmFrameSize32]byte // large enough for both 16-bit and 32-bit frames
|
||
f32 bool // true = f32le, false = s16le
|
||
err error
|
||
pos int // current sample frame position
|
||
total int // total frames (0 if unknown/unbounded)
|
||
}
|
||
|
||
func (f *ffmpegPipe) Stream(samples [][2]float64) (int, bool) {
|
||
n, ok := streamFromReader(f.reader, samples, f.buf[:], f.f32, &f.err)
|
||
f.pos += n
|
||
return n, ok
|
||
}
|
||
|
||
func (f *ffmpegPipe) Err() error { return f.err }
|
||
func (f *ffmpegPipe) Len() int { return f.total }
|
||
func (f *ffmpegPipe) Position() int { return f.pos }
|
||
|
||
// stop kills the running ffmpeg process and cleans up. When stdin is fed from
|
||
// src, src is closed first: os/exec runs a goroutine copying src -> ffmpeg
|
||
// stdin, and cmd.Wait() blocks until it returns. For an infinite radio stream
|
||
// that goroutine is parked in src.Read, so src must be closed to unblock it
|
||
// before Wait, otherwise stop hangs.
|
||
func (f *ffmpegPipe) stop() {
|
||
src := f.src
|
||
pipe := f.pipe
|
||
cmd := f.cmd
|
||
f.src = nil
|
||
f.pipe = nil
|
||
f.cmd = nil
|
||
|
||
if src != nil {
|
||
src.Close()
|
||
}
|
||
if pipe != nil {
|
||
pipe.Close()
|
||
}
|
||
if cmd != nil && cmd.Process != nil {
|
||
cmd.Process.Kill()
|
||
cmd.Wait()
|
||
}
|
||
}
|
||
|
||
func (f *ffmpegPipe) Close() error {
|
||
f.stop()
|
||
return nil
|
||
}
|
||
|
||
func (f *ffmpegPipe) bitDepth() int {
|
||
if f.f32 {
|
||
return 32
|
||
}
|
||
return 16
|
||
}
|
||
|
||
// waitForInitialAudio waits until ffmpeg has produced at least one PCM byte.
|
||
// This runs before a URL stream is handed to the speaker, so an idle or broken
|
||
// live stream cannot park the audio goroutine in Read and block future swaps.
|
||
func (f *ffmpegPipe) waitForInitialAudio(timeout time.Duration) error {
|
||
peekErr := make(chan error, 1)
|
||
go func() {
|
||
_, err := f.reader.Peek(1)
|
||
peekErr <- err
|
||
}()
|
||
|
||
timer := time.NewTimer(timeout)
|
||
defer timer.Stop()
|
||
|
||
select {
|
||
case err := <-peekErr:
|
||
if err != nil {
|
||
f.stop()
|
||
return fmt.Errorf("waiting for audio data: %w", err)
|
||
}
|
||
return nil
|
||
case <-timer.C:
|
||
f.stop()
|
||
<-peekErr // drain after stop unblocks the pipe reader
|
||
return fmt.Errorf("timed out waiting for audio data (%v)", timeout)
|
||
}
|
||
}
|
||
|
||
// ffmpegPipeStreamer reads PCM data incrementally from a running ffmpeg process.
|
||
// Used for live/infinite streams where seeking is not supported.
|
||
type ffmpegPipeStreamer struct {
|
||
ffmpegPipe
|
||
}
|
||
|
||
func (f *ffmpegPipeStreamer) Seek(int) error { return nil }
|
||
|
||
// decodeFFmpegPipeStream starts ffmpeg reading from src via stdin (pipe:0)
|
||
// instead of letting ffmpeg open the URL itself. Keeping the caller's reader
|
||
// chain in the data path means the ICY metadata reader stays attached, so live
|
||
// radio StreamTitle parsing keeps working for ffmpeg-only codecs (AAC, AAC+,
|
||
// Opus, ...). src is closed when the stream stops. Used for live/infinite HTTP
|
||
// streams; seeking is not supported.
|
||
func decodeFFmpegPipeStream(src io.ReadCloser, sr beep.SampleRate, bitDepth int) (*ffmpegPipeStreamer, beep.Format, error) {
|
||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to play this stream — install it with your package manager")
|
||
}
|
||
fp, format, err := startFFmpegPipe("pipe:0", src, sr, bitDepth)
|
||
if err != nil {
|
||
return nil, beep.Format{}, err
|
||
}
|
||
return &ffmpegPipeStreamer{ffmpegPipe: fp}, format, nil
|
||
}
|
||
|
||
// decodeFFmpegLocal starts ffmpeg as a streaming pipe for local files, giving
|
||
// instant playback start instead of buffering the entire file to memory.
|
||
// Seeking is supported by killing and restarting ffmpeg with a -ss offset.
|
||
// Duration is probed via ffprobe so the seek bar works.
|
||
func decodeFFmpegLocal(path string, sr beep.SampleRate, bitDepth int) (*localFFmpegStreamer, beep.Format, error) {
|
||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||
ext := filepath.Ext(path)
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to play %s files — install it with your package manager", ext)
|
||
}
|
||
|
||
_, _, precision := ffmpegPCMArgs(bitDepth)
|
||
total := probeFrames(path, sr)
|
||
|
||
s := &localFFmpegStreamer{ffmpegPipe: ffmpegPipe{total: total, f32: bitDepth == 32}, path: path, sr: sr}
|
||
if err := s.start(0); err != nil {
|
||
return nil, beep.Format{}, err
|
||
}
|
||
|
||
format := beep.Format{
|
||
SampleRate: sr,
|
||
NumChannels: 2,
|
||
Precision: precision,
|
||
}
|
||
return s, format, nil
|
||
}
|
||
|
||
// localFFmpegStreamer streams PCM from a running ffmpeg subprocess for local
|
||
// files. Unlike pcmStreamer it does not buffer the entire file — playback
|
||
// starts as soon as ffmpeg begins producing output. Seeking kills the current
|
||
// process and restarts with -ss (demuxer-level fast seek).
|
||
type localFFmpegStreamer struct {
|
||
ffmpegPipe
|
||
path string
|
||
sr beep.SampleRate
|
||
}
|
||
|
||
// start launches ffmpeg, optionally seeking to seekPos sample frames.
|
||
func (s *localFFmpegStreamer) start(seekPos int) error {
|
||
var args []string
|
||
if seekPos > 0 {
|
||
secs := float64(seekPos) / float64(s.sr)
|
||
args = append(args, "-ss", strconv.FormatFloat(secs, 'f', 3, 64))
|
||
}
|
||
pcmFmt, codec, _ := ffmpegPCMArgs(s.bitDepth())
|
||
args = append(args,
|
||
"-i", s.path,
|
||
"-f", pcmFmt,
|
||
"-acodec", codec,
|
||
"-ar", strconv.Itoa(int(s.sr)),
|
||
"-ac", "2",
|
||
"-loglevel", "error",
|
||
"pipe:1",
|
||
)
|
||
|
||
cmd := exec.Command("ffmpeg", args...)
|
||
pipe, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
return fmt.Errorf("ffmpeg pipe: %w", err)
|
||
}
|
||
if err := cmd.Start(); err != nil {
|
||
return fmt.Errorf("ffmpeg start: %w", err)
|
||
}
|
||
|
||
s.cmd = cmd
|
||
s.pipe = pipe
|
||
s.reader = bufio.NewReaderSize(pipe, pipeBufSize)
|
||
s.pos = seekPos
|
||
s.err = nil
|
||
return nil
|
||
}
|
||
|
||
func (s *localFFmpegStreamer) Seek(pos int) error {
|
||
if pos < 0 {
|
||
pos = 0
|
||
}
|
||
if s.total > 0 && pos > s.total {
|
||
pos = s.total
|
||
}
|
||
s.stop()
|
||
return s.start(pos)
|
||
}
|
||
|
||
// decodeNavFFmpeg starts ffmpeg with the navBuffer as stdin, returning a
|
||
// navFFmpegStreamer that begins producing PCM immediately as bytes arrive.
|
||
// Seeking kills ffmpeg, repositions the navBuffer, and restarts ffmpeg from
|
||
// the new position — no HTTP reconnect required.
|
||
func decodeNavFFmpeg(nb *navBuffer, sr beep.SampleRate, bitDepth int, totalFrames int) (*navFFmpegStreamer, beep.Format, error) {
|
||
if _, err := exec.LookPath("ffmpeg"); err != nil {
|
||
return nil, beep.Format{}, fmt.Errorf("ffmpeg is required to decode this format — install it with your package manager")
|
||
}
|
||
_, _, precision := ffmpegPCMArgs(bitDepth)
|
||
s := &navFFmpegStreamer{ffmpegPipe: ffmpegPipe{total: totalFrames, f32: bitDepth == 32}, nb: nb, sr: sr}
|
||
if err := s.start(0); err != nil {
|
||
return nil, beep.Format{}, err
|
||
}
|
||
format := beep.Format{
|
||
SampleRate: sr,
|
||
NumChannels: 2,
|
||
Precision: precision,
|
||
}
|
||
return s, format, nil
|
||
}
|
||
|
||
// navFFmpegStreamer streams PCM from a running ffmpeg subprocess whose stdin
|
||
// is a *navBuffer. Playback starts immediately — ffmpeg reads from the buffer
|
||
// as bytes arrive from the background download. Seeking kills the current
|
||
// ffmpeg process, repositions the navBuffer to the target byte offset, and
|
||
// restarts ffmpeg so it reads from that position onwards.
|
||
type navFFmpegStreamer struct {
|
||
ffmpegPipe
|
||
nb *navBuffer
|
||
sr beep.SampleRate
|
||
}
|
||
|
||
// start launches ffmpeg reading from nb at nb's current position.
|
||
func (s *navFFmpegStreamer) start(seekPos int) error {
|
||
pcmFmt, codec, _ := ffmpegPCMArgs(s.bitDepth())
|
||
cmd := exec.Command("ffmpeg",
|
||
"-i", "pipe:0",
|
||
"-f", pcmFmt,
|
||
"-acodec", codec,
|
||
"-ar", strconv.Itoa(int(s.sr)),
|
||
"-ac", "2",
|
||
"-loglevel", "error",
|
||
"pipe:1",
|
||
)
|
||
cmd.Stdin = s.nb
|
||
|
||
pipe, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
return fmt.Errorf("ffmpeg nav pipe: %w", err)
|
||
}
|
||
if err := cmd.Start(); err != nil {
|
||
return fmt.Errorf("ffmpeg nav start: %w", err)
|
||
}
|
||
|
||
s.cmd = cmd
|
||
s.pipe = pipe
|
||
s.reader = bufio.NewReaderSize(pipe, pipeBufSize)
|
||
s.pos = seekPos
|
||
s.err = nil
|
||
return nil
|
||
}
|
||
|
||
// Seek repositions playback to the given sample frame. Kills the current
|
||
// ffmpeg process, seeks the navBuffer to the proportional byte offset, and
|
||
// restarts ffmpeg reading from that position.
|
||
func (s *navFFmpegStreamer) Seek(targetFrame int) error {
|
||
if targetFrame < 0 {
|
||
targetFrame = 0
|
||
}
|
||
if s.total > 0 && targetFrame > s.total {
|
||
targetFrame = s.total
|
||
}
|
||
|
||
// Compute the byte offset in the navBuffer proportional to the seek position.
|
||
byteOffset := int64(0)
|
||
if s.total > 0 && s.nb.total > 0 {
|
||
ratio := float64(targetFrame) / float64(s.total)
|
||
byteOffset = int64(ratio * float64(s.nb.total))
|
||
}
|
||
|
||
s.stop()
|
||
|
||
// Reposition the navBuffer to the target byte offset.
|
||
// navBuffer.Seek blocks if the target hasn't downloaded yet.
|
||
if _, err := s.nb.Seek(byteOffset, io.SeekStart); err != nil {
|
||
return fmt.Errorf("nav ffmpeg seek: %w", err)
|
||
}
|
||
|
||
return s.start(targetFrame)
|
||
}
|
||
|
||
// ffmpegPCMArgs returns the ffmpeg format flag, codec name, and beep precision
|
||
// for the given bit depth. 32-bit uses float PCM (f32le) which preserves
|
||
// up to 24-bit audio without any truncation; 16-bit uses integer PCM (s16le).
|
||
func ffmpegPCMArgs(bitDepth int) (format, codec string, precision int) {
|
||
if bitDepth == 32 {
|
||
return "f32le", "pcm_f32le", 4
|
||
}
|
||
return "s16le", "pcm_s16le", 2
|
||
}
|
||
|
||
// probeFrames uses ffprobe to quickly read file duration from metadata and
|
||
// converts it to sample frames. This only reads the container header, so it
|
||
// returns almost instantly even for very large files.
|
||
func probeFrames(path string, sr beep.SampleRate) int {
|
||
out, err := exec.Command("ffprobe",
|
||
"-v", "error",
|
||
"-show_entries", "format=duration",
|
||
"-of", "default=noprint_wrappers=1:nokey=1",
|
||
path,
|
||
).Output()
|
||
if err != nil {
|
||
return 0
|
||
}
|
||
secs, err := strconv.ParseFloat(strings.TrimSpace(string(out)), 64)
|
||
if err != nil {
|
||
return 0
|
||
}
|
||
return int(secs * float64(sr))
|
||
}
|