diff --git a/ts/README.md b/ts/README.md new file mode 100644 index 0000000..139597f --- /dev/null +++ b/ts/README.md @@ -0,0 +1,2 @@ + + diff --git a/ts/checksum.go b/ts/checksum.go new file mode 100644 index 0000000..c5ac537 --- /dev/null +++ b/ts/checksum.go @@ -0,0 +1,102 @@ +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/ts/checksum_test.go b/ts/checksum_test.go new file mode 100644 index 0000000..fc4316e --- /dev/null +++ b/ts/checksum_test.go @@ -0,0 +1,12 @@ +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/ts/demuxer.go b/ts/demuxer.go new file mode 100644 index 0000000..d0f6a78 --- /dev/null +++ b/ts/demuxer.go @@ -0,0 +1,261 @@ +package ts + +import ( + "bytes" + "encoding/hex" + "fmt" + "time" + "github.com/nareix/av" + "github.com/nareix/codec/aacparser" + "github.com/nareix/codec/h264parser" + "io" +) + +type Demuxer struct { + R io.Reader + + pktidx int + pkts []av.Packet + + pat PAT + pmt *PMT + streams []*Stream + + probed bool +} + +func (self *Demuxer) Streams() (streams []av.CodecData, err error) { + if err = self.probe(); err != nil { + return + } + for _, stream := range self.streams { + streams = append(streams, stream.CodecData) + } + return +} + +func (self *Demuxer) probe() (err error) { + if self.probed { + return + } + for { + if self.pmt != nil { + n := 0 + for _, stream := range self.streams { + if stream.CodecData != nil { + n++ + } + } + if n == len(self.streams) { + break + } + } + if err = self.poll(); err != nil { + return + } + } + self.probed = true + return +} + +func (self *Demuxer) ReadPacket() (pkt av.Packet, err error) { + if err = self.probe(); err != nil { + return + } + + for self.pktidx == len(self.pkts) { + if err = self.poll(); err != nil { + return + } + } + if self.pktidx < len(self.pkts) { + pkt = self.pkts[self.pktidx] + self.pktidx++ + } + + return +} + +func (self *Demuxer) poll() (err error) { + self.pktidx = 0 + self.pkts = self.pkts[0:0] + for { + if err = self.readTSPacket(); err != nil { + return + } + if len(self.pkts) > 0 { + break + } + } + return +} + +func (self *Demuxer) readTSPacket() (err error) { + var header TSHeader + var n int + var data [188]byte + + if header, n, err = ReadTSPacket(self.R, data[:]); err != nil { + return + } + payload := data[:n] + + if header.PID == 0 { + if self.pat, err = ReadPAT(bytes.NewReader(payload)); err != nil { + return + } + } else { + if self.pmt == nil { + self.streams = []*Stream{} + + for _, entry := range self.pat.Entries { + if entry.ProgramMapPID == header.PID { + self.pmt = new(PMT) + if *self.pmt, err = ReadPMT(bytes.NewReader(payload)); err != nil { + return + } + for i, info := range self.pmt.ElementaryStreamInfos { + stream := &Stream{} + stream.idx = i + stream.demuxer = self + stream.pid = info.ElementaryPID + stream.streamType = info.StreamType + switch info.StreamType { + case ElementaryStreamTypeH264: + self.streams = append(self.streams, stream) + case ElementaryStreamTypeAdtsAAC: + self.streams = append(self.streams, stream) + } + } + } + } + + } else { + for _, stream := range self.streams { + if header.PID == stream.pid { + if err = stream.handleTSPacket(header, payload); err != nil { + return + } + } + } + + } + } + + return +} + +func (self *Stream) addPacket(payload []byte, timedelta time.Duration) { + dts := self.peshdr.DTS + pts := self.peshdr.PTS + if dts == 0 { + dts = pts + } + + demuxer := self.demuxer + pkt := av.Packet{ + Idx: int8(self.idx), + IsKeyFrame: self.tshdr.RandomAccessIndicator, + Time: time.Duration(dts)*time.Second / time.Duration(PTS_HZ) + timedelta, + Data: payload, + } + if pts != dts { + pkt.CompositionTime = time.Duration(pts-dts)*time.Second / time.Duration(PTS_HZ) + } + demuxer.pkts = append(demuxer.pkts, pkt) +} + +func (self *Stream) payloadEnd() (err error) { + payload := self.buf.Bytes() + + switch self.streamType { + case ElementaryStreamTypeAdtsAAC: + var config aacparser.MPEG4AudioConfig + var packets [][]byte + var totsamples int + if config, packets, totsamples, err = aacparser.SplitADTSFrames(payload); err != nil { + err = fmt.Errorf("ts: demuxer: SplitADTSFrames failed: %s", err) + return + } + config = config.Complete() + + if self.CodecData == nil { + bw := &bytes.Buffer{} + if err = aacparser.WriteMPEG4AudioConfig(bw, config); err != nil { + err = fmt.Errorf("WriteMPEG4AudioConfig failed: %s", err) + return + } + if self.CodecData, err = aacparser.NewCodecDataFromMPEG4AudioConfigBytes(bw.Bytes()); err != nil { + return + } + } + + frametime := time.Duration(totsamples / len(packets)) * time.Second / time.Duration(config.SampleRate) + timedelta := time.Duration(0) + for _, packet := range packets { + self.addPacket(packet, timedelta) + timedelta += frametime + } + + case ElementaryStreamTypeH264: + nalus, _ := h264parser.SplitNALUs(payload) + var sps, pps []byte + for _, nalu := range nalus { + if len(nalu) > 0 { + naltype := nalu[0] & 0x1f + switch naltype { + case 7: + sps = nalu + case 8: + pps = nalu + case 6: + case 9: + default: + if false { + fmt.Println("h264", len(nalus), "\n", hex.Dump(nalu)) + } + self.addPacket(nalu, time.Duration(0)) + } + } + } + + if self.CodecData == nil && len(sps) > 0 && len(pps) > 0 { + if self.CodecData, err = h264parser.NewCodecDataFromSPSAndPPS(sps, pps); err != nil { + return + } + } + } + + return +} + +func (self *Stream) handleTSPacket(header TSHeader, tspacket []byte) (err error) { + r := bytes.NewReader(tspacket) + lr := &io.LimitedReader{R: r, N: int64(len(tspacket))} + + if header.PayloadUnitStart && self.peshdr != nil && self.peshdr.DataLength == 0 { + if err = self.payloadEnd(); err != nil { + return + } + } + + if header.PayloadUnitStart { + self.buf = bytes.Buffer{} + if self.peshdr, err = ReadPESHeader(lr); err != nil { + return + } + self.tshdr = header + } + + if _, err = io.CopyN(&self.buf, lr, lr.N); err != nil { + return + } + + if self.buf.Len() == int(self.peshdr.DataLength) { + if err = self.payloadEnd(); err != nil { + return + } + } + + return +} diff --git a/ts/doc.txt b/ts/doc.txt new file mode 100644 index 0000000..0d698ce --- /dev/null +++ b/ts/doc.txt @@ -0,0 +1,52 @@ + +The first h264 packet + +00000000 00 00 00 01 09 f0 00 00 00 01 06 00 07 86 96 7f |................| + ^ ^ +00000010 80 00 00 40 80 00 00 01 06 05 11 03 87 f4 4e cd |...@..........N.| + ^ +00000020 0a 4b dc a1 94 3a c3 d4 9b 17 1f 00 80 00 00 00 |.K...:..........| +00000030 01 27 4d 40 0a a9 19 bf 2c b8 0b 50 10 10 13 0a |.'M@....,..P....| + ^ +00000040 d7 bd f0 10 00 00 00 01 28 fe 09 88 00 00 01 25 |........(......%| + ^ ^ +00000050 b8 20 20 bf ed 7e de 3b 56 7f 03 c5 d1 06 9a c3 |. ..~.;V.......| +00000060 74 e5 9d 18 e6 3f 9d 63 19 cc 46 fd 36 13 c5 b8 |t....?.c..F.6...| +00000070 7f 3e d9 39 1b 21 74 61 6d 9d b7 5b 71 e6 a2 de |.>.9.!tam..[q...| +00000080 68 0b bf 7e b8 d4 9d 2f 49 19 3b 11 6e 40 a3 0a |h..~.../I.;.n@..| +00000090 3f 02 c5 a3 47 1d ed 2a 3a a1 33 0e 15 2f 36 d8 |?...G..*:.3../6.| +000000a0 c7 2a eb c6 bf 0d 95 e4 42 3e 7b 5f 44 3f b9 a5 |.*......B>{_D?..| +000000b0 d7 a2 c6 f4 41 e8 82 b5 cc cd c9 a4 ec 29 6f 58 |....A........)oX| +000000c0 6c 78 38 7c b5 5b 27 97 9d 56 5f e0 00 00 01 25 |lx8|.['..V_....%| + ^ +000000d0 2b 82 02 0b af 43 64 9c f1 eb 03 7e 39 a0 b6 f8 |+....Cd....~9...| +000000e0 eb c4 88 d2 ef 79 1f 7c d6 4e 1a d3 5d 27 39 ca |.....y.|.N..]'9.| +000000f0 c6 0b a9 63 78 e2 19 41 71 d4 47 26 bc 58 e4 45 |...cx..Aq.G&.X.E| +00000100 db 77 7b 60 7f d9 e4 11 03 3d 05 0d 72 5e 0d a0 |.w{`.....=..r^..| +00000110 56 4f 1a a9 9e 4f 80 |VO...O.| + +[h264 @ 0x7fdf01802200] NAL 9(AUD) at 4/279 length 1 +[h264 @ 0x7fdf01802200] NAL 6(SEI) at 10/279 length 10 +[h264 @ 0x7fdf01802200] NAL 6(SEI) at 24/279 length 20 +[h264 @ 0x7fdf01802200] NAL 7(SPS) at 49/279 length 18 +[h264 @ 0x7fdf01802200] NAL 8(PPS) at 72/279 length 3 +[h264 @ 0x7fdf01802200] NAL 5(IDR) at 79/279 length 124 +[h264 @ 0x7fdf01802200] NAL 5(IDR) at 207/279 length 71 + +AUD is a must for QuickTime player. + +NAL_SLICE = 1, +NAL_DPA = 2, +NAL_DPB = 3, +NAL_DPC = 4, +NAL_IDR_SLICE = 5, +NAL_SEI = 6, +NAL_SPS = 7, +NAL_PPS = 8, +NAL_AUD = 9, +NAL_END_SEQUENCE = 10, +NAL_END_STREAM = 11, +NAL_FILLER_DATA = 12, +NAL_SPS_EXT = 13, +NAL_AUXILIARY_SLICE = 19, + diff --git a/ts/handler.go b/ts/handler.go new file mode 100644 index 0000000..d940a48 --- /dev/null +++ b/ts/handler.go @@ -0,0 +1,18 @@ +package ts + +import ( + "io" + "github.com/nareix/av" + "github.com/nareix/av/avutil" +) + +func Handler(h *avutil.RegisterHandler) { + h.Ext = ".ts" + h.ReaderDemuxer = func(r io.Reader) av.Demuxer { + return &Demuxer{R: r} + } + h.WriterMuxer = func(w io.Writer) av.Muxer { + return &Muxer{W: w} + } +} + diff --git a/ts/muxer.go b/ts/muxer.go new file mode 100644 index 0000000..e43ffbc --- /dev/null +++ b/ts/muxer.go @@ -0,0 +1,196 @@ +package ts + +import ( + "bytes" + "fmt" + "github.com/nareix/av" + "github.com/nareix/codec/aacparser" + "github.com/nareix/codec/h264parser" + "io" +) + +type Muxer struct { + W io.Writer + streams []*Stream + PaddingToMakeCounterCont bool + + tswPAT *TSWriter + tswPMT *TSWriter +} + +func (self *Muxer) isCodecSupported(codec av.CodecData) bool { + switch codec.Type() { + case av.H264, av.AAC: + return true + default: + return false + } +} + +func (self *Muxer) newStream(codec av.CodecData) (err error) { + if !self.isCodecSupported(codec) { + err = fmt.Errorf("codec type=%x is not supported", codec.Type()) + return + } + + stream := &Stream{ + muxer: self, + CodecData: codec, + tsw: &TSWriter{ + DiscontinuityIndicator: true, + PID: uint(len(self.streams) + 0x100), + }, + } + self.streams = append(self.streams, stream) + return +} + +func (self *Muxer) writePaddingTSPackets(tsw *TSWriter) (err error) { + for tsw.ContinuityCounter&0xf != 0x0 { + header := TSHeader{ + PID: tsw.PID, + ContinuityCounter: tsw.ContinuityCounter, + } + if _, err = WriteTSHeader(self.W, header, 0); err != nil { + return + } + tsw.ContinuityCounter++ + } + return +} + +func (self *Muxer) WriteTrailer() (err error) { + if self.PaddingToMakeCounterCont { + for _, stream := range self.streams { + if err = self.writePaddingTSPackets(stream.tsw); err != nil { + return + } + } + } + return +} + +func (self *Muxer) WritePATPMT() (err error) { + bufPAT := &bytes.Buffer{} + bufPMT := &bytes.Buffer{} + + pat := PAT{ + Entries: []PATEntry{ + {ProgramNumber: 1, ProgramMapPID: 0x1000}, + }, + } + WritePAT(bufPAT, pat) + + var elemStreams []ElementaryStreamInfo + for _, stream := range self.streams { + switch stream.Type() { + case av.AAC: + elemStreams = append(elemStreams, ElementaryStreamInfo{StreamType: ElementaryStreamTypeAdtsAAC, ElementaryPID: stream.tsw.PID}) + case av.H264: + elemStreams = append(elemStreams, ElementaryStreamInfo{StreamType: ElementaryStreamTypeH264, ElementaryPID: stream.tsw.PID}) + } + } + + pmt := PMT{ + PCRPID: 0x100, + ElementaryStreamInfos: elemStreams, + } + WritePMT(bufPMT, pmt) + + self.tswPMT = &TSWriter{ + PID: 0x1000, + DiscontinuityIndicator: true, + } + self.tswPAT = &TSWriter{ + PID: 0, + DiscontinuityIndicator: true, + } + if err = self.tswPAT.WriteTo(self.W, bufPAT.Bytes()); err != nil { + return + } + if err = self.tswPMT.WriteTo(self.W, bufPMT.Bytes()); err != nil { + return + } + + return +} + +func (self *Muxer) WriteHeader(streams []av.CodecData) (err error) { + self.streams = []*Stream{} + for _, stream := range streams { + if err = self.newStream(stream); err != nil { + return + } + } + + if err = self.WritePATPMT(); err != nil { + return + } + return +} + +func (self *Muxer) WritePacket(pkt av.Packet) (err error) { + if false { + fmt.Println("ts:", "in", pkt.Idx, pkt.Time, "len", len(pkt.Data)) + } + if err = self.writePacket(pkt); err != nil { + return + } + return +} + +func (self *Muxer) writePacket(pkt av.Packet) (err error) { + stream := self.streams[pkt.Idx] + + switch stream.Type() { + case av.AAC: + codec := stream.CodecData.(aacparser.CodecData) + data := pkt.Data + data = append(codec.MakeADTSHeader(1024, len(data)), data...) + + buf := &bytes.Buffer{} + pes := PESHeader{ + StreamId: StreamIdAAC, + PTS: timeToPesTs(pkt.Time), + } + WritePESHeader(buf, pes, len(data)) + buf.Write(data) + + stream.tsw.RandomAccessIndicator = true + stream.tsw.PCR = timeToPCR(pkt.Time) + if err = stream.tsw.WriteTo(self.W, buf.Bytes()); err != nil { + return + } + + case av.H264: + codec := stream.CodecData.(h264parser.CodecData) + buf := &bytes.Buffer{} + pes := PESHeader{ + StreamId: StreamIdH264, + PTS: timeToPesTs(pkt.Time + pkt.CompositionTime), + DTS: timeToPesTs(pkt.Time), + } + WritePESHeader(buf, pes, 0) + + if typ := h264parser.CheckNALUsType(pkt.Data); typ != h264parser.NALU_RAW { + err = fmt.Errorf("ts: h264 nalu format=%d invalid", typ) + return + } + nalus := [][]byte{} + if pkt.IsKeyFrame { + nalus = append([][]byte{codec.SPS(), codec.PPS()}) + } + nalus = append(nalus, pkt.Data) + h264parser.WalkNALUsAnnexb(nalus, func(b []byte) { + buf.Write(b) + }) + + stream.tsw.RandomAccessIndicator = pkt.IsKeyFrame + stream.tsw.PCR = timeToPCR(pkt.Time) + if err = stream.tsw.WriteTo(self.W, buf.Bytes()); err != nil { + return + } + } + + return +} diff --git a/ts/reader.go b/ts/reader.go new file mode 100644 index 0000000..5e786ff --- /dev/null +++ b/ts/reader.go @@ -0,0 +1,682 @@ +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 ReadTSHeader(r io.Reader) (self TSHeader, err error) { + var flags uint + + // sync(8) + // transport_error_indicator(1) + // payload_unit_start_indicator(1) + // transport_priority(1) + // pid(13) + // Scrambling control(2) + // Adaptation field flag(1) + // Continuity counter(4) + if flags, err = ReadUInt(r, 4); err != nil { + return + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "ts: flags %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {8, "sync"}, + {1, "transport_error_indicator"}, + {1, "payload_unit_start_indicator"}, + {1, "transport_priority"}, + {13, "pid"}, + {2, "scrambling_control"}, + {1, "adaptation_field_flag"}, + {1, "payload_flag"}, + {4, "continuity_counter"}, + }, + Val: flags, + Length: 32, + }) + } + + if flags&0x400000 != 0 { + // When set to '1' it indicates that this TS packet contains the first PES packet. + self.PayloadUnitStart = true + } + + if (flags&0xff000000)>>24 != 0x47 { + err = fmt.Errorf("invalid sync") + return + } + + self.PID = (flags & 0x1fff00) >> 8 + self.ContinuityCounter = flags & 0xf + + if flags&0x20 != 0 { + var flags, length uint + if length, err = ReadUInt(r, 1); err != nil { + return + } + if length > 0 { + lr := &io.LimitedReader{R: r, N: int64(length)} + if flags, err = ReadUInt(lr, 1); err != nil { + return + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "ts: ext_flags %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {1, "discontinuity_indicator"}, + {1, "random_access_indicator"}, + {1, "elementary_stream_priority_indicator"}, + {1, "pcr_flag"}, + {1, "opcr_flag"}, + {1, "splicing_point_flag"}, + {1, "transport_private_data_flag"}, + {1, "adaptation_field_extension_flag"}, + }, + Val: flags, + Length: 8, + }) + } + + // random_access_indicator + if flags&0x40 != 0 { + self.RandomAccessIndicator = true + } + + // PCR + if flags&0x10 != 0 { + var v uint64 + if v, err = ReadUInt64(lr, 6); err != nil { + return + } + // clock is 27MHz + self.PCR = UIntToPCR(v) + if DebugReader { + fmt.Fprintf(DebugOutput, "ts: PCR %d %f\n", self.PCR, float32(self.PCR)/PCR_HZ) + } + } + + // OPCR + if flags&0x08 != 0 { + var v uint64 + if v, err = ReadUInt64(lr, 6); err != nil { + return + } + self.OPCR = UIntToPCR(v) + } + + // Splice countdown + if flags&0x04 != 0 { + if _, err = ReadUInt(lr, 1); err != nil { + return + } + } + + // Transport private data + if flags&0x02 != 0 { + var length uint + if length, err = ReadUInt(lr, 1); err != nil { + return + } + + b := make([]byte, length) + if _, err = lr.Read(b); err != nil { + return + } + } + + // Adaptation extension + if lr.N > 0 { + if DebugReader { + // rubish + fmt.Fprintln(DebugOutput, "ts: skip", lr.N) + } + + if err = ReadDummy(lr, int(lr.N)); err != nil { + return + } + } + } + } + + return +} + +func ReadTSPacket(r io.Reader, data []byte) (self TSHeader, n int, err error) { + lr := &io.LimitedReader{R: r, N: 188} + if self, err = ReadTSHeader(lr); err != nil { + return + } + if DebugReader { + fmt.Fprintln(DebugOutput, "ts: data len", lr.N) + } + if n, err = lr.Read(data[:lr.N]); err != nil { + return + } + 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 DebugReader { + fmt.Fprintf(DebugOutput, "psi: pointer=%d\n", pointer) + } + + 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 + + if DebugReader { + fmt.Fprintf(DebugOutput, "psi: tableid=%d len=%d\n", self.TableId, length) + } + + 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 + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "psi: %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {2, "resverd"}, + {5, "version"}, + {1, "current_next_indicator"}, + }, + Val: flags, + Length: 8, + }) + } + + // 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 + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "psi: table_id=%x table_extension=%x secnum=%x lastsecnum=%x\n", + self.TableId, + self.TableIdExtension, + self.SecNum, + self.LastSecNum, + ) + } + + 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 + + if DebugReader { + fmt.Fprintf(DebugOutput, "pmt: %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {3, "reserved"}, + {13, "pcrpid"}, + }, + Val: flags, + Length: 16, + }) + } + + // Reserved(4)=0xf + // Reserved(2)=0x0 + // Program info length(10) + if flags, err = ReadUInt(lr, 2); err != nil { + return + } + length = flags & 0x3ff + + if DebugReader { + fmt.Fprintf(DebugOutput, "pmt: ProgramDescriptorsLen=%d\n", length) + } + + 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 + + if DebugReader { + fmt.Fprintf(DebugOutput, "pmt: info1 %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {3, "reserved"}, + {13, "elementary_pid"}, + }, + Val: flags, + Length: 16, + }) + } + + // Reserved(6) + // ES Info length(10) + if flags, err = ReadUInt(lr, 2); err != nil { + return + } + length = flags & 0x3ff + + if DebugReader { + fmt.Fprintf(DebugOutput, "pmt: info2 %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {6, "reserved"}, + {10, "es_info_length"}, + }, + Val: flags, + Length: 16, + }) + } + + 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 DebugReader { + fmt.Fprintf(DebugOutput, "pmt: ProgramDescriptors %v\n", self.ProgramDescriptors) + fmt.Fprintf(DebugOutput, "pmt: ElementaryStreamInfos %v\n", self.ElementaryStreamInfos) + } + + if err = cr.ReadCrc32UIntAndCheck(); err != nil { + if DebugReader { + fmt.Fprintf(DebugOutput, "pmt: %s\n", err) + } + 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 { + if DebugReader { + fmt.Fprintf(DebugOutput, "pat: %s\n", err) + } + return + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "pat: %v\n", self) + } + + 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 DebugReader { + fmt.Fprintf(DebugOutput, "pes: StreamId=%x length=%d\n", self.StreamId, length) + } + + 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 + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "pes: %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {2, "scrambling_control"}, + {1, "priority"}, + {1, "data_alignment_indicator"}, + {1, "copyright"}, + {1, "original_or_copy"}, + }, + Val: flags, + Length: 6, + }) + } + + // 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 + } + + if DebugReader { + fmt.Fprintf(DebugOutput, "pes: %s\n", FieldsDumper{ + Fields: []struct { + Length int + Desc string + }{ + {2, "pts_dts_flags"}, + {1, "escr_flag"}, + {1, "es_rate_flag"}, + {1, "dsm_trick_mode_flag"}, + {1, "additional_copy_info_flag"}, + {1, "pes_crc_flag"}, + {1, "pes_extension_flag"}, + }, + Val: flags, + Length: 8, + }) + } + + // 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 DebugReader { + fmt.Fprintf(DebugOutput, "pes: pts %d %f\n", self.PTS, float32(self.PTS)/float32(PTS_HZ)) + } + } + + if flags&0x40 != 0 && flags&0x80 != 0 { + var v uint64 + if v, err = ReadUInt64(lr, 5); err != nil { + return + } + self.DTS = PESUIntToTs(v) + if DebugReader { + fmt.Fprintf(DebugOutput, "pes: dts %d %f\n", self.DTS, float32(self.DTS)/float32(PTS_HZ)) + } + } + + // 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/ts/stream.go b/ts/stream.go new file mode 100644 index 0000000..3bd2a2a --- /dev/null +++ b/ts/stream.go @@ -0,0 +1,36 @@ +package ts + +import ( + "time" + "bytes" + "github.com/nareix/av" +) + +type Stream struct { + av.CodecData + + pid uint + buf bytes.Buffer + peshdr *PESHeader + tshdr TSHeader + + demuxer *Demuxer + muxer *Muxer + + streamId uint + streamType uint + + tsw *TSWriter + dataBuf *iovec + cacheSize int + + idx 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/ts/ts.go b/ts/ts.go new file mode 100644 index 0000000..dccb6a0 --- /dev/null +++ b/ts/ts.go @@ -0,0 +1,122 @@ +package ts + +import ( + "fmt" + "os" +) + +const ( + ElementaryStreamTypeH264 = 0x1B + ElementaryStreamTypeAdtsAAC = 0x0F +) + +type TSHeader struct { + PID uint + PCR uint64 + OPCR uint64 + ContinuityCounter uint + PayloadUnitStart bool + DiscontinuityIndicator bool + RandomAccessIndicator bool + HeaderLength uint +} + +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 +} + +var DebugOutput = os.Stdout +var DebugReader = false +var DebugWriter = false + +type FieldsDumper struct { + Fields []struct { + Length int + Desc string + } + Val uint + Length uint +} + +func (self FieldsDumper) String() (res string) { + pos := uint(self.Length) + for _, field := range self.Fields { + pos -= uint(field.Length) + val := (self.Val >> pos) & (1< 0 && self.Len > 0 { + data := self.data[self.idx] + + var b []byte + if n > len(data) { + b = data + } else { + b = data[:n] + } + + data = data[len(b):] + if len(data) == 0 { + self.idx++ + } else { + self.data[self.idx] = data + } + self.Len -= len(b) + n -= len(b) + written += len(b) + + if _, err = w.Write(b); err != nil { + return + } + } + return +} + +type sysiovec struct { + Base uintptr + Len uint64 +} + +type vecWriter struct { + fd uintptr + smallBytesBuf []byte + iov []sysiovec +} + +func (self *vecWriter) Write(p []byte) (written int, err error) { + iov := sysiovec{ + Len: uint64(len(p)), + } + + if len(p) < 16 { + iov.Base = uintptr(len(self.smallBytesBuf)) + self.smallBytesBuf = append(self.smallBytesBuf, p...) + } else { + iov.Base = uintptr(unsafe.Pointer(&p[0])) + } + + self.iov = append(self.iov, iov) + return +} + +func (self *vecWriter) Flush() (err error) { + for i := range self.iov { + iov := &self.iov[i] + if iov.Base < uintptr(len(self.smallBytesBuf)) { + iov.Base = uintptr(unsafe.Pointer(&self.smallBytesBuf[iov.Base])) + } + } + + N := 1024 + for i := 0; i < len(self.iov); i += N { + n := len(self.iov) - i + if n > N { + n = N + } + _, _, errno := syscall.Syscall(syscall.SYS_WRITEV, self.fd, uintptr(unsafe.Pointer(&self.iov[i])), uintptr(n)) + if errno != 0 { + err = fmt.Errorf("writev failed with error: %d", errno) + return + } + } + + if DebugWriter { + fmt.Printf("vecw: smallBytesBuf=%d iovNr=%d\n", len(self.smallBytesBuf), len(self.iov)) + } + + self.iov = self.iov[:0] + self.smallBytesBuf = self.smallBytesBuf[:0] + + return +} + +func newVecWriter(w io.Writer) (vecw *vecWriter) { + var err error + var f *os.File + + switch obj := w.(type) { + case *net.TCPConn: + f, err = obj.File() + if err != nil { + return + } + case *os.File: + f = obj + default: + return + } + + vecw = &vecWriter{ + fd: f.Fd(), + } + return +} diff --git a/ts/writer.go b/ts/writer.go new file mode 100644 index 0000000..85938ff --- /dev/null +++ b/ts/writer.go @@ -0,0 +1,556 @@ +package ts + +import ( + "bytes" + "fmt" + "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 makeRepeatValBytes(val byte, n int) []byte { + b := make([]byte, n) + for i := range b { + b[i] = val + } + return b +} + +func WriteRepeatVal(w io.Writer, val byte, n int) (err error) { + _, err = w.Write(makeRepeatValBytes(val, n)) + return +} + +func WriteTSHeader(w io.Writer, self TSHeader, dataLength int) (written int, err error) { + var flags, extFlags uint + + // sync(8) + // transport_error_indicator(1) + // payload_unit_start_indicator(1) + // transport_priority(1) + // pid(13) + // Scrambling control(2) + // Adaptation field flag(1) 0x20 + // Payload flag(1) 0x10 + // Continuity counter(4) + + flags = 0x47 << 24 + flags |= 0x10 + if self.PayloadUnitStart { + flags |= 0x400000 + } + flags |= (self.PID & 0x1fff) << 8 + flags |= self.ContinuityCounter & 0xf + + if DebugWriter { + fmt.Fprintf(DebugOutput, "tsw: pid=%x\n", self.PID) + } + + const PCR = 0x10 + const OPCR = 0x08 + const EXT = 0x20 + + if self.PCR != 0 { + extFlags |= PCR + } + if self.OPCR != 0 { + extFlags |= OPCR + } + if self.RandomAccessIndicator { + extFlags |= 0x40 + } + if self.DiscontinuityIndicator { + extFlags |= 0x80 + } + + if extFlags != 0 { + flags |= EXT + } + + // need padding + if dataLength < 184 { + flags |= EXT + } + + if err = WriteUInt(w, flags, 4); err != nil { + return + } + written += 4 + + if flags&EXT != 0 { + var length uint + + // Discontinuity indicator 1 0x80 + // Random Access indicator 1 0x40 + // Elementary stream priority indicator 1 0x20 + // PCR flag 1 0x10 + // OPCR flag 1 0x08 + + length = 1 // extFlags + if extFlags&PCR != 0 { + length += 6 + } + if extFlags&OPCR != 0 { + length += 6 + } + + paddingLength := 0 + // need padding + if int(length)+5+dataLength < 188 { + paddingLength = 188 - dataLength - 5 - int(length) + length = 188 - uint(dataLength) - 5 + } + + if DebugWriter { + fmt.Fprintf(DebugOutput, "tsw: header padding=%d\n", paddingLength) + } + + if err = WriteUInt(w, length, 1); err != nil { + return + } + if err = WriteUInt(w, extFlags, 1); err != nil { + return + } + + if extFlags&PCR != 0 { + if err = WriteUInt64(w, PCRToUInt(self.PCR), 6); err != nil { + return + } + } + + if extFlags&OPCR != 0 { + if err = WriteUInt64(w, PCRToUInt(self.OPCR), 6); err != nil { + return + } + } + + if paddingLength > 0 { + if err = WriteRepeatVal(w, 0xff, paddingLength); err != nil { + return + } + } + + written += int(length) + 1 + } + + return +} + +type TSWriter struct { + W io.Writer + PID uint + TSHeader + DisableHeaderPadding bool + DiscontinuityIndicator bool + + vecw *vecWriter +} + +func (self *TSWriter) EnableVecWriter() { + if self.vecw == nil { + self.vecw = newVecWriter(self.W) + + if DebugWriter && self.vecw != nil { + fmt.Fprintln(DebugOutput, "tsw: enabled vec writer") + } + } +} + +func (self *TSWriter) WriteIovec(data *iovec) (err error) { + if self.vecw != nil { + if err = self.WriteIovecTo(self.vecw, data); err != nil { + return + } + if err = self.vecw.Flush(); err != nil { + return + } + } else { + if err = self.WriteIovecTo(self.W, data); err != nil { + return + } + } + return +} + +func (self *TSWriter) WriteIovecTo(w io.Writer, data *iovec) (err error) { + for i := 0; data.Len > 0; i++ { + header := TSHeader{ + PID: self.PID, + ContinuityCounter: self.ContinuityCounter, + DiscontinuityIndicator: self.DiscontinuityIndicator, + } + + if i == 0 { + header.PayloadUnitStart = true + header.RandomAccessIndicator = self.RandomAccessIndicator + header.PCR = self.PCR + header.OPCR = self.OPCR + } + + requestLength := data.Len + if self.DisableHeaderPadding { + requestLength = 188 + } + var headerLength int + if headerLength, err = WriteTSHeader(w, header, requestLength); err != nil { + return + } + payloadLength := 188 - headerLength + if self.DisableHeaderPadding && data.Len < payloadLength { + data.Append(makeRepeatValBytes(0xff, payloadLength-data.Len)) + } + + if DebugWriter { + fmt.Fprintf(DebugOutput, "tsw: payloadLength=%d dataLength=%d\n", payloadLength, data.Len) + } + + if _, err = data.WriteTo(w, payloadLength); err != nil { + return + } + + self.ContinuityCounter++ + } + return +} + +func (self *TSWriter) WriteTo(w io.Writer, data []byte) (err error) { + iov := &iovec{} + iov.Append(data) + return self.WriteIovecTo(w, iov) +} + +func (self *TSWriter) Write(data []byte) (err error) { + iov := &iovec{} + iov.Append(data) + return self.WriteIovec(iov) +} + +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 + } + + if DebugWriter { + fmt.Fprintf(DebugOutput, "psiw: length=%d\n", length) + } + + // 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 WritePESHeader(w io.Writer, self PESHeader, dataLength int) (err error) { + // http://dvd.sourceforge.net/dvdinfo/pes-hdr.html + + var pts_dts_flags, header_length, packet_length uint + + // start code(24) 000001 + // StreamId(8) + // packet_length(16) + // resverd(6,2)=2,original_or_copy(0,1)=1 + // pts_dts_flags(6,2) + // header_length(8) + // pts(40)? + // dts(40)? + // data + + // start code(24) 000001 + if err = WriteUInt(w, 0x000001, 3); err != nil { + return + } + + // StreamId(8) + if err = WriteUInt(w, self.StreamId, 1); err != nil { + return + } + + const PTS = 1 << 7 + const DTS = 1 << 6 + + if self.PTS != 0 { + pts_dts_flags |= PTS + if self.DTS != 0 { + pts_dts_flags |= DTS + } + } + + if pts_dts_flags&PTS != 0 { + header_length += 5 + } + if pts_dts_flags&DTS != 0 { + header_length += 5 + } + + if dataLength > 0 { + packet_length = uint(dataLength) + header_length + 3 + } + // 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 err = WriteUInt(w, packet_length, 2); err != nil { + return + } + + // resverd(6,2)=2,original_or_copy(0,1)=1 + if err = WriteUInt(w, 2<<6|1, 1); err != nil { + return + } + + // pts_dts_flags(6,2) + if err = WriteUInt(w, pts_dts_flags, 1); err != nil { + return + } + + // header_length(8) + if err = WriteUInt(w, header_length, 1); err != nil { + return + } + + // pts(40)? + // dts(40)? + if pts_dts_flags&PTS != 0 { + if pts_dts_flags&DTS != 0 { + if err = WriteUInt64(w, PESTsToUInt(self.PTS)|3<<36, 5); err != nil { + return + } + if err = WriteUInt64(w, PESTsToUInt(self.DTS)|1<<36, 5); err != nil { + return + } + } else { + if err = WriteUInt64(w, PESTsToUInt(self.PTS)|2<<36, 5); err != nil { + return + } + } + } + + return +} + +func WritePESPacket(w *TSWriter, header PESHeader, data []byte) (err error) { + bw := &bytes.Buffer{} + if err = WritePESHeader(bw, header, len(data)); err != nil { + return + } + iov := &iovec{} + iov.Append(bw.Bytes()) + iov.Append(data) + if err = w.WriteIovec(iov); err != nil { + return + } + return +} + +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 WritePATPacket(w io.Writer, pat PAT) (err error) { + tsw := &TSWriter{ + W: w, + PID: 0, + DisableHeaderPadding: true, + } + bw := &bytes.Buffer{} + if err = WritePAT(bw, pat); err != nil { + return + } + if err = tsw.Write(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 +} + +func WritePMTPacket(w io.Writer, pmt PMT, pid uint) (err error) { + tsw := &TSWriter{ + W: w, + PID: pid, + DisableHeaderPadding: true, + } + bw := &bytes.Buffer{} + if err = WritePMT(bw, pmt); err != nil { + return + } + if err = tsw.Write(bw.Bytes()); err != nil { + return + } + return +} + + diff --git a/ts/writer_test.go b/ts/writer_test.go new file mode 100644 index 0000000..5450006 --- /dev/null +++ b/ts/writer_test.go @@ -0,0 +1,17 @@ +package ts + +import ( + "bytes" + "encoding/hex" + "testing" +) + +func TestWriteTSHeader(t *testing.T) { + bw := &bytes.Buffer{} + w := &TSWriter{ + W: bw, + PCR: 0x12345678, + } + w.Write([]byte{'h', 'e', 'l', 'o'}[:], false) + t.Logf("\n%s", hex.Dump(bw.Bytes())) +}