From b8b16441d192d9b7ed2fbc028cf45559e779a375 Mon Sep 17 00:00:00 2001 From: MickaelK Date: Tue, 2 Jun 2026 01:30:30 +1000 Subject: [PATCH] feature (plg_video_transcoder): libav + ffmpeg strategy --- .../pages/viewerpage/application_video.js | 1 + server/plugin/plg_video_transcoder/README.md | 5 + .../plg_video_transcoder/ffmpeg/constant.go | 13 + .../{ => ffmpeg}/handler.go | 46 +- .../transcoder.go} | 60 +-- .../plugin/plg_video_transcoder/index.cgo.go | 21 + server/plugin/plg_video_transcoder/index.go | 90 ++-- .../plg_video_transcoder/index.nocgo.go | 30 ++ .../plg_video_transcoder/libav/constant.go | 22 + .../plg_video_transcoder/libav/handler.go | 86 ++++ .../plg_video_transcoder/libav/pipeline.go | 470 ++++++++++++++++++ .../plg_video_transcoder/libav/transcoder.go | 101 ++++ .../plg_video_transcoder/transcoder_cgo.go | 102 ---- .../transcoder_pipeline_cgo.go | 421 ---------------- 14 files changed, 847 insertions(+), 621 deletions(-) create mode 100644 server/plugin/plg_video_transcoder/README.md create mode 100644 server/plugin/plg_video_transcoder/ffmpeg/constant.go rename server/plugin/plg_video_transcoder/{ => ffmpeg}/handler.go (66%) rename server/plugin/plg_video_transcoder/{transcoder_cli.go => ffmpeg/transcoder.go} (83%) create mode 100644 server/plugin/plg_video_transcoder/index.cgo.go create mode 100644 server/plugin/plg_video_transcoder/index.nocgo.go create mode 100644 server/plugin/plg_video_transcoder/libav/constant.go create mode 100644 server/plugin/plg_video_transcoder/libav/handler.go create mode 100644 server/plugin/plg_video_transcoder/libav/pipeline.go create mode 100644 server/plugin/plg_video_transcoder/libav/transcoder.go delete mode 100644 server/plugin/plg_video_transcoder/transcoder_cgo.go delete mode 100644 server/plugin/plg_video_transcoder/transcoder_pipeline_cgo.go diff --git a/public/assets/pages/viewerpage/application_video.js b/public/assets/pages/viewerpage/application_video.js index 6d87aba5..c3c425c3 100644 --- a/public/assets/pages/viewerpage/application_video.js +++ b/public/assets/pages/viewerpage/application_video.js @@ -140,6 +140,7 @@ export default function(render, { mime, getFilename, getDownloadUrl }) { const hls = new Hls({ debug: !!new URLSearchParams(location.search).get("debug"), manifestLoadPolicy: loadPolicy, + maxMaxBufferLength: 20, }); const sources = window.overrides["video-map-sources"]([{ src: getDownloadUrl(), diff --git a/server/plugin/plg_video_transcoder/README.md b/server/plugin/plg_video_transcoder/README.md new file mode 100644 index 00000000..ee38c030 --- /dev/null +++ b/server/plugin/plg_video_transcoder/README.md @@ -0,0 +1,5 @@ +The transcoder has 2 strategies that each have their own pro / cons: +- libav implemenetation in C without the ffmpeg wrapper +- ffmpeg implementation which relies on the ffmpeg CLI to be present + +Notes: the ffmpeg implementation is 50% slower and has a couple edge case in the audio transition between chunks that are not there in libav (that's why we separate the audio / video in the ffmpeg impl and have long chunks for audio) diff --git a/server/plugin/plg_video_transcoder/ffmpeg/constant.go b/server/plugin/plg_video_transcoder/ffmpeg/constant.go new file mode 100644 index 00000000..7a3f4ea6 --- /dev/null +++ b/server/plugin/plg_video_transcoder/ffmpeg/constant.go @@ -0,0 +1,13 @@ +package ffmpeg + +const ( + HLS_VIDEO_SEGMENT_LENGTH = 5 + HLS_AUDIO_SEGMENT_LENGTH = 8 * 10 + VIDEO_MAX_HEIGHT = 720 + AUDIO_BITRATE = 128000 +) + +var ( + VIDEO_CACHE_PATH = "data/cache/video/" + ENCODER string = "" +) diff --git a/server/plugin/plg_video_transcoder/handler.go b/server/plugin/plg_video_transcoder/ffmpeg/handler.go similarity index 66% rename from server/plugin/plg_video_transcoder/handler.go rename to server/plugin/plg_video_transcoder/ffmpeg/handler.go index 57c6ada1..4435cdf3 100644 --- a/server/plugin/plg_video_transcoder/handler.go +++ b/server/plugin/plg_video_transcoder/ffmpeg/handler.go @@ -1,4 +1,4 @@ -package plg_video_transcoder +package ffmpeg import ( "fmt" @@ -8,10 +8,41 @@ import ( "strconv" . "github.com/mickael-kerjean/filestash/server/common" + . "github.com/mickael-kerjean/filestash/server/middleware" "github.com/gorilla/mux" ) +func MasterPlaylist(cacheName string) string { + master := "#EXTM3U\n" + master += "#EXT-X-VERSION:6\n" + master += fmt.Sprintf(`#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",NAME="default",DEFAULT=YES,AUTOSELECT=YES,URI="%s"`+"\n", WithBase(fmt.Sprintf("/hls/audio.m3u8?path=%s", cacheName))) + master += `#EXT-X-STREAM-INF:BANDWIDTH=2500000,CODECS="avc1.64001f,mp4a.40.2",AUDIO="aud"` + "\n" + master += fmt.Sprintf(WithBase("/hls/video.m3u8?path=%s\n"), cacheName) + return master +} + +func RegisterRoutes(r *mux.Router, dir string, enc string) { + VIDEO_CACHE_PATH = dir + ENCODER = enc + r.PathPrefix(WithBase("/hls/audio.m3u8")).Handler(NewMiddlewareChain( + playlistAudioHandler, + []Middleware{SecureHeaders}, + )).Methods("GET") + r.PathPrefix(WithBase("/hls/video.m3u8")).Handler(NewMiddlewareChain( + playlistVideoHandler, + []Middleware{SecureHeaders}, + )).Methods("GET") + r.PathPrefix(WithBase("/hls/video_{segment}.ts")).Handler(NewMiddlewareChain( + hlsVideoHandler, + []Middleware{SecureHeaders}, + )).Methods("GET") + r.PathPrefix(WithBase("/hls/audio_{segment}.ts")).Handler(NewMiddlewareChain( + hlsAudioHandler, + []Middleware{SecureHeaders}, + )).Methods("GET") +} + func playlistVideoHandler(ctx *App, res http.ResponseWriter, req *http.Request) { cacheName := req.URL.Query().Get("path") cachePath := GetAbsolutePath(VIDEO_CACHE_PATH, cacheName) @@ -37,7 +68,7 @@ func playlistVideoHandler(ctx *App, res http.ResponseWriter, req *http.Request) float64(HLS_VIDEO_SEGMENT_LENGTH), duration-float64(i*HLS_VIDEO_SEGMENT_LENGTH), )) - response += fmt.Sprintf("/hls/video_%d.ts?path=%s\n", i, cacheName) + response += fmt.Sprintf(WithBase("/hls/video_%d.ts?path=%s\n"), i, cacheName) } response += "#EXT-X-ENDLIST\n" res.Header().Set("Content-Type", "application/x-mpegURL") @@ -69,7 +100,7 @@ func playlistAudioHandler(ctx *App, res http.ResponseWriter, req *http.Request) float64(HLS_AUDIO_SEGMENT_LENGTH), duration-float64(i*HLS_AUDIO_SEGMENT_LENGTH), )) - response += fmt.Sprintf("/hls/audio_%d.ts?path=%s\n", i, cacheName) + response += fmt.Sprintf(WithBase("/hls/audio_%d.ts?path=%s\n"), i, cacheName) } response += "#EXT-X-ENDLIST\n" res.Header().Set("Content-Type", "application/x-mpegURL") @@ -88,7 +119,7 @@ func hlsAudioHandler(ctx *App, res http.ResponseWriter, req *http.Request) { return } res.Header().Set("Content-Type", "video/mp2t") - if err := transcodeAudioSegment(cachePath, segmentNumber, res); err != nil { + if err := transcodeAudioSegment(req.Context(), cachePath, segmentNumber, res); err != nil { Log.Error("plg_video_transcoder::audio::run %s", err.Error()) } } @@ -100,17 +131,14 @@ func hlsVideoHandler(ctx *App, res http.ResponseWriter, req *http.Request) { res.WriteHeader(http.StatusBadRequest) return } - cachePath := GetAbsolutePath( - VIDEO_CACHE_PATH, - req.URL.Query().Get("path"), - ) + cachePath := GetAbsolutePath(VIDEO_CACHE_PATH, req.URL.Query().Get("path")) if _, err := os.Stat(cachePath); os.IsNotExist(err) { Log.Info("[plugin hls]: invalid video") res.WriteHeader(http.StatusServiceUnavailable) return } res.Header().Set("Content-Type", "video/mp2t") - if err := transcodeVideoSegment(cachePath, segmentNumber, res); err != nil { + if err := transcodeVideoSegment(req.Context(), cachePath, segmentNumber, res); err != nil { Log.Error("plg_video_transcoder::video::run %s", err.Error()) } } diff --git a/server/plugin/plg_video_transcoder/transcoder_cli.go b/server/plugin/plg_video_transcoder/ffmpeg/transcoder.go similarity index 83% rename from server/plugin/plg_video_transcoder/transcoder_cli.go rename to server/plugin/plg_video_transcoder/ffmpeg/transcoder.go index 8a101b11..4e947ebd 100644 --- a/server/plugin/plg_video_transcoder/transcoder_cli.go +++ b/server/plugin/plg_video_transcoder/ffmpeg/transcoder.go @@ -1,25 +1,18 @@ -//go:build !cgo - -package plg_video_transcoder +package ffmpeg import ( + "context" "encoding/json" "errors" "fmt" "io" "os/exec" "strconv" - "strings" . "github.com/mickael-kerjean/filestash/server/common" ) -const ( - VIDEO_MAX_HEIGHT = 720 - AUDIO_BITRATE = 128000 -) - -func transcodeVideoSegment(cachePath string, segmentNumber int, w io.Writer) error { +func transcodeVideoSegment(ctx context.Context, cachePath string, segmentNumber int, w io.Writer) error { start := segmentNumber * HLS_VIDEO_SEGMENT_LENGTH args := []string{ "-hide_banner", "-loglevel", "error", @@ -33,7 +26,7 @@ func transcodeVideoSegment(cachePath string, segmentNumber int, w io.Writer) err "-output_ts_offset", fmt.Sprintf("%d.00", start), } - switch video_encoder() { + switch ENCODER { case "h264_vaapi": args = append([]string{ "-init_hw_device", "vaapi=va:/dev/dri/renderD128", @@ -71,10 +64,10 @@ func transcodeVideoSegment(cachePath string, segmentNumber int, w io.Writer) err } args = append(args, "-f", "mpegts", "pipe:1") - return runFFmpeg(args, w) + return runFFmpeg(ctx, args, w) } -func transcodeAudioSegment(cachePath string, segmentNumber int, w io.Writer) error { +func transcodeAudioSegment(ctx context.Context, cachePath string, segmentNumber int, w io.Writer) error { start := segmentNumber * HLS_AUDIO_SEGMENT_LENGTH args := []string{ "-hide_banner", "-loglevel", "error", @@ -87,7 +80,27 @@ func transcodeAudioSegment(cachePath string, segmentNumber int, w io.Writer) err "-output_ts_offset", fmt.Sprintf("%d.00", start), "-f", "mpegts", "pipe:1", } - return runFFmpeg(args, w) + return runFFmpeg(ctx, args, w) +} + +func runFFmpeg(ctx context.Context, args []string, w io.Writer) error { + cmd := exec.CommandContext(ctx, "ffmpeg", args...) + cmd.Stdout = w + stderr, err := cmd.StderrPipe() + if err != nil { + return err + } + if err := cmd.Start(); err != nil { + return fmt.Errorf("ffmpeg start: %w", err) + } + msg, _ := io.ReadAll(stderr) + if err := cmd.Wait(); err != nil { + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("ffmpeg: %w: %s", err, msg) + } + return nil } func probeDuration(path string) (float64, error) { @@ -118,22 +131,3 @@ func probeDuration(path string) (float64, error) { } return d, nil } - -func runFFmpeg(args []string, w io.Writer) error { - cmd := exec.Command("ffmpeg", args...) - cmd.Stdout = w - stderr, err := cmd.StderrPipe() - if err != nil { - return err - } - if err := cmd.Start(); err != nil { - return fmt.Errorf("ffmpeg start: %w", err) - } - msg, _ := io.ReadAll(stderr) - if err := cmd.Wait(); err != nil { - if flat := strings.Join(strings.Fields(string(msg)), " "); !strings.Contains(flat, "Broken pipe") { - return fmt.Errorf("ffmpeg: %w: %s", err, flat) - } - } - return nil -} diff --git a/server/plugin/plg_video_transcoder/index.cgo.go b/server/plugin/plg_video_transcoder/index.cgo.go new file mode 100644 index 00000000..86de29c3 --- /dev/null +++ b/server/plugin/plg_video_transcoder/index.cgo.go @@ -0,0 +1,21 @@ +//go:build cgo + +package plg_video_transcoder + +import ( + "github.com/gorilla/mux" + + "github.com/mickael-kerjean/filestash/server/plugin/plg_video_transcoder/libav" +) + +func isActive() bool { + return true +} + +func servePlaylist(cacheName string) string { + return libav.MasterPlaylist(cacheName) +} + +func serveHLSChunks(r *mux.Router) { + libav.RegisterRoutes(r, VIDEO_CACHE_PATH, video_encoder()) +} diff --git a/server/plugin/plg_video_transcoder/index.go b/server/plugin/plg_video_transcoder/index.go index 765bfa16..05b07b75 100644 --- a/server/plugin/plg_video_transcoder/index.go +++ b/server/plugin/plg_video_transcoder/index.go @@ -9,15 +9,11 @@ import ( "time" . "github.com/mickael-kerjean/filestash/server/common" - . "github.com/mickael-kerjean/filestash/server/middleware" "github.com/gorilla/mux" ) const ( - HLS_VIDEO_SEGMENT_LENGTH = 10 - HLS_AUDIO_SEGMENT_LENGTH = 8 * 8 - CLEAR_CACHE_AFTER = 12 VIDEO_CACHE_PATH = "data/cache/video/" ) @@ -26,7 +22,7 @@ func init() { Hooks.Register.Onload(func() { blacklist_format() video_encoder() - if !plugin_enable() { + if !plugin_enable() || !isActive() { return } @@ -34,12 +30,16 @@ func init() { os.RemoveAll(cachePath) os.MkdirAll(cachePath, os.ModePerm) - Hooks.Register.ProcessFileContentBeforeSend(serverPlaylist) - Hooks.Register.HttpEndpoint(serverEndpoints) + Hooks.Register.ProcessFileContentBeforeSend(createPlaylist) + Hooks.Register.HttpEndpoint(func(r *mux.Router) error { + r.HandleFunc(OverrideVideoSourceMapper, createVideoMap) + serveHLSChunks(r) + return nil + }) }) } -func serverPlaylist(reader io.ReadCloser, ctx *App, res *http.ResponseWriter, req *http.Request) (io.ReadCloser, bool, error) { +func createPlaylist(reader io.ReadCloser, ctx *App, res *http.ResponseWriter, req *http.Request) (io.ReadCloser, bool, error) { query := req.URL.Query() if query.Get("transcode") != "hls" { return reader, false, nil @@ -54,59 +54,37 @@ func serverPlaylist(reader io.ReadCloser, ctx *App, res *http.ResponseWriter, re VIDEO_CACHE_PATH, cacheName, ) - f, err := os.OpenFile(cachePath, os.O_CREATE|os.O_RDWR, os.ModePerm) - if err != nil { - return reader, false, err + if _, err := os.Stat(cachePath); os.IsNotExist(err) { + f, err := os.OpenFile(cachePath, os.O_CREATE|os.O_RDWR, os.ModePerm) + if err != nil { + return reader, false, err + } + io.Copy(f, reader) + f.Close() + time.AfterFunc(CLEAR_CACHE_AFTER*time.Hour, func() { os.Remove(cachePath) }) } - io.Copy(f, reader) reader.Close() - f.Close() - time.AfterFunc(CLEAR_CACHE_AFTER*time.Hour, func() { os.Remove(cachePath) }) - response := "#EXTM3U\n" - response += "#EXT-X-VERSION:6\n" - response += fmt.Sprintf(`#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",NAME="default",DEFAULT=YES,AUTOSELECT=YES,URI="/hls/audio.m3u8?path=%s"`+"\n", cacheName) - response += `#EXT-X-STREAM-INF:BANDWIDTH=2500000,CODECS="avc1.64001f,mp4a.40.2",AUDIO="aud"` + "\n" - response += fmt.Sprintf("/hls/video.m3u8?path=%s\n", cacheName) (*res).Header().Set("Content-Type", "application/x-mpegURL") - return NewReadCloserFromBytes([]byte(response)), true, nil + return NewReadCloserFromBytes([]byte(servePlaylist(cacheName))), true, nil } -func serverEndpoints(r *mux.Router) error { - r.PathPrefix("/hls/audio.m3u8").Handler(NewMiddlewareChain( - playlistAudioHandler, - []Middleware{SecureHeaders}, - )).Methods("GET") - r.PathPrefix("/hls/video.m3u8").Handler(NewMiddlewareChain( - playlistVideoHandler, - []Middleware{SecureHeaders}, - )).Methods("GET") - r.PathPrefix("/hls/video_{segment}.ts").Handler(NewMiddlewareChain( - hlsVideoHandler, - []Middleware{SecureHeaders}, - )).Methods("GET") - r.PathPrefix("/hls/audio_{segment}.ts").Handler(NewMiddlewareChain( - hlsAudioHandler, - []Middleware{SecureHeaders}, - )).Methods("GET") - r.HandleFunc(OverrideVideoSourceMapper, func(res http.ResponseWriter, req *http.Request) { - res.Header().Set("Content-Type", GetMimeType(req.URL.String())) - if plugin_enable() == false { - return - } - res.Write([]byte(`window.overrides["video-map-sources"] = function(sources){`)) - res.Write([]byte(` return sources.map(function(source){`)) +func createVideoMap(res http.ResponseWriter, req *http.Request) { + res.Header().Set("Content-Type", GetMimeType(req.URL.String())) + if plugin_enable() == false { + return + } + res.Write([]byte(`window.overrides["video-map-sources"] = function(sources){`)) + res.Write([]byte(` return sources.map(function(source){`)) - blacklists := strings.Split(blacklist_format(), ",") - for i := 0; i < len(blacklists); i++ { - blacklists[i] = strings.TrimSpace(blacklists[i]) - res.Write([]byte(fmt.Sprintf(`if(source.type == "%s"){ return source; } `, GetMimeType("."+blacklists[i])))) - } - res.Write([]byte(` source.src = source.src + "&transcode=hls";`)) - res.Write([]byte(` source.type = "application/x-mpegURL";`)) - res.Write([]byte(` return source;`)) - res.Write([]byte(` })`)) - res.Write([]byte(`}`)) - }) - return nil + blacklists := strings.Split(blacklist_format(), ",") + for i := 0; i < len(blacklists); i++ { + blacklists[i] = strings.TrimSpace(blacklists[i]) + res.Write([]byte(fmt.Sprintf(`if(source.type == "%s"){ return source; } `, GetMimeType("."+blacklists[i])))) + } + res.Write([]byte(` source.src = source.src + "&transcode=hls";`)) + res.Write([]byte(` source.type = "application/x-mpegURL";`)) + res.Write([]byte(` return source;`)) + res.Write([]byte(` })`)) + res.Write([]byte(`}`)) } diff --git a/server/plugin/plg_video_transcoder/index.nocgo.go b/server/plugin/plg_video_transcoder/index.nocgo.go new file mode 100644 index 00000000..30035e7f --- /dev/null +++ b/server/plugin/plg_video_transcoder/index.nocgo.go @@ -0,0 +1,30 @@ +//go:build !cgo + +package plg_video_transcoder + +import ( + "os/exec" + + . "github.com/mickael-kerjean/filestash/server/common" + "github.com/mickael-kerjean/filestash/server/plugin/plg_video_transcoder/ffmpeg" + + "github.com/gorilla/mux" +) + +func isActive() bool { + for _, bin := range []string{"ffmpeg", "ffprobe"} { + if _, err := exec.LookPath(bin); err != nil { + Log.Warning("plg_video_transcoder: %q not found in $PATH, transcoding disabled", bin) + return false + } + } + return true +} + +func servePlaylist(cacheName string) string { + return ffmpeg.MasterPlaylist(cacheName) +} + +func serveHLSChunks(r *mux.Router) { + ffmpeg.RegisterRoutes(r, VIDEO_CACHE_PATH, video_encoder()) +} diff --git a/server/plugin/plg_video_transcoder/libav/constant.go b/server/plugin/plg_video_transcoder/libav/constant.go new file mode 100644 index 00000000..205038e1 --- /dev/null +++ b/server/plugin/plg_video_transcoder/libav/constant.go @@ -0,0 +1,22 @@ +package libav + +import ( + "runtime" +) + +const ( + VIDEO_MAX_HEIGHT = 720 + AUDIO_BITRATE = 128000 +) + +var ( + HLS_SEGMENT_LENGTH = 5 + VIDEO_CACHE_PATH = "data/cache/video/" + ENCODER string = "" +) + +func init() { + if runtime.NumCPU() <= 4 && runtime.GOARCH == "arm" { // eg: a raspberry pi + HLS_SEGMENT_LENGTH = 10 + } +} diff --git a/server/plugin/plg_video_transcoder/libav/handler.go b/server/plugin/plg_video_transcoder/libav/handler.go new file mode 100644 index 00000000..f3d83be1 --- /dev/null +++ b/server/plugin/plg_video_transcoder/libav/handler.go @@ -0,0 +1,86 @@ +package libav + +import ( + "fmt" + "math" + "net/http" + "os" + "strconv" + + . "github.com/mickael-kerjean/filestash/server/common" + . "github.com/mickael-kerjean/filestash/server/middleware" + + "github.com/gorilla/mux" +) + +func MasterPlaylist(cacheName string) string { + master := "#EXTM3U\n" + master += "#EXT-X-VERSION:3\n" + master += `#EXT-X-STREAM-INF:BANDWIDTH=2628000,CODECS="avc1.64001f,mp4a.40.2"` + "\n" + master += fmt.Sprintf(WithBase("/hls/index.m3u8?path=%s\n"), cacheName) + return master +} + +func RegisterRoutes(r *mux.Router, dir string, enc string) { + VIDEO_CACHE_PATH = dir + ENCODER = enc + r.PathPrefix(WithBase("/hls/index.m3u8")).Handler(NewMiddlewareChain( + playlistHandler, + []Middleware{SecureHeaders}, + )).Methods("GET") + r.PathPrefix(WithBase("/hls/segment_{segment}.ts")).Handler(NewMiddlewareChain( + segmentHandler, + []Middleware{SecureHeaders}, + )).Methods("GET") +} + +func playlistHandler(ctx *App, res http.ResponseWriter, req *http.Request) { + cacheName := req.URL.Query().Get("path") + cachePath := GetAbsolutePath(VIDEO_CACHE_PATH, cacheName) + if _, err := os.Stat(cachePath); os.IsNotExist(err) { + res.WriteHeader(http.StatusServiceUnavailable) + return + } + duration, err := probeDuration(cachePath) + if err != nil { + res.WriteHeader(http.StatusInternalServerError) + return + } + + response := "#EXTM3U\n" + response += "#EXT-X-VERSION:3\n" + response += "#EXT-X-MEDIA-SEQUENCE:0\n" + response += "#EXT-X-ALLOW-CACHE:YES\n" + response += "#EXT-X-PLAYLIST-TYPE:VOD\n" + response += fmt.Sprintf("#EXT-X-TARGETDURATION:%d\n", HLS_SEGMENT_LENGTH) + total := int(math.Ceil(duration / float64(HLS_SEGMENT_LENGTH))) + for i := 0; i < total; i++ { + response += fmt.Sprintf("#EXTINF:%.4f, nodesc\n", math.Min( + float64(HLS_SEGMENT_LENGTH), + duration-float64(i*HLS_SEGMENT_LENGTH), + )) + response += fmt.Sprintf(WithBase("/hls/segment_%d.ts?path=%s\n"), i, cacheName) + } + response += "#EXT-X-ENDLIST\n" + res.Header().Set("Content-Type", "application/x-mpegURL") + res.Write([]byte(response)) +} + +func segmentHandler(ctx *App, res http.ResponseWriter, req *http.Request) { + segmentNumber, err := strconv.Atoi(mux.Vars(req)["segment"]) + if err != nil { + Log.Info("[plugin hls] invalid segment request '%s'", mux.Vars(req)["segment"]) + res.WriteHeader(http.StatusBadRequest) + return + } + cachePath := GetAbsolutePath(VIDEO_CACHE_PATH, req.URL.Query().Get("path")) + if _, err := os.Stat(cachePath); os.IsNotExist(err) { + Log.Info("[plugin hls]: invalid video") + res.WriteHeader(http.StatusServiceUnavailable) + return + } + res.Header().Set("Content-Type", "video/mp2t") + if err := transcodeSegment(cachePath, segmentNumber, res); err != nil { + Log.Error("plg_video_transcoder::segment::run %s", err.Error()) + } +} diff --git a/server/plugin/plg_video_transcoder/libav/pipeline.go b/server/plugin/plg_video_transcoder/libav/pipeline.go new file mode 100644 index 00000000..1b65b2a9 --- /dev/null +++ b/server/plugin/plg_video_transcoder/libav/pipeline.go @@ -0,0 +1,470 @@ +package libav + +import ( + "errors" + "fmt" + "io" + "math" + + "github.com/asticode/go-astikit" + "github.com/mickael-kerjean/go-astiav" +) + +type stream struct { + inStream *astiav.Stream + decCtx *astiav.CodecContext + encCodec *astiav.Codec + encCtx *astiav.CodecContext + encOpts *astiav.Dictionary + outStream *astiav.Stream + + srcCtx *astiav.BuffersrcFilterContext + sinkCtx *astiav.BuffersinkFilterContext + hwDevice *astiav.HardwareDeviceContext + + feedStart, feedEnd int64 + trimStart, trimEnd int64 + trim bool + lastDts int64 + done bool +} + +type pipeline struct { + inFmt *astiav.FormatContext + outFmt *astiav.FormatContext + streams []*stream + byIndex map[int]*stream + closer *astikit.Closer +} + +func NewPipeline(path string) (p *pipeline, err error) { + p = &pipeline{closer: astikit.NewCloser(), byIndex: map[int]*stream{}} + defer func() { + if err != nil { + p.Close() + } + }() + + p.inFmt = astiav.AllocFormatContext() + if p.inFmt == nil { + return p, errors.New("input fmt ctx is nil") + } + p.closer.Add(p.inFmt.Free) + if err = p.inFmt.OpenInput(path, nil, nil); err != nil { + return p, fmt.Errorf("open input: %w", err) + } + p.closer.Add(p.inFmt.CloseInput) + if err = p.inFmt.FindStreamInfo(nil); err != nil { + return p, fmt.Errorf("find stream info: %w", err) + } + return p, nil +} + +func (p *pipeline) Device(s *stream, hwType astiav.HardwareDeviceType, hwPixFmt astiav.PixelFormat, w, h int) error { + hwDevice, err := astiav.CreateHardwareDeviceContext(hwType, "", nil, 0) + if err != nil { + return fmt.Errorf("%s device: %w", hwType.Name(), err) + } + p.closer.Add(hwDevice.Free) + hwFrames := astiav.AllocHardwareFramesContext(hwDevice) + if hwFrames == nil { + return fmt.Errorf("%s frames context is nil", hwType.Name()) + } + defer hwFrames.Free() + hwFrames.SetHardwarePixelFormat(hwPixFmt) + hwFrames.SetSoftwarePixelFormat(astiav.PixelFormatNv12) + hwFrames.SetWidth(w) + hwFrames.SetHeight(h) + if err := hwFrames.Initialize(); err != nil { + return fmt.Errorf("%s frames init: %w", hwType.Name(), err) + } + s.encCtx.SetPixelFormat(hwPixFmt) + s.encCtx.SetHardwareFramesContext(hwFrames) + s.hwDevice = hwDevice + return nil +} + +func (p *pipeline) AddStream(mediaType astiav.MediaType, encName string) (*stream, error) { + s := &stream{} + inStream, decCodec, err := p.inFmt.FindBestStream(mediaType, -1, -1) + if err != nil { + return nil, fmt.Errorf("no %s stream: %w", mediaType, err) + } + s.inStream = inStream + + if s.encCodec = astiav.FindEncoderByName(encName); s.encCodec == nil { + return nil, fmt.Errorf("encoder %q not found", encName) + } + if s.encCtx = astiav.AllocCodecContext(s.encCodec); s.encCtx == nil { + return nil, errors.New("enc ctx is nil") + } + s.encOpts = astiav.NewDictionary() + p.closer.Add(s.encOpts.Free) + p.closer.Add(s.encCtx.Free) + + if mediaType == astiav.MediaTypeVideo && encName == "h264_v4l2m2m" { + decCodec = astiav.FindDecoderByName("h264_v4l2m2m") + } + if s.decCtx = astiav.AllocCodecContext(decCodec); s.decCtx == nil { + return nil, errors.New("dec ctx is nil") + } + p.closer.Add(s.decCtx.Free) + if err = s.inStream.CodecParameters().ToCodecContext(s.decCtx); err != nil { + return nil, fmt.Errorf("dec params: %w", err) + } + tb := s.inStream.TimeBase() + s.decCtx.SetTimeBase(tb) + s.encCtx.SetTimeBase(tb) + + switch mediaType { + case astiav.MediaTypeVideo: + s.decCtx.SetFramerate(p.inFmt.GuessFrameRate(s.inStream, nil)) + s.encCtx.SetFramerate(s.decCtx.Framerate()) + s.encCtx.SetSampleAspectRatio(s.decCtx.SampleAspectRatio()) + case astiav.MediaTypeAudio: + channelLayout := s.decCtx.ChannelLayout() + if v := s.encCodec.ChannelLayouts(); len(v) > 0 { + channelLayout = v[0] + } + sampleFmt := s.decCtx.SampleFormat() + if v := s.encCodec.SampleFormats(); len(v) > 0 { + sampleFmt = v[0] + } + s.encCtx.SetChannelLayout(channelLayout) + s.encCtx.SetSampleFormat(sampleFmt) + s.encCtx.SetSampleRate(s.decCtx.SampleRate()) + } + if err = s.decCtx.Open(decCodec, nil); err != nil { + return nil, fmt.Errorf("open decoder: %w", err) + } + + p.streams = append(p.streams, s) + p.byIndex[s.inStream.Index()] = s + return s, nil +} + +func (p *pipeline) Close() { + p.closer.Close() +} + +func (p *pipeline) Open(w io.Writer) (err error) { + if p.outFmt, err = astiav.AllocOutputFormatContext(nil, "mpegts", ""); err != nil { + return fmt.Errorf("alloc out fmt: %w", err) + } + p.closer.Add(p.outFmt.Free) + ioCtx, err := astiav.AllocIOContext( + 65536, true, nil, nil, + func(b []byte) (int, error) { + return w.Write(b) + }, + ) + if err != nil { + return fmt.Errorf("alloc io ctx: %w", err) + } + p.outFmt.SetPb(ioCtx) + p.closer.Add(ioCtx.Free) + + for _, s := range p.streams { + if err := s.encCtx.Open(s.encCodec, s.encOpts); err != nil { + return fmt.Errorf("open encoder: %w", err) + } + if s.outStream = p.outFmt.NewStream(nil); s.outStream == nil { + return errors.New("out stream is nil") + } + if err := s.outStream.CodecParameters().FromCodecContext(s.encCtx); err != nil { + return fmt.Errorf("out stream params: %w", err) + } + s.outStream.SetTimeBase(s.encCtx.TimeBase()) + } + return nil +} + +func (p *pipeline) Build(s *stream, graphSpec string) (err error) { + if s == nil { + return nil + } + fg := astiav.AllocFilterGraph() + if fg == nil { + return errors.New("filter graph is nil") + } + p.closer.Add(fg.Free) + + var ( + srcName = "" + sinkName = "" + srcParams = astiav.AllocBuffersrcFilterContextParameters() + ) + srcParams.SetTimeBase(s.inStream.TimeBase()) + defer srcParams.Free() + switch s.decCtx.MediaType() { + case astiav.MediaTypeVideo: + srcName, sinkName = "buffer", "buffersink" + srcParams.SetWidth(s.decCtx.Width()) + srcParams.SetHeight(s.decCtx.Height()) + srcParams.SetPixelFormat(s.decCtx.PixelFormat()) + srcParams.SetSampleAspectRatio(s.decCtx.SampleAspectRatio()) + case astiav.MediaTypeAudio: + srcName, sinkName = "abuffer", "abuffersink" + srcParams.SetChannelLayout(s.decCtx.ChannelLayout()) + srcParams.SetSampleFormat(s.decCtx.SampleFormat()) + srcParams.SetSampleRate(s.decCtx.SampleRate()) + default: + return fmt.Errorf("unsupported media type %s", s.decCtx.MediaType()) + } + srcF, sinkF := astiav.FindFilterByName(srcName), astiav.FindFilterByName(sinkName) + if srcF == nil || sinkF == nil { + return fmt.Errorf("filter %s/%s missing", srcName, sinkName) + } + if s.srcCtx, err = fg.NewBuffersrcFilterContext(srcF, "in"); err != nil { + return fmt.Errorf("new buffersrc: %w", err) + } + if s.sinkCtx, err = fg.NewBuffersinkFilterContext(sinkF, "out"); err != nil { + return fmt.Errorf("new buffersink: %w", err) + } + if err := s.srcCtx.SetParameters(srcParams); err != nil { + return fmt.Errorf("buffersrc params: %w", err) + } + if err := s.srcCtx.Initialize(nil); err != nil { + return fmt.Errorf("buffersrc init: %w", err) + } + + seg, err := fg.ParseSegment(graphSpec) + if err != nil { + return fmt.Errorf("segment parse %q: %w", graphSpec, err) + } + defer seg.Free() + if err := seg.CreateFilters(0); err != nil { + return fmt.Errorf("segment create filters: %w", err) + } + if err := seg.ApplyOpts(0); err != nil { + return fmt.Errorf("segment apply opts: %w", err) + } + if s.hwDevice != nil { + for _, chain := range seg.Chains() { + for _, params := range chain.Filters() { + if fc := params.FilterContext(); fc != nil { + fc.SetHardwareDeviceContext(s.hwDevice) + } + } + } + } + if err := seg.Init(0); err != nil { + return fmt.Errorf("segment init: %w", err) + } + segIn, segOut := astiav.AllocFilterInOut(), astiav.AllocFilterInOut() + defer segIn.Free() + defer segOut.Free() + if err := seg.Link(0, segIn, segOut); err != nil { + return fmt.Errorf("segment link: %w", err) + } + for io := segIn; io != nil; io = io.Next() { + if err := s.srcCtx.FilterContext().LinkTo(0, io.FilterContext(), 0); err != nil { + return fmt.Errorf("link buffersrc: %w", err) + } + } + for io := segOut; io != nil; io = io.Next() { + if err := io.FilterContext().LinkTo(0, s.sinkCtx.FilterContext(), 0); err != nil { + return fmt.Errorf("link buffersink: %w", err) + } + } + if err := fg.Configure(); err != nil { + return fmt.Errorf("filter configure: %w", err) + } + return nil +} + +func (p *pipeline) Run(startSec, endSec int) error { + for _, s := range p.streams { + tb := s.inStream.TimeBase() + startPts := int64(startSec) * int64(tb.Den()) / int64(tb.Num()) + endPts := int64(endSec) * int64(tb.Den()) / int64(tb.Num()) + s.feedStart, s.feedEnd = startPts, endPts + s.trim, s.done = false, false + s.lastDts = math.MinInt64 + if s.decCtx.MediaType() == astiav.MediaTypeAudio { + const marginSec = 0.5 + margin := int64(marginSec * float64(tb.Den()) / float64(tb.Num())) + if s.feedStart = startPts - margin; s.feedStart < 0 { + s.feedStart = 0 + } + s.feedEnd = endPts + margin + s.trimStart, s.trimEnd = startPts, endPts + s.trim = true + } + } + + seekSec := float64(startSec) - 0.5 + if seekSec < 0 { + seekSec = 0 + } + seekTs := int64(seekSec * float64(astiav.TimeBase)) + if err := p.inFmt.SeekFrame(-1, seekTs, astiav.NewSeekFlags(astiav.SeekFlagBackward)); err != nil { + return fmt.Errorf("seek: %w", err) + } + if err := p.outFmt.WriteHeader(nil); err != nil { + return fmt.Errorf("write header: %w", err) + } + + pkt := astiav.AllocPacket() + defer pkt.Free() + encPkt := astiav.AllocPacket() + defer encPkt.Free() + decFrame := astiav.AllocFrame() + defer decFrame.Free() + filtFrame := astiav.AllocFrame() + defer filtFrame.Free() + + drained := func(err error) bool { + return errors.Is(err, astiav.ErrEagain) || errors.Is(err, astiav.ErrEof) + } + + drainEncoder := func(s *stream) error { + for { + if err := s.encCtx.ReceivePacket(encPkt); err != nil { + if drained(err) { + return nil + } + return err + } + if s.trim && encPkt.Pts() != astiav.NoPtsValue && (encPkt.Pts() < s.trimStart || encPkt.Pts() >= s.trimEnd) { + encPkt.Unref() + continue + } + encPkt.SetStreamIndex(s.outStream.Index()) + encPkt.RescaleTs(s.encCtx.TimeBase(), s.outStream.TimeBase()) + if encPkt.Dts() <= s.lastDts { + encPkt.Unref() + continue + } + s.lastDts = encPkt.Dts() + err := p.outFmt.WriteInterleavedFrame(encPkt) + encPkt.Unref() + if err != nil { + return err + } + } + } + + pushThroughFilter := func(s *stream, f *astiav.Frame) error { + if err := s.srcCtx.AddFrame(f, astiav.NewBuffersrcFlags(astiav.BuffersrcFlagKeepRef)); err != nil { + return err + } + for { + if err := s.sinkCtx.GetFrame(filtFrame, astiav.NewBuffersinkFlags()); err != nil { + if drained(err) { + return nil + } + return err + } + err := s.encCtx.SendFrame(filtFrame) + filtFrame.Unref() + if err != nil { + return err + } + if err := drainEncoder(s); err != nil { + return err + } + } + } + + drainDecoder := func(s *stream) error { + for { + if err := s.decCtx.ReceiveFrame(decFrame); err != nil { + if drained(err) { + return nil + } + return fmt.Errorf("receive frame: %w", err) + } + pts := decFrame.Pts() + if pts == math.MinInt64 { // h264 in avi should fall back to dts. + pts = decFrame.PktDts() + } + if pts >= s.feedEnd { + decFrame.Unref() + s.done = true + return nil + } + if pts < s.feedStart { + decFrame.Unref() + continue + } + decFrame.SetPts(pts) + err := pushThroughFilter(s, decFrame) + decFrame.Unref() + if err != nil { + return fmt.Errorf("filter: %w", err) + } + } + } + + allDone := func() bool { + for _, s := range p.streams { + if !s.done { + return false + } + } + return true + } + + for !allDone() { + if err := p.inFmt.ReadFrame(pkt); err != nil { + if errors.Is(err, astiav.ErrEof) { + break + } + return fmt.Errorf("read frame: %w", err) + } + s := p.byIndex[pkt.StreamIndex()] + if s == nil || s.done { + pkt.Unref() + continue + } + err := s.decCtx.SendPacket(pkt) + pkt.Unref() + if err != nil && !errors.Is(err, astiav.ErrInvaliddata) { + return fmt.Errorf("send packet: %w", err) + } + if err := drainDecoder(s); err != nil { + return err + } + } + + for _, s := range p.streams { + if !s.done { + if err := s.decCtx.SendPacket(nil); err != nil { + return fmt.Errorf("flush decoder: %w", err) + } + if err := drainDecoder(s); err != nil { + return err + } + } + if err := pushThroughFilter(s, nil); err != nil { + return fmt.Errorf("flush filter: %w", err) + } + if err := s.encCtx.SendFrame(nil); err != nil { + return fmt.Errorf("flush encoder: %w", err) + } + if err := drainEncoder(s); err != nil { + return fmt.Errorf("drain: %w", err) + } + } + if err := p.outFmt.WriteTrailer(); err != nil { + return fmt.Errorf("write trailer: %w", err) + } + return nil +} + +func probeDuration(path string) (float64, error) { + inFmt := astiav.AllocFormatContext() + if inFmt == nil { + return 0, errors.New("input format context is nil") + } + defer inFmt.Free() + if err := inFmt.OpenInput(path, nil, nil); err != nil { + return 0, err + } + defer inFmt.CloseInput() + if err := inFmt.FindStreamInfo(nil); err != nil { + return 0, err + } + return float64(inFmt.Duration()) / float64(astiav.TimeBase), nil +} diff --git a/server/plugin/plg_video_transcoder/libav/transcoder.go b/server/plugin/plg_video_transcoder/libav/transcoder.go new file mode 100644 index 00000000..6668937a --- /dev/null +++ b/server/plugin/plg_video_transcoder/libav/transcoder.go @@ -0,0 +1,101 @@ +package libav + +import ( + "fmt" + "io" + + . "github.com/mickael-kerjean/filestash/server/common" + + "github.com/mickael-kerjean/go-astiav" +) + +func init() { + astiav.SetLogLevel(astiav.LogLevelFatal) +} + +func transcodeSegment(cachePath string, segmentNumber int, w io.Writer) error { + p, err := NewPipeline(cachePath) + if err != nil { + return err + } + defer p.Close() + + video, err := p.AddStream(astiav.MediaTypeVideo, ENCODER) + if err != nil { + return err + } + videoGraph, err := configureVideoEncoder(p, video) + if err != nil { + return err + } + + audio, _ := p.AddStream(astiav.MediaTypeAudio, "aac") + configureAudioEncoder(audio) + + if err := p.Open(w); err != nil { + return err + } + if err := p.Build(video, videoGraph); err != nil { + return err + } + if audio != nil { + if err := p.Build(audio, audioGraphSpec(audio)); err != nil { + return err + } + } + return p.Run( + segmentNumber*HLS_SEGMENT_LENGTH, + (segmentNumber+1)*HLS_SEGMENT_LENGTH, + ) +} + +func configureVideoEncoder(p *pipeline, s *stream) (string, error) { + outH := min(VIDEO_MAX_HEIGHT, s.decCtx.Height()) &^ 1 + outW := (s.decCtx.Width()*outH/s.decCtx.Height() + 1) &^ 1 + s.encCtx.SetWidth(outW) + s.encCtx.SetHeight(outH) + s.encOpts.Set("force_key_frames", fmt.Sprintf("expr:gte(t,n_forced*%d.000)", HLS_SEGMENT_LENGTH), 0) + + switch s.encCodec.Name() { + case "h264_vaapi": + if err := p.Device(s, astiav.HardwareDeviceTypeVAAPI, astiav.PixelFormatVaapi, outW, outH); err != nil { + return "", err + } + return fmt.Sprintf("format=nv12,hwupload,scale_vaapi=%d:%d", outW, outH), nil + case "h264_nvenc": + if err := p.Device(s, astiav.HardwareDeviceTypeCUDA, astiav.PixelFormatCuda, outW, outH); err != nil { + return "", err + } + return fmt.Sprintf("format=nv12,hwupload_cuda,scale_cuda=%d:%d", outW, outH), nil + case "h264_v4l2m2m": + s.encCtx.SetPixelFormat(astiav.PixelFormatYuv420P) + s.encCtx.SetBitRate(2500000) + if outW == s.decCtx.Width() && outH == s.decCtx.Height() { + return "format=yuv420p", nil + } + return fmt.Sprintf("scale=%d:%d,format=yuv420p", outW, outH), nil + case "libx264": + s.encCtx.SetPixelFormat(astiav.PixelFormatYuv420P) + s.encOpts.Set("preset", "veryfast", 0) + s.encOpts.Set("x264opts", "subme=0:me_range=4:rc_lookahead=10:me=dia:no_chroma_me:8x8dct=0:partitions=none", 0) + return fmt.Sprintf("scale=%d:%d,format=yuv420p", outW, outH), nil + default: + return "", ErrNotImplemented + } +} + +func configureAudioEncoder(s *stream) { + if s == nil { + return + } + s.encCtx.SetBitRate(AUDIO_BITRATE) +} + +func audioGraphSpec(s *stream) string { + return fmt.Sprintf( + "aformat=sample_fmts=%s:channel_layouts=%s,asetnsamples=n=%d:p=0", + s.encCtx.SampleFormat().Name(), + s.encCtx.ChannelLayout().String(), + s.encCtx.FrameSize(), + ) +} diff --git a/server/plugin/plg_video_transcoder/transcoder_cgo.go b/server/plugin/plg_video_transcoder/transcoder_cgo.go deleted file mode 100644 index a5a76274..00000000 --- a/server/plugin/plg_video_transcoder/transcoder_cgo.go +++ /dev/null @@ -1,102 +0,0 @@ -//go:build cgo - -package plg_video_transcoder - -import ( - "fmt" - "io" - - . "github.com/mickael-kerjean/filestash/server/common" - - "github.com/mickael-kerjean/go-astiav" -) - -const ( - VIDEO_MAX_HEIGHT = 720 - AUDIO_BITRATE = 128000 -) - -func init() { - astiav.SetLogLevel(astiav.LogLevelFatal) -} - -func transcodeVideoSegment(cachePath string, segmentNumber int, w io.Writer) error { - p, err := NewPipeline(cachePath, astiav.MediaTypeVideo, video_encoder()) - if err != nil { - return err - } - defer p.Close() - - outH := min(VIDEO_MAX_HEIGHT, p.decCtx.Height()) &^ 1 - outW := (p.decCtx.Width()*outH/p.decCtx.Height() + 1) &^ 1 - p.encCtx.SetWidth(outW) - p.encCtx.SetHeight(outH) - p.encOpts.Set("force_key_frames", fmt.Sprintf("expr:gte(t,n_forced*%d.000)", HLS_VIDEO_SEGMENT_LENGTH), 0) - - var graphSpec string - switch p.encCodec.Name() { - case "h264_vaapi": - if err := p.Device(astiav.HardwareDeviceTypeVAAPI, astiav.PixelFormatVaapi, outW, outH); err != nil { - return err - } - graphSpec = fmt.Sprintf("format=nv12,hwupload,scale_vaapi=%d:%d", outW, outH) - case "h264_nvenc": - if err := p.Device(astiav.HardwareDeviceTypeCUDA, astiav.PixelFormatCuda, outW, outH); err != nil { - return err - } - graphSpec = fmt.Sprintf("format=nv12,hwupload_cuda,scale_cuda=%d:%d", outW, outH) - case "h264_v4l2m2m": - p.encCtx.SetPixelFormat(astiav.PixelFormatYuv420P) - p.encCtx.SetBitRate(2500000) - if outW == p.decCtx.Width() && outH == p.decCtx.Height() { - graphSpec = "format=yuv420p" - } else { - graphSpec = fmt.Sprintf("scale=%d:%d,format=yuv420p", outW, outH) - } - case "libx264": - p.encCtx.SetPixelFormat(astiav.PixelFormatYuv420P) - p.encOpts.Set("preset", "veryfast", 0) - p.encOpts.Set("x264opts", "subme=0:me_range=4:rc_lookahead=10:me=dia:no_chroma_me:8x8dct=0:partitions=none", 0) - graphSpec = fmt.Sprintf("scale=%d:%d,format=yuv420p", outW, outH) - default: - return ErrNotImplemented - } - - if err := p.Open(w); err != nil { - return err - } - if err := p.Build(graphSpec); err != nil { - return err - } - return p.Run( - segmentNumber*HLS_VIDEO_SEGMENT_LENGTH, - (segmentNumber+1)*HLS_VIDEO_SEGMENT_LENGTH, - ) -} - -func transcodeAudioSegment(cachePath string, segmentNumber int, w io.Writer) error { - p, err := NewPipeline(cachePath, astiav.MediaTypeAudio, "aac") - if err != nil { - return err - } - defer p.Close() - - p.encCtx.SetBitRate(AUDIO_BITRATE) - if err = p.Open(w); err != nil { - return err - } - graphSpec := fmt.Sprintf( - "aformat=sample_fmts=%s:channel_layouts=%s,asetnsamples=n=%d:p=0", - p.encCtx.SampleFormat().Name(), - p.encCtx.ChannelLayout().String(), - p.encCtx.FrameSize(), - ) - - if err = p.Build(graphSpec); err != nil { - return err - } - return p.Run( - segmentNumber*HLS_AUDIO_SEGMENT_LENGTH, - (segmentNumber+1)*HLS_AUDIO_SEGMENT_LENGTH, - ) -} diff --git a/server/plugin/plg_video_transcoder/transcoder_pipeline_cgo.go b/server/plugin/plg_video_transcoder/transcoder_pipeline_cgo.go deleted file mode 100644 index ebbd2358..00000000 --- a/server/plugin/plg_video_transcoder/transcoder_pipeline_cgo.go +++ /dev/null @@ -1,421 +0,0 @@ -//go:build cgo - -package plg_video_transcoder - -import ( - "errors" - "fmt" - "io" - "math" - - "github.com/asticode/go-astikit" - "github.com/mickael-kerjean/go-astiav" -) - -type pipeline struct { - closer *astikit.Closer - - inFmt *astiav.FormatContext - inStream *astiav.Stream - decCtx *astiav.CodecContext - - encCodec *astiav.Codec - encCtx *astiav.CodecContext - encOpts *astiav.Dictionary - - outFmt *astiav.FormatContext - outStream *astiav.Stream - - srcCtx *astiav.BuffersrcFilterContext - sinkCtx *astiav.BuffersinkFilterContext - - hwDevice *astiav.HardwareDeviceContext -} - -func (p *pipeline) Device(hwType astiav.HardwareDeviceType, hwPixFmt astiav.PixelFormat, w, h int) error { - hwDevice, err := astiav.CreateHardwareDeviceContext(hwType, "", nil, 0) - if err != nil { - return fmt.Errorf("%s device: %w", hwType.Name(), err) - } - p.closer.Add(hwDevice.Free) - hwFrames := astiav.AllocHardwareFramesContext(hwDevice) - if hwFrames == nil { - return fmt.Errorf("%s frames context is nil", hwType.Name()) - } - defer hwFrames.Free() - hwFrames.SetHardwarePixelFormat(hwPixFmt) - hwFrames.SetSoftwarePixelFormat(astiav.PixelFormatNv12) - hwFrames.SetWidth(w) - hwFrames.SetHeight(h) - if err := hwFrames.Initialize(); err != nil { - return fmt.Errorf("%s frames init: %w", hwType.Name(), err) - } - p.encCtx.SetPixelFormat(hwPixFmt) - p.encCtx.SetHardwareFramesContext(hwFrames) - p.hwDevice = hwDevice - return nil -} - -func NewPipeline(path string, mediaType astiav.MediaType, encName string) (p *pipeline, err error) { - // stage 1: encoder side - p = &pipeline{closer: astikit.NewCloser()} - if p.encCodec = astiav.FindEncoderByName(encName); p.encCodec == nil { - return nil, fmt.Errorf("encoder %q not found", encName) - } - if p.encCtx = astiav.AllocCodecContext(p.encCodec); p.encCtx == nil { - return p, errors.New("enc ctx is nil") - } - p.closer.Add(p.encCtx.Free) - p.encOpts = astiav.NewDictionary() - p.closer.Add(p.encOpts.Free) - defer func() { - if err != nil { - p.Close() - p = nil - } - }() - - // stage 2: input side - p.inFmt = astiav.AllocFormatContext() - if p.inFmt == nil { - return p, errors.New("input fmt ctx is nil") - } - p.closer.Add(p.inFmt.Free) - if err = p.inFmt.OpenInput(path, nil, nil); err != nil { - return p, fmt.Errorf("open input: %w", err) - } - p.closer.Add(p.inFmt.CloseInput) - if err = p.inFmt.FindStreamInfo(nil); err != nil { - return p, fmt.Errorf("find stream info: %w", err) - } - - // stage 3: decoder side - stream, decCodec, err := p.inFmt.FindBestStream(mediaType, -1, -1) - if err != nil { - return p, fmt.Errorf("no %s stream: %w", mediaType, err) - } - p.inStream = stream - if mediaType == astiav.MediaTypeVideo && encName == "h264_v4l2m2m" { - decCodec = astiav.FindDecoderByName(encName) - } - p.decCtx = astiav.AllocCodecContext(decCodec) - if p.decCtx == nil { - return p, errors.New("dec ctx is nil") - } - p.closer.Add(p.decCtx.Free) - if err = p.inStream.CodecParameters().ToCodecContext(p.decCtx); err != nil { - return p, fmt.Errorf("dec params: %w", err) - } - tb := p.inStream.TimeBase() - p.decCtx.SetTimeBase(tb) - p.encCtx.SetTimeBase(tb) - - // stage 4: per-media config - switch mediaType { - case astiav.MediaTypeVideo: - p.decCtx.SetFramerate(p.inFmt.GuessFrameRate(p.inStream, nil)) - p.encCtx.SetFramerate(p.decCtx.Framerate()) - p.encCtx.SetSampleAspectRatio(p.decCtx.SampleAspectRatio()) - case astiav.MediaTypeAudio: - channelLayout := p.decCtx.ChannelLayout() - if v := p.encCodec.ChannelLayouts(); len(v) > 0 { - channelLayout = v[0] - } - sampleFmt := p.decCtx.SampleFormat() - if v := p.encCodec.SampleFormats(); len(v) > 0 { - sampleFmt = v[0] - } - p.encCtx.SetChannelLayout(channelLayout) - p.encCtx.SetSampleFormat(sampleFmt) - p.encCtx.SetSampleRate(p.decCtx.SampleRate()) - } - if err = p.decCtx.Open(decCodec, nil); err != nil { - return p, fmt.Errorf("open decoder: %w", err) - } - return p, nil -} - -func (p *pipeline) Close() { - p.closer.Close() -} - -func (p *pipeline) Open(w io.Writer) (err error) { - if p.outFmt, err = astiav.AllocOutputFormatContext(nil, "mpegts", ""); err != nil { - return fmt.Errorf("alloc out fmt: %w", err) - } - p.closer.Add(p.outFmt.Free) - if err := p.encCtx.Open(p.encCodec, p.encOpts); err != nil { - return fmt.Errorf("open encoder: %w", err) - } - ioCtx, err := astiav.AllocIOContext( - 65536, true, nil, nil, - func(b []byte) (int, error) { - return w.Write(b) - }, - ) - if err != nil { - return fmt.Errorf("alloc io ctx: %w", err) - } - p.outFmt.SetPb(ioCtx) - p.closer.Add(ioCtx.Free) - if p.outStream = p.outFmt.NewStream(nil); p.outStream == nil { - return errors.New("out stream is nil") - } - if err := p.outStream.CodecParameters().FromCodecContext(p.encCtx); err != nil { - return fmt.Errorf("out stream params: %w", err) - } - p.outStream.SetTimeBase(p.encCtx.TimeBase()) - return nil -} - -func (p *pipeline) Build(graphSpec string) (err error) { - fg := astiav.AllocFilterGraph() - if fg == nil { - return errors.New("filter graph is nil") - } - p.closer.Add(fg.Free) - - var ( - srcName = "" - sinkName = "" - srcParams = astiav.AllocBuffersrcFilterContextParameters() - ) - srcParams.SetTimeBase(p.inStream.TimeBase()) - defer srcParams.Free() - switch p.decCtx.MediaType() { - case astiav.MediaTypeVideo: - srcName, sinkName = "buffer", "buffersink" - srcParams.SetWidth(p.decCtx.Width()) - srcParams.SetHeight(p.decCtx.Height()) - srcParams.SetPixelFormat(p.decCtx.PixelFormat()) - srcParams.SetSampleAspectRatio(p.decCtx.SampleAspectRatio()) - case astiav.MediaTypeAudio: - srcName, sinkName = "abuffer", "abuffersink" - srcParams.SetChannelLayout(p.decCtx.ChannelLayout()) - srcParams.SetSampleFormat(p.decCtx.SampleFormat()) - srcParams.SetSampleRate(p.decCtx.SampleRate()) - default: - return fmt.Errorf("unsupported media type %s", p.decCtx.MediaType()) - } - srcF, sinkF := astiav.FindFilterByName(srcName), astiav.FindFilterByName(sinkName) - if srcF == nil || sinkF == nil { - return fmt.Errorf("filter %s/%s missing", srcName, sinkName) - } - if p.srcCtx, err = fg.NewBuffersrcFilterContext(srcF, "in"); err != nil { - return fmt.Errorf("new buffersrc: %w", err) - } - if p.sinkCtx, err = fg.NewBuffersinkFilterContext(sinkF, "out"); err != nil { - return fmt.Errorf("new buffersink: %w", err) - } - if err := p.srcCtx.SetParameters(srcParams); err != nil { - return fmt.Errorf("buffersrc params: %w", err) - } - if err := p.srcCtx.Initialize(nil); err != nil { - return fmt.Errorf("buffersrc init: %w", err) - } - - seg, err := fg.ParseSegment(graphSpec) - if err != nil { - return fmt.Errorf("segment parse %q: %w", graphSpec, err) - } - defer seg.Free() - if err := seg.CreateFilters(0); err != nil { - return fmt.Errorf("segment create filters: %w", err) - } - if err := seg.ApplyOpts(0); err != nil { - return fmt.Errorf("segment apply opts: %w", err) - } - if p.hwDevice != nil { - for _, chain := range seg.Chains() { - for _, params := range chain.Filters() { - if fc := params.FilterContext(); fc != nil { - fc.SetHardwareDeviceContext(p.hwDevice) - } - } - } - } - if err := seg.Init(0); err != nil { - return fmt.Errorf("segment init: %w", err) - } - segIn, segOut := astiav.AllocFilterInOut(), astiav.AllocFilterInOut() - defer segIn.Free() - defer segOut.Free() - if err := seg.Link(0, segIn, segOut); err != nil { - return fmt.Errorf("segment link: %w", err) - } - for io := segIn; io != nil; io = io.Next() { - if err := p.srcCtx.FilterContext().LinkTo(0, io.FilterContext(), 0); err != nil { - return fmt.Errorf("link buffersrc: %w", err) - } - } - for io := segOut; io != nil; io = io.Next() { - if err := io.FilterContext().LinkTo(0, p.sinkCtx.FilterContext(), 0); err != nil { - return fmt.Errorf("link buffersink: %w", err) - } - } - if err := fg.Configure(); err != nil { - return fmt.Errorf("filter configure: %w", err) - } - return nil -} - -func (p *pipeline) Run(startSec, endSec int) error { - tb := p.inStream.TimeBase() - startPts := int64(startSec) * int64(tb.Den()) / int64(tb.Num()) - endPts := int64(endSec) * int64(tb.Den()) / int64(tb.Num()) - - if err := p.inFmt.SeekFrame(p.inStream.Index(), startPts, astiav.NewSeekFlags(astiav.SeekFlagBackward)); err != nil { - return fmt.Errorf("seek: %w", err) - } - if err := p.outFmt.WriteHeader(nil); err != nil { - return fmt.Errorf("write header: %w", err) - } - - pkt := astiav.AllocPacket() - defer pkt.Free() - encPkt := astiav.AllocPacket() - defer encPkt.Free() - decFrame := astiav.AllocFrame() - defer decFrame.Free() - filtFrame := astiav.AllocFrame() - defer filtFrame.Free() - - drained := func(err error) bool { - return errors.Is(err, astiav.ErrEagain) || errors.Is(err, astiav.ErrEof) - } - - lastDts := int64(math.MinInt64) - drainEncoder := func() error { - for { - if err := p.encCtx.ReceivePacket(encPkt); err != nil { - if drained(err) { - return nil - } - return err - } - encPkt.SetStreamIndex(p.outStream.Index()) - encPkt.RescaleTs(p.encCtx.TimeBase(), p.outStream.TimeBase()) - if encPkt.Dts() <= lastDts { - encPkt.Unref() - continue - } - lastDts = encPkt.Dts() - err := p.outFmt.WriteInterleavedFrame(encPkt) - encPkt.Unref() - if err != nil { - return err - } - } - } - - pushThroughFilter := func(f *astiav.Frame) error { - if err := p.srcCtx.AddFrame(f, astiav.NewBuffersrcFlags(astiav.BuffersrcFlagKeepRef)); err != nil { - return err - } - for { - if err := p.sinkCtx.GetFrame(filtFrame, astiav.NewBuffersinkFlags()); err != nil { - if drained(err) { - return nil - } - return err - } - err := p.encCtx.SendFrame(filtFrame) - filtFrame.Unref() - if err != nil { - return err - } - if err := drainEncoder(); err != nil { - return err - } - } - } - - drainDecoder := func() (bool, error) { - for { - if err := p.decCtx.ReceiveFrame(decFrame); err != nil { - if drained(err) { - return false, nil - } - return false, fmt.Errorf("receive frame: %w", err) - } - pts := decFrame.Pts() - if pts == math.MinInt64 { // h264 in avi should fall back to dts. - pts = decFrame.PktDts() - } - if pts >= endPts { - decFrame.Unref() - return true, nil - } - if pts < startPts { - decFrame.Unref() - continue - } - decFrame.SetPts(pts) - err := pushThroughFilter(decFrame) - decFrame.Unref() - if err != nil { - return false, fmt.Errorf("filter: %w", err) - } - } - } - - done := false - for !done { - if err := p.inFmt.ReadFrame(pkt); err != nil { - if errors.Is(err, astiav.ErrEof) { - break - } - return fmt.Errorf("read frame: %w", err) - } - if pkt.StreamIndex() != p.inStream.Index() { - pkt.Unref() - continue - } - err := p.decCtx.SendPacket(pkt) - pkt.Unref() - if err != nil && !errors.Is(err, astiav.ErrInvaliddata) { - return fmt.Errorf("send packet: %w", err) - } - if done, err = drainDecoder(); err != nil { - return err - } - } - - if !done { - if err := p.decCtx.SendPacket(nil); err != nil { - return fmt.Errorf("flush decoder: %w", err) - } - if _, err := drainDecoder(); err != nil { - return err - } - } - if err := pushThroughFilter(nil); err != nil { - return fmt.Errorf("flush filter: %w", err) - } - if err := p.encCtx.SendFrame(nil); err != nil { - return fmt.Errorf("flush encoder: %w", err) - } - if err := drainEncoder(); err != nil { - return fmt.Errorf("drain: %w", err) - } - if err := p.outFmt.WriteTrailer(); err != nil { - return fmt.Errorf("write trailer: %w", err) - } - return nil -} - -func probeDuration(path string) (float64, error) { - inFmt := astiav.AllocFormatContext() - if inFmt == nil { - return 0, errors.New("input format context is nil") - } - defer inFmt.Free() - if err := inFmt.OpenInput(path, nil, nil); err != nil { - return 0, err - } - defer inFmt.CloseInput() - if err := inFmt.FindStreamInfo(nil); err != nil { - return 0, err - } - return float64(inFmt.Duration()) / float64(astiav.TimeBase), nil -}