apps/agent
1
0
Fork 0
mirror of https://github.com/kerberos-io/agent.git synced 2025-10-06 00:26:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #199 from kerberos-io/feature/new-method-to-calc-pre-recording-start-time

Browse Source 
Feature/new method to calc pre recording start time
This commit is contained in:
Cédric Verstraeten
2025-07-05 17:08:38 +02:00
committed by GitHub
parent 448d4a946d 5823abed95
commit 0e259f0e7a
3 changed files with 42 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -478,6 +478,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
Time: pts2, Time: pts2,
TimeLegacy: pts, TimeLegacy: pts,
CompositionTime: pts2, CompositionTime: pts2,
CurrentTime: time.Now().UnixMilli(),
Idx: g.AudioG711Index, Idx: g.AudioG711Index,
IsVideo: false, IsVideo: false,
IsAudio: true, IsAudio: true,
@@ -519,6 +520,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
Time: pts2, Time: pts2,
TimeLegacy: pts, TimeLegacy: pts,
CompositionTime: pts2, CompositionTime: pts2,
CurrentTime: time.Now().UnixMilli(),
Idx: g.AudioG711Index, Idx: g.AudioG711Index,
IsVideo: false, IsVideo: false,
IsAudio: true, IsAudio: true,
@@ -631,19 +633,13 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
return return
} }
// Extract DTS from RTP packets
//dts2, err := dtsExtractor.Extract(filteredAU, pts2)
//if err != nil {
// log.Log.Error("capture.golibrtsp.Start(): " + err.Error())
// return
//}
pkt := packets.Packet{ pkt := packets.Packet{
IsKeyFrame: idrPresent, IsKeyFrame: idrPresent,
Packet: rtppkt, Packet: rtppkt,
Data: enc, Data: enc,
Time: pts2, Time: pts2,
TimeLegacy: pts, TimeLegacy: pts,
CurrentTime: time.Now().UnixMilli(),
CompositionTime: pts2, CompositionTime: pts2,
Idx: g.VideoH264Index, Idx: g.VideoH264Index,
IsVideo: true, IsVideo: true,
@@ -788,6 +784,7 @@ func (g *Golibrtsp) Start(ctx context.Context, streamType string, queue *packets
Data: enc, Data: enc,
Time: pts2, Time: pts2,
TimeLegacy: pts, TimeLegacy: pts,
CurrentTime: time.Now().UnixMilli(),
CompositionTime: pts2, CompositionTime: pts2,
Idx: g.VideoH265Index, Idx: g.VideoH265Index,
IsVideo: true, IsVideo: true,

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

@@ -68,10 +68,16 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
postRecording := config.Capture.PostRecording * 1000 // number of seconds to record. postRecording := config.Capture.PostRecording * 1000 // number of seconds to record.
maxRecordingPeriod := config.Capture.MaxLengthRecording * 1000 // maximum number of seconds to record. maxRecordingPeriod := config.Capture.MaxLengthRecording * 1000 // maximum number of seconds to record.
// Synchronise the last synced time // We will calculate the maxRecordingPeriod based on the preRecording and postRecording values.
now := time.Now().UnixMilli() if maxRecordingPeriod == 0 {
startRecording := now // If maxRecordingPeriod is not set, we will use the preRecording and postRecording values
timestamp := now maxRecordingPeriod = preRecording + postRecording
}
if maxRecordingPeriod < preRecording+postRecording {
log.Log.Error("capture.main.HandleRecordStream(): maxRecordingPeriod is less than preRecording + postRecording, this is not allowed. Setting maxRecordingPeriod to preRecording + postRecording.")
maxRecordingPeriod = preRecording + postRecording
}
if config.FriendlyName != "" { if config.FriendlyName != "" {
config.Name = config.FriendlyName config.Name = config.FriendlyName
@@ -105,8 +111,6 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
// Do not do anything! // Do not do anything!
log.Log.Info("capture.main.HandleRecordStream(continuous): start recording") log.Log.Info("capture.main.HandleRecordStream(continuous): start recording")
now = time.Now().Unix()
timestamp = now
start := false start := false
// If continuous record the full length // If continuous record the full length
@@ -114,6 +118,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
// Recording file name // Recording file name
fullName := "" fullName := ""
var startRecording int64 = 0 // start recording timestamp in milliseconds
// Get as much packets we need. // Get as much packets we need.
var cursorError error var cursorError error
var pkt packets.Packet var pkt packets.Packet
@@ -132,7 +138,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
now := time.Now().UnixMilli() now := time.Now().UnixMilli()
if start && // If already recording and current frame is a keyframe and we should stop recording if start && // If already recording and current frame is a keyframe and we should stop recording
nextPkt.IsKeyFrame && (timestamp+postRecording-now <= 0 || now-startRecording >= maxRecordingPeriod-1000) { nextPkt.IsKeyFrame && (startRecording+postRecording-now <= 0 || now-startRecording > maxRecordingPeriod-500) {
pts := convertPTS(pkt.TimeLegacy) pts := convertPTS(pkt.TimeLegacy)
if pkt.IsVideo { if pkt.IsVideo {
@@ -236,7 +242,6 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
} }
start = true start = true
timestamp = now
// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token // timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
// 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4 // 1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4
@@ -247,7 +252,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
// - Number of changes // - Number of changes
// - Token // - Token
startRecording = time.Now().UnixMilli() startRecording = pkt.CurrentTime
startRecordingSeconds := startRecording / 1000 // convert to seconds startRecordingSeconds := startRecording / 1000 // convert to seconds
startRecordingMilliseconds := startRecording % 1000 // convert to milliseconds startRecordingMilliseconds := startRecording % 1000 // convert to milliseconds
s := strconv.FormatInt(startRecordingSeconds, 10) + "_" + // start timestamp in seconds s := strconv.FormatInt(startRecordingSeconds, 10) + "_" + // start timestamp in seconds
@@ -409,16 +414,12 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
log.Log.Info("capture.main.HandleRecordStream(motiondetection): Start motion based recording ") log.Log.Info("capture.main.HandleRecordStream(motiondetection): Start motion based recording ")
var lastDuration int64 = 0 // last duration in milliseconds var lastRecordingTime int64 = 0 // last recording timestamp in milliseconds
var lastRecordingTime int64 = 0 // last recording time in milliseconds
var displayTime int64 = 0 // display time in milliseconds var displayTime int64 = 0 // display time in milliseconds
var videoTrack uint32 var videoTrack uint32
var audioTrack uint32 var audioTrack uint32
streams, _ := rtspClient.GetVideoStreams()
videoStream := streams[0] // We will use the first video stream, as we only expect one video stream.
for motion := range communication.HandleMotion { for motion := range communication.HandleMotion {
// Get as much packets we need. // Get as much packets we need.
@@ -427,43 +428,24 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
var nextPkt packets.Packet var nextPkt packets.Packet
recordingCursor := queue.Oldest() // Start from the latest packet in the queue) recordingCursor := queue.Oldest() // Start from the latest packet in the queue)
timestamp = time.Now().UnixMilli() now := time.Now().UnixMilli()
startRecording = time.Now().UnixMilli() // we mark the current time when the record started. motionTimestamp := now
// If we have prerecording we will substract the number of seconds. start := false
// Taking into account FPS = GOP size (Keyfram interval)
var preRecordingDelta int64 = 0
if preRecording > 0 {
fps := videoStream.FPS if cursorError == nil {
queueSize := queue.GetSize() pkt, cursorError = recordingCursor.ReadPacket()
// Based on the GOP size and FPS we can calculate the pre-recording time.
// It might be that the queue size is 0, in that case we will not calculate the pre-recording time.
queuedAvailablePreRecording := preRecording
if queueSize > 0 && fps > 0 {
queuedAvailablePreRecording = int64(queueSize) / int64(fps) * 1000 // convert to milliseconds
}
timeBetweenNowAndLastRecording := startRecording - lastRecordingTime
if lastRecordingTime == 0 {
timeBetweenNowAndLastRecording = 0
} }
// Might be that recordings are coming short after each other. displayTime = pkt.CurrentTime
// Therefore we do some math with the current time and the last recording time. startRecording := pkt.CurrentTime
if timeBetweenNowAndLastRecording >= preRecording {
displayTime = startRecording - preRecording + 1000 // We have more packets in the queue (which might still be older than where we close the previous recording).
preRecordingDelta = preRecording // In that case we will use the last recording time to determine the start time of the recording, otherwise
} else if timeBetweenNowAndLastRecording < preRecording { // we will have duplicate frames in the recording.
// If the time between now and the last recording is less than the pre-recording time, if startRecording < lastRecordingTime {
// we will use the pre-recording time. displayTime = lastRecordingTime
if lastRecordingTime == 0 && queuedAvailablePreRecording < preRecording { startRecording = lastRecordingTime
displayTime = startRecording - queuedAvailablePreRecording
} else {
preRecordingDelta = timeBetweenNowAndLastRecording
displayTime = startRecording - preRecordingDelta
}
}
} }
// timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token // timestamp_microseconds_instanceName_regionCoordinates_numberOfChanges_token
@@ -527,12 +509,6 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.") log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
} }
start := false
if cursorError == nil {
pkt, cursorError = recordingCursor.ReadPacket()
}
for cursorError == nil { for cursorError == nil {
nextPkt, cursorError = recordingCursor.ReadPacket() nextPkt, cursorError = recordingCursor.ReadPacket()
@@ -540,24 +516,23 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + cursorError.Error()) log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + cursorError.Error())
} }
now := time.Now().UnixMilli() now = time.Now().UnixMilli()
select { select {
case motion := <-communication.HandleMotion: case motion := <-communication.HandleMotion:
timestamp = now motionTimestamp = now
log.Log.Info("capture.main.HandleRecordStream(motiondetection): motion detected while recording. Expanding recording.") log.Log.Info("capture.main.HandleRecordStream(motiondetection): motion detected while recording. Expanding recording.")
numberOfChanges := motion.NumberOfChanges numberOfChanges := motion.NumberOfChanges
log.Log.Info("capture.main.HandleRecordStream(motiondetection): Received message with recording data, detected changes to save: " + strconv.Itoa(numberOfChanges)) log.Log.Info("capture.main.HandleRecordStream(motiondetection): Received message with recording data, detected changes to save: " + strconv.Itoa(numberOfChanges))
default: default:
} }
if (timestamp+postRecording+(preRecording-preRecordingDelta)-now < 0 || now-startRecording > maxRecordingPeriod-preRecordingDelta) && nextPkt.IsKeyFrame { if (motionTimestamp+postRecording-now < 0 || now-startRecording > maxRecordingPeriod-500) && nextPkt.IsKeyFrame {
log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0 - " + strconv.FormatInt(timestamp+postRecording-now, 10) + " < 0") log.Log.Info("capture.main.HandleRecordStream(motiondetection): timestamp+postRecording-now < 0 - " + strconv.FormatInt(motionTimestamp+postRecording-now, 10) + " < 0")
log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-1000 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-1000, 10)) log.Log.Info("capture.main.HandleRecordStream(motiondetection): now-startRecording > maxRecordingPeriod-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 10))
log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(timestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10)) log.Log.Info("capture.main.HandleRecordStream(motiondetection): closing recording (timestamp: " + strconv.FormatInt(motionTimestamp, 10) + ", postRecording: " + strconv.FormatInt(postRecording, 10) + ", now: " + strconv.FormatInt(now, 10) + ", startRecording: " + strconv.FormatInt(startRecording, 10) + ", maxRecordingPeriod: " + strconv.FormatInt(maxRecordingPeriod, 10))
break break
} }
if pkt.IsKeyFrame && !start && (pkt.Time >= lastDuration || pkt.Time == 0) { if pkt.IsKeyFrame && !start && pkt.CurrentTime >= startRecording {
// We start the recording if we have a keyframe and the last duration is 0 or less than the current packet time. // We start the recording if we have a keyframe and the last duration is 0 or less than the current packet time.
// It could be start we start from the beginning of the recording. // It could be start we start from the beginning of the recording.
log.Log.Debug("capture.main.HandleRecordStream(motiondetection): write frames") log.Log.Debug("capture.main.HandleRecordStream(motiondetection): write frames")
@@ -590,8 +565,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
// Update the last duration and last recording time. // Update the last duration and last recording time.
// This is used to determine if we need to start a new recording. // This is used to determine if we need to start a new recording.
lastDuration = pkt.Time lastRecordingTime = pkt.CurrentTime
lastRecordingTime = time.Now().UnixMilli()
// This will close the recording and write the last packet. // This will close the recording and write the last packet.
mp4Video.Close(&config) mp4Video.Close(&config)

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

@@ -17,6 +17,7 @@ type Packet struct {
CompositionTime int64 // packet presentation time minus decode time for H264 B-Frame CompositionTime int64 // packet presentation time minus decode time for H264 B-Frame
Time int64 // packet decode time Time int64 // packet decode time
TimeLegacy time.Duration TimeLegacy time.Duration
CurrentTime int64 // current time in milliseconds (UNIX timestamp)
Data []byte // packet data Data []byte // packet data
Gopsize int // size of the GOP Gopsize int // size of the GOP
} }