Merge pull request #252 from kerberos-io/fix/close-mp4-after-started

fix/close-mp4-after-started

machinery/src/capture/main.go

@@ -518,21 +518,8 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				if len(spsNALUS) == 0 || len(ppsNALUS) == 0 {
 					log.Log.Warning("capture.main.HandleRecordStream(motiondetection): missing SPS/PPS at recording start")
 				}
-				// Create a video file, and set the dimensions.
-				mp4Video := video.NewMP4(fullName, spsNALUS, ppsNALUS, vpsNALUS, configuration.Config.Capture.MaxLengthRecording)
-				mp4Video.SetWidth(width)
-				mp4Video.SetHeight(height)
-
-				if videoCodec == "H264" {
-					videoTrack = mp4Video.AddVideoTrack("H264")
-				} else if videoCodec == "H265" {
-					videoTrack = mp4Video.AddVideoTrack("H265")
-				}
-				if audioCodec == "AAC" {
-					audioTrack = mp4Video.AddAudioTrack("AAC")
-				} else if audioCodec == "PCM_MULAW" {
-					log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
-				}
+				// Create the MP4 only once the first keyframe arrives.
+				var mp4Video *video.MP4
 
 				for cursorError == nil {
 
@@ -551,7 +538,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 					default:
 					}
 
-					if (motionTimestamp+postRecording-now < 0 || now-startRecording > maxRecordingPeriod-500) && nextPkt.IsKeyFrame {
+					if start && (motionTimestamp+postRecording-now < 0 || now-startRecording > maxRecordingPeriod-500) && nextPkt.IsKeyFrame {
 						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-500 - " + strconv.FormatInt(now-startRecording, 10) + " > " + strconv.FormatInt(maxRecordingPeriod-500, 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))
@@ -561,20 +548,44 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 						// 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.
 						log.Log.Debug("capture.main.HandleRecordStream(motiondetection): write frames")
+						log.Log.Debug("capture.main.HandleRecordStream(motiondetection): recording started on keyframe")
+
+						// Align duration timers with the first keyframe.
+						startRecording = pkt.CurrentTime
+
+						// Create a video file, and set the dimensions.
+						mp4Video = video.NewMP4(fullName, spsNALUS, ppsNALUS, vpsNALUS, configuration.Config.Capture.MaxLengthRecording)
+						mp4Video.SetWidth(width)
+						mp4Video.SetHeight(height)
+
+						if videoCodec == "H264" {
+							videoTrack = mp4Video.AddVideoTrack("H264")
+						} else if videoCodec == "H265" {
+							videoTrack = mp4Video.AddVideoTrack("H265")
+						}
+						if audioCodec == "AAC" {
+							audioTrack = mp4Video.AddAudioTrack("AAC")
+						} else if audioCodec == "PCM_MULAW" {
+							log.Log.Debug("capture.main.HandleRecordStream(continuous): no AAC audio codec detected, skipping audio track.")
+						}
 						start = true
 					}
 					if start {
 						pts := convertPTS(pkt.TimeLegacy)
 						if pkt.IsVideo {
 							log.Log.Debug("capture.main.HandleRecordStream(motiondetection): add video sample")
-							if err := mp4Video.AddSampleToTrack(videoTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-								log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
+							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 err := mp4Video.AddSampleToTrack(audioTrack, pkt.IsKeyFrame, pkt.Data, pts); err != nil {
-									log.Log.Error("capture.main.HandleRecordStream(motiondetection): " + err.Error())
+								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..
@@ -592,6 +603,11 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat
 				// This is used to determine if we need to start a new recording.
 				lastRecordingTime = pkt.CurrentTime
 
+				if mp4Video == nil {
+					log.Log.Warning("capture.main.HandleRecordStream(motiondetection): recording closed without keyframe; no MP4 created")
+					continue
+				}
+
 				// This will close the recording and write the last packet.
 				if len(mp4Video.SPSNALUs) == 0 && len(configuration.Config.Capture.IPCamera.SPSNALUs) > 0 {
 					mp4Video.SPSNALUs = configuration.Config.Capture.IPCamera.SPSNALUs

machinery/src/video/mp4.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/Eyevinn/mp4ff/avc"
 	mp4ff "github.com/Eyevinn/mp4ff/mp4"
 	"github.com/kerberos-io/agent/machinery/src/encryption"
 	"github.com/kerberos-io/agent/machinery/src/log"
@@ -559,11 +560,16 @@ func (mp4 *MP4) Close(config *models.Config) {
 	case "H264", "AVC1":
 		init.AddEmptyTrack(videoTimescale, "video", "und")
 		includePS := true
-		spsNALUs := sanitizeParameterSets(mp4.SPSNALUs)
-		ppsNALUs := sanitizeParameterSets(mp4.PPSNALUs)
+		spsNALUs, ppsNALUs := normalizeH264ParameterSets(mp4.SPSNALUs, mp4.PPSNALUs)
+		log.Log.Debug("mp4.Close(): AVC parameter sets: SPS=" + formatNaluDebug(spsNALUs) + ", PPS=" + formatNaluDebug(ppsNALUs))
 		err := init.Moov.Traks[0].SetAVCDescriptor("avc1", spsNALUs, ppsNALUs, includePS)
 		if err != nil {
 			log.Log.Error("mp4.Close(): error setting AVC descriptor: " + err.Error())
+			if fallbackErr := addAVCDescriptorFallback(init.Moov.Traks[0], spsNALUs, ppsNALUs, uint16(mp4.width), uint16(mp4.height)); fallbackErr != nil {
+				log.Log.Error("mp4.Close(): error setting AVC descriptor fallback: " + fallbackErr.Error())
+			} else {
+				log.Log.Warning("mp4.Close(): AVC descriptor fallback used due to SPS parse error")
+			}
 		}
 		init.Moov.Traks[0].Tkhd.Duration = actualVideoDuration
 		init.Moov.Traks[0].Tkhd.Width = mp4ff.Fixed32(uint32(mp4.width) << 16)
@@ -580,9 +586,8 @@ func (mp4 *MP4) Close(config *models.Config) {
 	case "H265", "HVC1":
 		init.AddEmptyTrack(videoTimescale, "video", "und")
 		includePS := true
-		vpsNALUs := sanitizeParameterSets(mp4.VPSNALUs)
-		spsNALUs := sanitizeParameterSets(mp4.SPSNALUs)
-		ppsNALUs := sanitizeParameterSets(mp4.PPSNALUs)
+		vpsNALUs, spsNALUs, ppsNALUs := normalizeH265ParameterSets(mp4.VPSNALUs, mp4.SPSNALUs, mp4.PPSNALUs)
+		log.Log.Debug("mp4.Close(): HEVC parameter sets: VPS=" + formatNaluDebug(vpsNALUs) + ", SPS=" + formatNaluDebug(spsNALUs) + ", PPS=" + formatNaluDebug(ppsNALUs))
 		err := init.Moov.Traks[0].SetHEVCDescriptor("hvc1", vpsNALUs, spsNALUs, ppsNALUs, [][]byte{}, includePS)
 		if err != nil {
 			log.Log.Error("mp4.Close(): error setting HEVC descriptor: " + err.Error())
@@ -854,6 +859,156 @@ func sanitizeParameterSets(nalus [][]byte) [][]byte {
 	return clean
 }
 
+// normalizeH264ParameterSets splits Annex B blobs and extracts SPS/PPS NALUs.
+func normalizeH264ParameterSets(spsIn [][]byte, ppsIn [][]byte) ([][]byte, [][]byte) {
+	all := make([][]byte, 0, len(spsIn)+len(ppsIn))
+	all = append(all, spsIn...)
+	all = append(all, ppsIn...)
+	var spsOut [][]byte
+	var ppsOut [][]byte
+	for _, blob := range all {
+		for _, nalu := range splitParamSetNALUs(blob) {
+			nalu = removeAnnexBStartCode(nalu)
+			if len(nalu) == 0 {
+				continue
+			}
+			typ := nalu[0] & 0x1F
+			switch typ {
+			case 7:
+				spsOut = append(spsOut, nalu)
+			case 8:
+				ppsOut = append(ppsOut, nalu)
+			}
+		}
+	}
+	if len(spsOut) == 0 {
+		spsOut = sanitizeParameterSets(spsIn)
+	}
+	if len(ppsOut) == 0 {
+		ppsOut = sanitizeParameterSets(ppsIn)
+	}
+	return spsOut, ppsOut
+}
+
+// normalizeH265ParameterSets splits Annex B blobs and extracts VPS/SPS/PPS NALUs.
+func normalizeH265ParameterSets(vpsIn [][]byte, spsIn [][]byte, ppsIn [][]byte) ([][]byte, [][]byte, [][]byte) {
+	all := make([][]byte, 0, len(vpsIn)+len(spsIn)+len(ppsIn))
+	all = append(all, vpsIn...)
+	all = append(all, spsIn...)
+	all = append(all, ppsIn...)
+	var vpsOut [][]byte
+	var spsOut [][]byte
+	var ppsOut [][]byte
+	for _, blob := range all {
+		for _, nalu := range splitParamSetNALUs(blob) {
+			nalu = removeAnnexBStartCode(nalu)
+			if len(nalu) == 0 {
+				continue
+			}
+			typ := (nalu[0] >> 1) & 0x3F
+			switch typ {
+			case 32:
+				vpsOut = append(vpsOut, nalu)
+			case 33:
+				spsOut = append(spsOut, nalu)
+			case 34:
+				ppsOut = append(ppsOut, nalu)
+			}
+		}
+	}
+	if len(vpsOut) == 0 {
+		vpsOut = sanitizeParameterSets(vpsIn)
+	}
+	if len(spsOut) == 0 {
+		spsOut = sanitizeParameterSets(spsIn)
+	}
+	if len(ppsOut) == 0 {
+		ppsOut = sanitizeParameterSets(ppsIn)
+	}
+	return vpsOut, spsOut, ppsOut
+}
+
+// splitParamSetNALUs splits Annex B parameter set blobs; raw NALUs are returned as-is.
+func splitParamSetNALUs(blob []byte) [][]byte {
+	if len(blob) == 0 {
+		return nil
+	}
+	if findStartCode(blob, 0) >= 0 {
+		return splitNALUs(blob)
+	}
+	return [][]byte{blob}
+}
+
+func formatNaluDebug(nalus [][]byte) string {
+	if len(nalus) == 0 {
+		return "none"
+	}
+	parts := make([]string, 0, len(nalus))
+	for _, nalu := range nalus {
+		if len(nalu) == 0 {
+			parts = append(parts, "len=0")
+			continue
+		}
+		max := 8
+		if len(nalu) < max {
+			max = len(nalu)
+		}
+		parts = append(parts, fmt.Sprintf("len=%d head=%x", len(nalu), nalu[:max]))
+	}
+	return strings.Join(parts, "; ")
+}
+
+func addAVCDescriptorFallback(trak *mp4ff.TrakBox, spsNALUs, ppsNALUs [][]byte, width, height uint16) error {
+	if trak == nil || trak.Mdia == nil || trak.Mdia.Minf == nil || trak.Mdia.Minf.Stbl == nil || trak.Mdia.Minf.Stbl.Stsd == nil {
+		return fmt.Errorf("missing trak stsd")
+	}
+	if len(spsNALUs) == 0 {
+		return fmt.Errorf("no SPS NALU available")
+	}
+	decConfRec, err := buildAVCDecConfRecFromSPS(spsNALUs, ppsNALUs)
+	if err != nil {
+		return err
+	}
+	if width == 0 && trak.Tkhd != nil {
+		width = uint16(uint32(trak.Tkhd.Width) >> 16)
+	}
+	if height == 0 && trak.Tkhd != nil {
+		height = uint16(uint32(trak.Tkhd.Height) >> 16)
+	}
+	if width > 0 && height > 0 && trak.Tkhd != nil {
+		trak.Tkhd.Width = mp4ff.Fixed32(uint32(width) << 16)
+		trak.Tkhd.Height = mp4ff.Fixed32(uint32(height) << 16)
+	}
+	avcC := &mp4ff.AvcCBox{DecConfRec: *decConfRec}
+	avcx := mp4ff.CreateVisualSampleEntryBox("avc1", width, height, avcC)
+	trak.Mdia.Minf.Stbl.Stsd.AddChild(avcx)
+	return nil
+}
+
+func buildAVCDecConfRecFromSPS(spsNALUs, ppsNALUs [][]byte) (*avc.DecConfRec, error) {
+	if len(spsNALUs) == 0 {
+		return nil, fmt.Errorf("no SPS NALU available")
+	}
+	sps := spsNALUs[0]
+	if len(sps) < 4 {
+		return nil, fmt.Errorf("SPS too short: len=%d", len(sps))
+	}
+	// SPS NALU: byte 0 is NAL header, next 3 bytes are profile/compat/level.
+	dec := &avc.DecConfRec{
+		AVCProfileIndication: sps[1],
+		ProfileCompatibility: sps[2],
+		AVCLevelIndication:   sps[3],
+		SPSnalus:             spsNALUs,
+		PPSnalus:             ppsNALUs,
+		ChromaFormat:         1,
+		BitDepthLumaMinus1:   0,
+		BitDepthChromaMinus1: 0,
+		NumSPSExt:            0,
+		NoTrailingInfo:       true,
+	}
+	return dec, nil
+}
+
 // splitNALUs splits Annex B data into raw NAL units without start codes.
 func splitNALUs(data []byte) [][]byte {
 	var nalus [][]byte