add: live encode progress tracking and read-only status web UI
Stream ffmpeg -progress during the video encode into an in-memory tracker (internal/status) so the long-running step is no longer a black box: percent, fps, speed, and ETA are derived from the source duration and updated ~1/s. Progress prints to stdout only (no logs.db row) to avoid burying real events. Expose it over HTTP (internal/server, default :8080, set http_addr to "" to disable): GET /status returns the live snapshot, the pending input queue, and recent non-debug events; GET / serves an embedded dashboard that polls /status every second. The server shuts down on the same SIGINT/SIGTERM context as the watcher.
This commit is contained in:
Esa Kataja
@@ -0,0 +1,201 @@
|
||||
// Package status tracks the live state of the single in-flight encode job.
|
||||
// The watcher processes one file at a time, so one mutex-guarded value is
|
||||
// enough — no per-job table, no concurrency design.
|
||||
package status
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"io"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Phase labels for the current job.
|
||||
const (
|
||||
PhaseIdle = "idle"
|
||||
PhaseProbing = "probing"
|
||||
PhaseAudio = "audio"
|
||||
PhaseEncoding = "encoding"
|
||||
)
|
||||
|
||||
// Tracker holds live progress for the active job. Safe for concurrent use:
|
||||
// the encode goroutine writes, HTTP/log readers call Snapshot.
|
||||
type Tracker struct {
|
||||
mu sync.RWMutex
|
||||
file string
|
||||
phase string
|
||||
totalSec float64 // source duration; 0 until known
|
||||
outTime float64 // encoded position in seconds
|
||||
fps float64
|
||||
speed float64
|
||||
startedAt time.Time
|
||||
}
|
||||
|
||||
// Snapshot is an immutable view of the tracker for readers.
|
||||
type Snapshot struct {
|
||||
File string `json:"file"`
|
||||
Phase string `json:"phase"`
|
||||
Percent float64 `json:"percent"`
|
||||
FPS float64 `json:"fps"`
|
||||
Speed float64 `json:"speed"`
|
||||
ElapsedSec int `json:"elapsed_sec"`
|
||||
ETASec int `json:"eta_sec"`
|
||||
StartedAt time.Time `json:"started_at"`
|
||||
}
|
||||
|
||||
func New() *Tracker {
|
||||
return &Tracker{phase: PhaseIdle}
|
||||
}
|
||||
|
||||
// Begin marks the start of a new job, resetting all progress fields.
|
||||
func (t *Tracker) Begin(file string) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.file = file
|
||||
t.phase = PhaseProbing
|
||||
t.totalSec, t.outTime, t.fps, t.speed = 0, 0, 0, 0
|
||||
t.startedAt = time.Now()
|
||||
}
|
||||
|
||||
func (t *Tracker) SetPhase(p string) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.phase = p
|
||||
}
|
||||
|
||||
// SetTotal records the source duration in seconds (from ffprobe).
|
||||
func (t *Tracker) SetTotal(seconds float64) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.totalSec = seconds
|
||||
}
|
||||
|
||||
// Update records one ffmpeg -progress sample.
|
||||
func (t *Tracker) Update(outTimeSec, fps, speed float64) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.outTime, t.fps, t.speed = outTimeSec, fps, speed
|
||||
}
|
||||
|
||||
// Idle clears the tracker when no job is running.
|
||||
func (t *Tracker) Idle() {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.file = ""
|
||||
t.phase = PhaseIdle
|
||||
t.totalSec, t.outTime, t.fps, t.speed = 0, 0, 0, 0
|
||||
t.startedAt = time.Time{}
|
||||
}
|
||||
|
||||
func (t *Tracker) Snapshot() Snapshot {
|
||||
t.mu.RLock()
|
||||
defer t.mu.RUnlock()
|
||||
s := Snapshot{
|
||||
File: t.file,
|
||||
Phase: t.phase,
|
||||
FPS: t.fps,
|
||||
Speed: t.speed,
|
||||
StartedAt: t.startedAt,
|
||||
}
|
||||
if !t.startedAt.IsZero() {
|
||||
s.ElapsedSec = int(time.Since(t.startedAt).Seconds())
|
||||
}
|
||||
if t.totalSec > 0 {
|
||||
s.Percent = t.outTime / t.totalSec * 100
|
||||
if s.Percent > 100 {
|
||||
s.Percent = 100
|
||||
}
|
||||
if t.speed > 0 {
|
||||
remaining := t.totalSec - t.outTime
|
||||
if remaining < 0 {
|
||||
remaining = 0
|
||||
}
|
||||
s.ETASec = int(remaining / t.speed)
|
||||
}
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// ProgressSample is one completed ffmpeg -progress block.
|
||||
type ProgressSample struct {
|
||||
OutTimeSec float64
|
||||
FPS float64
|
||||
Speed float64
|
||||
Done bool // progress=end
|
||||
}
|
||||
|
||||
// ScanProgress reads ffmpeg `-progress` key=value output from r and calls
|
||||
// onSample once per block (each block is terminated by a "progress=" line).
|
||||
// Returns when r is exhausted.
|
||||
func ScanProgress(r io.Reader, onSample func(ProgressSample)) error {
|
||||
sc := bufio.NewScanner(r)
|
||||
var cur ProgressSample
|
||||
for sc.Scan() {
|
||||
if parseProgressLine(sc.Text(), &cur) {
|
||||
onSample(cur)
|
||||
cur = ProgressSample{}
|
||||
}
|
||||
}
|
||||
return sc.Err()
|
||||
}
|
||||
|
||||
// parseProgressLine folds one "key=value" line into s. Returns true when the
|
||||
// line closes a block (key == "progress").
|
||||
//
|
||||
// ponytail: ffmpeg field naming drifts between builds — out_time_us is the
|
||||
// modern microsecond field; out_time_ms is historically ALSO microseconds (a
|
||||
// known mislabel); out_time is the "HH:MM:SS.ffffff" string. Prefer out_time_us,
|
||||
// fall back to the others only if it hasn't set a value this block. Verify
|
||||
// against the ffmpeg in the Docker image if percentages look off.
|
||||
func parseProgressLine(line string, s *ProgressSample) (complete bool) {
|
||||
k, v, ok := strings.Cut(line, "=")
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
k, v = strings.TrimSpace(k), strings.TrimSpace(v)
|
||||
switch k {
|
||||
case "out_time_us":
|
||||
if us, err := strconv.ParseFloat(v, 64); err == nil {
|
||||
s.OutTimeSec = us / 1e6
|
||||
}
|
||||
case "out_time_ms":
|
||||
if s.OutTimeSec == 0 {
|
||||
if ms, err := strconv.ParseFloat(v, 64); err == nil {
|
||||
s.OutTimeSec = ms / 1e6 // really microseconds, see note above
|
||||
}
|
||||
}
|
||||
case "out_time":
|
||||
if s.OutTimeSec == 0 {
|
||||
s.OutTimeSec = parseTimecode(v)
|
||||
}
|
||||
case "fps":
|
||||
if f, err := strconv.ParseFloat(v, 64); err == nil {
|
||||
s.FPS = f
|
||||
}
|
||||
case "speed":
|
||||
if f, err := strconv.ParseFloat(strings.TrimSuffix(v, "x"), 64); err == nil {
|
||||
s.Speed = f // "N/A" leaves it 0
|
||||
}
|
||||
case "progress":
|
||||
s.Done = v == "end"
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// parseTimecode parses "HH:MM:SS.ffffff" into seconds; 0 on bad input.
|
||||
func parseTimecode(tc string) float64 {
|
||||
parts := strings.Split(tc, ":")
|
||||
if len(parts) != 3 {
|
||||
return 0
|
||||
}
|
||||
h, err1 := strconv.ParseFloat(parts[0], 64)
|
||||
m, err2 := strconv.ParseFloat(parts[1], 64)
|
||||
sec, err3 := strconv.ParseFloat(parts[2], 64)
|
||||
if err1 != nil || err2 != nil || err3 != nil {
|
||||
return 0
|
||||
}
|
||||
return h*3600 + m*60 + sec
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package status
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestScanProgress(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
input string
|
||||
wantOutSec, wantFPS, wantSpd float64
|
||||
wantDone bool
|
||||
}{
|
||||
{
|
||||
name: "out_time_us preferred over ms and string",
|
||||
input: "frame=120\nfps=24.00\nout_time_us=5000000\nout_time_ms=9999999\nout_time=00:00:09.000000\nspeed=1.02x\nprogress=continue\n",
|
||||
wantOutSec: 5, wantFPS: 24, wantSpd: 1.02, wantDone: false,
|
||||
},
|
||||
{
|
||||
name: "out_time string fallback when no us field",
|
||||
input: "fps=12\nout_time=00:01:30.500000\nspeed=0.09x\nprogress=continue\n",
|
||||
wantOutSec: 90.5, wantFPS: 12, wantSpd: 0.09, wantDone: false,
|
||||
},
|
||||
{
|
||||
name: "end block",
|
||||
input: "out_time_us=7200000000\nspeed=2x\nprogress=end\n",
|
||||
wantOutSec: 7200, wantFPS: 0, wantSpd: 2, wantDone: true,
|
||||
},
|
||||
{
|
||||
name: "speed N/A leaves zero",
|
||||
input: "out_time_us=1000000\nspeed=N/A\nprogress=continue\n",
|
||||
wantOutSec: 1, wantFPS: 0, wantSpd: 0, wantDone: false,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
var last ProgressSample
|
||||
n := 0
|
||||
if err := ScanProgress(strings.NewReader(tt.input), func(s ProgressSample) { last = s; n++ }); err != nil {
|
||||
t.Fatalf("ScanProgress: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("got %d samples, want 1", n)
|
||||
}
|
||||
if last.OutTimeSec != tt.wantOutSec || last.FPS != tt.wantFPS || last.Speed != tt.wantSpd || last.Done != tt.wantDone {
|
||||
t.Errorf("got %+v, want out=%v fps=%v spd=%v done=%v", last, tt.wantOutSec, tt.wantFPS, tt.wantSpd, tt.wantDone)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestScanProgressMultipleBlocks(t *testing.T) {
|
||||
in := "out_time_us=1000000\nspeed=1x\nprogress=continue\nout_time_us=2000000\nspeed=1x\nprogress=end\n"
|
||||
var got []ProgressSample
|
||||
if err := ScanProgress(strings.NewReader(in), func(s ProgressSample) { got = append(got, s) }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("got %d blocks, want 2", len(got))
|
||||
}
|
||||
if got[0].OutTimeSec != 1 || got[1].OutTimeSec != 2 || got[0].Done || !got[1].Done {
|
||||
t.Errorf("blocks = %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSnapshotPercentAndETA(t *testing.T) {
|
||||
tr := New()
|
||||
tr.Begin("episode.mkv")
|
||||
tr.SetTotal(100)
|
||||
tr.Update(25, 10, 0.5) // 25% done, 75s left at 0.5x -> 150s ETA
|
||||
s := tr.Snapshot()
|
||||
if s.Percent != 25 {
|
||||
t.Errorf("Percent = %v, want 25", s.Percent)
|
||||
}
|
||||
if s.ETASec != 150 {
|
||||
t.Errorf("ETASec = %v, want 150", s.ETASec)
|
||||
}
|
||||
if s.File != "episode.mkv" || s.Phase != PhaseProbing {
|
||||
t.Errorf("File/Phase = %q/%q", s.File, s.Phase)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSnapshotIdleNoDivByZero(t *testing.T) {
|
||||
s := New().Snapshot() // no Begin, totalSec 0
|
||||
if s.Percent != 0 || s.ETASec != 0 || s.Phase != PhaseIdle {
|
||||
t.Errorf("idle snapshot = %+v", s)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user