itten/go-mxl-player
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

stats window

Browse Source
This commit is contained in:
Dmitry Sergeev
2026-09-01 22:59:09 +03:00
parent ca25bf88a7
commit a8d277ee3d
13 changed files with 524 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/adapter/mxl/audio.go
+1
View File
@@ -170,6 +170,7 @@ func (r *audioReader) ReadAudio(ctx context.Context) (playback.AudioFrame, error
Index: frame.Index,
SampleCount: frame.SampleCount,
Channels: frame.Channels,
Label: frame.Label,
SampleRateNumerator: r.rateNumerator,
SampleRateDenominator: r.rateDenominator,
Samples: frame.Samples,
internal/adapter/mxl/sync.go
+11 -7
View File
@@ -133,18 +133,22 @@ func (r *syncReader) ReadSync(
}
return playback.SyncFrame{
Video: playback.VideoFrame{
Index: video.Index,
Width: video.Width,
Height: video.Height,
Stride: video.Stride,
Size: video.Size,
Invalid: video.Invalid,
Payload: video.Payload,
Index: video.Index,
Width: video.Width,
Height: video.Height,
Stride: video.Stride,
Size: video.Size,
Invalid: video.Invalid,
Label: video.Label,
FrameRateNumerator: video.FrameRateNumerator,
FrameRateDenominator: video.FrameRateDenominator,
Payload: video.Payload,
},
Audio: playback.AudioFrame{
Index: audio.Index,
SampleCount: audio.SampleCount,
Channels: audio.Channels,
Label: audio.Label,
SampleRateNumerator: r.rateNumerator,
SampleRateDenominator: r.rateDenominator,
Samples: audio.Samples,
internal/adapter/mxl/video.go
+10 -7
View File
@@ -113,13 +113,16 @@ func (r *videoReader) ReadVideo(
frame, err := r.source.ReadOnceCtx(ctx, r.readTimeout)
if err == nil {
return playback.VideoFrame{
Index: frame.Index,
Width: frame.Width,
Height: frame.Height,
Stride: frame.Stride,
Size: frame.Size,
Invalid: frame.Invalid,
Payload: frame.Payload,
Index: frame.Index,
Width: frame.Width,
Height: frame.Height,
Stride: frame.Stride,
Size: frame.Size,
Invalid: frame.Invalid,
Label: frame.Label,
FrameRateNumerator: frame.FrameRateNumerator,
FrameRateDenominator: frame.FrameRateDenominator,
Payload: frame.Payload,
}, nil
}
if ctx.Err() != nil {
internal/playback/audio.go
+1
View File
@@ -13,6 +13,7 @@ type AudioFrame struct {
Index uint64
SampleCount uint64
Channels uint64
Label string
SampleRateNumerator int64
SampleRateDenominator int64
internal/playback/media_stats.go
+136
View File
@@ -0,0 +1,136 @@
package playback
import (
"context"
"sync"
"time"
)
type VideoMediaStats struct {
Available bool
Label string
Index uint64
Width uint32
Height uint32
Stride uint32
PayloadSize uint32
DeclaredFPS float64
ReceivedFPS float64
FrameDT time.Duration
Invalid uint64
}
type AudioMediaStats struct {
Available bool
Label string
Index uint64
SampleRateHz float64
Channels uint64
SampleCount uint64
BatchDuration time.Duration
}
type MediaStatsSnapshot struct {
Video VideoMediaStats
Audio AudioMediaStats
}
type MediaStatsStore struct {
mu sync.RWMutex
snapshot MediaStatsSnapshot
now func() time.Time
lastVideoAt time.Time
videoWindowAt time.Time
videoWindowCount uint64
}
func NewMediaStatsStore() *MediaStatsStore {
return &MediaStatsStore{now: time.Now}
}
func (s *MediaStatsStore) ObserveVideo(frame VideoFrame) {
now := s.now()
s.mu.Lock()
defer s.mu.Unlock()
stats := &s.snapshot.Video
stats.Available = true
stats.Label = frame.Label
stats.Index = frame.Index
stats.Width = frame.Width
stats.Height = frame.Height
stats.Stride = frame.Stride
stats.PayloadSize = frame.Size
if frame.FrameRateDenominator > 0 {
stats.DeclaredFPS = float64(frame.FrameRateNumerator) /
float64(frame.FrameRateDenominator)
}
if !s.lastVideoAt.IsZero() {
stats.FrameDT = now.Sub(s.lastVideoAt)
}
s.lastVideoAt = now
if frame.Invalid {
stats.Invalid++
}
if s.videoWindowAt.IsZero() {
s.videoWindowAt = now
}
s.videoWindowCount++
if elapsed := now.Sub(s.videoWindowAt); elapsed >= time.Second {
stats.ReceivedFPS = float64(s.videoWindowCount) / elapsed.Seconds()
s.videoWindowAt = now
s.videoWindowCount = 0
}
}
func (s *MediaStatsStore) ObserveAudio(frame AudioFrame) {
s.mu.Lock()
defer s.mu.Unlock()
stats := &s.snapshot.Audio
stats.Available = true
stats.Label = frame.Label
stats.Index = frame.Index
stats.Channels = frame.Channels
stats.SampleCount = frame.SampleCount
if frame.SampleRateDenominator > 0 {
stats.SampleRateHz = float64(frame.SampleRateNumerator) /
float64(frame.SampleRateDenominator)
}
if stats.SampleRateHz > 0 {
stats.BatchDuration = time.Duration(
float64(time.Second) * float64(frame.SampleCount) / stats.SampleRateHz,
)
}
}
func (s *MediaStatsStore) Snapshot() MediaStatsSnapshot {
s.mu.RLock()
defer s.mu.RUnlock()
return s.snapshot
}
type VideoStatsSink struct {
Stats *MediaStatsStore
Sink VideoSink
}
func (s VideoStatsSink) ConsumeVideo(ctx context.Context, frame VideoFrame) error {
if s.Stats != nil {
s.Stats.ObserveVideo(frame)
}
return s.Sink.ConsumeVideo(ctx, frame)
}
type AudioStatsSink struct {
Stats *MediaStatsStore
Sink AudioSink
}
func (s AudioStatsSink) ConsumeAudio(ctx context.Context, frame AudioFrame) error {
if s.Stats != nil {
s.Stats.ObserveAudio(frame)
}
return s.Sink.ConsumeAudio(ctx, frame)
}
internal/playback/media_stats_test.go
+116
View File
@@ -0,0 +1,116 @@
package playback
import (
"context"
"testing"
"time"
)
func TestMediaStatsStoreObservesVideo(t *testing.T) {
store := NewMediaStatsStore()
now := time.Date(2026, time.September, 1, 12, 0, 0, 0, time.UTC)
store.now = func() time.Time { return now }
frame := VideoFrame{
Index: 10,
Width: 1920,
Height: 1080,
Stride: 5120,
Size: 5_529_600,
Label: "Main video",
FrameRateNumerator: 30000,
FrameRateDenominator: 1001,
}
store.ObserveVideo(frame)
now = now.Add(40 * time.Millisecond)
frame.Index++
frame.Invalid = true
store.ObserveVideo(frame)
got := store.Snapshot().Video
if !got.Available || got.Label != "Main video" || got.Index != 11 {
t.Fatalf("video stats = %#v", got)
}
if got.Width != 1920 || got.Height != 1080 || got.Stride != 5120 {
t.Fatalf("video dimensions = %#v", got)
}
if got.DeclaredFPS < 29.96 || got.DeclaredFPS > 29.98 {
t.Fatalf("declared FPS = %v", got.DeclaredFPS)
}
if got.FrameDT != 40*time.Millisecond || got.Invalid != 1 {
t.Fatalf("video timing = %#v", got)
}
}
func TestMediaStatsStoreCalculatesReceivedFPS(t *testing.T) {
store := NewMediaStatsStore()
now := time.Date(2026, time.September, 1, 12, 0, 0, 0, time.UTC)
store.now = func() time.Time { return now }
store.ObserveVideo(VideoFrame{})
now = now.Add(500 * time.Millisecond)
store.ObserveVideo(VideoFrame{})
now = now.Add(500 * time.Millisecond)
store.ObserveVideo(VideoFrame{})
if got := store.Snapshot().Video.ReceivedFPS; got != 3 {
t.Fatalf("received FPS = %v, want 3", got)
}
}
func TestMediaStatsStoreObservesAudio(t *testing.T) {
store := NewMediaStatsStore()
store.ObserveAudio(AudioFrame{
Index: 100,
SampleCount: 480,
Channels: 2,
Label: "Programme audio",
SampleRateNumerator: 48_000,
SampleRateDenominator: 1,
})
got := store.Snapshot().Audio
if !got.Available || got.Label != "Programme audio" || got.Index != 100 {
t.Fatalf("audio stats = %#v", got)
}
if got.SampleRateHz != 48_000 || got.Channels != 2 || got.SampleCount != 480 {
t.Fatalf("audio format = %#v", got)
}
if got.BatchDuration != 10*time.Millisecond {
t.Fatalf("batch duration = %v, want 10ms", got.BatchDuration)
}
}
type recordingVideoStatsSink struct{ frame VideoFrame }
func (s *recordingVideoStatsSink) ConsumeVideo(_ context.Context, frame VideoFrame) error {
s.frame = frame
return nil
}
type recordingAudioStatsSink struct{ frame AudioFrame }
func (s *recordingAudioStatsSink) ConsumeAudio(_ context.Context, frame AudioFrame) error {
s.frame = frame
return nil
}
func TestStatsSinksObserveAndForwardFrames(t *testing.T) {
store := NewMediaStatsStore()
videoDownstream := &recordingVideoStatsSink{}
audioDownstream := &recordingAudioStatsSink{}
video := VideoFrame{Index: 7, Label: "video"}
audio := AudioFrame{Index: 8, Label: "audio"}
if err := (VideoStatsSink{Stats: store, Sink: videoDownstream}).ConsumeVideo(context.Background(), video); err != nil {
t.Fatalf("ConsumeVideo() error = %v", err)
}
if err := (AudioStatsSink{Stats: store, Sink: audioDownstream}).ConsumeAudio(context.Background(), audio); err != nil {
t.Fatalf("ConsumeAudio() error = %v", err)
}
if videoDownstream.frame.Index != video.Index || audioDownstream.frame.Index != audio.Index {
t.Fatalf("forwarded frames = %#v, %#v", videoDownstream.frame, audioDownstream.frame)
}
snapshot := store.Snapshot()
if snapshot.Video.Label != "video" || snapshot.Audio.Label != "audio" {
t.Fatalf("stats snapshot = %#v", snapshot)
}
}
internal/playback/video.go
+10 -7
View File
@@ -8,13 +8,16 @@ import "context"
// reader is closed. Consumers must finish reading Payload before returning
// control to the worker
type VideoFrame struct {
Index uint64
Width uint32
Height uint32
Stride uint32
Size uint32
Invalid bool
Payload []byte
Index uint64
Width uint32
Height uint32
Stride uint32
Size uint32
Invalid bool
Label string
FrameRateNumerator int64
FrameRateDenominator int64
Payload []byte
}
// VideoReader reads frames from a video source.
internal/source/source.go
+68 -34
View File
@@ -11,6 +11,7 @@ import (
)
type flowDef struct {
Label string `json:"label"`
FrameWidth int `json:"frame_width"`
FrameHeight int `json:"frame_height"`
MediaType string `json:"media_type"`
@@ -21,14 +22,31 @@ type flowDef struct {
} `json:"grain_rate"`
}
func flowLabel(inst *mxl.Instance, flowID string) string {
definition, err := inst.FlowDef(flowID)
if err != nil {
return ""
}
var metadata struct {
Label string `json:"label"`
}
if json.Unmarshal([]byte(definition), &metadata) != nil {
return ""
}
return metadata.Label
}
type Frame struct {
Index uint64
Width uint32
Height uint32
Stride uint32
Size uint32
Invalid bool
Payload []byte
Index uint64
Width uint32
Height uint32
Stride uint32
Size uint32
Invalid bool
Label string
FrameRateNumerator int64
FrameRateDenominator int64
Payload []byte
}
type Source struct {
@@ -40,6 +58,7 @@ type Source struct {
stride uint32
width uint32
height uint32
label string
idx uint64
}
@@ -109,6 +128,7 @@ func Open(domain, flowID string) (*Source, error) {
stride: info.Config.Discrete.SliceSizes[0],
width: uint32(fd.FrameWidth),
height: uint32(fd.FrameHeight),
label: fd.Label,
idx: idx,
}, nil
}
@@ -157,13 +177,16 @@ func (s *Source) ReadOnceCtx(
switch {
case err == nil:
frame := Frame{
Index: g.Index,
Width: s.width,
Height: s.height,
Stride: s.stride,
Size: g.GrainSize,
Invalid: g.Invalid(),
Payload: g.Payload,
Index: g.Index,
Width: s.width,
Height: s.height,
Stride: s.stride,
Size: g.GrainSize,
Invalid: g.Invalid(),
Label: s.label,
FrameRateNumerator: s.rate.Num,
FrameRateDenominator: s.rate.Den,
Payload: g.Payload,
}
s.idx++
return frame, nil
@@ -222,12 +245,14 @@ type AudioSource struct {
rate mxl.Rational
chans uint64
idx uint64
label string
}
type AudioFrame struct {
Index uint64
SampleCount uint64
Channels uint64
Label string
Samples [][]byte // per-channel byte slices (F32, deinterleaved)
}
@@ -293,6 +318,7 @@ func OpenAudio(domain, flowID string) (*AudioSource, error) {
rate: rate,
chans: channels,
idx: idx,
label: flowLabel(inst, flowID),
}, nil
}
@@ -326,6 +352,7 @@ func (s *AudioSource) ReadAudioOnceCtx(
Index: s.idx,
SampleCount: batch,
Channels: s.chans,
Label: s.label,
Samples: samples,
}
s.idx += batch
@@ -401,15 +428,16 @@ func (s *AudioSource) Rate() mxl.Rational { return s.rate }
func (s *AudioSource) Channels() uint64 { return s.chans }
type SyncSource struct {
inst *mxl.Instance
vr *mxl.Reader
ar *mxl.Reader
group *mxl.SyncGroup
rate mxl.Rational // video rate
aRate mxl.Rational
chans uint64
idx uint64
width, height, stride uint32
inst *mxl.Instance
vr *mxl.Reader
ar *mxl.Reader
group *mxl.SyncGroup
rate mxl.Rational // video rate
aRate mxl.Rational
chans uint64
idx uint64
width, height, stride uint32
videoLabel, audioLabel string
}
// OpenSameDomainSync opens a native MXL synchronization group.
@@ -577,17 +605,19 @@ func OpenSameDomainSync(domain, videoFlow, audioFlow string) (*SyncSource, error
}
return &SyncSource{
inst: inst,
vr: vr,
ar: ar,
group: group,
rate: vRate,
aRate: aRate,
chans: channels,
idx: idx,
width: uint32(fd.FrameWidth),
height: uint32(fd.FrameHeight),
stride: vInfo.Config.Discrete.SliceSizes[0],
inst: inst,
vr: vr,
ar: ar,
group: group,
rate: vRate,
aRate: aRate,
chans: channels,
idx: idx,
width: uint32(fd.FrameWidth),
height: uint32(fd.FrameHeight),
stride: vInfo.Config.Discrete.SliceSizes[0],
videoLabel: fd.Label,
audioLabel: flowLabel(inst, audioFlow),
}, nil
}
@@ -665,6 +695,9 @@ func (s *SyncSource) NextSync(ctx context.Context, timeout time.Duration) (Frame
Index: g.Index, Width: s.width, Height: s.height,
Stride: s.stride, Size: g.GrainSize,
Invalid: g.Invalid(), Payload: g.Payload,
Label: s.videoLabel,
FrameRateNumerator: s.rate.Num,
FrameRateDenominator: s.rate.Den,
}
s.idx++
@@ -680,6 +713,7 @@ func (s *SyncSource) NextSync(ctx context.Context, timeout time.Duration) (Frame
aFrame := AudioFrame{
Index: aIdx, SampleCount: audioBatch,
Channels: s.chans, Samples: samples,
Label: s.audioLabel,
}
return vFrame, aFrame, nil
case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly), errors.Is(err, mxl.ErrOutOfRangeLate):