diff --git a/cmd/tapo/tapo.go b/cmd/tapo/tapo.go index a48cde45..527cc898 100644 --- a/cmd/tapo/tapo.go +++ b/cmd/tapo/tapo.go @@ -15,11 +15,5 @@ func handle(url string) (streamer.Producer, error) { if err := conn.Dial(); err != nil { return nil, err } - if err := conn.Play(); err != nil { - return nil, err - } - if err := conn.Handle(); err != nil { - return nil, err - } return conn, nil } diff --git a/pkg/mpegts/checksum.go b/pkg/mpegts/checksum.go new file mode 100644 index 00000000..eafc8c6d --- /dev/null +++ b/pkg/mpegts/checksum.go @@ -0,0 +1,54 @@ +package mpegts + +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 calcCRC32(crc uint32, data []byte) uint32 { + for _, b := range data { + crc = ieeeCrc32Tbl[b^byte(crc)] ^ (crc >> 8) + } + return crc +} diff --git a/pkg/mpegts/helpers.go b/pkg/mpegts/helpers.go index 7d706628..ccf09bda 100644 --- a/pkg/mpegts/helpers.go +++ b/pkg/mpegts/helpers.go @@ -14,6 +14,7 @@ const ( ) const ( + StreamTypePrivate = 0x06 // PCMU or PCMA from FFmpeg StreamTypeAAC = 0x0F StreamTypeH264 = 0x1B StreamTypePCMATapo = 0x90 @@ -93,7 +94,7 @@ func (p *PES) GetPacket() (pkt *rtp.Packet) { flags := p.Payload[1] optSize := p.Payload[2] // optional fields - p.Payload = p.Payload[minHeaderSize+optSize:] + payload := p.Payload[minHeaderSize+optSize:] switch p.StreamType { case StreamTypeH264: @@ -101,7 +102,7 @@ func (p *PES) GetPacket() (pkt *rtp.Packet) { const hasPTS = 0b1000_0000 if flags&hasPTS != 0 { - ts = uint32(ParseTime(p.Payload[minHeaderSize:])) + ts = ParseTime(p.Payload[minHeaderSize:]) } pkt = &rtp.Packet{ @@ -109,12 +110,12 @@ func (p *PES) GetPacket() (pkt *rtp.Packet) { PayloadType: p.StreamType, Timestamp: ts, }, - Payload: h264.AnnexB2AVC(p.Payload), + Payload: h264.AnnexB2AVC(payload), } case StreamTypePCMATapo: p.Sequence++ - p.Timestamp += uint32(len(p.Payload)) + p.Timestamp += uint32(len(payload)) pkt = &rtp.Packet{ Header: rtp.Header{ @@ -123,7 +124,7 @@ func (p *PES) GetPacket() (pkt *rtp.Packet) { SequenceNumber: p.Sequence, Timestamp: p.Timestamp, }, - Payload: p.Payload, + Payload: payload, } } @@ -155,9 +156,8 @@ func (p *PES) GetPacket() (pkt *rtp.Packet) { return } -func ParseTime(b []byte) time.Duration { - ts := (uint64(b[0]) >> 1 & 0x7 << 30) | (uint64(b[1]) << 22) | (uint64(b[2]) >> 1 & 0x7F << 15) | (uint64(b[3]) << 7) | (uint64(b[4]) >> 1 & 0x7F) - return time.Duration(ts) +func ParseTime(b []byte) uint32 { + return (uint32(b[0]&0x0E) << 29) | (uint32(b[1]) << 22) | (uint32(b[2]&0xFE) << 14) | (uint32(b[3]) << 7) | (uint32(b[4]) >> 1) } func GetMedia(pkt *rtp.Packet) *streamer.Media { diff --git a/pkg/mpegts/mpegts_test.go b/pkg/mpegts/mpegts_test.go new file mode 100644 index 00000000..840732e8 --- /dev/null +++ b/pkg/mpegts/mpegts_test.go @@ -0,0 +1,14 @@ +package mpegts + +import ( + "github.com/stretchr/testify/assert" + "testing" +) + +func TestTime(t *testing.T) { + w := NewWriter() + w.WriteTime(0xFFFFFFFF) + assert.Equal(t, []byte{0x27, 0xFF, 0xFF, 0xFF, 0xFF}, w.Bytes()) + ts := ParseTime(w.Bytes()) + assert.Equal(t, uint32(0xFFFFFFFF), ts) +} diff --git a/pkg/mpegts/writer.go b/pkg/mpegts/writer.go new file mode 100644 index 00000000..188509e3 --- /dev/null +++ b/pkg/mpegts/writer.go @@ -0,0 +1,219 @@ +package mpegts + +type Writer struct { + b []byte // packets buffer + m int // crc start + + pid []uint16 + counter []byte + streamType []byte + timestamp []uint32 +} + +func NewWriter() *Writer { + return &Writer{} +} + +func (w *Writer) AddPES(pid uint16, streamType byte) { + w.pid = append(w.pid, pid) + w.streamType = append(w.streamType, streamType) + w.counter = append(w.counter, 0) + w.timestamp = append(w.timestamp, 0) +} + +func (w *Writer) WriteByte(b byte) { + w.b = append(w.b, b) +} + +func (w *Writer) WriteUint16(i uint16) { + w.b = append(w.b, byte(i>>8), byte(i)) +} + +func (w *Writer) WriteTime(t uint32) { + const onlyPTS = 0x20 + // [>>32 <<3] [>>24 <<2] [>>16 <<2] [>>8 <<1] [<<1] + w.b = append(w.b, onlyPTS|byte(t>>29)|1, byte(t>>22), byte(t>>14)|1, byte(t>>7), byte(t<<1)|1) +} + +func (w *Writer) WriteBytes(b []byte) { + w.b = append(w.b, b...) +} + +func (w *Writer) MarkChecksum() { + w.m = len(w.b) +} + +func (w *Writer) WriteChecksum() { + crc := calcCRC32(0xFFFFFFFF, w.b[w.m:]) + w.b = append(w.b, byte(crc), byte(crc>>8), byte(crc>>16), byte(crc>>24)) +} + +func (w *Writer) FinishPacket() { + if n := len(w.b) % PacketSize; n != 0 { + w.b = append(w.b, make([]byte, PacketSize-n)...) + } +} + +func (w *Writer) Bytes() []byte { + if len(w.b)%PacketSize != 0 { + panic("wrong packet size") + } + return w.b +} + +func (w *Writer) Reset() { + w.b = nil +} + +const isUnitStart = 0x4000 +const flagHasAdaptation = 0x20 +const flagHasPayload = 0x10 +const lenIsProgramTable = 0xB000 +const tableFlags = 0xC1 +const tableHeader = 0xE000 +const tableLength = 0xF000 + +const patPID = 0 +const patTableID = 0 +const patTableExtID = 1 + +func (w *Writer) WritePAT() { + w.WriteByte(SyncByte) + w.WriteUint16(isUnitStart | patPID) // PAT PID + w.WriteByte(flagHasPayload) // flags... + + w.WriteByte(0) // Pointer field + + w.MarkChecksum() + w.WriteByte(patTableID) // Table ID + w.WriteUint16(lenIsProgramTable | 13) // Section length + w.WriteUint16(patTableExtID) // Table ID extension + w.WriteByte(tableFlags) // flags... + w.WriteByte(0) // Section number + w.WriteByte(0) // Last section number + + w.WriteUint16(1) // Program num (usual 1) + w.WriteUint16(tableHeader + pmtPID) + + w.WriteChecksum() + + w.FinishPacket() +} + +const pmtPID = 18 +const pmtTableID = 2 +const pmtTableExtID = 1 + +func (w *Writer) WritePMT() { + w.WriteByte(SyncByte) + w.WriteUint16(isUnitStart | pmtPID) // PMT PID + w.WriteByte(flagHasPayload) // flags... + + w.WriteByte(0) // Pointer field + + tableLen := 13 + uint16(len(w.pid))*5 + + w.MarkChecksum() + w.WriteByte(pmtTableID) // Table ID + w.WriteUint16(lenIsProgramTable | tableLen) // Section length + w.WriteUint16(pmtTableExtID) // Table ID extension + w.WriteByte(tableFlags) // flags... + w.WriteByte(0) // Section number + w.WriteByte(0) // Last section number + + w.WriteUint16(tableHeader | w.pid[0]) // PID + w.WriteUint16(tableLength | 0) // Info length + + for i, pid := range w.pid { + w.WriteByte(w.streamType[i]) + w.WriteUint16(tableHeader | pid) // PID + w.WriteUint16(tableLength | 0) // Info len + } + + w.WriteChecksum() + + w.FinishPacket() +} + +const pesHeaderSize = PacketSize - 18 + +func (w *Writer) WritePES(pid uint16, streamID byte, payload []byte) { + w.WriteByte(SyncByte) + w.WriteUint16(isUnitStart | pid) + + // check if payload lower then max first packet size + if len(payload) < PacketSize-18 { + w.WriteByte(flagHasAdaptation | flagHasPayload) + + // for 183 payload will be zero + adSize := PacketSize - 18 - 1 - byte(len(payload)) + w.WriteByte(adSize) + w.WriteBytes(make([]byte, adSize)) + } else { + w.WriteByte(flagHasPayload) + } + + w.WriteByte(0) + w.WriteByte(0) + w.WriteByte(1) + + w.WriteByte(streamID) + w.WriteUint16(uint16(8 + len(payload))) + + w.WriteByte(0x80) + w.WriteByte(0x80) // only PTS + w.WriteByte(5) // optional size + + switch w.streamType[0] { + case StreamTypePCMATapo: + w.timestamp[0] += uint32(len(payload) * 45 / 8) + } + + w.WriteTime(w.timestamp[0]) + + if len(payload) < PacketSize-18 { + w.WriteBytes(payload) + return + } + + w.WriteBytes(payload[:pesHeaderSize]) + + payload = payload[pesHeaderSize:] + var counter byte + + for { + counter++ + + if len(payload) > PacketSize-4 { + // payload more then maximum size + w.WriteByte(SyncByte) + w.WriteUint16(pid) + w.WriteByte(flagHasPayload | counter&0xF) + w.WriteBytes(payload[:PacketSize-4]) + + payload = payload[PacketSize-4:] + } else if len(payload) == PacketSize-4 { + // payload equal maximum size (last packet) + w.WriteByte(SyncByte) + w.WriteUint16(pid) + w.WriteByte(flagHasPayload | counter&0xF) + w.WriteBytes(payload) + + break + } else { + // payload lower than maximum size (last packet) + w.WriteByte(SyncByte) + w.WriteUint16(pid) + w.WriteByte(flagHasAdaptation | flagHasPayload | counter&0xF) + + // for 183 payload will be zero + adSize := PacketSize - 4 - 1 - byte(len(payload)) + w.WriteByte(adSize) + w.WriteBytes(make([]byte, adSize)) + + w.WriteBytes(payload) + + break + } + } +} diff --git a/pkg/tapo/backchannel.go b/pkg/tapo/backchannel.go new file mode 100644 index 00000000..cec8e395 --- /dev/null +++ b/pkg/tapo/backchannel.go @@ -0,0 +1,53 @@ +package tapo + +import ( + "bytes" + "github.com/AlexxIT/go2rtc/pkg/mpegts" + "github.com/AlexxIT/go2rtc/pkg/streamer" + "github.com/pion/rtp" + "strconv" +) + +func (c *Client) backchannelWriter() streamer.WriterFunc { + w := mpegts.NewWriter() + w.AddPES(68, mpegts.StreamTypePCMATapo) + w.WritePAT() + w.WritePMT() + + return func(packet *rtp.Packet) (err error) { + // don't know why 68 and 192 + w.WritePES(68, 192, packet.Payload) + err = c.WriteBackchannel(w.Bytes()) + w.Reset() + return + } +} + +func (c *Client) SetupBackchannel() (err error) { + // if conn1 is not used - we will use it for backchannel + // or we need to start another conn for session2 + if c.session1 != "" { + if c.conn2, err = c.newConn(); err != nil { + return + } + } else { + c.conn2 = c.conn1 + } + + c.session2, err = c.Request(c.conn2, []byte(`{"params":{"talk":{"mode":"aec"},"method":"get"},"seq":3,"type":"request"}`)) + return +} + +func (c *Client) WriteBackchannel(body []byte) (err error) { + // TODO: fixme (size) + buf := bytes.NewBuffer(nil) + buf.WriteString("----client-stream-boundary--\r\n") + buf.WriteString("Content-Type: audio/mp2t\r\n") + buf.WriteString("X-If-Encrypt: 0\r\n") + buf.WriteString("X-Session-Id: " + c.session2 + "\r\n") + buf.WriteString("Content-Length: " + strconv.Itoa(len(body)) + "\r\n\r\n") + buf.Write(body) + + _, err = buf.WriteTo(c.conn2) + return +} diff --git a/pkg/tapo/client.go b/pkg/tapo/client.go index 16f4064d..a5e5444a 100644 --- a/pkg/tapo/client.go +++ b/pkg/tapo/client.go @@ -1,9 +1,11 @@ package tapo import ( + "bytes" "crypto/aes" "crypto/cipher" "crypto/md5" + "encoding/json" "errors" "fmt" "github.com/AlexxIT/go2rtc/pkg/mpegts" @@ -14,7 +16,6 @@ import ( "net/http" "net/url" "strconv" - "strings" ) type Client struct { @@ -25,10 +26,13 @@ type Client struct { medias []*streamer.Media tracks map[byte]*streamer.Track - conn net.Conn - reader *multipart.Reader + conn1 net.Conn + conn2 net.Conn decrypt func(b []byte) []byte + + session1 string + session2 string } // block ciphers using cipher block chaining. @@ -42,9 +46,14 @@ func NewClient(url string) *Client { } func (c *Client) Dial() (err error) { + c.conn1, err = c.newConn() + return +} + +func (c *Client) newConn() (net.Conn, error) { u, err := url.Parse(c.url) if err != nil { - return + return nil, err } // support raw username/password @@ -66,20 +75,28 @@ func (c *Client) Dial() (err error) { req, err := http.NewRequest("POST", u.String(), nil) if err != nil { - return + return nil, err } req.Header.Set("Content-Type", "multipart/mixed; boundary=--client-stream-boundary--") res, err := tcp.Do(req) if err != nil { - return + return nil, err } if res.StatusCode != http.StatusOK { - return errors.New(res.Status) + return nil, errors.New(res.Status) } + if c.decrypt == nil { + c.newDectypter(res, username, password) + } + + return res.Body.(net.Conn), nil +} + +func (c *Client) newDectypter(res *http.Response, username, password string) { // extract nonce from response // cipher="AES_128_CBC" username="admin" padding="PKCS7_16" algorithm="MD5" nonce="***" nonce := res.Header.Get("Key-Exchange") @@ -106,52 +123,34 @@ func (c *Client) Dial() (err error) { padSize := int(b[len(b)-1]) return b[:len(b)-padSize] } - - c.conn = res.Body.(net.Conn) - - boundary := res.Header.Get("Content-Type") - _, boundary, _ = strings.Cut(boundary, "boundary=") - - c.reader = multipart.NewReader(c.conn, boundary) - - return nil } -func (c *Client) Play() (err error) { - // audio: default, disable, enable - body := []byte( - "----client-stream-boundary--\r\n" + - "Content-Type: application/json\r\nContent-Length: 120\r\n\r\n" + - `{"params":{"preview":{"audio":["default"],"channels":[0],"resolutions":["HD"]},"method":"get"},"seq":1,"type":"request"}` + - "\r\n", - ) +func (c *Client) SetupStream() (err error) { + if c.session1 != "" { + return + } - _, err = c.conn.Write(body) - return nil + // audio: default, disable, enable + c.session1, err = c.Request(c.conn1, []byte(`{"params":{"preview":{"audio":["default"],"channels":[0],"resolutions":["HD"]},"method":"get"},"seq":1,"type":"request"}`)) + return } // Handle - first run will be in probe state func (c *Client) Handle() error { - if c.tracks == nil { - c.tracks = map[byte]*streamer.Track{} - } + mpReader := multipart.NewReader(c.conn1, "--device-stream-boundary--") + tsReader := mpegts.NewReader() - reader := mpegts.NewReader() - - probe := streamer.NewProbe(c.medias == nil) - for probe == nil || probe.Active() { - p, err := c.reader.NextRawPart() + for { + p, err := mpReader.NextRawPart() if err != nil { return err } - ct := p.Header.Get("Content-Type") - if ct != "video/mp2t" { + if ct := p.Header.Get("Content-Type"); ct != "video/mp2t" { continue } cl := p.Header.Get("Content-Length") - size, err := strconv.Atoi(cl) if err != nil { return err @@ -169,37 +168,62 @@ func (c *Client) Handle() error { } body = c.decrypt(body) - reader.SetBuffer(body) + tsReader.SetBuffer(body) for { - pkt := reader.GetPacket() + pkt := tsReader.GetPacket() if pkt == nil { break } - track := c.tracks[pkt.PayloadType] - if track == nil { - // count track on probe state even if not support it - probe.Append(pkt.PayloadType) - - media := mpegts.GetMedia(pkt) - if media == nil { - continue // unsupported codec - } - - track = streamer.NewTrack2(media, nil) - - c.medias = append(c.medias, media) - c.tracks[pkt.PayloadType] = track + if track := c.tracks[pkt.PayloadType]; track != nil { + _ = track.WriteRTP(pkt) } - - _ = track.WriteRTP(pkt) } } - - return nil } -func (c *Client) Close() error { - return c.conn.Close() +func (c *Client) Close() (err error) { + if c.conn1 != nil { + err = c.conn1.Close() + } + if c.conn2 != nil { + _ = c.conn2.Close() + } + return +} + +func (c *Client) Request(conn net.Conn, body []byte) (string, error) { + // TODO: fixme (size) + buf := bytes.NewBuffer(nil) + buf.WriteString("----client-stream-boundary--\r\n") + buf.WriteString("Content-Type: application/json\r\n") + buf.WriteString("Content-Length: " + strconv.Itoa(len(body)) + "\r\n\r\n") + buf.Write(body) + buf.WriteString("\r\n") + + if _, err := buf.WriteTo(conn); err != nil { + return "", err + } + + mpReader := multipart.NewReader(conn, "--device-stream-boundary--") + + for { + p, err := mpReader.NextRawPart() + if err != nil { + return "", err + } + + var v struct { + Params struct { + SessionID string `json:"session_id"` + } `json:"params"` + } + + if err = json.NewDecoder(p).Decode(&v); err != nil { + return "", err + } + + return v.Params.SessionID, nil + } } diff --git a/pkg/tapo/producer.go b/pkg/tapo/producer.go index cc9e5c44..bcb9aa3d 100644 --- a/pkg/tapo/producer.go +++ b/pkg/tapo/producer.go @@ -1,18 +1,82 @@ package tapo -import "github.com/AlexxIT/go2rtc/pkg/streamer" +import ( + "github.com/AlexxIT/go2rtc/pkg/mpegts" + "github.com/AlexxIT/go2rtc/pkg/streamer" +) func (c *Client) GetMedias() []*streamer.Media { + // producer should have persistent medias + if c.medias == nil { + // don't know if all Tapo has this capabilities... + c.medias = []*streamer.Media{ + { + Kind: streamer.KindVideo, + Direction: streamer.DirectionSendonly, + Codecs: []*streamer.Codec{ + {Name: streamer.CodecH264, ClockRate: 90000, PayloadType: streamer.PayloadTypeRAW}, + }, + }, + { + Kind: streamer.KindAudio, + Direction: streamer.DirectionSendonly, + Codecs: []*streamer.Codec{ + {Name: streamer.CodecPCMA, ClockRate: 8000, PayloadType: 8}, + }, + }, + { + Kind: streamer.KindAudio, + Direction: streamer.DirectionRecvonly, + Codecs: []*streamer.Codec{ + {Name: streamer.CodecPCMA, ClockRate: 8000, PayloadType: 8}, + }, + }, + } + } + return c.medias } -func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track { +func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) (track *streamer.Track) { for _, track := range c.tracks { if track.Codec == codec { return track } } - return nil + + if c.tracks == nil { + c.tracks = map[byte]*streamer.Track{} + } + + if media.Kind == streamer.KindVideo { + if err := c.SetupStream(); err != nil { + return nil + } + + track = streamer.NewTrack2(media, codec) + c.tracks[mpegts.StreamTypeH264] = track + } else { + if media.Direction == streamer.DirectionSendonly { + if err := c.SetupStream(); err != nil { + return nil + } + + track = streamer.NewTrack2(media, codec) + c.tracks[mpegts.StreamTypePCMATapo] = track + } else { + if err := c.SetupBackchannel(); err != nil { + return nil + } + + if w := c.backchannelWriter(); w != nil { + track = streamer.NewTrack2(media, codec) + track.Bind(w) + c.tracks[0] = track + } + } + } + + return } func (c *Client) Start() error {