Fix TS muxer: individual continuity counters, proper PCR timing, periodic PAT/PMT

format/ts/muxer.go

@@ -2,12 +2,14 @@ package ts
 
 import (
 	"fmt"
+	"io"
+	"log"
+	"time"
+
 	"github.com/datarhei/joy4/av"
 	"github.com/datarhei/joy4/codec/aacparser"
 	"github.com/datarhei/joy4/codec/h264parser"
 	"github.com/datarhei/joy4/format/ts/tsio"
-	"io"
-	"time"
 )
 
 var CodecTypes = []av.CodecType{av.H264, av.AAC}
@@ -25,20 +27,47 @@ type Muxer struct {
 	nalus   [][]byte
 
 	tswpat, tswpmt *tsio.TSWriter
+
+	// Add counters for periodic PAT/PMT sending
+	patpmtCounter int
+	lastPATPMT    time.Time
+
+	// Separate counters for PAT and PMT
+	patCounter uint
+	pmtCounter uint
+
+	// Counter for relative time
+	startTime   time.Time
+	packetCount uint64
+
+	// For tracking PCR generation
+	pcrPID      uint16
+	lastPCRTime time.Duration
+
+	// For tracking timestamps
+	lastVideoTime time.Duration
+	lastAudioTime time.Duration
 }
 
 func NewMuxer(w io.Writer) *Muxer {
-	return &Muxer{
-		w:       w,
-		psidata: make([]byte, 188),
-		peshdr:  make([]byte, tsio.MaxPESHeaderLength),
-		tshdr:   make([]byte, tsio.MaxTSHeaderLength),
-		adtshdr: make([]byte, aacparser.ADTSHeaderLength),
-		nalus:   make([][]byte, 16),
-		datav:   make([][]byte, 16),
-		tswpmt:  tsio.NewTSWriter(tsio.PMT_PID),
-		tswpat:  tsio.NewTSWriter(tsio.PAT_PID),
+	muxer := &Muxer{
+		w:          w,
+		psidata:    make([]byte, 188),
+		peshdr:     make([]byte, tsio.MaxPESHeaderLength),
+		tshdr:      make([]byte, tsio.MaxTSHeaderLength),
+		adtshdr:    make([]byte, aacparser.ADTSHeaderLength),
+		nalus:      make([][]byte, 16),
+		datav:      make([][]byte, 16),
+		tswpmt:     tsio.NewTSWriter(tsio.PMT_PID),
+		tswpat:     tsio.NewTSWriter(tsio.PAT_PID),
+		lastPATPMT: time.Now(),
 	}
+
+	// Return separate counters for PAT/PMT
+	muxer.tswpat.SetGlobalCounter(&muxer.patCounter)
+	muxer.tswpmt.SetGlobalCounter(&muxer.pmtCounter)
+
+	return muxer
 }
 
 func (self *Muxer) newStream(codec av.CodecData) (err error) {
@@ -135,6 +164,7 @@ func (self *Muxer) WritePATPMT() (err error) {
 		return
 	}
 
+	self.lastPATPMT = time.Now()
 	return
 }
 
@@ -146,15 +176,66 @@ func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) {
 		}
 	}
 
+	// Set the PCR PID to the first stream's PID (usually video)
+	if len(self.streams) > 0 {
+		self.pcrPID = self.streams[0].pid
+	}
+
 	if err = self.WritePATPMT(); err != nil {
 		return
 	}
 	return
 }
 
 func (self *Muxer) WritePacket(pkt av.Packet) (err error) {
+	// Initialize startTime on the first packet
+	if self.startTime.IsZero() {
+		self.startTime = time.Now()
+	}
+
+	// Periodically send PAT/PMT
+	if time.Since(self.lastPATPMT) > 1*time.Second {
+		if err = self.WritePATPMT(); err != nil {
+			return
+		}
+	}
+
 	stream := self.streams[pkt.Idx]
-	pkt.Time += time.Second
+
+	// Correct PCR generation logic.
+	// The PCR should be derived from the actual stream timestamps and sent periodically.
+	var pcrTime time.Duration
+	if stream.pid == self.pcrPID {
+		// MPEG-TS spec suggests PCR interval should be <= 100ms. We use 80ms.
+		if self.lastPCRTime == 0 || (pkt.Time-self.lastPCRTime) > 80*time.Millisecond {
+			pcrTime = pkt.Time
+			self.lastPCRTime = pkt.Time
+		}
+	}
+
+	// Diagnostics for timestamp gaps
+	if stream.Type() == av.H264 {
+		// Check interval between video frames
+		if self.lastVideoTime > 0 {
+			timeDiff := pkt.Time - self.lastVideoTime
+			if timeDiff > time.Second {
+				log.Printf("WARNING: Large video time gap: %v between frames", timeDiff)
+			}
+		}
+		self.lastVideoTime = pkt.Time
+	} else if stream.Type() == av.AAC {
+		// Check interval between audio packets
+		if self.lastAudioTime > 0 {
+			timeDiff := pkt.Time - self.lastAudioTime
+			if timeDiff > time.Second {
+				log.Printf("WARNING: Large audio time gap: %v between packets", timeDiff)
+			}
+		}
+		self.lastAudioTime = pkt.Time
+	}
+
+	// Use original timestamps without complex conversion
+	originalTime := pkt.Time
 
 	switch stream.Type() {
 	case av.AAC:
@@ -166,7 +247,8 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) {
 		self.datav[1] = self.adtshdr
 		self.datav[2] = pkt.Data
 
-		if err = stream.tsw.WritePackets(self.w, self.datav[:3], pkt.Time, true, false); err != nil {
+		// Do not send PCR with audio packets. It's handled by the video stream.
+		if err = stream.tsw.WritePackets(self.w, self.datav[:3], 0, true, false); err != nil {
 			return
 		}
 
@@ -196,10 +278,16 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) {
 		n := tsio.FillPESHeader(self.peshdr, tsio.StreamIdH264, -1, pkt.Time+pkt.CompositionTime, pkt.Time)
 		datav[0] = self.peshdr[:n]
 
-		if err = stream.tsw.WritePackets(self.w, datav, pkt.Time, pkt.IsKeyFrame, false); err != nil {
+		// Pass the correctly calculated pcrTime. It will be non-zero only when needed.
+		if err = stream.tsw.WritePackets(self.w, datav, pcrTime, pkt.IsKeyFrame, false); err != nil {
 			return
 		}
 	}
 
+	// Increment packet count
+	self.packetCount++
+
+	// Restore original time for correctness
+	pkt.Time = originalTime
 	return
 }

format/ts/tsio/tsio.go

@@ -2,9 +2,10 @@ package tsio
 
 import (
 	"fmt"
-	"github.com/datarhei/joy4/utils/bits/pio"
 	"io"
 	"time"
+
+	"github.com/datarhei/joy4/utils/bits/pio"
 )
 
 const (
@@ -374,6 +375,10 @@ func FillPSI(h []byte, tableid uint8, tableext uint16, datalen int) (n int) {
 
 func TimeToPCR(tm time.Duration) (pcr uint64) {
 	// base(33)+resverd(6)+ext(9)
+	// Добавляем wraparound для предотвращения переполнения
+	if tm > 26*time.Hour {
+		tm = tm % (26*time.Hour + 30*time.Minute)
+	}
 	ts := uint64(tm * PCR_HZ / time.Second)
 	base := ts / 300
 	ext := ts % 300
@@ -390,6 +395,10 @@ func PCRToTime(pcr uint64) (tm time.Duration) {
 }
 
 func TimeToTs(tm time.Duration) (v uint64) {
+	// Добавляем wraparound для PTS (каждые 26.5 часов)
+	if tm > 26*time.Hour {
+		tm = tm % (26*time.Hour + 30*time.Minute)
+	}
 	ts := uint64(tm * PTS_HZ / time.Second)
 	// 0010	PTS 32..30 1	PTS 29..15 1 PTS 14..00 1
 	v = ((ts>>30)&0x7)<<33 | ((ts>>15)&0x7fff)<<17 | (ts&0x7fff)<<1 | 0x100010001
@@ -501,6 +510,8 @@ type TSWriter struct {
 	w                 io.Writer
 	ContinuityCounter uint
 	tshdr             []byte
+	// Глобальный счетчик continuity для синхронизации
+	globalCounter *uint
 }
 
 func NewTSWriter(pid uint16) *TSWriter {
@@ -514,24 +525,43 @@ func NewTSWriter(pid uint16) *TSWriter {
 	return w
 }
 
+// Метод для установки глобального счетчика
+func (self *TSWriter) SetGlobalCounter(counter *uint) {
+	self.globalCounter = counter
+}
+
 func (self *TSWriter) WritePackets(w io.Writer, datav [][]byte, pcr time.Duration, sync bool, paddata bool) (err error) {
 	datavlen := pio.VecLen(datav)
 	writev := make([][]byte, len(datav))
 	writepos := 0
 
 	for writepos < datavlen {
 		self.tshdr[1] = self.tshdr[1] & 0x1f
-		self.tshdr[3] = byte(self.ContinuityCounter)&0xf | 0x30
+
+		// Используем синхронизированный счетчик для PAT/PMT
+		if self.globalCounter != nil {
+			self.tshdr[3] = byte(*self.globalCounter)&0xf | 0x30
+			*self.globalCounter++
+		} else {
+			self.tshdr[3] = byte(self.ContinuityCounter)&0xf | 0x30
+			self.ContinuityCounter++
+		}
+
 		self.tshdr[5] = 0 // flags
 		hdrlen := 6
-		self.ContinuityCounter++
 
 		if writepos == 0 {
 			self.tshdr[1] = 0x40 | self.tshdr[1] // Payload Unit Start Indicator
 			if pcr != 0 {
 				hdrlen += 6
 				self.tshdr[5] = 0x10 | self.tshdr[5] // PCR flag (Discontinuity indicator 0x80)
-				pio.PutU48BE(self.tshdr[6:12], TimeToPCR(pcr))
+				// Исправляем обработку PCR - добавляем wraparound для длинных стримов
+				pcrValue := pcr
+				if pcr > 26*time.Hour {
+					// Wraparound для PCR каждые 26.5 часов
+					pcrValue = pcr % (26*time.Hour + 30*time.Minute)
+				}
+				pio.PutU48BE(self.tshdr[6:12], TimeToPCR(pcrValue))
 			}
 			if sync {
 				self.tshdr[5] = 0x40 | self.tshdr[5] // Random Access indicator