From b8afa6ca0cc809b53689e035cec19dd0eba67c2f Mon Sep 17 00:00:00 2001 From: nareix Date: Fri, 15 Jul 2016 00:26:57 +0800 Subject: [PATCH] ts: rewrite muxer --- format/ts/checksum.go | 102 ---------- format/ts/checksum_test.go | 12 -- format/ts/demuxer.go | 8 +- format/ts/muxer.go | 223 ++++++--------------- format/ts/reader.go | 385 ------------------------------------- format/ts/stream.go | 13 +- format/ts/ts.go | 92 --------- format/ts/tsio/tsio.go | 3 + format/ts/writer.go | 199 ------------------- 9 files changed, 68 insertions(+), 969 deletions(-) delete mode 100644 format/ts/checksum.go delete mode 100644 format/ts/checksum_test.go delete mode 100644 format/ts/reader.go delete mode 100644 format/ts/ts.go delete mode 100644 format/ts/writer.go diff --git a/format/ts/checksum.go b/format/ts/checksum.go deleted file mode 100644 index c5ac537..0000000 --- a/format/ts/checksum.go +++ /dev/null @@ -1,102 +0,0 @@ -package ts - -import ( - "encoding/hex" - "fmt" - "io" -) - -var ieeeCrc32Tbl = []uint32{ - 0x00000000, 0xB71DC104, 0x6E3B8209, 0xD926430D, 0xDC760413, 0x6B6BC517, - 0xB24D861A, 0x0550471E, 0xB8ED0826, 0x0FF0C922, 0xD6D68A2F, 0x61CB4B2B, - 0x649B0C35, 0xD386CD31, 0x0AA08E3C, 0xBDBD4F38, 0x70DB114C, 0xC7C6D048, - 0x1EE09345, 0xA9FD5241, 0xACAD155F, 0x1BB0D45B, 0xC2969756, 0x758B5652, - 0xC836196A, 0x7F2BD86E, 0xA60D9B63, 0x11105A67, 0x14401D79, 0xA35DDC7D, - 0x7A7B9F70, 0xCD665E74, 0xE0B62398, 0x57ABE29C, 0x8E8DA191, 0x39906095, - 0x3CC0278B, 0x8BDDE68F, 0x52FBA582, 0xE5E66486, 0x585B2BBE, 0xEF46EABA, - 0x3660A9B7, 0x817D68B3, 0x842D2FAD, 0x3330EEA9, 0xEA16ADA4, 0x5D0B6CA0, - 0x906D32D4, 0x2770F3D0, 0xFE56B0DD, 0x494B71D9, 0x4C1B36C7, 0xFB06F7C3, - 0x2220B4CE, 0x953D75CA, 0x28803AF2, 0x9F9DFBF6, 0x46BBB8FB, 0xF1A679FF, - 0xF4F63EE1, 0x43EBFFE5, 0x9ACDBCE8, 0x2DD07DEC, 0x77708634, 0xC06D4730, - 0x194B043D, 0xAE56C539, 0xAB068227, 0x1C1B4323, 0xC53D002E, 0x7220C12A, - 0xCF9D8E12, 0x78804F16, 0xA1A60C1B, 0x16BBCD1F, 0x13EB8A01, 0xA4F64B05, - 0x7DD00808, 0xCACDC90C, 0x07AB9778, 0xB0B6567C, 0x69901571, 0xDE8DD475, - 0xDBDD936B, 0x6CC0526F, 0xB5E61162, 0x02FBD066, 0xBF469F5E, 0x085B5E5A, - 0xD17D1D57, 0x6660DC53, 0x63309B4D, 0xD42D5A49, 0x0D0B1944, 0xBA16D840, - 0x97C6A5AC, 0x20DB64A8, 0xF9FD27A5, 0x4EE0E6A1, 0x4BB0A1BF, 0xFCAD60BB, - 0x258B23B6, 0x9296E2B2, 0x2F2BAD8A, 0x98366C8E, 0x41102F83, 0xF60DEE87, - 0xF35DA999, 0x4440689D, 0x9D662B90, 0x2A7BEA94, 0xE71DB4E0, 0x500075E4, - 0x892636E9, 0x3E3BF7ED, 0x3B6BB0F3, 0x8C7671F7, 0x555032FA, 0xE24DF3FE, - 0x5FF0BCC6, 0xE8ED7DC2, 0x31CB3ECF, 0x86D6FFCB, 0x8386B8D5, 0x349B79D1, - 0xEDBD3ADC, 0x5AA0FBD8, 0xEEE00C69, 0x59FDCD6D, 0x80DB8E60, 0x37C64F64, - 0x3296087A, 0x858BC97E, 0x5CAD8A73, 0xEBB04B77, 0x560D044F, 0xE110C54B, - 0x38368646, 0x8F2B4742, 0x8A7B005C, 0x3D66C158, 0xE4408255, 0x535D4351, - 0x9E3B1D25, 0x2926DC21, 0xF0009F2C, 0x471D5E28, 0x424D1936, 0xF550D832, - 0x2C769B3F, 0x9B6B5A3B, 0x26D61503, 0x91CBD407, 0x48ED970A, 0xFFF0560E, - 0xFAA01110, 0x4DBDD014, 0x949B9319, 0x2386521D, 0x0E562FF1, 0xB94BEEF5, - 0x606DADF8, 0xD7706CFC, 0xD2202BE2, 0x653DEAE6, 0xBC1BA9EB, 0x0B0668EF, - 0xB6BB27D7, 0x01A6E6D3, 0xD880A5DE, 0x6F9D64DA, 0x6ACD23C4, 0xDDD0E2C0, - 0x04F6A1CD, 0xB3EB60C9, 0x7E8D3EBD, 0xC990FFB9, 0x10B6BCB4, 0xA7AB7DB0, - 0xA2FB3AAE, 0x15E6FBAA, 0xCCC0B8A7, 0x7BDD79A3, 0xC660369B, 0x717DF79F, - 0xA85BB492, 0x1F467596, 0x1A163288, 0xAD0BF38C, 0x742DB081, 0xC3307185, - 0x99908A5D, 0x2E8D4B59, 0xF7AB0854, 0x40B6C950, 0x45E68E4E, 0xF2FB4F4A, - 0x2BDD0C47, 0x9CC0CD43, 0x217D827B, 0x9660437F, 0x4F460072, 0xF85BC176, - 0xFD0B8668, 0x4A16476C, 0x93300461, 0x242DC565, 0xE94B9B11, 0x5E565A15, - 0x87701918, 0x306DD81C, 0x353D9F02, 0x82205E06, 0x5B061D0B, 0xEC1BDC0F, - 0x51A69337, 0xE6BB5233, 0x3F9D113E, 0x8880D03A, 0x8DD09724, 0x3ACD5620, - 0xE3EB152D, 0x54F6D429, 0x7926A9C5, 0xCE3B68C1, 0x171D2BCC, 0xA000EAC8, - 0xA550ADD6, 0x124D6CD2, 0xCB6B2FDF, 0x7C76EEDB, 0xC1CBA1E3, 0x76D660E7, - 0xAFF023EA, 0x18EDE2EE, 0x1DBDA5F0, 0xAAA064F4, 0x738627F9, 0xC49BE6FD, - 0x09FDB889, 0xBEE0798D, 0x67C63A80, 0xD0DBFB84, 0xD58BBC9A, 0x62967D9E, - 0xBBB03E93, 0x0CADFF97, 0xB110B0AF, 0x060D71AB, 0xDF2B32A6, 0x6836F3A2, - 0x6D66B4BC, 0xDA7B75B8, 0x035D36B5, 0xB440F7B1, 0x00000001, -} - -func updateIeeeCrc32(crc uint32, data []byte) uint32 { - for _, b := range data { - crc = ieeeCrc32Tbl[b^byte(crc)] ^ (crc >> 8) - } - return crc -} - -type Crc32Reader struct { - R io.Reader - Crc32 uint32 -} - -var DebugCrc32 = false - -func (self *Crc32Reader) Read(b []byte) (n int, err error) { - if n, err = self.R.Read(b); err != nil { - return - } - if DebugCrc32 { - fmt.Println("crc32: update", hex.EncodeToString(b)) - } - self.Crc32 = updateIeeeCrc32(self.Crc32, b) - return -} - -func (self *Crc32Reader) ReadCrc32UIntAndCheck() (err error) { - if err = ReadDummy(self, 4); err != nil { - return - } - if self.Crc32 != 0 { - err = fmt.Errorf("crc32(%x) != 0", self.Crc32) - return - } - return -} - -type Crc32Writer struct { - W io.Writer - Crc32 uint32 -} - -func (self *Crc32Writer) Write(b []byte) (n int, err error) { - if n, err = self.W.Write(b); err != nil { - return - } - self.Crc32 = updateIeeeCrc32(self.Crc32, b) - return -} diff --git a/format/ts/checksum_test.go b/format/ts/checksum_test.go deleted file mode 100644 index fc4316e..0000000 --- a/format/ts/checksum_test.go +++ /dev/null @@ -1,12 +0,0 @@ -package ts - -import ( - "testing" -) - -func TestChecksum(t *testing.T) { - b := []byte("hello world") - b = append(b, []byte{0xbb, 0x08, 0xec, 0x87}...) - crc := updateIeeeCrc32(0xffffffff, b) - t.Logf("%x", crc) -} diff --git a/format/ts/demuxer.go b/format/ts/demuxer.go index 866fc18..354bdf1 100644 --- a/format/ts/demuxer.go +++ b/format/ts/demuxer.go @@ -113,9 +113,9 @@ func (self *Demuxer) initPMT(payload []byte) (err error) { stream.pid = info.ElementaryPID stream.streamType = info.StreamType switch info.StreamType { - case ElementaryStreamTypeH264: + case tsio.ElementaryStreamTypeH264: self.streams = append(self.streams, stream) - case ElementaryStreamTypeAdtsAAC: + case tsio.ElementaryStreamTypeAdtsAAC: self.streams = append(self.streams, stream) } } @@ -215,7 +215,7 @@ func (self *Stream) payloadEnd() (n int, err error) { self.data = nil switch self.streamType { - case ElementaryStreamTypeAdtsAAC: + case tsio.ElementaryStreamTypeAdtsAAC: var config aacparser.MPEG4AudioConfig delta := time.Duration(0) @@ -235,7 +235,7 @@ func (self *Stream) payloadEnd() (n int, err error) { payload = payload[framelen:] } - case ElementaryStreamTypeH264: + case tsio.ElementaryStreamTypeH264: nalus, _ := h264parser.SplitNALUs(payload) var sps, pps []byte for _, nalu := range nalus { diff --git a/format/ts/muxer.go b/format/ts/muxer.go index fcbb3af..1f2e35c 100644 --- a/format/ts/muxer.go +++ b/format/ts/muxer.go @@ -1,32 +1,47 @@ package ts import ( - "bytes" + "time" "bufio" "fmt" "github.com/nareix/joy4/av" + "github.com/nareix/joy4/format/ts/tsio" "github.com/nareix/pio" "github.com/nareix/joy4/codec/aacparser" "github.com/nareix/joy4/codec/h264parser" "io" ) +var CodecTypes = []av.CodecType{av.H264, av.AAC} + type Muxer struct { w *bufio.Writer streams []*Stream PaddingToMakeCounterCont bool + psidata []byte peshdr []byte tshdr []byte adtshdr []byte datav [][]byte nalus [][]byte - tswPAT *TSWriter - tswPMT *TSWriter + tswpat, tswpmt *tsio.TSWriter } -var CodecTypes = []av.CodecType{av.H264, av.AAC} +func NewMuxer(w io.Writer) *Muxer { + return &Muxer{ + w: bufio.NewWriterSize(w, pio.RecommendBufioSize), + 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), + } +} func (self *Muxer) newStream(codec av.CodecData) (err error) { ok := false @@ -46,7 +61,7 @@ func (self *Muxer) newStream(codec av.CodecData) (err error) { muxer: self, CodecData: codec, pid: pid, - tsw: NewTSWriter(uint16(pid)), + tsw: tsio.NewTSWriter(pid), } self.streams = append(self.streams, stream) return @@ -85,50 +100,50 @@ func (self *Muxer) WriteTrailer() (err error) { return } -func NewMuxer(w io.Writer) *Muxer { - return &Muxer{ - w: bufio.NewWriterSize(w, pio.RecommendBufioSize), - peshdr: make([]byte, MaxPESHeaderLength), - tshdr: make([]byte, MaxTSHeaderLength), - adtshdr: make([]byte, aacparser.ADTSHeaderLength), - nalus: make([][]byte, 16), - datav: make([][]byte, 16), - } +func (self *Muxer) ChangeWriter(w io.Writer) (err error) { + return } func (self *Muxer) WritePATPMT() (err error) { - bufPAT := &bytes.Buffer{} - bufPMT := &bytes.Buffer{} - - pat := PAT{ - Entries: []PATEntry{ - {ProgramNumber: 1, ProgramMapPID: 0x1000}, + pat := tsio.PAT{ + Entries: []tsio.PATEntry{ + {ProgramNumber: 1, ProgramMapPID: tsio.PMT_PID}, }, } - WritePAT(bufPAT, pat) + patlen := pat.Marshal(self.psidata[tsio.PSIHeaderLength:]) + self.datav[0] = self.psidata[:tsio.PSIHeaderLength+patlen] + if err = self.tswpat.WritePackets(self.w, self.datav[:1], 0, false, true); err != nil { + return + } - var elemStreams []ElementaryStreamInfo + var elemStreams []tsio.ElementaryStreamInfo for _, stream := range self.streams { switch stream.Type() { case av.AAC: - elemStreams = append(elemStreams, ElementaryStreamInfo{StreamType: ElementaryStreamTypeAdtsAAC, ElementaryPID: uint(stream.pid)}) + elemStreams = append(elemStreams, tsio.ElementaryStreamInfo{ + StreamType: tsio.ElementaryStreamTypeAdtsAAC, + ElementaryPID: stream.pid, + }) case av.H264: - elemStreams = append(elemStreams, ElementaryStreamInfo{StreamType: ElementaryStreamTypeH264, ElementaryPID: uint(stream.pid)}) + elemStreams = append(elemStreams, tsio.ElementaryStreamInfo{ + StreamType: tsio.ElementaryStreamTypeH264, + ElementaryPID: stream.pid, + }) } } - pmt := PMT{ + pmt := tsio.PMT{ PCRPID: 0x100, ElementaryStreamInfos: elemStreams, } - WritePMT(bufPMT, pmt) - - self.tswPMT = NewTSWriter(0x1000) - self.tswPAT = NewTSWriter(0) - if err = self.tswPAT.WritePackets(self.w, [][]byte{bufPAT.Bytes()}, 0, false, true); err != nil { + pmtlen := pmt.Len() + if pmtlen+tsio.PSIHeaderLength > len(self.psidata) { + err = fmt.Errorf("ts: pmt too large") return } - if err = self.tswPMT.WritePackets(self.w, [][]byte{bufPMT.Bytes()}, 0, false, true); err != nil { + pmt.Marshal(self.psidata[tsio.PSIHeaderLength:]) + self.datav[0] = self.psidata[:tsio.PSIHeaderLength+pmtlen] + if err = self.tswpmt.WritePackets(self.w, self.datav[:1], 0, false, true); err != nil { return } @@ -149,144 +164,21 @@ func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) { return } -const MaxPESHeaderLength = 19 -const MaxTSHeaderLength = 12 - -func FillPESHeader(h []byte, streamid uint8, datalength int, pts, dts uint64) (n int) { - h[0] = 0 - h[1] = 0 - h[2] = 1 - h[3] = streamid - - const PTS = 1 << 7 - const DTS = 1 << 6 - - var pts_dts_flags uint8 - if pts != 0 { - pts_dts_flags |= PTS - if dts != 0 { - pts_dts_flags |= DTS - } - } - - if pts_dts_flags&PTS != 0 { - n += 5 - } - if pts_dts_flags&DTS != 0 { - n += 5 - } - - var packet_length uint16 - // packet_length(16) if zero then variable length - // Specifies the number of bytes remaining in the packet after this field. Can be zero. - // If the PES packet length is set to zero, the PES packet can be of any length. - // A value of zero for the PES packet length can be used only when the PES packet payload is a **video** elementary stream. - if datalength >= 0 { - packet_length = uint16(datalength + n + 3) - } - pio.PutU16BE(h[4:6], packet_length) - - h[6] = 2<<6|1 // resverd(6,2)=2,original_or_copy(0,1)=1 - h[7] = pts_dts_flags - h[8] = uint8(n) - - // pts(40)? - // dts(40)? - if pts_dts_flags&PTS != 0 { - if pts_dts_flags&DTS != 0 { - pio.PutU40BE(h[9:14], PESTsToUInt(pts)|3<<36) - pio.PutU40BE(h[14:19], PESTsToUInt(dts)|1<<36) - } else { - pio.PutU40BE(h[9:14], PESTsToUInt(pts)|2<<36) - } - } - - n += 9 - return -} - -func NewTSWriter(pid uint16) *TSWriter { - w := &TSWriter{} - w.tshdr = make([]byte, 188) - w.tshdr[0] = 0x47 - pio.PutU16BE(w.tshdr[1:3], pid&0x1fff) - for i := 6; i < 188; i++ { - w.tshdr[i] = 0xff - } - return w -} - -func (self *TSWriter) WritePackets(w io.Writer, datav [][]byte, pcr uint64, 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 - 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], PCRToUInt(pcr)) - } - if sync { - self.tshdr[5] = 0x40|self.tshdr[5] // Random Access indicator - } - } - - padtail := 0 - end := writepos + 188 - hdrlen - if end > datavlen { - if paddata { - padtail = end - datavlen - } else { - hdrlen += end - datavlen - } - end = datavlen - } - n := pio.VecSliceNoNew(datav, writev, writepos, end) - - self.tshdr[4] = byte(hdrlen)-5 // length - if _, err = w.Write(self.tshdr[:hdrlen]); err != nil { - return - } - for i := 0; i < n; i++ { - if _, err = w.Write(writev[i]); err != nil { - return - } - } - if padtail > 0 { - if _, err = w.Write(self.tshdr[188-padtail:188]); err != nil { - return - } - } - - writepos = end - } - - return -} - func (self *Muxer) WritePacket(pkt av.Packet) (err error) { stream := self.streams[pkt.Idx] + pkt.Time += time.Second switch stream.Type() { case av.AAC: codec := stream.CodecData.(aacparser.CodecData) - n := FillPESHeader(self.peshdr, StreamIdAAC, len(self.adtshdr)+len(pkt.Data), timeToPesTs(pkt.Time), 0) + n := tsio.FillPESHeader(self.peshdr, tsio.StreamIdAAC, len(self.adtshdr)+len(pkt.Data), pkt.Time, 0) self.datav[0] = self.peshdr[:n] aacparser.FillADTSHeader(self.adtshdr, codec.Config, 1024, len(pkt.Data)) self.datav[1] = self.adtshdr self.datav[2] = pkt.Data - if err = stream.tsw.WritePackets(self.w, self.datav[:3], timeToPCR(pkt.Time), true, false); err != nil { + if err = stream.tsw.WritePackets(self.w, self.datav[:3], pkt.Time, true, false); err != nil { return } @@ -304,16 +196,19 @@ func (self *Muxer) WritePacket(pkt av.Packet) (err error) { } datav := self.datav[:1] - h264parser.WalkNALUsAnnexb(nalus, func(b []byte) { - datav = append(datav, b) - }) + for i, nalu := range nalus { + if i == 0 { + datav = append(datav, h264parser.AUDBytes) + } else { + datav = append(datav, h264parser.StartCodeBytes) + } + datav = append(datav, nalu) + } - pts := timeToPesTs(pkt.Time+pkt.CompositionTime) - dts := timeToPesTs(pkt.Time) - n := FillPESHeader(self.peshdr, StreamIdH264, -1, pts, dts) + 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, timeToPCR(pkt.Time), pkt.IsKeyFrame, false); err != nil { + if err = stream.tsw.WritePackets(self.w, datav, pkt.Time, pkt.IsKeyFrame, false); err != nil { return } } diff --git a/format/ts/reader.go b/format/ts/reader.go deleted file mode 100644 index 0074d02..0000000 --- a/format/ts/reader.go +++ /dev/null @@ -1,385 +0,0 @@ -package ts - -import ( - "fmt" - "io" - "io/ioutil" -) - -func ReadUInt(r io.Reader, n int) (res uint, err error) { - var b [4]byte - if _, err = r.Read(b[0:n]); err != nil { - return - } - for i := 0; i < n; i++ { - res <<= 8 - res |= uint(b[i]) - } - return -} - -func ReadDummy(r io.Reader, n int) (err error) { - _, err = io.CopyN(ioutil.Discard, r, int64(n)) - return -} - -func ReadUInt64(r io.Reader, n int) (res uint64, err error) { - var res32 uint - if n > 4 { - if res32, err = ReadUInt(r, n-4); err != nil { - return - } - res |= uint64(res32) << 32 - n = 4 - } - if res32, err = ReadUInt(r, n); err != nil { - return - } - res |= uint64(res32) - return -} - -func ReadPSI(r io.Reader) (self PSI, lr *io.LimitedReader, cr *Crc32Reader, err error) { - var flags, pointer, length uint - - // pointer field - if pointer, err = ReadUInt(r, 1); err != nil { - return - } - - if pointer != 0 { - if err = ReadDummy(r, int(pointer)); err != nil { - return - } - } - - cr = &Crc32Reader{R: r, Crc32: 0xffffffff} - - // table_id - if self.TableId, err = ReadUInt(cr, 1); err != nil { - return - } - - // reserved(4)=0xb - // section_length(10) - if flags, err = ReadUInt(cr, 2); err != nil { - return - } - length = flags & 0x3FF - - lr = &io.LimitedReader{R: cr, N: int64(length)} - - // Table ID extension(16) - if self.TableIdExtension, err = ReadUInt(lr, 2); err != nil { - return - } - - // resverd(2)=3 - // version(5) - // Current_next_indicator(1) - if flags, err = ReadUInt(lr, 1); err != nil { - return - } - - // section_number(8) - if self.SecNum, err = ReadUInt(lr, 1); err != nil { - return - } - - // last_section_number(8) - if self.LastSecNum, err = ReadUInt(lr, 1); err != nil { - return - } - - lr.N -= 4 - return -} - -func ReadPMT(r io.Reader) (self PMT, err error) { - readDescs := func(lr *io.LimitedReader) (res []Descriptor, err error) { - var desc Descriptor - for lr.N > 0 { - if desc.Tag, err = ReadUInt(lr, 1); err != nil { - return - } - var length uint - if length, err = ReadUInt(lr, 1); err != nil { - return - } - desc.Data = make([]byte, length) - if _, err = lr.Read(desc.Data); err != nil { - return - } - res = append(res, desc) - } - return - } - - var lr *io.LimitedReader - var cr *Crc32Reader - //var psi PSI - - if _, lr, cr, err = ReadPSI(r); err != nil { - return - } - - var flags, length uint - - // 111(3) - // PCRPID(13) - if flags, err = ReadUInt(lr, 2); err != nil { - return - } - self.PCRPID = flags & 0x1fff - - // Reserved(4)=0xf - // Reserved(2)=0x0 - // Program info length(10) - if flags, err = ReadUInt(lr, 2); err != nil { - return - } - length = flags & 0x3ff - - if length > 0 { - lr := &io.LimitedReader{R: lr, N: int64(length)} - if self.ProgramDescriptors, err = readDescs(lr); err != nil { - return - } - } - - for lr.N > 0 { - var info ElementaryStreamInfo - if info.StreamType, err = ReadUInt(lr, 1); err != nil { - return - } - - // Reserved(3) - // Elementary PID(13) - if flags, err = ReadUInt(lr, 2); err != nil { - return - } - info.ElementaryPID = flags & 0x1fff - - // Reserved(6) - // ES Info length(10) - if flags, err = ReadUInt(lr, 2); err != nil { - return - } - length = flags & 0x3ff - - if length > 0 { - lr := &io.LimitedReader{R: lr, N: int64(length)} - if info.Descriptors, err = readDescs(lr); err != nil { - return - } - } - self.ElementaryStreamInfos = append(self.ElementaryStreamInfos, info) - } - - if err = cr.ReadCrc32UIntAndCheck(); err != nil { - return - } - - return -} - -func ReadPAT(r io.Reader) (self PAT, err error) { - var lr *io.LimitedReader - var cr *Crc32Reader - //var psi PSI - - if _, lr, cr, err = ReadPSI(r); err != nil { - return - } - - for lr.N > 0 { - entry := PATEntry{} - if entry.ProgramNumber, err = ReadUInt(lr, 2); err != nil { - return - } - if entry.ProgramNumber == 0 { - if entry.NetworkPID, err = ReadUInt(lr, 2); err != nil { - return - } - entry.NetworkPID &= 0x1fff - } else { - if entry.ProgramMapPID, err = ReadUInt(lr, 2); err != nil { - return - } - entry.ProgramMapPID &= 0x1fff - } - self.Entries = append(self.Entries, entry) - } - - if err = cr.ReadCrc32UIntAndCheck(); err != nil { - return - } - - return -} - -func ReadPESHeader(r io.Reader) (res *PESHeader, err error) { - var flags, length uint - self := &PESHeader{} - - // http://dvd.sourceforge.net/dvdinfo/pes-hdr.html - - // start code 000001 - if flags, err = ReadUInt(r, 3); err != nil { - return - } - if flags != 0x000001 { - err = fmt.Errorf("invalid PES header") - return - } - - if self.StreamId, err = ReadUInt(r, 1); err != nil { - return - } - - if length, err = ReadUInt(r, 2); err != nil { - return - } - - if length == 0 { - length = 1 << 31 - } - lrAll := &io.LimitedReader{R: r, N: int64(length)} - lr := lrAll - - // 10(2) - // PES scrambling control(2) - // PES priority(1) - // data alignment indicator(1) - // copyright(1) - // original or copy(1) - if _, err = ReadUInt(lr, 1); err != nil { - return - } - - // PTS DTS flags(2) - // ESCR flag(1) - // ES rate flag(1) - // DSM trick mode flag(1) - // additional copy info flag(1) - // PES CRC flag(1) - // PES extension flag(1) - if flags, err = ReadUInt(lr, 1); err != nil { - return - } - - // PES header data length(8) - if length, err = ReadUInt(lr, 1); err != nil { - return - } - lr = &io.LimitedReader{R: lr, N: int64(length)} - - if flags&0x80 != 0 { - var v uint64 - if v, err = ReadUInt64(lr, 5); err != nil { - return - } - self.PTS = PESUIntToTs(v) - } - - if flags&0x40 != 0 && flags&0x80 != 0 { - var v uint64 - if v, err = ReadUInt64(lr, 5); err != nil { - return - } - self.DTS = PESUIntToTs(v) - } - - // ESCR flag - if flags&0x20 != 0 { - if _, err = ReadUInt64(lr, 6); err != nil { - return - } - } - - // ES rate flag - if flags&0x10 != 0 { - if _, err = ReadUInt64(lr, 3); err != nil { - return - } - } - - // additional copy info flag - if flags&0x04 != 0 { - if _, err = ReadUInt(lr, 1); err != nil { - return - } - } - - // PES CRC flag - if flags&0x02 != 0 { - if _, err = ReadUInt(lr, 2); err != nil { - return - } - } - - // PES extension flag - if flags&0x01 != 0 { - var flags uint - - // PES private data flag(1) - // pack header field flag(1) - // program packet sequence counter flag(1) - // P-STD buffer flag(1) - // 111(3) - // PES extension flag 2(1) - if flags, err = ReadUInt(lr, 1); err != nil { - return - } - - // PES private data flag(1) - if flags&0x80 != 0 { - // if set to 1 16 bytes of user defined data is appended to the header data field - if err = ReadDummy(lr, 16); err != nil { - return - } - } - - // pack header field flag(1) - if flags&0x40 != 0 { - // if set to 1 the 8-bit pack field length value is appended to the header data field - if err = ReadDummy(lr, 1); err != nil { - return - } - } - - // program packet sequence counter flag(1) - if flags&0x20 != 0 { - if err = ReadDummy(lr, 2); err != nil { - return - } - } - - // P-STD buffer flag(1) - if flags&0x10 != 0 { - if err = ReadDummy(lr, 2); err != nil { - return - } - } - - // PES extension flag 2(1) - if flags&0x01 != 0 { - if err = ReadDummy(lr, 2); err != nil { - return - } - } - } - - if lr.N > 0 { - if err = ReadDummy(lr, int(lr.N)); err != nil { - return - } - } - - if lrAll.N < 65536 { - self.DataLength = uint(lrAll.N) - } - - res = self - return -} diff --git a/format/ts/stream.go b/format/ts/stream.go index e649481..36b8831 100644 --- a/format/ts/stream.go +++ b/format/ts/stream.go @@ -2,15 +2,13 @@ package ts import ( "time" - "bytes" "github.com/nareix/joy4/av" + "github.com/nareix/joy4/format/ts/tsio" ) type Stream struct { av.CodecData - buf bytes.Buffer - demuxer *Demuxer muxer *Muxer @@ -18,7 +16,7 @@ type Stream struct { streamId uint8 streamType uint8 - tsw *TSWriter + tsw *tsio.TSWriter idx int iskeyframe bool @@ -27,10 +25,3 @@ type Stream struct { datalen int } -func timeToPesTs(tm time.Duration) uint64 { - return uint64(tm*PTS_HZ/time.Second) + PTS_HZ -} - -func timeToPCR(tm time.Duration) uint64 { - return uint64(tm*PCR_HZ/time.Second) + PCR_HZ -} diff --git a/format/ts/ts.go b/format/ts/ts.go deleted file mode 100644 index 651f14c..0000000 --- a/format/ts/ts.go +++ /dev/null @@ -1,92 +0,0 @@ -package ts - -import ( - "io" -) - -const ( - ElementaryStreamTypeH264 = 0x1B - ElementaryStreamTypeAdtsAAC = 0x0F -) - -type PATEntry struct { - ProgramNumber uint - NetworkPID uint - ProgramMapPID uint -} - -type PAT struct { - Entries []PATEntry -} - -type PMT struct { - PCRPID uint - ProgramDescriptors []Descriptor - ElementaryStreamInfos []ElementaryStreamInfo -} - -type Descriptor struct { - Tag uint - Data []byte -} - -type ElementaryStreamInfo struct { - StreamType uint - ElementaryPID uint - Descriptors []Descriptor -} - -type PSI struct { - TableIdExtension uint - TableId uint - SecNum uint - LastSecNum uint -} - -const ( - StreamIdH264 = 0xe0 - StreamIdAAC = 0xc0 -) - -type PESHeader struct { - StreamId uint // H264=0xe0 AAC=0xc0 - DataLength uint - PTS uint64 - DTS uint64 - ESCR uint64 -} - -func PESUIntToTs(v uint64) (ts uint64) { - // 0010 PTS 32..30 1 PTS 29..15 1 PTS 14..00 1 - return (((v >> 33) & 0x7) << 30) | (((v >> 17) & 0x7fff) << 15) | ((v >> 1) & 0x7fff) -} - -func PESTsToUInt(ts uint64) (v uint64) { - // 0010 PTS 32..30 1 PTS 29..15 1 PTS 14..00 1 - return ((ts>>30)&0x7)<<33 | ((ts>>15)&0x7fff)<<17 | (ts&0x7fff)<<1 | 0x100010001 -} - -const ( - PTS_HZ = 90000 - PCR_HZ = 27000000 -) - -func UIntToPCR(v uint64) uint64 { - // base(33)+resverd(6)+ext(9) - base := v >> 15 - ext := v & 0x1ff - return base*300 + ext -} - -func PCRToUInt(pcr uint64) uint64 { - base := pcr / 300 - ext := pcr % 300 - return base<<15 | 0x3f<<9 | ext -} - -type TSWriter struct { - w io.Writer - ContinuityCounter uint - tshdr []byte -} - diff --git a/format/ts/tsio/tsio.go b/format/ts/tsio/tsio.go index 2850cdb..fb2ba6a 100644 --- a/format/ts/tsio/tsio.go +++ b/format/ts/tsio/tsio.go @@ -15,6 +15,7 @@ const ( const ( PAT_PID = 0 + PMT_PID = 0x1000 ) const TableIdPMT = 2 @@ -327,6 +328,8 @@ func ParsePSI(h []byte) (tableid uint8, tableext uint16, hdrlen int, datalen int return } +const PSIHeaderLength = 9 + func FillPSI(h []byte, tableid uint8, tableext uint16, data []byte) (n int) { // pointer(8) h[n] = 0 diff --git a/format/ts/writer.go b/format/ts/writer.go deleted file mode 100644 index 9ac3f83..0000000 --- a/format/ts/writer.go +++ /dev/null @@ -1,199 +0,0 @@ -package ts - -import ( - "bytes" - "io" -) - -func WriteUInt64(w io.Writer, val uint64, n int) (err error) { - var b [8]byte - for i := n - 1; i >= 0; i-- { - b[i] = byte(val) - val >>= 8 - } - if _, err = w.Write(b[:n]); err != nil { - return - } - return -} - -func WriteUInt(w io.Writer, val uint, n int) (err error) { - return WriteUInt64(w, uint64(val), n) -} - -func WritePSI(w io.Writer, self PSI, data []byte) (err error) { - // pointer(8) - // table_id(8) - // reserved(4)=0xb,section_length(10) - // Table ID extension(16) - // section_syntax_indicator(1)=1,private_bit(1)=1,reserved(2)=3,unused(2)=0,section_length(10) - // section_number(8) - // last_section_number(8) - // data - // crc(32) - - // pointer(8) - if err = WriteUInt(w, 0, 1); err != nil { - return - } - - cw := &Crc32Writer{W: w, Crc32: 0xffffffff} - - // table_id(8) - if err = WriteUInt(cw, self.TableId, 1); err != nil { - return - } - - // section_syntax_indicator(1)=1,private_bit(1)=0,reserved(2)=3,unused(2)=0,section_length(10) - var flags, length uint - length = 2 + 3 + 4 + uint(len(data)) - flags = 0xa<<12 | length - if err = WriteUInt(cw, flags, 2); err != nil { - return - } - - // Table ID extension(16) - if err = WriteUInt(cw, self.TableIdExtension, 2); err != nil { - return - } - - // resverd(2)=3,version(5)=0,Current_next_indicator(1)=1 - flags = 0x3<<6 | 1 - if err = WriteUInt(cw, flags, 1); err != nil { - return - } - - // section_number(8) - if err = WriteUInt(cw, self.SecNum, 1); err != nil { - return - } - - // last_section_number(8) - if err = WriteUInt(cw, self.LastSecNum, 1); err != nil { - return - } - - // data - if _, err = cw.Write(data); err != nil { - return - } - - // crc(32) - if err = WriteUInt(w, bswap32(uint(cw.Crc32)), 4); err != nil { - return - } - - return -} - -func bswap32(v uint) uint { - return (v >> 24) | ((v>>16)&0xff)<<8 | ((v>>8)&0xff)<<16 | (v&0xff)<<24 -} - -func WritePAT(w io.Writer, self PAT) (err error) { - bw := &bytes.Buffer{} - - for _, entry := range self.Entries { - if err = WriteUInt(bw, entry.ProgramNumber, 2); err != nil { - return - } - if entry.ProgramNumber == 0 { - if err = WriteUInt(bw, entry.NetworkPID&0x1fff|7<<13, 2); err != nil { - return - } - } else { - if err = WriteUInt(bw, entry.ProgramMapPID&0x1fff|7<<13, 2); err != nil { - return - } - } - } - - psi := PSI{ - TableIdExtension: 1, - } - if err = WritePSI(w, psi, bw.Bytes()); err != nil { - return - } - - return -} - -func WritePMT(w io.Writer, self PMT) (err error) { - writeDescs := func(w io.Writer, descs []Descriptor) (err error) { - for _, desc := range descs { - if err = WriteUInt(w, desc.Tag, 1); err != nil { - return - } - if err = WriteUInt(w, uint(len(desc.Data)), 1); err != nil { - return - } - if _, err = w.Write(desc.Data); err != nil { - return - } - } - return - } - - writeBody := func(w io.Writer) (err error) { - if err = WriteUInt(w, self.PCRPID|7<<13, 2); err != nil { - return - } - - bw := &bytes.Buffer{} - if err = writeDescs(bw, self.ProgramDescriptors); err != nil { - return - } - - if err = WriteUInt(w, 0xf<<12|uint(bw.Len()), 2); err != nil { - return - } - if _, err = w.Write(bw.Bytes()); err != nil { - return - } - - for _, info := range self.ElementaryStreamInfos { - if err = WriteUInt(w, info.StreamType, 1); err != nil { - return - } - - // Reserved(3) - // Elementary PID(13) - if err = WriteUInt(w, info.ElementaryPID|7<<13, 2); err != nil { - return - } - - bw := &bytes.Buffer{} - if err = writeDescs(bw, info.Descriptors); err != nil { - return - } - - // Reserved(6) - // ES Info length length(10) - if err = WriteUInt(w, uint(bw.Len())|0x3c<<10, 2); err != nil { - return - } - - if _, err = w.Write(bw.Bytes()); err != nil { - return - } - } - - return - } - - bw := &bytes.Buffer{} - if err = writeBody(bw); err != nil { - return - } - - psi := PSI{ - TableId: 2, - TableIdExtension: 1, - } - if err = WritePSI(w, psi, bw.Bytes()); err != nil { - return - } - - return -} -