Merge pull request #597 from streamplace/eli/better-bframe-detection

media: fix bframe detection

Makefile

···
 		-D "gst-plugins-ugly:gpl=enabled" \
 		-D "x264:asm=enabled" \
 		-D "gstreamer-full:gst-full=enabled" \
-		-D "gstreamer-full:gst-full-plugins=libgstopusparse.a;libgstcodectimestamper.a;libgstrtp.a;libgstaudioresample.a;libgstlibav.a;libgstmatroska.a;libgstmultifile.a;libgstjpeg.a;libgstaudiotestsrc.a;libgstaudioconvert.a;libgstaudioparsers.a;libgstfdkaac.a;libgstisomp4.a;libgstapp.a;libgstvideoconvertscale.a;libgstvideobox.a;libgstvideorate.a;libgstpng.a;libgstcompositor.a;libgstaudiorate.a;libgstx264.a;libgstopus.a;libgstvideotestsrc.a;libgstvideoparsersbad.a;libgstaudioparsers.a;libgstmpegtsmux.a;libgstmpegtsdemux.a;libgstplayback.a;libgsttypefindfunctions.a;libgstcoretracers.a" \
+		-D "gstreamer-full:gst-full-plugins=libgstopusparse.a;libgstcodectimestamper.a;libgstrtp.a;libgstaudioresample.a;libgstlibav.a;libgstmatroska.a;libgstmultifile.a;libgstjpeg.a;libgstaudiotestsrc.a;libgstaudioconvert.a;libgstaudioparsers.a;libgstfdkaac.a;libgstisomp4.a;libgstapp.a;libgstvideoconvertscale.a;libgstvideobox.a;libgstvideorate.a;libgstpng.a;libgstcompositor.a;libgstaudiorate.a;libgstx264.a;libgstopus.a;libgstvideotestsrc.a;libgstvideoparsersbad.a;libgstaudioparsers.a;libgstmpegtsmux.a;libgstmpegtsdemux.a;libgstplayback.a;libgsttypefindfunctions.a;libgstcoretracers.a;libgstcodec2json.a" \
 		-D "gstreamer-full:gst-full-libraries=gstreamer-controller-1.0,gstreamer-plugins-base-1.0,gstreamer-pbutils-1.0" \
 		-D "gstreamer-full:gst-full-elements=coreelements:concat,filesrc,filesink,queue,queue2,multiqueue,typefind,tee,capsfilter,fakesink,identity" \
 		-D "gstreamer-full:bad=enabled" \
···
 		-D "glib:glib_assert=false" \
 		-D "glib:glib_assert=false" \
 		-D "gst-libav:glib_assert=false" \
-		-D "gst-plugins-good:adaptivedemux2=disabled"
+		-D "gst-plugins-good:adaptivedemux2=disabled" \
+		-D "gst-plugins-bad:codec2json=enabled"
 
 STATIC_OPTS = \
 	$(BASE_OPTS) \

pkg/config/config.go

···
 	PLCURL                 string
 	SQLLogging             bool
 	SentryDSN              string
+	LivepeerDebug          bool
 }
 
 func (cli *CLI) NewFlagSet(name string) *flag.FlagSet {
···
 	fs.StringVar(&cli.PLCURL, "plc-url", "https://plc.directory", "url of the plc directory")
 	fs.BoolVar(&cli.SQLLogging, "sql-logging", false, "enable sql logging")
 	fs.StringVar(&cli.SentryDSN, "sentry-dsn", "", "sentry dsn for error reporting")
+	fs.BoolVar(&cli.LivepeerDebug, "livepeer-debug", false, "log livepeer segments to $SP_DATA_DIR/livepeer-debug")
 
 	lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError)
 	_ = starter.NewLivepeerConfig(lpFlags)
···
 	}
 	if cli.LivepeerGateway {
 		gatewayPath := cli.DataFilePath([]string{"livepeer", "gateway"})
+		err = fs.Set("livepeer.rtmp-addr", "127.0.0.1:0")
+		if err != nil {
+			return err
+		}
 		err = fs.Set("livepeer.data-dir", gatewayPath)
 		if err != nil {
 			return err

pkg/director/stream_session.go

···
 
 	if ss.lp == nil {
 		var err error
-		ss.lp, err = livepeer.NewLivepeerSession(ctx, spseg.Creator, ss.cli.LivepeerGatewayURL)
+		ss.lp, err = livepeer.NewLivepeerSession(ctx, ss.cli, spseg.Creator, ss.cli.LivepeerGatewayURL)
 		if err != nil {
 			return err
 		}

pkg/livepeer/livepeer.go

···
 	"mime"
 	"mime/multipart"
 	"net/http"
+	"os"
 	"strings"
 	"time"
 
 	"golang.org/x/net/context/ctxhttp"
 	"stream.place/streamplace/pkg/aqhttp"
+	"stream.place/streamplace/pkg/config"
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/media"
 	"stream.place/streamplace/pkg/renditions"
···
 	Count      int
 	GatewayURL string
 	Guard      chan struct{}
+	CLI        *config.CLI
 }
 
 // borrowed from catalyst-api
···
 	return string(res)
 }
 
-func NewLivepeerSession(ctx context.Context, did string, gatewayURL string) (*LivepeerSession, error) {
-	sessionID := RandomTrailer(8)
+func NewLivepeerSession(ctx context.Context, cli *config.CLI, did string, gatewayURL string) (*LivepeerSession, error) {
+	sessionID := fmt.Sprintf("%s-%s", did, RandomTrailer(8))
+	sessionID = strings.ReplaceAll(sessionID, ":", "")
+	sessionID = strings.ReplaceAll(sessionID, ".", "")
 	return &LivepeerSession{
-		SessionID:  strings.ReplaceAll(fmt.Sprintf("%s-%s", did, sessionID), ":", ""),
+		SessionID:  sessionID,
 		Count:      0,
 		GatewayURL: gatewayURL,
 		Guard:      make(chan struct{}, SegmentsInFlight),
+		CLI:        cli,
 	}, nil
 }
 
···
 	req.Header.Set("Content-Resolution", fmt.Sprintf("%dx%d", width, height))
 	req.Header.Set("Livepeer-Transcode-Configuration", string(bs))
 
+	if ls.CLI.LivepeerDebug {
+		debugDir := ls.CLI.DataFilePath([]string{"livepeer-debug"})
+		err = os.MkdirAll(debugDir, 0755)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create debug directory: %w", err)
+		}
+		debugFile := fmt.Sprintf("%s/livepeer-debug/%s-%06d-input.ts", ls.CLI.DataDir, sessionIDRen, seqNo)
+		err = os.WriteFile(debugFile, tsSeg.Bytes(), 0644)
+		if err != nil {
+			return nil, fmt.Errorf("failed to write debug file: %w", err)
+		}
+		bs, err := json.MarshalIndent(req.Header, "", "  ")
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal livepeer profile: %w", err)
+		}
+		configFile := fmt.Sprintf("%s/livepeer-debug/%s-%06d-config.json", ls.CLI.DataDir, sessionIDRen, seqNo)
+		err = os.WriteFile(configFile, bs, 0644)
+		if err != nil {
+			return nil, fmt.Errorf("failed to write debug file: %w", err)
+		}
+		log.Log(ctx, "wrote debug file", "file", debugFile)
+	}
+
 	resp, err := ctxhttp.Do(ctx, &aqhttp.Client, req)
 	if err != nil {
 		<-ls.Guard
···
 			}
 			mp4Bs := bytes.Buffer{}
 			audioReader := bytes.NewReader(audioSeg.Bytes())
+			if ls.CLI.LivepeerDebug {
+				debugFile := fmt.Sprintf("%s/livepeer-debug/%s-%06d-output-%s", ls.CLI.DataDir, sessionIDRen, seqNo, p.FileName())
+				err = os.WriteFile(debugFile, tsSeg.Bytes(), 0644)
+				if err != nil {
+					return nil, fmt.Errorf("failed to write debug file: %w", err)
+				}
+				log.Log(ctx, "wrote debug file", "file", debugFile)
+			}
 			err = media.MPEGTSVideoMP4AudioToMP4(ctx, p, audioReader, &mp4Bs)
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert ts to mp4: %w", err)

pkg/media/media_data_parser.go

···
 package media
 
 import (
+	"bufio"
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
+	"io"
 	"strconv"
 	"strings"
 
···
 	defer cancel()
 	pipelineSlice := []string{
 		"appsrc name=appsrc ! qtdemux name=demux",
-		"demux.video_0 ! queue ! h264parse name=videoparse disable-passthrough=true config-interval=-1 ! h264timestamper ! appsink sync=false name=videoappsink",
-		"demux.audio_0 ! queue ! opusparse name=audioparse ! appsink sync=false name=audioappsink",
+		"demux.video_0 ! queue ! h264parse name=videoparse disable-passthrough=true config-interval=-1 ! h2642json ! appsink sync=false name=jsonappsink",
+		"demux.audio_0 ! queue ! opusparse name=audioparse ! fakesink sync=false",
 	}
 
 	pipeline, err := gst.NewPipelineFromString(strings.Join(pipelineSlice, "\n"))
···
 		return nil, fmt.Errorf("error connecting pad-add: %w", err)
 	}
 
-	audioSinkElem, err := pipeline.GetElementByName("audioappsink")
-	if err != nil {
-		return nil, fmt.Errorf("failed to get audioappsink element: %w", err)
-	}
-	audioSink := app.SinkFromElement(audioSinkElem)
-	if audioSink == nil {
-		return nil, fmt.Errorf("failed to get audioappsink element: %w", err)
-	}
-
-	audioSink.SetCallbacks(&app.SinkCallbacks{
-		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
-			sample := sink.PullSample()
-			if sample == nil {
-				return gst.FlowOK
-			}
-
-			return gst.FlowOK
-		},
-	})
-
-	videoSinkElem, err := pipeline.GetElementByName("videoappsink")
+	jsonSinkElem, err := pipeline.GetElementByName("jsonappsink")
 	if err != nil {
 		return nil, fmt.Errorf("failed to get videoappsink element: %w", err)
 	}
-	videoSink := app.SinkFromElement(videoSinkElem)
-	if videoSink == nil {
+	jsonSink := app.SinkFromElement(jsonSinkElem)
+	if jsonSink == nil {
 		return nil, fmt.Errorf("failed to get videoappsink element: %w", err)
 	}
 
 	hasBFrames := false
-	videoSink.SetCallbacks(&app.SinkCallbacks{
+
+	r, w := io.Pipe()
+	bufW := bufio.NewWriter(w)
+	decoder := json.NewDecoder(r)
+
+	decodeErr := make(chan error)
+	go func() {
+		for {
+			var obj map[string]any
+			err := decoder.Decode(&obj)
+			if err == io.EOF {
+				decodeErr <- nil
+				break // End of stream
+			}
+			if err != nil {
+				decodeErr <- err
+				break
+			}
+			// https://github.com/GStreamer/gstreamer/blob/68fa54c7616b93d5b7cc5febaa388546fcd617e0/subprojects/gst-plugins-bad/ext/codec2json/gsth2642json.c#L836
+			header, ok := obj["slice header"].(map[string]any)
+			if !ok {
+				continue
+			}
+			// https://github.com/GStreamer/gstreamer/blob/68fa54c7616b93d5b7cc5febaa388546fcd617e0/subprojects/gst-plugins-bad/ext/codec2json/gsth2642json.c#L622
+			flag, ok := header["direct spatial mv pred flag"].(bool)
+			if ok && flag {
+				hasBFrames = true
+			}
+		}
+		close(decodeErr)
+	}()
+
+	jsonSink.SetCallbacks(&app.SinkCallbacks{
 		NewSampleFunc: func(sink *app.Sink) gst.FlowReturn {
 			sample := sink.PullSample()
 			if sample == nil {
 				return gst.FlowOK
 			}
 
-			buf := sample.GetBuffer()
-			pts := buf.PresentationTimestamp().String()
-			dts := buf.DecodingTimestamp().String()
-
-			if pts != dts {
-				hasBFrames = true
+			buf := sample.GetBuffer().Bytes()
+			_, err := bufW.Write(buf)
+			if err != nil {
+				log.Error(ctx, "failed to write to buffer", "error", err)
+				return gst.FlowError
 			}
 
 			return gst.FlowOK
···
 	}()
 
 	<-ctx.Done()
+
+	err = w.Close()
+	if err != nil {
+		return nil, fmt.Errorf("error closing writer: %w", err)
+	}
+
+	err = <-decodeErr
+	if err != nil {
+		return nil, fmt.Errorf("error decoding JSON object: %w", err)
+	}
 
 	if videoMetadata == nil {
 		return nil, fmt.Errorf("no video metadata")

pkg/media/media_data_parser_test.go

···
 )
 
 func TestMediaDataParser(t *testing.T) {
+	segmentsWithoutBFrames := []string{
+		remote.RemoteFixture("d63d26050db9a60c0944b4c2e2b1d052c4350a2a8a877324c7b0b7e7a0c1ae27/bframe-false-positive.mp4"),
+		getFixture("sample-segment.mp4"),
+	}
 	withNoGSTLeaks(t, func() {
-		// Open input file
-		inputFile, err := os.Open(getFixture("sample-segment.mp4"))
-		require.NoError(t, err)
-		defer inputFile.Close()
-		bs, err := io.ReadAll(inputFile)
-		require.NoError(t, err)
+		for _, segment := range segmentsWithoutBFrames {
+			// Open input file
+			inputFile, err := os.Open(segment)
+			require.NoError(t, err)
+			defer inputFile.Close()
+			bs, err := io.ReadAll(inputFile)
+			require.NoError(t, err)
 
-		ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"GStreamerFunc": {"ParseSegmentMediaData": 9}})
-		mediaData, err := ParseSegmentMediaData(ctx, bs)
-		require.NoError(t, err)
-		require.NotNil(t, mediaData)
-		require.False(t, mediaData.Video[0].BFrames, "Video should not have BFrames")
-		require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty")
+			ctx := log.WithDebugValue(context.Background(), map[string]map[string]int{"GStreamerFunc": {"ParseSegmentMediaData": 9}})
+			mediaData, err := ParseSegmentMediaData(ctx, bs)
+			require.NoError(t, err)
+			require.NotNil(t, mediaData)
+			require.False(t, mediaData.Video[0].BFrames, "Video should not have BFrames")
+			require.Greater(t, mediaData.Duration, int64(0), "Video duration should not be empty")
+		}
 	})
 }