kerberos-io/agent
1
0
Fork 0
mirror of https://github.com/kerberos-io/agent.git synced 2026-08-23 15:08:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #280 from kerberos-io/fix/dts-pts-correction

Browse Source 
fix/dts-pts-correction
This commit is contained in:
Cédric Verstraeten
2026-06-08 18:23:19 +02:00
committed by GitHub
parent 1a477bf42d 5f7ede40ca
commit b8c05aa3e2
5 changed files with 104 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
machinery/src/capture/gortsplib.go
View File

@@ -517,6 +517,35 @@ func (g *Golibrtsp) ConnectBackChannel(ctx context.Context, ctxRunAgent context.
return
}
// dtsExtractor abstracts the codec-specific DTS extractors from mediacommon
// (h264.DTSExtractor2 and h265.DTSExtractor2), which expose the same method.
type dtsExtractor interface {
Extract(au [][]byte, pts int64) (int64, error)
}
// compositionOffsetMs returns the composition time offset (PTS - DTS) in
// milliseconds for a coded access unit. Streams that contain B-frames deliver
// access units in decode order with non-monotonic PTS; the fragmented MP4
// writer needs a monotonic DTS timeline plus a per-sample composition offset
// so browsers (Media Source Extensions) can decode the chained segments.
//
// It returns 0 when the codec has no frame reordering (the common case, e.g.
// baseline "IPPP" streams) or when extraction fails, making it a safe no-op.
func compositionOffsetMs(ext dtsExtractor, au [][]byte, pts int64, clockRate int) int64 {
if ext == nil || clockRate <= 0 {
return 0
}
dts, err := ext.Extract(au, pts)
if err != nil {
return 0
}
offset := pts - dts
if offset <= 0 {
return 0
}
return offset * 1000 / int64(clockRate)
}
// Start the RTSP client, and start reading packets.
func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets.Queue, configuration *models.Configuration, communication *models.Communication) (err error) {
log.Log.Debug("capture.golibrtsp.Start(): started")
@@ -602,7 +631,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
var filteredAU [][]byte
if g.VideoH264Media != nil && g.VideoH264Forma != nil {
//dtsExtractor := h264.NewDTSExtractor2()
// Extracts DTS from the bitstream to support B-frame H264 streams.
// Created once per stream (tracks reorder state across access units).
h264DTSExtractor := h264.NewDTSExtractor2()
g.Client.OnPacketRTP(g.VideoH264Media, g.VideoH264Forma, func(rtppkt *rtp.Packet) {
@@ -742,6 +773,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
return
}
// Composition time offset (PTS - DTS) in milliseconds. Non-zero
// only for streams with B-frames; the MP4 writer uses it to keep a
// monotonic decode timeline and present frames in PTS order.
compositionOffset := compositionOffsetMs(h264DTSExtractor, au, pts2, g.VideoH264Forma.ClockRate())
pkt := packets.Packet{
IsKeyFrame: idrPresent,
Packet: rtppkt,
@@ -749,7 +785,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
Time: pts2,
TimeLegacy: pts,
CurrentTime: time.Now().UnixMilli(),
CompositionTime: pts2,
CompositionTime: compositionOffset,
Idx: g.VideoH264Index,
IsVideo: true,
IsAudio: false,
@@ -817,6 +853,11 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
// called when a video RTP packet arrives for H265
if g.VideoH265Media != nil && g.VideoH265Forma != nil {
// Extracts DTS from the bitstream to support B-frame H265 streams.
// Created once per stream (tracks reorder state across access units).
h265DTSExtractor := h265.NewDTSExtractor2()
g.Client.OnPacketRTP(g.VideoH265Media, g.VideoH265Forma, func(rtppkt *rtp.Packet) {
// This will check if we need to stop the thread,
@@ -860,6 +901,10 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
}
}
// Preserve the decoded access unit (in decode order) for DTS
// extraction before we rewrite it into the filtered/annexb form.
decodedAU := au
filteredAU = [][]byte{
{byte(h265.NALUType_AUD_NUT) << 1, 1, 0x50},
}
@@ -902,6 +947,9 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
return
}
// Composition time offset (PTS - DTS) in milliseconds; see H264 handler.
compositionOffset := compositionOffsetMs(h265DTSExtractor, decodedAU, pts2, g.VideoH265Forma.ClockRate())
pkt := packets.Packet{
IsKeyFrame: isRandomAccess,
Packet: rtppkt,
@@ -909,7 +957,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
Time: pts2,
TimeLegacy: pts,
CurrentTime: time.Now().UnixMilli(),
CompositionTime: pts2,
CompositionTime: compositionOffset,
Idx: g.VideoH265Index,
IsVideo: true,
IsAudio: false,

113
machinery/src/capture/main.go
View File

@@ -140,23 +140,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
if start && // If already recording and current frame is a keyframe and we should stop recording
nextPkt.IsKeyFrame && (startRecording+postRecording-now <= 0 || now-startRecording > maxRecordingPeriod-500) {
pts := convertPTS(pkt.TimeLegacy)
if pkt.IsVideo {
// Write the last packet
if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
}
} else if pkt.IsAudio {
// Write the last packet
if pkt.Codec == "AAC" {
if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
}
} else if pkt.Codec == "PCM_MULAW" {
// TODO: transcode to AAC, some work to do..
log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
}
}
// Write the last packet before closing the recording.
writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
// Close mp4
if len(mp4Video.SPSNALUs) == 0 && len(configuration.Config.Capture.IPCamera.SPSNALUs) > 0 {
@@ -311,43 +296,12 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
}
pts := convertPTS(pkt.TimeLegacy)
if pkt.IsVideo {
if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
}
} else if pkt.IsAudio {
if pkt.Codec == "AAC" {
if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
}
} else if pkt.Codec == "PCM_MULAW" {
// TODO: transcode to AAC, some work to do..
// We might need to use ffmpeg to transcode the audio to AAC.
// For now we will skip the audio track.
log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
}
}
writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
recordingStatus = "started"
} else if start {
pts := convertPTS(pkt.TimeLegacy)
if pkt.IsVideo {
// New method using new mp4 library
if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
}
} else if pkt.IsAudio {
if pkt.Codec == "AAC" {
if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(continuous): " + err.Error())
}
} else if pkt.Codec == "PCM_MULAW" {
// TODO: transcode to AAC, some work to do..
log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
}
}
writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
}
pkt = nextPkt
}
@@ -571,29 +525,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
start = true
}
if start {
pts := convertPTS(pkt.TimeLegacy)
if pkt.IsVideo {
log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add video sample")
if mp4Video != nil {
if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
}
}
} else if pkt.IsAudio {
log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add audio sample")
if pkt.Codec == "AAC" {
if mp4Video != nil {
if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
}
}
} else if pkt.Codec == "PCM_MULAW" {
// TODO: transcode to AAC, some work to do..
// We might need to use ffmpeg to transcode the audio to AAC.
// For now we will skip the audio track.
log.Log.Debug("capture.main.HandleRecordStream(motiondetection): no AAC audio codec detected, skipping audio track.")
}
}
writeSampleToMP4(mp4Video, videoTrack, audioTrack, pkt)
}
pkt = nextPkt
@@ -867,6 +799,41 @@ func convertPTS(v time.Duration) uint64 {
return uint64(v.Milliseconds())
}
// writeSampleToMP4 writes a single capture packet to the fragmented MP4.
//
// For video it derives the decode timestamp (DTS) from the packet PTS using the
// per-packet composition offset (PTS - DTS), which is non-zero only for streams
// that contain B-frames. Passing the monotonic DTS as the sample timestamp keeps
// the fragment timeline (tfdt/sidx) monotonic, while the composition offset is
// forwarded so frames are still presented in PTS order.
func writeSampleToMP4(mp4Video *video.MP4, videoTrack, audioTrack uint32, pkt packets.Packet) {
if mp4Video == nil {
return
}
pts := convertPTS(pkt.TimeLegacy)
if pkt.IsVideo {
compositionOffset := pkt.CompositionTime
dts := pts
if compositionOffset > 0 && uint64(compositionOffset) <= pts {
dts = pts - uint64(compositionOffset)
}
if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, dts, compositionOffset); err != nil {
log.Log.Error("capture.main.writeSampleToMP4(): " + err.Error())
}
} else if pkt.IsAudio {
if pkt.Codec == "AAC" {
if err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts, 0); err != nil {
log.Log.Error("capture.main.writeSampleToMP4(): " + err.Error())
}
} else if pkt.Codec == "PCM_MULAW" {
// TODO: transcode to AAC, some work to do..
log.Log.Debug("capture.main.writeSampleToMP4(): no AAC audio codec detected, skipping audio track.")
}
}
}
/*func convertPTS2(v int64) uint64 {
return uint64(v) / 100
}*/

2
machinery/src/packets/packet.go
View File

@@ -14,7 +14,7 @@ type Packet struct {
IsKeyFrame bool // video packet is key frame
Idx int8 // stream index in container format
Codec string // codec name
CompositionTime int64 // packet presentation time minus decode time for H264 B-Frame
CompositionTime int64 // composition offset (PTS - DTS) in milliseconds, non-zero for H264/H265 B-frames
Time int64 // packet decode time
TimeLegacy time.Duration
CurrentTime int64 // current time in milliseconds (UNIX timestamp)

13
machinery/src/video/mp4.go
View File

@@ -266,7 +266,16 @@ func (mp4 *MP4) flushPendingVideoSample(nextPTS uint64) bool {
return true
}
func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64) error {
// AddSampleToTrack appends a sample to the given track.
//
// For video, pts is the decode timestamp (DTS, in milliseconds) and
// compositionOffset is the composition time offset (PTS - DTS, in milliseconds).
// The offset is non-zero only for streams that contain B-frames; it is written
// as the sample's signed composition time offset so the decoder presents frames
// in PTS order while the fragment timeline stays monotonic in DTS.
//
// For audio, pts is the sample timestamp and compositionOffset should be 0.
func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, pts uint64, compositionOffset int64) error {
if isKeyframe && trackID == uint32(mp4.VideoTrack) {
mp4.TotalKeyframesReceived++
@@ -375,7 +384,7 @@ func (mp4 *MP4) AddSampleToTrack(trackID uint32, isKeyframe bool, data []byte, p
fullSample.Sample = mp4ff.Sample{
Size: uint32(len(fullSample.Data)),
Flags: flags,
CompositionTimeOffset: 0, // No composition time offset for video
CompositionTimeOffset: int32(compositionOffset), // PTS-DTS, non-zero for B-frames
}
mp4.VideoFullSample = &fullSample
mp4.PendingSampleIsKeyframe = isKeyframe

2
machinery/src/video/mp4_duration_test.go
View File

@@ -49,7 +49,7 @@ func TestMP4Duration(t *testing.T) {
for i := 0; i < numFrames; i++ {
pts := uint64(i) * frameDuration
isKeyframe := i%gopSize == 0
err := mp4Video.AddSampleToTrack(videoTrack, isKeyframe, makeFrame(isKeyframe), pts)
err := mp4Video.AddSampleToTrack(videoTrack, isKeyframe, makeFrame(isKeyframe), pts, 0)
if err != nil {
t.Fatalf("AddSampleToTrack failed at frame %d: %v", i, err)
}