diff --git a/format/ts/muxer.go b/format/ts/muxer.go index 8e7b123d..834a789f 100644 --- a/format/ts/muxer.go +++ b/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,6 +176,11 @@ 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 } @@ -153,8 +188,54 @@ func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) { } 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 } diff --git a/format/ts/tsio/tsio.go b/format/ts/tsio/tsio.go index b31696b4..652a182c 100644 --- a/format/ts/tsio/tsio.go +++ b/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,6 +525,11 @@ 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)) @@ -521,17 +537,31 @@ func (self *TSWriter) WritePackets(w io.Writer, datav [][]byte, pcr time.Duratio 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