rpi: route original absolute timestamp of packets (#1300) (#4382)

Author: aler9

README.md

@@ -1709,10 +1709,11 @@ The command inserted into `runOnDemand` will start only when a client requests t
 
 ### Route absolute timestamps
 
-Some streaming protocols allow to route absolute timestamps, associated with each frame, that are useful for synchronizing several video or data streams together. In particular, _MediaMTX_ supports receiving absolute timestamps with the following protocols:
+Some streaming protocols allow to route absolute timestamps, associated with each frame, that are useful for synchronizing several video or data streams together. In particular, _MediaMTX_ supports receiving absolute timestamps with the following protocols and devices:
 
 * HLS (through the `EXT-X-PROGRAM-DATE-TIME` tag in playlists)
 * RTSP (through RTCP reports, when `rtspAbsoluteTimestamp` is `true` in settings)
+* Raspberry Pi Camera
 
 and supports sending absolute timestamps with the following protocols:
 

internal/staticsources/rpicamera/camera.go

@@ -8,14 +8,34 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strconv"
+	"syscall"
 	"time"
+	"unsafe"
 
 	"github.com/bluenviron/mediacommon/v2/pkg/codecs/h264"
 )
 
+func ntpTime() syscall.Timespec {
+	var t syscall.Timespec
+	syscall.Syscall(syscall.SYS_CLOCK_GETTIME, 0, uintptr(unsafe.Pointer(&t)), 0)
+	return t
+}
+
+func monotonicTime() syscall.Timespec {
+	var t syscall.Timespec
+	syscall.Syscall(syscall.SYS_CLOCK_GETTIME, 1, uintptr(unsafe.Pointer(&t)), 0)
+	return t
+}
+
+func multiplyAndDivide(v, m, d int64) int64 {
+	secs := v / d
+	dec := v % d
+	return (secs*m + dec*m/d)
+}
+
 type camera struct {
-	Params params
-	OnData func(time.Duration, [][]byte)
+	params params
+	onData func(int64, time.Time, [][]byte)
 
 	cmd      *exec.Cmd
 	pipeOut  *pipe
@@ -70,7 +90,7 @@ func (c *camera) initialize() error {
 
 	go c.run()
 
-	c.pipeOut.write(append([]byte{'c'}, c.Params.serialize()...))
+	c.pipeOut.write(append([]byte{'c'}, c.params.serialize()...))
 
 	return nil
 }
@@ -162,17 +182,27 @@ outer:
 			return fmt.Errorf(string(buf[1:]))
 
 		case 'b':
-			tmp := uint64(buf[8])<<56 | uint64(buf[7])<<48 | uint64(buf[6])<<40 | uint64(buf[5])<<32 |
-				uint64(buf[4])<<24 | uint64(buf[3])<<16 | uint64(buf[2])<<8 | uint64(buf[1])
-			dts := time.Duration(tmp) * time.Microsecond
+			dts := int64(buf[8])<<56 | int64(buf[7])<<48 | int64(buf[6])<<40 | int64(buf[5])<<32 |
+				int64(buf[4])<<24 | int64(buf[3])<<16 | int64(buf[2])<<8 | int64(buf[1])
 
 			var nalus h264.AnnexB
 			err = nalus.Unmarshal(buf[9:])
 			if err != nil {
 				return err
 			}
 
-			c.OnData(dts, nalus)
+			unixNTP := ntpTime()
+			unixMono := monotonicTime()
+
+			// subtract from NTP the delay from now to the moment the frame was taken
+			ntp := time.Unix(int64(unixNTP.Sec), int64(unixNTP.Nsec))
+			deltaT := time.Duration(unixMono.Nano()-dts*1e3) * time.Nanosecond
+			ntp = ntp.Add(-deltaT)
+
+			c.onData(
+				multiplyAndDivide(dts, 90000, 1e6),
+				ntp,
+				nalus)
 
 		default:
 			return fmt.Errorf("unexpected data from pipe: '0x%.2x'", buf[0])

internal/staticsources/rpicamera/camera_disabled.go

@@ -8,8 +8,8 @@ import (
 )
 
 type camera struct {
-	Params params
-	OnData func(time.Duration, [][]byte)
+	params params
+	onData func(int64, time.Time, [][]byte)
 }
 
 func (c *camera) initialize() error {

internal/staticsources/rpicamera/source.go

@@ -14,12 +14,6 @@ import (
 	"github.com/bluenviron/mediamtx/internal/unit"
 )
 
-func multiplyAndDivide(v, m, d int64) int64 {
-	secs := v / d
-	dec := v % d
-	return (secs*m + dec*m/d)
-}
-
 func paramsFromConf(logLevel conf.LogLevel, cnf *conf.Path) params {
 	return params{
 		LogLevel: func() string {
@@ -95,7 +89,7 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 	medias := []*description.Media{medi}
 	var stream *stream.Stream
 
-	onData := func(dts time.Duration, au [][]byte) {
+	onData := func(pts int64, ntp time.Time, au [][]byte) {
 		if stream == nil {
 			res := s.Parent.SetReady(defs.PathSourceStaticSetReadyReq{
 				Desc:               &description.Session{Medias: medias},
@@ -110,8 +104,8 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 
 		stream.WriteUnit(medi, medi.Formats[0], &unit.H264{
 			Base: unit.Base{
-				NTP: time.Now(),
-				PTS: multiplyAndDivide(int64(dts), 90000, int64(time.Second)),
+				PTS: pts,
+				NTP: ntp,
 			},
 			AU: au,
 		})
@@ -124,8 +118,8 @@ func (s *Source) Run(params defs.StaticSourceRunParams) error {
 	}()
 
 	cam := &camera{
-		Params: paramsFromConf(s.LogLevel, params.Conf),
-		OnData: onData,
+		params: paramsFromConf(s.LogLevel, params.Conf),
+		onData: onData,
 	}
 	err := cam.initialize()
 	if err != nil {