connect to Kerberos Hub if required, even if camera is not working

machinery/src/cloud/Cloud.go

@@ -128,8 +128,10 @@ func HandleUpload(configuration *models.Configuration, communication *models.Com
 							log.Log.Error("HandleUpload: " + err.Error())
 						}
 					} else {
-						delay = 5 * time.Second // slow down
-						log.Log.Error("HandleUpload: " + err.Error())
+						delay = 20 * time.Second // slow down
+						if err != nil {
+							log.Log.Error("HandleUpload: " + err.Error())
+						}
 					}
 
 					time.Sleep(delay)
@@ -215,36 +217,36 @@ func GetSystemInfo() (models.System, error) {
 func HandleHeartBeat(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time) {
 	log.Log.Debug("HandleHeartBeat: started")
 
-	config := configuration.Config
+loop:
+	for {
 
-	if config.Offline == "true" {
-		log.Log.Debug("HandleHeartBeat: stopping as Offline is enabled.")
-	} else {
+		config := configuration.Config
 
-		url := config.HeartbeatURI
-		key := ""
-		username := ""
-		vaultURI := ""
+		if config.Offline == "true" {
+			log.Log.Debug("HandleHeartBeat: stopping as Offline is enabled.")
+		} else {
+
+			url := config.HeartbeatURI
+			key := ""
+			username := ""
+			vaultURI := ""
 
-		username = config.S3.Username
-		if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" {
 			username = config.S3.Username
-			key = config.S3.Publickey
-		} else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" {
-			key = config.KStorage.CloudKey
-			username = config.KStorage.Directory
-		}
+			if config.Cloud == "s3" && config.S3 != nil && config.S3.Publickey != "" {
+				username = config.S3.Username
+				key = config.S3.Publickey
+			} else if config.Cloud == "kstorage" && config.KStorage != nil && config.KStorage.CloudKey != "" {
+				key = config.KStorage.CloudKey
+				username = config.KStorage.Directory
+			}
 
-		// This is the new way ;)
-		if config.HubURI != "" {
-			url = config.HubURI + "/devices/heartbeat"
-		}
-		if config.HubKey != "" {
-			key = config.HubKey
-		}
-
-	loop:
-		for {
+			// This is the new way ;)
+			if config.HubURI != "" {
+				url = config.HubURI + "/devices/heartbeat"
+			}
+			if config.HubKey != "" {
+				key = config.HubKey
+			}
 
 			if key != "" {
 				// Check if we have a friendly name or not.
@@ -292,39 +294,44 @@ func HandleHeartBeat(configuration *models.Configuration, communication *models.
 				// Congert to string
 				macs, _ := json.Marshal(system.MACs)
 				ips, _ := json.Marshal(system.IPs)
+				cameraConnected := "true"
+				if communication.CameraConnected == false {
+					cameraConnected = "false"
+				}
 
 				var object = fmt.Sprintf(`{
-					"key" : "%s",
-					"version" : "3.0.0",
-					"release" : "%s",
-					"cpuid" : "%s",
-					"clouduser" : "%s",
-					"cloudpublickey" : "%s",
-					"cameraname" : "%s",
-					"enterprise" : %t,
-					"hostname" : "%s",
-					"architecture" : "%s",
-					"totalMemory" : "%d",
-					"usedMemory" : "%d",
-					"freeMemory" : "%d",
-					"processMemory" : "%d",
-					"mac_list" : %s,
-					"ip_list" : %s,
-					"board" : "",
-					"disk1size" : "%s",
-					"disk3size" : "%s",
-					"diskvdasize" :  "%s",
-					"uptime" : "%s",
-					"boot_time" : "%s",
-					"siteID" : "%s",
-					"onvif" : "%s",
-					"numberoffiles" : "33",
-					"timestamp" : 1564747908,
-					"cameratype" : "IPCamera",
-					"docker" : true,
-					"kios" : false,
-					"raspberrypi" : false
-				}`, config.Key, system.Version, system.CPUId, username, key, name, isEnterprise, system.Hostname, system.Architecture, system.TotalMemory, system.UsedMemory, system.FreeMemory, system.ProcessUsedMemory, macs, ips, "0", "0", "0", uptimeString, boottimeString, config.HubSite, onvifEnabled)
+						"key" : "%s",
+						"version" : "3.0.0",
+						"release" : "%s",
+						"cpuid" : "%s",
+						"clouduser" : "%s",
+						"cloudpublickey" : "%s",
+						"cameraname" : "%s",
+						"enterprise" : %t,
+						"hostname" : "%s",
+						"architecture" : "%s",
+						"totalMemory" : "%d",
+						"usedMemory" : "%d",
+						"freeMemory" : "%d",
+						"processMemory" : "%d",
+						"mac_list" : %s,
+						"ip_list" : %s,
+						"board" : "",
+						"disk1size" : "%s",
+						"disk3size" : "%s",
+						"diskvdasize" :  "%s",
+						"uptime" : "%s",
+						"boot_time" : "%s",
+						"siteID" : "%s",
+						"onvif" : "%s",
+						"cameraConnected": "%s",
+						"numberoffiles" : "33",
+						"timestamp" : 1564747908,
+						"cameratype" : "IPCamera",
+						"docker" : true,
+						"kios" : false,
+						"raspberrypi" : false
+					}`, config.Key, system.Version, system.CPUId, username, key, name, isEnterprise, system.Hostname, system.Architecture, system.TotalMemory, system.UsedMemory, system.FreeMemory, system.ProcessUsedMemory, macs, ips, "0", "0", "0", uptimeString, boottimeString, config.HubSite, onvifEnabled, cameraConnected)
 
 				var jsonStr = []byte(object)
 				buffy := bytes.NewBuffer(jsonStr)
@@ -365,14 +372,14 @@ func HandleHeartBeat(configuration *models.Configuration, communication *models.
 			} else {
 				log.Log.Error("HandleHeartBeat: Disabled as we do not have a public key defined.")
 			}
+		}
 
-			// This will check if we need to stop the thread,
-			// because of a reconfiguration.
-			select {
-			case <-communication.HandleHeartBeat:
-				break loop
-			case <-time.After(15 * time.Second):
-			}
+		// This will check if we need to stop the thread,
+		// because of a reconfiguration.
+		select {
+		case <-communication.HandleHeartBeat:
+			break loop
+		case <-time.After(15 * time.Second):
 		}
 	}
 

machinery/src/components/Kerberos.go

@@ -64,6 +64,9 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun
 	subDecoder := &ffmpeg.VideoDecoder{}
 	cameraSettings := &models.Camera{}
 
+	// Handle heartbeats
+	go cloud.HandleHeartBeat(configuration, communication, uptimeStart)
+
 	// Run the agent and fire up all the other
 	// goroutines which do image capture, motion detection, onvif, etc.
 
@@ -75,13 +78,16 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun
 		}
 		// We will re open the configuration, might have changed :O!
 		OpenConfig(configuration)
+
+		// We will override the configuration with the environment variables
+		OverrideWithEnvironmentVariables(configuration)
 	}
 	log.Log.Debug("Bootstrap: finished")
 }
 
 func RunAgent(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time, cameraSettings *models.Camera, decoder *ffmpeg.VideoDecoder, subDecoder *ffmpeg.VideoDecoder) string {
-	log.Log.Debug("RunAgent: bootstrapping agent")
 
+	log.Log.Debug("RunAgent: bootstrapping agent")
 	config := configuration.Config
 
 	// Currently only support H264 encoded cameras, this will change.
@@ -185,9 +191,6 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 		communication.HandleONVIF = make(chan models.OnvifAction, 1)
 		mqttClient := routers.ConfigureMQTT(configuration, communication)
 
-		// Handle heartbeats
-		go cloud.HandleHeartBeat(configuration, communication, uptimeStart)
-
 		// Handle the camera stream
 		go capture.HandleStream(infile, queue, communication)
 
@@ -241,14 +244,14 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 
 		// Here we are cleaning up everything!
 		if configuration.Config.Offline != "true" {
-			communication.HandleHeartBeat <- "stop"
 			communication.HandleUpload <- "stop"
 		}
 		communication.HandleStream <- "stop"
 		if subStreamEnabled {
 			communication.HandleSubStream <- "stop"
 		}
-		time.Sleep(time.Second * 1)
+
+		time.Sleep(time.Second * 3)
 
 		infile.Close()
 		infile = nil
@@ -278,6 +281,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 		// Waiting for some seconds to make sure everything is properly closed.
 		log.Log.Info("RunAgent: waiting 3 seconds to make sure everything is properly closed.")
 		time.Sleep(time.Second * 3)
+
 	} else {
 		log.Log.Error("Something went wrong while opening RTSP: " + err.Error())
 		time.Sleep(time.Second * 3)

machinery/src/models/Communication.go

@@ -34,4 +34,5 @@ type Communication struct {
 	Decoder               *ffmpeg.VideoDecoder
 	SubDecoder            *ffmpeg.VideoDecoder
 	Image                 string
+	CameraConnected       bool
 }

machinery/src/rtsp/client.go

@@ -0,0 +1,247 @@
+package rtsp
+
+import (
+	"fmt"
+	"image"
+	"image/jpeg"
+	"log"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/bluenviron/gortsplib/v3"
+	"github.com/bluenviron/gortsplib/v3/pkg/base"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats"
+	"github.com/bluenviron/gortsplib/v3/pkg/formats/rtph265"
+	"github.com/bluenviron/gortsplib/v3/pkg/url"
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+
+	"github.com/pion/rtp"
+)
+
+func CreateClient() {
+	c := &gortsplib.Client{
+		OnRequest: func(req *base.Request) {
+			//log.Log.Info(logger.Debug, "c->s %v", req)
+		},
+		OnResponse: func(res *base.Response) {
+			//s.Log(logger.Debug, "s->c %v", res)
+		},
+		OnTransportSwitch: func(err error) {
+			//s.Log(logger.Warn, err.Error())
+		},
+		OnPacketLost: func(err error) {
+			//s.Log(logger.Warn, err.Error())
+		},
+		OnDecodeError: func(err error) {
+			//s.Log(logger.Warn, err.Error())
+		},
+	}
+
+	u, err := url.Parse("rtsp://admin:admin@192.168.1.111") //"rtsp://seing:bud-edPTQc@109.159.199.103:554/rtsp/defaultPrimary?mtu=1440&streamType=m") //
+	if err != nil {
+		panic(err)
+	}
+
+	err = c.Start(u.Scheme, u.Host)
+	if err != nil {
+		//return err
+	}
+	defer c.Close()
+
+	medias, baseURL, _, err := c.Describe(u)
+	if err != nil {
+		//return err
+	}
+	fmt.Println(medias)
+
+	// find the H264 media and format
+	var forma *formats.H265
+	medi := medias.FindFormat(&forma)
+	if medi == nil {
+		panic("media not found")
+	}
+
+	// setup RTP/H264 -> H264 decoder
+	rtpDec := forma.CreateDecoder()
+	// setup H264 -> MPEG-TS muxer
+	//pegtsMuxer, err := newMPEGTSMuxer(forma.SPS, forma.PPS)
+	if err != nil {
+		panic(err)
+	}
+
+	// setup H264 -> raw frames decoder
+	/*h264RawDec, err := newH264Decoder()
+	if err != nil {
+		panic(err)
+	}
+	defer h264RawDec.close()
+
+	// if SPS and PPS are present into the SDP, send them to the decoder
+	if forma.SPS != nil {
+		h264RawDec.decode(forma.SPS)
+	}
+	if forma.PPS != nil {
+		h264RawDec.decode(forma.PPS)
+	}*/
+
+	readErr := make(chan error)
+	go func() {
+		readErr <- func() error {
+			// Get codecs
+			for _, medi := range medias {
+				for _, forma := range medi.Formats {
+					fmt.Println(forma)
+				}
+			}
+
+			err = c.SetupAll(medias, baseURL)
+			if err != nil {
+				return err
+			}
+
+			for _, medi := range medias {
+				for _, forma := range medi.Formats {
+					c.OnPacketRTP(medi, forma, func(pkt *rtp.Packet) {
+
+						au, pts, err := rtpDec.Decode(pkt)
+						if err != nil {
+							if err != rtph265.ErrNonStartingPacketAndNoPrevious && err != rtph265.ErrMorePacketsNeeded {
+								log.Printf("ERR: %v", err)
+							}
+							return
+						}
+
+						for _, nalu := range au {
+							log.Printf("received NALU with PTS %v and size %d\n", pts, len(nalu))
+						}
+
+						/*// extract access unit from RTP packets
+						// DecodeUntilMarker is necessary for the DTS extractor to work
+						if pkt.PayloadType == 96 {
+							au, pts, err := rtpDec.DecodeUntilMarker(pkt)
+
+							if err != nil {
+								if err != rtph264.ErrNonStartingPacketAndNoPrevious && err != rtph264.ErrMorePacketsNeeded {
+									log.Printf("ERR: %v", err)
+								}
+								return
+							}
+
+							// encode the access unit into MPEG-TS
+							mpegtsMuxer.encode(au, pts)
+
+							for _, nalu := range au {
+								// convert NALUs into RGBA frames
+								img, err := h264RawDec.decode(nalu)
+								if err != nil {
+									panic(err)
+								}
+
+								// wait for a frame
+								if img == nil {
+									continue
+								}
+
+								// convert frame to JPEG and save to file
+								err = saveToFile(img)
+								if err != nil {
+									panic(err)
+								}
+							}
+						}*/
+
+					})
+				}
+			}
+
+			_, err = c.Play(nil)
+			if err != nil {
+				return err
+			}
+
+			return c.Wait()
+		}()
+	}()
+
+	for {
+		select {
+		case err := <-readErr:
+			fmt.Println(err)
+		}
+	}
+}
+
+func saveToFile(img image.Image) error {
+	// create file
+	fname := strconv.FormatInt(time.Now().UnixNano()/int64(time.Millisecond), 10) + ".jpg"
+	f, err := os.Create(fname)
+	if err != nil {
+		panic(err)
+	}
+	defer f.Close()
+
+	log.Println("saving", fname)
+
+	// convert to jpeg
+	return jpeg.Encode(f, img, &jpeg.Options{
+		Quality: 60,
+	})
+}
+
+// extract SPS and PPS without decoding RTP packets
+func rtpH264ExtractSPSPPS(pkt *rtp.Packet) ([]byte, []byte) {
+	if len(pkt.Payload) < 1 {
+		return nil, nil
+	}
+
+	typ := h264.NALUType(pkt.Payload[0] & 0x1F)
+
+	switch typ {
+	case h264.NALUTypeSPS:
+		return pkt.Payload, nil
+
+	case h264.NALUTypePPS:
+		return nil, pkt.Payload
+
+	case h264.NALUTypeSTAPA:
+		payload := pkt.Payload[1:]
+		var sps []byte
+		var pps []byte
+
+		for len(payload) > 0 {
+			if len(payload) < 2 {
+				break
+			}
+
+			size := uint16(payload[0])<<8 | uint16(payload[1])
+			payload = payload[2:]
+
+			if size == 0 {
+				break
+			}
+
+			if int(size) > len(payload) {
+				return nil, nil
+			}
+
+			nalu := payload[:size]
+			payload = payload[size:]
+
+			typ = h264.NALUType(nalu[0] & 0x1F)
+
+			switch typ {
+			case h264.NALUTypeSPS:
+				sps = nalu
+
+			case h264.NALUTypePPS:
+				pps = nalu
+			}
+		}
+
+		return sps, pps
+
+	default:
+		return nil, nil
+	}
+}

machinery/src/rtsp/h264decoder.go

@@ -0,0 +1,140 @@
+package rtsp
+
+import (
+	"fmt"
+	"image"
+	"unsafe"
+)
+
+// #cgo pkg-config: libavcodec libavutil libswscale
+// #include <libavcodec/avcodec.h>
+// #include <libavutil/imgutils.h>
+// #include <libswscale/swscale.h>
+import "C"
+
+func frameData(frame *C.AVFrame) **C.uint8_t {
+	return (**C.uint8_t)(unsafe.Pointer(&frame.data[0]))
+}
+
+func frameLineSize(frame *C.AVFrame) *C.int {
+	return (*C.int)(unsafe.Pointer(&frame.linesize[0]))
+}
+
+// h264Decoder is a wrapper around ffmpeg's H264 decoder.
+type h264Decoder struct {
+	codecCtx    *C.AVCodecContext
+	srcFrame    *C.AVFrame
+	swsCtx      *C.struct_SwsContext
+	dstFrame    *C.AVFrame
+	dstFramePtr []uint8
+}
+
+// newH264Decoder allocates a new h264Decoder.
+func newH264Decoder() (*h264Decoder, error) {
+	codec := C.avcodec_find_decoder(C.AV_CODEC_ID_H264)
+	if codec == nil {
+		return nil, fmt.Errorf("avcodec_find_decoder() failed")
+	}
+
+	codecCtx := C.avcodec_alloc_context3(codec)
+	if codecCtx == nil {
+		return nil, fmt.Errorf("avcodec_alloc_context3() failed")
+	}
+
+	res := C.avcodec_open2(codecCtx, codec, nil)
+	if res < 0 {
+		C.avcodec_close(codecCtx)
+		return nil, fmt.Errorf("avcodec_open2() failed")
+	}
+
+	srcFrame := C.av_frame_alloc()
+	if srcFrame == nil {
+		C.avcodec_close(codecCtx)
+		return nil, fmt.Errorf("av_frame_alloc() failed")
+	}
+
+	return &h264Decoder{
+		codecCtx: codecCtx,
+		srcFrame: srcFrame,
+	}, nil
+}
+
+// close closes the decoder.
+func (d *h264Decoder) close() {
+	if d.dstFrame != nil {
+		C.av_frame_free(&d.dstFrame)
+	}
+
+	if d.swsCtx != nil {
+		C.sws_freeContext(d.swsCtx)
+	}
+
+	C.av_frame_free(&d.srcFrame)
+	C.avcodec_close(d.codecCtx)
+}
+
+func (d *h264Decoder) decode(nalu []byte) (image.Image, error) {
+	nalu = append([]uint8{0x00, 0x00, 0x00, 0x01}, []uint8(nalu)...)
+
+	// send frame to decoder
+	var avPacket C.AVPacket
+	avPacket.data = (*C.uint8_t)(C.CBytes(nalu))
+	defer C.free(unsafe.Pointer(avPacket.data))
+	avPacket.size = C.int(len(nalu))
+	res := C.avcodec_send_packet(d.codecCtx, &avPacket)
+	if res < 0 {
+		return nil, nil
+	}
+
+	// receive frame if available
+	res = C.avcodec_receive_frame(d.codecCtx, d.srcFrame)
+	if res < 0 {
+		return nil, nil
+	}
+
+	// if frame size has changed, allocate needed objects
+	if d.dstFrame == nil || d.dstFrame.width != d.srcFrame.width || d.dstFrame.height != d.srcFrame.height {
+		if d.dstFrame != nil {
+			C.av_frame_free(&d.dstFrame)
+		}
+
+		if d.swsCtx != nil {
+			C.sws_freeContext(d.swsCtx)
+		}
+
+		d.dstFrame = C.av_frame_alloc()
+		d.dstFrame.format = C.AV_PIX_FMT_RGBA
+		d.dstFrame.width = d.srcFrame.width
+		d.dstFrame.height = d.srcFrame.height
+		d.dstFrame.color_range = C.AVCOL_RANGE_JPEG
+		res = C.av_frame_get_buffer(d.dstFrame, 1)
+		if res < 0 {
+			return nil, fmt.Errorf("av_frame_get_buffer() err")
+		}
+
+		d.swsCtx = C.sws_getContext(d.srcFrame.width, d.srcFrame.height, C.AV_PIX_FMT_YUV420P,
+			d.dstFrame.width, d.dstFrame.height, (int32)(d.dstFrame.format), C.SWS_BILINEAR, nil, nil, nil)
+		if d.swsCtx == nil {
+			return nil, fmt.Errorf("sws_getContext() err")
+		}
+
+		dstFrameSize := C.av_image_get_buffer_size((int32)(d.dstFrame.format), d.dstFrame.width, d.dstFrame.height, 1)
+		d.dstFramePtr = (*[1 << 30]uint8)(unsafe.Pointer(d.dstFrame.data[0]))[:dstFrameSize:dstFrameSize]
+	}
+
+	// convert frame from YUV420 to RGB
+	res = C.sws_scale(d.swsCtx, frameData(d.srcFrame), frameLineSize(d.srcFrame),
+		0, d.srcFrame.height, frameData(d.dstFrame), frameLineSize(d.dstFrame))
+	if res < 0 {
+		return nil, fmt.Errorf("sws_scale() err")
+	}
+
+	// embed frame into an image.Image
+	return &image.RGBA{
+		Pix:    d.dstFramePtr,
+		Stride: 4 * (int)(d.dstFrame.width),
+		Rect: image.Rectangle{
+			Max: image.Point{(int)(d.dstFrame.width), (int)(d.dstFrame.height)},
+		},
+	}, nil
+}

machinery/src/rtsp/mp4muxer.go

@@ -0,0 +1,15 @@
+package rtsp
+
+// mp4Muxer allows to save a H264 stream into a Mp4 file.
+type mp4Muxer struct {
+	sps []byte
+	pps []byte
+}
+
+// newMp4Muxer allocates a mp4Muxer.
+func newMp4Muxer(sps []byte, pps []byte) (*mp4Muxer, error) {
+	return &mp4Muxer{
+		sps: sps,
+		pps: pps,
+	}, nil
+}

machinery/src/rtsp/mpegtsmuxer.go

@@ -0,0 +1,173 @@
+package rtsp
+
+import (
+	"bufio"
+	"context"
+	"log"
+	"os"
+	"time"
+
+	"github.com/asticode/go-astits"
+	"github.com/bluenviron/mediacommon/pkg/codecs/h264"
+)
+
+// mpegtsMuxer allows to save a H264 stream into a MPEG-TS file.
+type mpegtsMuxer struct {
+	sps []byte
+	pps []byte
+
+	f                *os.File
+	b                *bufio.Writer
+	mux              *astits.Muxer
+	dtsExtractor     *h264.DTSExtractor
+	firstIDRReceived bool
+	startDTS         time.Duration
+}
+
+// newMPEGTSMuxer allocates a mpegtsMuxer.
+func newMPEGTSMuxer(sps []byte, pps []byte) (*mpegtsMuxer, error) {
+	f, err := os.Create("mystream.ts")
+	if err != nil {
+		return nil, err
+	}
+	b := bufio.NewWriter(f)
+
+	mux := astits.NewMuxer(context.Background(), b)
+	mux.AddElementaryStream(astits.PMTElementaryStream{
+		ElementaryPID: 256,
+		StreamType:    astits.StreamTypeH264Video,
+	})
+	mux.SetPCRPID(256)
+
+	return &mpegtsMuxer{
+		sps: sps,
+		pps: pps,
+		f:   f,
+		b:   b,
+		mux: mux,
+	}, nil
+}
+
+// close closes all the mpegtsMuxer resources.
+func (e *mpegtsMuxer) close() {
+	e.b.Flush()
+	e.f.Close()
+}
+
+// encode encodes a H264 access unit into MPEG-TS.
+func (e *mpegtsMuxer) encode(au [][]byte, pts time.Duration) error {
+	// prepend an AUD. This is required by some players
+	filteredNALUs := [][]byte{
+		{byte(h264.NALUTypeAccessUnitDelimiter), 240},
+	}
+
+	nonIDRPresent := false
+	idrPresent := false
+
+	for _, nalu := range au {
+		typ := h264.NALUType(nalu[0] & 0x1F)
+		switch typ {
+		case h264.NALUTypeSPS:
+			e.sps = append([]byte(nil), nalu...)
+			continue
+
+		case h264.NALUTypePPS:
+			e.pps = append([]byte(nil), nalu...)
+			continue
+
+		case h264.NALUTypeAccessUnitDelimiter:
+			continue
+
+		case h264.NALUTypeIDR:
+			idrPresent = true
+
+		case h264.NALUTypeNonIDR:
+			nonIDRPresent = true
+		}
+
+		filteredNALUs = append(filteredNALUs, nalu)
+	}
+
+	au = filteredNALUs
+
+	if !nonIDRPresent && !idrPresent {
+		return nil
+	}
+
+	// add SPS and PPS before every group that contains an IDR
+	if idrPresent {
+		au = append([][]byte{e.sps, e.pps}, au...)
+	}
+
+	var dts time.Duration
+
+	if !e.firstIDRReceived {
+		// skip samples silently until we find one with a IDR
+		if !idrPresent {
+			return nil
+		}
+
+		e.firstIDRReceived = true
+		e.dtsExtractor = h264.NewDTSExtractor()
+
+		var err error
+		dts, err = e.dtsExtractor.Extract(au, pts)
+		if err != nil {
+			return err
+		}
+
+		e.startDTS = dts
+		dts = 0
+		pts -= e.startDTS
+
+	} else {
+		var err error
+		dts, err = e.dtsExtractor.Extract(au, pts)
+		if err != nil {
+			return err
+		}
+
+		dts -= e.startDTS
+		pts -= e.startDTS
+	}
+
+	oh := &astits.PESOptionalHeader{
+		MarkerBits: 2,
+	}
+
+	if dts == pts {
+		oh.PTSDTSIndicator = astits.PTSDTSIndicatorOnlyPTS
+		oh.PTS = &astits.ClockReference{Base: int64(pts.Seconds() * 90000)}
+	} else {
+		oh.PTSDTSIndicator = astits.PTSDTSIndicatorBothPresent
+		oh.DTS = &astits.ClockReference{Base: int64(dts.Seconds() * 90000)}
+		oh.PTS = &astits.ClockReference{Base: int64(pts.Seconds() * 90000)}
+	}
+
+	// encode into Annex-B
+	annexb, err := h264.AnnexBMarshal(au)
+	if err != nil {
+		return err
+	}
+
+	// write TS packet
+	_, err = e.mux.WriteData(&astits.MuxerData{
+		PID: 256,
+		AdaptationField: &astits.PacketAdaptationField{
+			RandomAccessIndicator: idrPresent,
+		},
+		PES: &astits.PESData{
+			Header: &astits.PESHeader{
+				OptionalHeader: oh,
+				StreamID:       224, // video
+			},
+			Data: annexb,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	log.Println("wrote TS packet")
+	return nil
+}