testing

av/pktque/filters.go

@@ -170,6 +170,20 @@ func (self *AVSync) check(i int) (start time.Duration, end time.Duration, correc
 	return
 }
 
+type CalcDuration struct {
+	LastTime map[int8]time.Duration
+}
+
+func (self *CalcDuration) ModifyPacket(pkt *av.Packet, streams []av.CodecData, videoidx int, audioidx int) (drop bool, err error) {
+	if tmp, ok := self.LastTime[pkt.Idx]; ok && tmp != 0 {
+		pkt.Duration = pkt.Time - self.LastTime[pkt.Idx]
+	} else if pkt.Time < 100*time.Millisecond {
+		pkt.Duration = pkt.Time
+	}
+	self.LastTime[pkt.Idx] = pkt.Time
+	return
+}
+
 // Make packets reading speed as same as walltime, effect like ffmpeg -re option.
 type Walltime struct {
 	firsttime time.Time

example/test/main.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"github.com/deepch/vdk/format/ts"
+	"log"
+	"os"
+)
+
+func main() {
+	f, _ := os.Open("edb9708f29b24ba9b175808d6b9df9c6541e25766d4a40209a8f903948b72f3f.ts")
+	m := ts.NewDemuxer(f)
+	var i int
+	for {
+		p, err := m.ReadPacket()
+		if err != nil {
+			return
+		}
+		if p.IsKeyFrame {
+			i = 0
+		}
+		log.Println(i, p.Time, p.Data[4:10], len(p.Data))
+		i++
+
+	}
+}

format/mp4/muxer.go

@@ -3,23 +3,22 @@ package mp4
 import (
 	"bufio"
 	"fmt"
-	"io"
-	"time"
-
-	"github.com/deepch/vdk/codec/h265parser"
-
 	"github.com/deepch/vdk/av"
 	"github.com/deepch/vdk/codec/aacparser"
 	"github.com/deepch/vdk/codec/h264parser"
+	"github.com/deepch/vdk/codec/h265parser"
 	"github.com/deepch/vdk/format/mp4/mp4io"
 	"github.com/deepch/vdk/utils/bits/pio"
+	"io"
+	"time"
 )
 
 type Muxer struct {
-	w       io.WriteSeeker
-	bufw    *bufio.Writer
-	wpos    int64
-	streams []*Stream
+	w                  io.WriteSeeker
+	bufw               *bufio.Writer
+	wpos               int64
+	streams            []*Stream
+	NegativeTsMakeZero bool
 }
 
 func NewMuxer(w io.WriteSeeker) *Muxer {
@@ -206,8 +205,12 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) {
 
 func (self *Stream) writePacket(pkt av.Packet, rawdur time.Duration) (err error) {
 	if rawdur < 0 {
-		err = fmt.Errorf("mp4: stream#%d time=%v < lasttime=%v", pkt.Idx, pkt.Time, self.lastpkt.Time)
-		return
+		if self.muxer.NegativeTsMakeZero {
+			rawdur = 0
+		} else {
+			err = fmt.Errorf("mp4: stream#%d time=%v < lasttime=%v", pkt.Idx, pkt.Time, self.lastpkt.Time)
+			return
+		}
 	}
 
 	if _, err = self.muxer.bufw.Write(pkt.Data); err != nil {

format/mp4f/muxer.go

@@ -3,6 +3,7 @@ package mp4f
 import (
 	"bufio"
 	"fmt"
+	"log"
 	"os"
 	"time"
 
@@ -238,14 +239,15 @@ func (self *Muxer) WriteTrailer() (err error) {
 	return
 }
 
-func (element *Muxer) WriteHeader(streams []av.CodecData) (err error) {
+func (element *Muxer) WriteHeader(streams []av.CodecData) error {
 	element.streams = []*Stream{}
 	for _, stream := range streams {
-		if err = element.newStream(stream); err != nil {
-			return
+		if err := element.newStream(stream); err != nil {
+			log.Println("WriteHeader", err)
 		}
 	}
-	return
+
+	return nil
 }
 
 func (element *Muxer) GetInit(streams []av.CodecData) (string, []byte) {
@@ -285,6 +287,9 @@ func (element *Muxer) GetInit(streams []av.CodecData) (string, []byte) {
 }
 
 func (element *Muxer) WritePacket(pkt av.Packet, GOP bool) (bool, []byte, error) {
+	if pkt.Idx+1 > int8(len(element.streams)) {
+		return false, nil, nil
+	}
 	stream := element.streams[pkt.Idx]
 	if GOP {
 		ts := time.Duration(0)

format/mp4m/muxer.go

@@ -14,10 +14,11 @@ import (
 )
 
 type Muxer struct {
-	w       io.WriteSeeker
-	bufw    *bufio.Writer
-	wpos    int64
-	streams []*Stream
+	w                  io.WriteSeeker
+	bufw               *bufio.Writer
+	wpos               int64
+	streams            []*Stream
+	NegativeTsMakeZero bool
 }
 
 func NewMuxer(w io.WriteSeeker) *Muxer {
@@ -181,8 +182,12 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) {
 
 func (self *Stream) writePacket(pkt av.Packet, rawdur time.Duration) (err error) {
 	if rawdur < 0 {
-		err = fmt.Errorf("mp4: stream#%d time=%v < lasttime=%v", pkt.Idx, pkt.Time, self.lastpkt.Time)
-		return
+		if self.muxer.NegativeTsMakeZero {
+			rawdur = 0
+		} else {
+			err = fmt.Errorf("mp4: stream#%d time=%v < lasttime=%v", pkt.Idx, pkt.Time, self.lastpkt.Time)
+			return
+		}
 	}
 
 	if _, err = self.muxer.bufw.Write(pkt.Data); err != nil {

format/nvr/muxer.go

@@ -4,11 +4,15 @@ import (
 	"bytes"
 	"encoding/binary"
 	"encoding/gob"
+	"errors"
 	"fmt"
 	"github.com/deepch/vdk/av"
 	"github.com/deepch/vdk/codec/aacparser"
 	"github.com/deepch/vdk/codec/h264parser"
 	"github.com/deepch/vdk/format/mp4"
+	"github.com/google/uuid"
+	"github.com/moby/sys/mountinfo"
+	"github.com/shirou/gopsutil/v3/disk"
 	"os"
 	"path/filepath"
 	"strings"
@@ -16,7 +20,7 @@ import (
 )
 
 var MIME = []byte{11, 22, 111, 222, 11, 22, 111, 222}
-var listTag = []string{"{host_name}", "{stream_name}", "{channel_name}", "{stream_id}", "{channel_id}", "{start_year}", "{start_month}", "{start_day}", "{start_minute}", "{start_second}", "{start_millisecond}", "{start_unix_second}", "{start_unix_millisecond}", "{start_time}", "{start_pts}", "{end_year}", "{end_month}", "{end_day}", "{end_minute}", "{end_second}", "{end_millisecond}", "{start_unix_second}", "{start_unix_millisecond}", "{end_time}", "{end_pts}", "{duration_second}", "{duration_millisecond}"}
+var listTag = []string{"{server_id}", "{host_name}", "{host_name_short}", "{host_name_long}", "{stream_name}", "{channel_name}", "{stream_id}", "{channel_id}", "{start_year}", "{start_month}", "{start_day}", "{start_hour}", "{start_minute}", "{start_second}", "{start_millisecond}", "{start_unix_second}", "{start_unix_millisecond}", "{start_time}", "{start_pts}", "{end_year}", "{end_month}", "{end_day}", "{end_hour}", "{end_minute}", "{end_second}", "{end_millisecond}", "{start_unix_second}", "{start_unix_millisecond}", "{end_time}", "{end_pts}", "{duration_second}", "{duration_millisecond}"}
 
 const (
 	MP4 = "mp4"
@@ -24,19 +28,20 @@ const (
 )
 
 type Muxer struct {
-	muxer                                                  *mp4.Muxer
-	format                                                 string
-	limit                                                  int
-	d                                                      *os.File
-	m                                                      *os.File
-	dur                                                    time.Duration
-	h                                                      int
-	gof                                                    *Gof
-	patch                                                  string
-	start, end                                             time.Time
-	pstart, pend                                           time.Duration
-	started                                                bool
-	hostName, streamName, channelName, streamID, channelID string
+	muxer                                                            *mp4.Muxer
+	format                                                           string
+	limit                                                            int
+	d                                                                *os.File
+	m                                                                *os.File
+	dur                                                              time.Duration
+	h                                                                int
+	gof                                                              *Gof
+	patch                                                            string
+	mpoint                                                           []string
+	start, end                                                       time.Time
+	pstart, pend                                                     time.Duration
+	started                                                          bool
+	serverID, streamName, channelName, streamID, channelID, hostname string
 }
 
 type Gof struct {
@@ -64,18 +69,21 @@ func init() {
 
 }
 
-func NewMuxer(hostName, streamName, channelName, streamID, channelID, patch string, format string, limit int) (m *Muxer, err error) {
+func NewMuxer(serverID, streamName, channelName, streamID, channelID string, mpoint []string, patch, format string, limit int) (m *Muxer, err error) {
+	hostname, _ := os.Hostname()
 	m = &Muxer{
+		mpoint:      mpoint,
 		patch:       patch,
 		h:           -1,
 		gof:         &Gof{},
 		format:      format,
 		limit:       limit,
-		hostName:    hostName,
+		serverID:    serverID,
 		streamName:  streamName,
 		channelName: channelName,
 		streamID:    streamID,
 		channelID:   channelID,
+		hostname:    hostname,
 	}
 	return
 }
@@ -83,7 +91,7 @@ func NewMuxer(hostName, streamName, channelName, streamID, channelID, patch stri
 func (m *Muxer) WriteHeader(streams []av.CodecData) (err error) {
 	m.gof.Streams = streams
 	if m.format == MP4 {
-		m.OpenMP4()
+		return m.OpenMP4()
 	}
 
 	return
@@ -111,11 +119,15 @@ func (m *Muxer) WritePacket(pkt av.Packet) (err error) {
 func (m *Muxer) writePacketMP4(pkt av.Packet) (err error) {
 	if pkt.IsKeyFrame && m.dur > time.Duration(m.limit)*time.Second {
 		m.pstart = pkt.Time
-		m.OpenMP4()
+		if err = m.OpenMP4(); err != nil {
+			return
+		}
 		m.dur = 0
+
 	}
 	m.dur += pkt.Duration
 	m.pend = pkt.Time
+
 	return m.muxer.WritePacket(pkt)
 }
 
@@ -186,10 +198,19 @@ func (m *Muxer) OpenNVR() (err error) {
 func (m *Muxer) OpenMP4() (err error) {
 	m.WriteTrailer()
 	m.start = time.Now().UTC()
-	if m.d, err = os.CreateTemp("", "rtspvideo.*.mp4"); err != nil {
+
+	d, err := m.filePatch()
+	if err != nil {
+		return
+	}
+	if err = os.MkdirAll(filepath.Dir(d), 0755); err != nil {
+		return
+	}
+	if m.d, err = os.Create(filepath.Join(filepath.Dir(d), fmt.Sprintf("tmp_%s_%d.mp4", uuid.New(), time.Now().Unix()))); err != nil {
 		return
 	}
 	m.muxer = mp4.NewMuxer(m.d)
+	m.muxer.NegativeTsMakeZero = true
 	if err = m.muxer.WriteHeader(m.gof.Streams); err != nil {
 		return
 	}
@@ -197,13 +218,40 @@ func (m *Muxer) OpenMP4() (err error) {
 	return
 }
 
-func (m *Muxer) filePatch() string {
-	ts := m.patch
+func (m *Muxer) filePatch() (string, error) {
+	var (
+		mu = float64(100)
+		ui = -1
+	)
+
+	for i, i2 := range m.mpoint {
+		if m, err := mountinfo.Mounted(i2); err == nil && m {
+			if d, err := disk.Usage(i2); err == nil {
+				if d.UsedPercent < mu {
+					ui = i
+					mu = d.UsedPercent
+				}
+			}
+		}
+	}
+
+	if ui == -1 {
+		return "", errors.New("not mount ready")
+	}
+
+	ts := filepath.Join(m.mpoint[ui], m.patch)
 	m.end = time.Now().UTC()
+
 	for _, s := range listTag {
 		switch s {
+		case "{server_id}":
+			ts = strings.Replace(ts, "{server_id}", m.serverID, -1)
 		case "{host_name}":
-			ts = strings.Replace(ts, "{host_name}", m.hostName, -1)
+			ts = strings.Replace(ts, "{host_name}", m.hostname, -1)
+		case "{host_name_short}":
+			ts = strings.Replace(ts, "{host_name_short}", m.hostname, -1)
+		case "{host_name_long}":
+			ts = strings.Replace(ts, "{host_name_long}", m.hostname, -1)
 		case "{stream_name}":
 			ts = strings.Replace(ts, "{stream_name}", m.streamName, -1)
 		case "{channel_name}":
@@ -215,13 +263,15 @@ func (m *Muxer) filePatch() string {
 		case "{start_year}":
 			ts = strings.Replace(ts, "{start_year}", fmt.Sprintf("%d", m.start.Year()), -1)
 		case "{start_month}":
-			ts = strings.Replace(ts, "{start_month}", fmt.Sprintf("%02d", int(m.start.Month())), -1)
+			ts = strings.Replace(ts, "{start_month}", fmt.Sprintf("%d", int(m.start.Month())), -1)
 		case "{start_day}":
-			ts = strings.Replace(ts, "{start_day}", fmt.Sprintf("%02d", m.start.Day()), -1)
+			ts = strings.Replace(ts, "{start_day}", fmt.Sprintf("%d", m.start.Day()), -1)
+		case "{start_hour}":
+			ts = strings.Replace(ts, "{start_hour}", fmt.Sprintf("%d", m.start.Hour()), -1)
 		case "{start_minute}":
-			ts = strings.Replace(ts, "{start_minute}", fmt.Sprintf("%02d", m.start.Minute()), -1)
+			ts = strings.Replace(ts, "{start_minute}", fmt.Sprintf("%d", m.start.Minute()), -1)
 		case "{start_second}":
-			ts = strings.Replace(ts, "{start_second}", fmt.Sprintf("%02d", m.start.Second()), -1)
+			ts = strings.Replace(ts, "{start_second}", fmt.Sprintf("%d", m.start.Second()), -1)
 		case "{start_millisecond}":
 			ts = strings.Replace(ts, "{start_millisecond}", fmt.Sprintf("%d", m.start.Nanosecond()/1000/1000), -1)
 		case "{start_unix_millisecond}":
@@ -235,13 +285,15 @@ func (m *Muxer) filePatch() string {
 		case "{end_year}":
 			ts = strings.Replace(ts, "{end_year}", fmt.Sprintf("%d", m.end.Year()), -1)
 		case "{end_month}":
-			ts = strings.Replace(ts, "{end_month}", fmt.Sprintf("%02d", int(m.end.Month())), -1)
+			ts = strings.Replace(ts, "{end_month}", fmt.Sprintf("%d", int(m.end.Month())), -1)
 		case "{end_day}":
-			ts = strings.Replace(ts, "{end_day}", fmt.Sprintf("%02d", m.end.Day()), -1)
+			ts = strings.Replace(ts, "{end_day}", fmt.Sprintf("%d", m.end.Day()), -1)
+		case "{end_hour}":
+			ts = strings.Replace(ts, "{end_hour}", fmt.Sprintf("%d", m.end.Hour()), -1)
 		case "{end_minute}":
-			ts = strings.Replace(ts, "{end_minute}", fmt.Sprintf("%02d", m.end.Minute()), -1)
+			ts = strings.Replace(ts, "{end_minute}", fmt.Sprintf("%d", m.end.Minute()), -1)
 		case "{end_second}":
-			ts = strings.Replace(ts, "{end_second}", fmt.Sprintf("%02d", m.end.Second()), -1)
+			ts = strings.Replace(ts, "{end_second}", fmt.Sprintf("%d", m.end.Second()), -1)
 		case "{end_millisecond}":
 			ts = strings.Replace(ts, "{end_millisecond}", fmt.Sprintf("%d", m.start.Nanosecond()/1000/1000), -1)
 		case "{end_unix_millisecond}":
@@ -259,7 +311,7 @@ func (m *Muxer) filePatch() string {
 		}
 	}
 
-	return ts
+	return ts, nil
 }
 
 func (m *Muxer) WriteTrailer() (err error) {
@@ -271,12 +323,12 @@ func (m *Muxer) WriteTrailer() (err error) {
 	}
 	if m.d != nil {
 		if m.format == MP4 {
-			p := m.filePatch()
-			if err = os.MkdirAll(filepath.Dir(p), 0755); err != nil {
-				return
+			p, err := m.filePatch()
+			if err != nil {
+				return err
 			}
-			if err = os.Rename(m.d.Name(), p); err != nil {
-				return
+			if err = os.Rename(m.d.Name(), filepath.Join(filepath.Dir(m.d.Name()), filepath.Base(p))); err != nil {
+				return err
 			}
 		}
 		err = m.d.Close()

format/rtsp/client.go

@@ -40,9 +40,10 @@ const (
 )
 
 type Client struct {
-	DebugRtsp bool
-	DebugRtp  bool
-	Headers   []string
+	DebugRtsp    bool
+	DebugRtp     bool
+	DisableAudio bool
+	Headers      []string
 
 	SkipErrRtpBlock bool
 
@@ -1076,12 +1077,44 @@ func (self *Stream) handleRtpPacket(packet []byte) (err error) {
 		err = fmt.Errorf("rtp: packet too short")
 		return
 	}
-	payloadOffset := 12 + int(packet[0]&0xf)*4
+
+	timestamp := binary.BigEndian.Uint32(packet[4:8])
+
+	/*
+		Test offset
+	*/
+	Padding := (packet[0]>>5)&1 == 1
+	Extension := (packet[0]>>4)&1 == 1
+	CSRCCnt := int(packet[0] & 0x0f)
+
+	RTPHeaderSize := 12
+
+	payloadOffset := RTPHeaderSize
+	end := len(packet)
+	if end-payloadOffset >= 4*CSRCCnt {
+		payloadOffset += 4 * CSRCCnt
+	}
+
+	if Extension && end-payloadOffset >= 4 {
+		extLen := 4 * int(binary.BigEndian.Uint16(packet[payloadOffset+2:]))
+		payloadOffset += 4
+		if end-payloadOffset >= extLen {
+			payloadOffset += extLen
+		}
+	}
+
+	if Padding && end-payloadOffset > 0 {
+		paddingLen := int(packet[end-1])
+		if end-payloadOffset >= paddingLen {
+			end -= paddingLen
+		}
+	}
+
 	if payloadOffset > len(packet) {
 		err = fmt.Errorf("rtp: packet too short")
 		return
 	}
-	timestamp := binary.BigEndian.Uint32(packet[4:8])
+
 	payload := packet[payloadOffset:]
 
 	/*

format/rtsp/server.go

@@ -0,0 +1 @@
+package rtsp

format/ts/demuxer.go

@@ -284,7 +284,11 @@ func (self *Stream) payloadEnd() (n int, err error) {
 						b := make([]byte, 4+len(nalu))
 						pio.PutU32BE(b[0:4], uint32(len(nalu)))
 						copy(b[4:], nalu)
-						self.addPacket(b, time.Duration(0), (1000*time.Millisecond)/time.Duration(self.fps))
+						fps := self.fps
+						if self.fps == 0 {
+							fps = 25
+						}
+						self.addPacket(b, time.Duration(0), (1000*time.Millisecond)/time.Duration(fps))
 						n++
 					}
 				}