From cc69ce4b46da3cc61cdcfe1b15f05f489d5a1f4e Mon Sep 17 00:00:00 2001 From: kirpicheff <79604900020@yandex.ru> Date: Mon, 21 Jul 2025 00:50:53 +0300 Subject: [PATCH 1/3] Fix TS muxer: individual continuity counters, proper PCR timing, periodic PAT/PMT --- format/ts/muxer.go | 111 +++++++++++++++++++++++++++++++++++------ format/ts/tsio/tsio.go | 38 ++++++++++++-- 2 files changed, 130 insertions(+), 19 deletions(-) diff --git a/format/ts/muxer.go b/format/ts/muxer.go index 8e7b123d..d6a0be9c 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,43 @@ type Muxer struct { nalus [][]byte tswpat, tswpmt *tsio.TSWriter + + // Добавляем счетчики для периодической отправки PAT/PMT + patpmtCounter int + lastPATPMT time.Time + + // Отдельные счетчики для PAT и PMT + patCounter uint + pmtCounter uint + + // Счетчик для относительного времени + startTime time.Time + packetCount uint64 + + // Для отслеживания временных меток + 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(), } + + // Возвращаем отдельные счетчики для 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 +160,7 @@ func (self *Muxer) WritePATPMT() (err error) { return } + self.lastPATPMT = time.Now() return } @@ -153,8 +179,48 @@ func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) { } func (self *Muxer) WritePacket(pkt av.Packet) (err error) { + // Инициализируем startTime при первом пакете + if self.startTime.IsZero() { + self.startTime = time.Now() + } + + // ВОЗВРАЩАЕМ периодическую отправку 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 + + // Логирование для диагностики временных меток + if stream.Type() == av.H264 { + // Проверяем промежуток между видео кадрами + 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 + + log.Printf("VIDEO packet: time=%v, isKeyFrame=%v, compositionTime=%v, packetCount=%d", + pkt.Time, pkt.IsKeyFrame, pkt.CompositionTime, self.packetCount) + } else if stream.Type() == av.AAC { + // Проверяем промежуток между аудио пакетами + 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 + + log.Printf("AUDIO packet: time=%v, packetCount=%d", pkt.Time, self.packetCount) + } + + // Используем оригинальные временные метки без сложной конвертации + originalTime := pkt.Time switch stream.Type() { case av.AAC: @@ -166,7 +232,12 @@ 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 { + // ВОЗВРАЩАЕМ PCR для аудио пакетов с простым временем + pcrTime := time.Duration(0) + if self.packetCount%100 == 0 { // Каждые 100 пакетов + pcrTime = time.Duration(self.packetCount) * time.Millisecond // Простое время + } + if err = stream.tsw.WritePackets(self.w, self.datav[:3], pcrTime, true, false); err != nil { return } @@ -196,10 +267,20 @@ 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 { + // ВОЗВРАЩАЕМ PCR для видео с простым временем + pcrTime := time.Duration(0) + if pkt.IsKeyFrame && self.packetCount%100 == 0 { + pcrTime = time.Duration(self.packetCount) * time.Millisecond // Простое время + } + if err = stream.tsw.WritePackets(self.w, datav, pcrTime, pkt.IsKeyFrame, false); err != nil { return } } + // Увеличиваем счетчик пакетов + self.packetCount++ + + // Восстанавливаем оригинальное время для корректности + 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 From 0e15b916d6da2ef006666863a269d23609c763c1 Mon Sep 17 00:00:00 2001 From: kirpicheff <79604900020@yandex.ru> Date: Mon, 21 Jul 2025 20:53:50 +0300 Subject: [PATCH 2/3] remove debug logs --- format/ts/muxer.go | 41 ++++++++++++++++++----------------------- 1 file changed, 18 insertions(+), 23 deletions(-) diff --git a/format/ts/muxer.go b/format/ts/muxer.go index d6a0be9c..cc38754d 100644 --- a/format/ts/muxer.go +++ b/format/ts/muxer.go @@ -28,19 +28,19 @@ type Muxer struct { tswpat, tswpmt *tsio.TSWriter - // Добавляем счетчики для периодической отправки PAT/PMT + // Add counters for periodic PAT/PMT sending patpmtCounter int lastPATPMT time.Time - // Отдельные счетчики для PAT и PMT + // Separate counters for PAT and PMT patCounter uint pmtCounter uint - // Счетчик для относительного времени + // Counter for relative time startTime time.Time packetCount uint64 - // Для отслеживания временных меток + // For tracking timestamps lastVideoTime time.Duration lastAudioTime time.Duration } @@ -59,7 +59,7 @@ func NewMuxer(w io.Writer) *Muxer { lastPATPMT: time.Now(), } - // Возвращаем отдельные счетчики для PAT/PMT + // Return separate counters for PAT/PMT muxer.tswpat.SetGlobalCounter(&muxer.patCounter) muxer.tswpmt.SetGlobalCounter(&muxer.pmtCounter) @@ -179,12 +179,12 @@ func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) { } func (self *Muxer) WritePacket(pkt av.Packet) (err error) { - // Инициализируем startTime при первом пакете + // Initialize startTime on the first packet if self.startTime.IsZero() { self.startTime = time.Now() } - // ВОЗВРАЩАЕМ периодическую отправку PAT/PMT + // Periodically send PAT/PMT if time.Since(self.lastPATPMT) > 1*time.Second { if err = self.WritePATPMT(); err != nil { return @@ -193,9 +193,9 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { stream := self.streams[pkt.Idx] - // Логирование для диагностики временных меток + // 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 { @@ -203,11 +203,8 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { } } self.lastVideoTime = pkt.Time - - log.Printf("VIDEO packet: time=%v, isKeyFrame=%v, compositionTime=%v, packetCount=%d", - pkt.Time, pkt.IsKeyFrame, pkt.CompositionTime, self.packetCount) } else if stream.Type() == av.AAC { - // Проверяем промежуток между аудио пакетами + // Check interval between audio packets if self.lastAudioTime > 0 { timeDiff := pkt.Time - self.lastAudioTime if timeDiff > time.Second { @@ -215,11 +212,9 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { } } self.lastAudioTime = pkt.Time - - log.Printf("AUDIO packet: time=%v, packetCount=%d", pkt.Time, self.packetCount) } - // Используем оригинальные временные метки без сложной конвертации + // Use original timestamps without complex conversion originalTime := pkt.Time switch stream.Type() { @@ -232,10 +227,10 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { self.datav[1] = self.adtshdr self.datav[2] = pkt.Data - // ВОЗВРАЩАЕМ PCR для аудио пакетов с простым временем + // Return PCR for audio packets with simple time pcrTime := time.Duration(0) - if self.packetCount%100 == 0 { // Каждые 100 пакетов - pcrTime = time.Duration(self.packetCount) * time.Millisecond // Простое время + if self.packetCount%100 == 0 { // Every 100 packets + pcrTime = time.Duration(self.packetCount) * time.Millisecond // Simple time } if err = stream.tsw.WritePackets(self.w, self.datav[:3], pcrTime, true, false); err != nil { return @@ -267,20 +262,20 @@ 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] - // ВОЗВРАЩАЕМ PCR для видео с простым временем + // Return PCR for video with simple time pcrTime := time.Duration(0) if pkt.IsKeyFrame && self.packetCount%100 == 0 { - pcrTime = time.Duration(self.packetCount) * time.Millisecond // Простое время + pcrTime = time.Duration(self.packetCount) * time.Millisecond // Simple time } 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 } From 9d75cebb38b3996e02640fff1177e60acc056eb6 Mon Sep 17 00:00:00 2001 From: kirpicheff <79604900020@yandex.ru> Date: Sun, 27 Jul 2025 11:05:10 +0300 Subject: [PATCH 3/3] Fix PCR generation in TS muxer to use actual stream timestamps - Replace packet counter-based PCR with timestamp-based PCR generation - Set PCR PID to first stream (usually video) for proper synchronization - Send PCR every 80ms based on actual video timestamps - Remove incorrect PCR generation from audio packets - This fixes 'picture is too late' warnings and improves stream stability --- format/ts/muxer.go | 34 +++++++++++++++++++++++----------- 1 file changed, 23 insertions(+), 11 deletions(-) diff --git a/format/ts/muxer.go b/format/ts/muxer.go index cc38754d..834a789f 100644 --- a/format/ts/muxer.go +++ b/format/ts/muxer.go @@ -40,6 +40,10 @@ type Muxer struct { startTime time.Time packetCount uint64 + // For tracking PCR generation + pcrPID uint16 + lastPCRTime time.Duration + // For tracking timestamps lastVideoTime time.Duration lastAudioTime time.Duration @@ -172,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 } @@ -193,6 +202,17 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { stream := self.streams[pkt.Idx] + // 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 @@ -227,12 +247,8 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { self.datav[1] = self.adtshdr self.datav[2] = pkt.Data - // Return PCR for audio packets with simple time - pcrTime := time.Duration(0) - if self.packetCount%100 == 0 { // Every 100 packets - pcrTime = time.Duration(self.packetCount) * time.Millisecond // Simple time - } - if err = stream.tsw.WritePackets(self.w, self.datav[:3], pcrTime, 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 } @@ -262,11 +278,7 @@ 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] - // Return PCR for video with simple time - pcrTime := time.Duration(0) - if pkt.IsKeyFrame && self.packetCount%100 == 0 { - pcrTime = time.Duration(self.packetCount) * time.Millisecond // Simple time - } + // 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 }