feature (plg_video_transcoder): libav + ffmpeg strategy

This commit is contained in:
MickaelK
2026-06-02 01:30:30 +10:00
parent fc5cd5a01f
commit b8b16441d1
14 changed files with 847 additions and 621 deletions
@@ -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(),
@@ -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)
@@ -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 = ""
)
@@ -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())
}
}
@@ -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
}
@@ -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())
}
+34 -56
View File
@@ -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(`}`))
}
@@ -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())
}
@@ -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
}
}
@@ -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())
}
}
@@ -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
}
@@ -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(),
)
}
@@ -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,
)
}
@@ -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
}