udpate stuff

This commit is contained in:
2026-06-11 09:48:17 -07:00
parent 6bd80a0b3a
commit a2df93097a
14 changed files with 294 additions and 26 deletions

View File

@@ -88,6 +88,7 @@ func (a *App) HandleSetSource(w http.ResponseWriter, r *http.Request) {
}
a.mu.Lock()
a.Status.ChannelName = ""
a.Status.Transitioning = false
a.mu.Unlock()
}
@@ -139,6 +140,7 @@ func (a *App) HandleSetChannel(w http.ResponseWriter, r *http.Request) {
a.Status.Source = "iptv"
a.Status.ChannelName = found.Name
a.Status.Live = true
a.Status.Transitioning = true
a.mu.Unlock()
log.Printf("[admin] channel set to %s (%s)", found.Name, found.ID)
@@ -187,3 +189,4 @@ func (a *App) HandleProcessStatus(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(status)
}

View File

@@ -17,6 +17,17 @@ func (a *App) HandleStatus(w http.ResponseWriter, r *http.Request) {
status := a.Status
a.mu.RUnlock()
// Attach live stream health metrics when FFmpeg is running
status.Health = a.FFmpeg.Health()
// Auto-clear transitioning once FFmpeg is producing health data
if status.Transitioning && status.Health != nil && status.Health.Speed != "" {
a.mu.Lock()
a.Status.Transitioning = false
a.mu.Unlock()
status.Transitioning = false
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(status)
}

View File

@@ -105,13 +105,17 @@ func generateID(name, url string) string {
}
// FilterChannels returns channels matching the given search term and/or group.
// Always returns a non-nil slice so it encodes as [] rather than null in JSON.
func FilterChannels(channels []models.Channel, search string, group string) []models.Channel {
if search == "" && group == "" {
if channels == nil {
return []models.Channel{}
}
return channels
}
searchLower := strings.ToLower(search)
var result []models.Channel
result := []models.Channel{}
for _, ch := range channels {
if group != "" && !strings.EqualFold(ch.Group, group) {
@@ -127,9 +131,10 @@ func FilterChannels(channels []models.Channel, search string, group string) []mo
}
// GetGroups returns a deduplicated, sorted list of group names from the channel list.
// Always returns a non-nil slice so it encodes as [] rather than null in JSON.
func GetGroups(channels []models.Channel) []string {
seen := make(map[string]bool)
var groups []string
groups := []string{}
for _, ch := range channels {
if ch.Group != "" && !seen[ch.Group] {

View File

@@ -9,11 +9,24 @@ type Channel struct {
StreamURL string `json:"stream_url"`
}
// StreamHealth holds real-time FFmpeg progress metrics for diagnosing issues.
// Speed < 1x indicates the upstream IPTV source can't deliver fast enough.
// Drop frames > 0 with speed ~1x suggests local processing issues.
type StreamHealth struct {
Speed string `json:"speed"` // "1.00x" — upstream delivery rate
FPS string `json:"fps"` // frames per second being processed
Bitrate string `json:"bitrate"` // output bitrate
DropFrames string `json:"drop_frames"` // frames dropped by FFmpeg
DupFrames string `json:"dup_frames"` // frames duplicated by FFmpeg
}
// StreamStatus represents the current streaming state.
type StreamStatus struct {
Source string `json:"source"` // "obs" or "iptv"
ChannelName string `json:"channel_name"` // current IPTV channel name (empty if OBS)
Live bool `json:"live"`
Source string `json:"source"` // "obs" or "iptv"
ChannelName string `json:"channel_name"` // current IPTV channel name (empty if OBS)
Live bool `json:"live"`
Transitioning bool `json:"transitioning"` // true while switching to a new channel
Health *StreamHealth `json:"health,omitempty"`
}
// ProcessInfo represents the status of a managed child process.

View File

@@ -1,9 +1,12 @@
package process
import (
"bufio"
"fmt"
"io"
"log"
"os/exec"
"strings"
"sync"
"time"
@@ -17,6 +20,7 @@ type FFmpegManager struct {
startTime time.Time
lastError string
streamURL string
health models.StreamHealth
}
// NewFFmpegManager creates a new FFmpeg process manager.
@@ -24,7 +28,7 @@ func NewFFmpegManager() *FFmpegManager {
return &FFmpegManager{}
}
// Start spawns an FFmpeg process to pull from streamURL and push RTMP to MediaMTX.
// Start spawns an FFmpeg process to pull from streamURL and push to MediaMTX.
// It kills any existing FFmpeg process first.
func (f *FFmpegManager) Start(streamURL string) error {
f.mu.Lock()
@@ -34,7 +38,11 @@ func (f *FFmpegManager) Start(streamURL string) error {
f.stopLocked()
f.cmd = exec.Command("ffmpeg",
"-re",
"-loglevel", "warning",
"-progress", "pipe:1",
"-reconnect", "1",
"-reconnect_streamed", "1",
"-reconnect_delay_max", "5",
"-i", streamURL,
"-c", "copy",
"-f", "flv",
@@ -42,6 +50,17 @@ func (f *FFmpegManager) Start(streamURL string) error {
)
f.streamURL = streamURL
f.lastError = ""
f.health = models.StreamHealth{}
// Capture stdout for progress stats, stderr for errors
stdout, err := f.cmd.StdoutPipe()
if err != nil {
return fmt.Errorf("stdout pipe: %w", err)
}
stderr, err := f.cmd.StderrPipe()
if err != nil {
return fmt.Errorf("stderr pipe: %w", err)
}
if err := f.cmd.Start(); err != nil {
f.lastError = err.Error()
@@ -51,18 +70,59 @@ func (f *FFmpegManager) Start(streamURL string) error {
f.startTime = time.Now()
log.Printf("[ffmpeg] started with pid %d, source: %s", f.cmd.Process.Pid, streamURL)
// Monitor in background
go f.monitor()
// Pass cmd reference so goroutines can detect process replacement
cmd := f.cmd
go f.parseProgress(stdout)
go f.logStderr(stderr)
go f.monitor(cmd)
return nil
}
func (f *FFmpegManager) monitor() {
err := f.cmd.Wait()
// parseProgress reads FFmpeg -progress output and updates health stats.
func (f *FFmpegManager) parseProgress(r io.Reader) {
scanner := bufio.NewScanner(r)
for scanner.Scan() {
line := scanner.Text()
k, v, ok := strings.Cut(line, "=")
if !ok {
continue
}
f.mu.Lock()
switch k {
case "speed":
f.health.Speed = v
case "fps":
f.health.FPS = v
case "bitrate":
f.health.Bitrate = v
case "drop_frames":
f.health.DropFrames = v
case "dup_frames":
f.health.DupFrames = v
}
f.mu.Unlock()
}
}
func (f *FFmpegManager) logStderr(r io.Reader) {
scanner := bufio.NewScanner(r)
for scanner.Scan() {
log.Printf("[ffmpeg] %s", scanner.Text())
}
}
func (f *FFmpegManager) monitor(cmd *exec.Cmd) {
err := cmd.Wait()
f.mu.Lock()
defer f.mu.Unlock()
// If the process was replaced by a new Start() call, don't touch state
if f.cmd != cmd {
return
}
if err != nil {
f.lastError = err.Error()
log.Printf("[ffmpeg] exited: %v", err)
@@ -96,6 +156,19 @@ func (f *FFmpegManager) stopLocked() error {
return nil
}
// Health returns the latest stream health stats from FFmpeg progress.
func (f *FFmpegManager) Health() *models.StreamHealth {
f.mu.Lock()
defer f.mu.Unlock()
if f.cmd == nil || f.health.Speed == "" {
return nil
}
h := f.health // copy
return &h
}
// Status returns the current FFmpeg process status.
func (f *FFmpegManager) Status() models.ProcessInfo {
f.mu.Lock()

View File

@@ -1,7 +1,9 @@
package process
import (
"bufio"
"fmt"
"io"
"log"
"os/exec"
"sync"
@@ -54,6 +56,10 @@ func (m *MediaMTXManager) startLocked() error {
m.stopped = false
m.stopCh = make(chan struct{})
// Capture stdout/stderr for logging
stdout, _ := m.cmd.StdoutPipe()
stderr, _ := m.cmd.StderrPipe()
if err := m.cmd.Start(); err != nil {
m.lastError = err.Error()
return fmt.Errorf("start mediamtx: %w", err)
@@ -62,12 +68,22 @@ func (m *MediaMTXManager) startLocked() error {
m.startTime = time.Now()
log.Printf("[mediamtx] started with pid %d", m.cmd.Process.Pid)
go m.logOutput("stdout", stdout)
go m.logOutput("stderr", stderr)
// Monitor in background and auto-restart on unexpected exit
go m.monitor()
return nil
}
func (m *MediaMTXManager) logOutput(name string, r io.Reader) {
scanner := bufio.NewScanner(r)
for scanner.Scan() {
log.Printf("[mediamtx:%s] %s", name, scanner.Text())
}
}
func (m *MediaMTXManager) monitor() {
err := m.cmd.Wait()